fujin/music-agregator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add indexer cache with River queue for scheduled refresh

Browse Source
This commit is contained in:
Alexander
2026-05-07 21:41:17 +02:00
parent 2041c154cf
commit 79f3f145de
22 changed files with 686 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/config/config.go
+12
View File
@@ -2,6 +2,7 @@ package config
import (
"fmt"
"time"
)
const (
@@ -21,11 +22,16 @@ type Config struct {
Port string `yaml:"port"`
} `yaml:"app"`
Database struct {
URL string `yaml:"url"`
} `yaml:"database"`
Indexer struct {
Url string `yaml:"url"`
Port string `yaml:"port"`
Type IndexerType `yaml:"type"`
ApiKey string `yaml:"api_key"`
Cache CacheConfig `yaml:"cache"`
} `yaml:"indexer"`
Torrent struct {
@@ -40,6 +46,12 @@ type Config struct {
} `yaml:"metadata"`
}
type CacheConfig struct {
Enabled bool `yaml:"enabled"`
RefreshInterval time.Duration `yaml:"refresh_interval"`
TTL time.Duration `yaml:"ttl"`
}
func (t *IndexerType) UnmarshalYAML(unmarshal func(any) error) error {
var value string
if err := unmarshal(&value); err != nil {
internal/indexer/cache.go
+79
View File
@@ -0,0 +1,79 @@
package indexer
import (
"sync"
"time"
"github.com/rs/zerolog/log"
)
type CacheEntry struct {
Key string
URL string
Result SearchResult
CreatedAt time.Time
TTL time.Duration
RefreshInterval time.Duration
}
func (e *CacheEntry) IsExpired() bool {
return time.Now().After(e.CreatedAt.Add(e.TTL))
}
type IndexerCache struct {
entries map[string]*CacheEntry
mu sync.RWMutex
}
func NewIndexerCache() *IndexerCache {
return &IndexerCache{
entries: make(map[string]*CacheEntry),
}
}
func (c *IndexerCache) Get(key string) (*CacheEntry, bool) {
c.mu.RLock()
defer c.mu.RUnlock()
entry, ok := c.entries[key]
if !ok {
log.Trace().Str("key", key).Msg("cache miss")
return nil, false
}
if entry.IsExpired() {
log.Trace().Str("key", key).Msg("cache expired")
return nil, false
}
log.Trace().Str("key", key).Int("items", len(entry.Result.Items)).Msg("cache hit")
return entry, true
}
func (c *IndexerCache) Add(entry CacheEntry) {
c.mu.Lock()
defer c.mu.Unlock()
c.entries[entry.Key] = &entry
log.Debug().Str("key", entry.Key).Int("items", len(entry.Result.Items)).Dur("ttl", entry.TTL).Dur("refresh", entry.RefreshInterval).Msg("cache entry added")
}
func (c *IndexerCache) Update(key string, result SearchResult) {
c.mu.Lock()
defer c.mu.Unlock()
if entry, ok := c.entries[key]; ok {
entry.Result = result
log.Debug().Str("key", key).Int("items", len(result.Items)).Msg("cache entry updated")
} else {
log.Warn().Str("key", key).Msg("cache update for missing entry")
}
}
func (c *IndexerCache) Remove(key string) {
c.mu.Lock()
defer c.mu.Unlock()
delete(c.entries, key)
log.Debug().Str("key", key).Msg("cache entry removed")
}
internal/indexer/cache_worker.go
+65
View File
@@ -0,0 +1,65 @@
package indexer
import (
"context"
"time"
"github.com/jackc/pgx/v5"
"github.com/riverqueue/river"
"github.com/rs/zerolog/log"
)
type CacheRefreshArgs struct {
Key string `json:"key"`
URL string `json:"url"`
TTLExpires time.Time `json:"ttl_expires"`
RefreshInterval time.Duration `json:"refresh_interval"`
}
func (CacheRefreshArgs) Kind() string { return "indexer_cache_refresh" }
type CacheRefreshWorker struct {
river.WorkerDefaults[CacheRefreshArgs]
Cache *IndexerCache
Indexer Indexer
RiverClient *river.Client[pgx.Tx]
}
func (w *CacheRefreshWorker) Work(ctx context.Context, job *river.Job[CacheRefreshArgs]) error {
args := job.Args
log.Trace().Str("key", args.Key).Int64("job_id", job.ID).Time("ttl_expires", args.TTLExpires).Msg("cache refresh worker started")
if time.Now().After(args.TTLExpires) {
w.Cache.Remove(args.Key)
log.Debug().Str("key", args.Key).Msg("cache entry TTL expired, removed")
return nil
}
log.Trace().Str("key", args.Key).Str("url", args.URL).Msg("fetching fresh data from indexer")
start := time.Now()
result, err := w.Indexer.FetchURL(args.URL)
if err != nil {
retryAt := time.Now().Add(5 * time.Minute)
log.Error().Err(err).Str("key", args.Key).Time("retry_at", retryAt).Msg("cache refresh failed, scheduling retry")
w.RiverClient.Insert(ctx, args, &river.InsertOpts{
ScheduledAt: retryAt,
})
return nil
}
log.Trace().Str("key", args.Key).Int("items", len(result.Items)).Dur("duration", time.Since(start)).Msg("fresh data fetched")
w.Cache.Update(args.Key, result)
nextRefresh := time.Now().Add(args.RefreshInterval)
_, err = w.RiverClient.Insert(ctx, args, &river.InsertOpts{
ScheduledAt: nextRefresh,
})
if err != nil {
log.Error().Err(err).Str("key", args.Key).Msg("failed to schedule next cache refresh")
} else {
log.Trace().Str("key", args.Key).Time("next_refresh", nextRefresh).Msg("next refresh scheduled")
}
log.Debug().Str("key", args.Key).Int("items", len(result.Items)).Msg("cache refreshed")
return nil
}
internal/indexer/cached_indexer.go
+90
View File
@@ -0,0 +1,90 @@
package indexer
import (
"context"
"time"
"github.com/jackc/pgx/v5"
"github.com/riverqueue/river"
"github.com/rs/zerolog/log"
"homelab.lan/music-agregator/internal/config"
)
type CachedIndexer struct {
inner Indexer
cache *IndexerCache
riverClient *river.Client[pgx.Tx]
cfg config.CacheConfig
}
func NewCachedIndexer(inner Indexer, cache *IndexerCache, riverClient *river.Client[pgx.Tx], cfg config.CacheConfig) *CachedIndexer {
return &CachedIndexer{
inner: inner,
cache: cache,
riverClient: riverClient,
cfg: cfg,
}
}
func (c *CachedIndexer) Search(query string, limit int32, tracker string) (SearchResult, error) {
key := query + "|" + tracker
log.Trace().Str("key", key).Str("query", query).Str("tracker", tracker).Msg("cached indexer search")
if entry, ok := c.cache.Get(key); ok {
log.Debug().Str("key", key).Int("items", len(entry.Result.Items)).Msg("returning cached result")
return entry.Result, nil
}
log.Trace().Str("key", key).Msg("cache miss, fetching from indexer")
result, err := c.inner.Search(query, limit, tracker)
if err != nil {
log.Error().Err(err).Str("key", key).Msg("cached indexer fetch failed")
return SearchResult{}, err
}
url := c.inner.BuildSearchURL(query, limit, tracker)
log.Trace().Str("key", key).Str("url", url).Int("items", len(result.Items)).Msg("caching result")
c.cache.Add(CacheEntry{
Key: key,
URL: url,
Result: result,
CreatedAt: time.Now(),
TTL: c.cfg.TTL,
RefreshInterval: c.cfg.RefreshInterval,
})
scheduleAt := time.Now().Add(c.cfg.RefreshInterval)
_, err = c.riverClient.Insert(context.Background(), CacheRefreshArgs{
Key: key,
URL: url,
TTLExpires: time.Now().Add(c.cfg.TTL),
RefreshInterval: c.cfg.RefreshInterval,
}, &river.InsertOpts{
ScheduledAt: scheduleAt,
})
if err != nil {
log.Error().Err(err).Str("key", key).Msg("failed to schedule cache refresh job")
} else {
log.Debug().Str("key", key).Time("scheduled_at", scheduleAt).Msg("cache refresh job scheduled")
}
log.Debug().Str("key", key).Dur("ttl", c.cfg.TTL).Dur("refresh", c.cfg.RefreshInterval).Int("items", len(result.Items)).Msg("cached indexer search complete")
return result, nil
}
func (c *CachedIndexer) FetchURL(url string) (SearchResult, error) {
log.Trace().Str("url", url).Msg("cached indexer fetch URL passthrough")
return c.inner.FetchURL(url)
}
func (c *CachedIndexer) BuildSearchURL(query string, limit int32, tracker string) string {
return c.inner.BuildSearchURL(query, limit, tracker)
}
func (c *CachedIndexer) Capabilities(indexerName string) (IndexerCapabilities, error) {
log.Trace().Str("indexer", indexerName).Msg("cached indexer capabilities passthrough")
return c.inner.Capabilities(indexerName)
}
internal/indexer/indexer.go
+2
View File
@@ -2,5 +2,7 @@ package indexer
type Indexer interface {
Search(query string, limit int32, indexer string) (SearchResult, error)
FetchURL(url string) (SearchResult, error)
BuildSearchURL(query string, limit int32, tracker string) string
Capabilities(indexerName string) (IndexerCapabilities, error)
}
internal/indexer/jackett.go
+36 -5
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"io"
"net/http"
"net/url"
"time"
"github.com/rs/zerolog/log"
@@ -20,21 +21,42 @@ func NewIndexer(cfg config.Config) Indexer {
return &JacketIndexer{
cfg: cfg,
client: &http.Client{
Timeout: time.Second * 10,
Timeout: 60 * time.Second,
},
}
}
func (indexer *JacketIndexer) Search(query string, limit int32, tracker string) (SearchResult, error) {
func (indexer *JacketIndexer) BuildSearchURL(query string, limit int32, tracker string) string {
searchTracker := "all"
if len(tracker) != 0 {
searchTracker = tracker
}
url := indexer.cfg.Indexer.Url
uri := fmt.Sprintf("%v/api/v2.0/indexers/%v/results/torznab?apikey=%v&limit=%d&cat=3010,3040&q=%v&t=search", url, searchTracker, indexer.cfg.Indexer.ApiKey, limit, query)
uri := fmt.Sprintf("%v/api/v2.0/indexers/%v/results/torznab?apikey=%v&cat=3010,3040&q=%v&t=search",
indexer.cfg.Indexer.Url, searchTracker, indexer.cfg.Indexer.ApiKey, url.QueryEscape(query))
if limit > 0 {
uri += fmt.Sprintf("&limit=%d", limit)
}
log.Trace().Str("tracker", searchTracker).Str("query", query).Int32("limit", limit).Msg("jackett request")
return uri
}
func (indexer *JacketIndexer) Search(query string, limit int32, tracker string) (SearchResult, error) {
uri := indexer.BuildSearchURL(query, limit, tracker)
return indexer.FetchURL(uri)
}
type JackettError struct {
Code string `xml:"code,attr"`
Description string `xml:"description,attr"`
}
func (e *JackettError) Error() string {
return fmt.Sprintf("jackett error %s: %s", e.Code, e.Description)
}
func (indexer *JacketIndexer) FetchURL(uri string) (SearchResult, error) {
log.Trace().Str("uri", uri).Msg("jackett request")
req, err := http.NewRequest("GET", uri, nil)
if err != nil {
@@ -62,6 +84,15 @@ func (indexer *JacketIndexer) Search(query string, limit int32, tracker string)
Dur("duration", time.Since(start)).
Msg("jackett response")
if resp.StatusCode != http.StatusOK {
var jackettErr JackettError
if xmlErr := xml.Unmarshal(body, &jackettErr); xmlErr == nil && jackettErr.Code != "" {
log.Error().Str("code", jackettErr.Code).Str("description", jackettErr.Description).Msg("jackett returned error")
return SearchResult{}, &jackettErr
}
return SearchResult{}, fmt.Errorf("jackett returned HTTP %d", resp.StatusCode)
}
var searchResult SearchResult
if err := xml.Unmarshal(body, &searchResult); err != nil {
log.Error().Err(err).Msg("error parsing search XML")
internal/indexer/server.go
+8 -4
View File
@@ -3,6 +3,8 @@ package indexer
import (
"context"
"github.com/jackc/pgx/v5"
"github.com/riverqueue/river"
"github.com/rs/zerolog/log"
"google.golang.org/grpc"
@@ -15,10 +17,10 @@ type IndexerServer struct {
pb.UnimplementedIndexerServiceServer
}
func NewIndexerServer(cfg config.Config) (*IndexerServer, error) {
service, err := NewIndexerService(cfg)
func NewIndexerServer(cfg config.Config, riverClient *river.Client[pgx.Tx]) (*IndexerServer, error) {
service, err := NewIndexerService(cfg, riverClient)
if err != nil {
log.Err(err).Msg("Failed to initialize IndexerService")
log.Err(err).Msg("failed to initialize IndexerService")
return nil, err
}
@@ -32,7 +34,9 @@ func (server *IndexerServer) Search(ctx context.Context, req *pb.SearchRequest)
Str("tracker", req.GetTracker()).
Msg("search started")
resp, err := server.service.Search(req)
log.Trace().Str("query", req.GetQuery()).Msg("fetching results from indexer")
resp, err := server.service.Search(req.GetQuery(), req.GetLimit(), req.GetTracker())
if err != nil {
log.Error().Err(err).Str("query", req.GetQuery()).Msg("search failed")
return nil, err
internal/indexer/service.go
+17 -7
View File
@@ -3,6 +3,8 @@ package indexer
import (
"fmt"
"github.com/jackc/pgx/v5"
"github.com/riverqueue/river"
"github.com/rs/zerolog/log"
pb "homelab.lan/music-agregator/gen/music_agregator/indexer/v1"
@@ -13,20 +15,28 @@ type IndexerService struct {
indexer Indexer
}
func NewIndexerService(cfg config.Config) (*IndexerService, error) {
func NewIndexerService(cfg config.Config, riverClient *river.Client[pgx.Tx]) (*IndexerService, error) {
var idx Indexer
switch cfg.Indexer.Type {
case config.IndexerTypeJackett:
indexer := NewIndexer(cfg)
return &IndexerService{indexer: indexer}, nil
idx = NewIndexer(cfg)
default:
return nil, fmt.Errorf("Unable to create the indexer for type: %v", cfg.Indexer.Type)
return nil, fmt.Errorf("unable to create the indexer for type: %v", cfg.Indexer.Type)
}
if cfg.Indexer.Cache.Enabled && riverClient != nil {
cache := NewIndexerCache()
idx = NewCachedIndexer(idx, cache, riverClient, cfg.Indexer.Cache)
log.Info().Dur("ttl", cfg.Indexer.Cache.TTL).Dur("refresh", cfg.Indexer.Cache.RefreshInterval).Msg("indexer cache enabled")
}
return &IndexerService{indexer: idx}, nil
}
func (service *IndexerService) Search(req *pb.SearchRequest) (*pb.SearchResponse, error) {
log.Trace().Str("query", req.GetQuery()).Msg("fetching results from indexer")
func (service *IndexerService) Search(query string, limit int32, indexer string) (*pb.SearchResponse, error) {
searchResult, err := service.indexer.Search(req.GetQuery(), req.GetLimit(), req.GetTracker())
searchResult, err := service.indexer.Search(query, limit, indexer)
if err != nil {
log.Error().Err(err).Msg("failed to search in indexer")
return nil, err
internal/metadata/client.go
+1 -1
View File
@@ -10,7 +10,7 @@ import (
pb "homelab.lan/music-agregator/gen/metadata/v1"
)
func newMetadataClient(endpoint string) (pb.MetadataServiceClient, *grpc.ClientConn, error) {
func NewMetadataClient(endpoint string) (pb.MetadataServiceClient, *grpc.ClientConn, error) {
log.Trace().Str("endpoint", endpoint).Msg("connecting to metadata service")
conn, err := grpc.NewClient(endpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))
internal/metadata/server.go
+9 -4
View File
@@ -17,7 +17,7 @@ type MetadataServer struct {
}
func NewMetadataServer(cfg config.Config) (*MetadataServer, error) {
client, conn, err := newMetadataClient(cfg.Metadata.Endpoint)
client, conn, err := NewMetadataClient(cfg.Metadata.Endpoint)
if err != nil {
log.Err(err).Msg("failed to initialize MetadataServer")
return nil, err
@@ -36,7 +36,7 @@ func (s *MetadataServer) Register(server *grpc.Server) {
pb.RegisterMetadataServiceServer(server, s)
}
func (s *MetadataServer) GetArtist(ctx context.Context, req *pb.GetArtistRequest) (*pb.Artist, error) {
func (s *MetadataServer) GetArtist(ctx context.Context, req *pb.GetArtistRequest) (*pb.GetArtistResponse, error) {
log.Debug().Msg("metadata GetArtist")
return s.client.GetArtist(ctx, req)
}
@@ -46,7 +46,7 @@ func (s *MetadataServer) SearchArtists(ctx context.Context, req *pb.SearchArtist
return s.client.SearchArtists(ctx, req)
}
func (s *MetadataServer) GetAlbum(ctx context.Context, req *pb.GetAlbumRequest) (*pb.Album, error) {
func (s *MetadataServer) GetAlbum(ctx context.Context, req *pb.GetAlbumRequest) (*pb.GetAlbumResponse, error) {
log.Debug().Msg("metadata GetAlbum")
return s.client.GetAlbum(ctx, req)
}
@@ -56,7 +56,7 @@ func (s *MetadataServer) GetArtistAlbums(ctx context.Context, req *pb.GetArtistA
return s.client.GetArtistAlbums(ctx, req)
}
func (s *MetadataServer) GetTrack(ctx context.Context, req *pb.GetTrackRequest) (*pb.Track, error) {
func (s *MetadataServer) GetTrack(ctx context.Context, req *pb.GetTrackRequest) (*pb.GetTrackResponse, error) {
log.Debug().Msg("metadata GetTrack")
return s.client.GetTrack(ctx, req)
}
@@ -66,6 +66,11 @@ func (s *MetadataServer) GetAlbumTracks(ctx context.Context, req *pb.GetAlbumTra
return s.client.GetAlbumTracks(ctx, req)
}
func (s *MetadataServer) SearchAlbums(ctx context.Context, req *pb.SearchAlbumsRequest) (*pb.SearchAlbumsResponse, error) {
log.Debug().Str("query", req.GetQuery()).Str("artist", req.GetArtist()).Msg("metadata SearchAlbums")
return s.client.SearchAlbums(ctx, req)
}
func (s *MetadataServer) SyncArtist(ctx context.Context, req *pb.SyncArtistRequest) (*pb.SyncArtistResponse, error) {
log.Debug().Msg("metadata SyncArtist")
return s.client.SyncArtist(ctx, req)
internal/server.go
+37
View File
@@ -0,0 +1,37 @@
package internal
import (
"context"
"github.com/jackc/pgx/v5"
"github.com/riverqueue/river"
"github.com/rs/zerolog/log"
"google.golang.org/grpc"
pb "homelab.lan/music-agregator/gen/music_agregator/v1"
"homelab.lan/music-agregator/internal/config"
)
type MusicAgregatorServer struct {
service *MusicAgregatorService
pb.UnimplementedMusicAgregatorServiceServer
}
func NewMusicAgregatorServer(cfg config.Config, riverClient *river.Client[pgx.Tx]) (*MusicAgregatorServer, error) {
service, err := NewMusicAgregatorService(cfg, riverClient)
if err != nil {
log.Err(err).Msg("failed to create MusicAgregatorService")
return nil, err
}
return &MusicAgregatorServer{
service: service,
}, nil
}
func (s *MusicAgregatorServer) MonitorAlbum(ctx context.Context, req *pb.MonitorAlbumRequest) (*pb.MonitorAlbumResponse, error) {
return s.service.MonitorAlbum(ctx, req)
}
func (s *MusicAgregatorServer) Register(server *grpc.Server) {
pb.RegisterMusicAgregatorServiceServer(server, s)
}
internal/service.go
+91
View File
@@ -0,0 +1,91 @@
package internal
import (
"context"
"github.com/jackc/pgx/v5"
"github.com/riverqueue/river"
"github.com/rs/zerolog/log"
"google.golang.org/grpc"
metadataPb "homelab.lan/music-agregator/gen/metadata/v1"
pb "homelab.lan/music-agregator/gen/music_agregator/v1"
"homelab.lan/music-agregator/internal/config"
"homelab.lan/music-agregator/internal/indexer"
"homelab.lan/music-agregator/internal/metadata"
)
type MusicAgregatorService struct {
config config.Config
metadataClient metadataPb.MetadataServiceClient
metadataConn *grpc.ClientConn
indexer *indexer.IndexerService
}
func NewMusicAgregatorService(cfg config.Config, riverClient *river.Client[pgx.Tx]) (*MusicAgregatorService, error) {
indexer, err := indexer.NewIndexerService(cfg, riverClient)
if err != nil {
log.Err(err).Msg("failed to create IndexerService")
return nil, err
}
metadataClient, conn, err := metadata.NewMetadataClient(cfg.Metadata.Endpoint)
if err != nil {
log.Err(err).Msg("failed to create metadata client")
return nil, err
}
return &MusicAgregatorService{
config: cfg,
metadataClient: metadataClient,
metadataConn: conn,
indexer: indexer,
}, nil
}
func (s *MusicAgregatorService) Close() {
if s.metadataConn != nil {
s.metadataConn.Close()
}
}
func (service *MusicAgregatorService) MonitorAlbum(ctx context.Context, req *pb.MonitorAlbumRequest) (*pb.MonitorAlbumResponse, error) {
resp, err := service.metadataClient.GetAlbum(ctx, &metadataPb.GetAlbumRequest{
Identifier: &metadataPb.GetAlbumRequest_Id{Id: req.GetAlbumId()},
})
if err != nil {
log.Error().Err(err).Str("album_id", req.GetAlbumId()).Msg("metadata GetAlbum failed")
return nil, err
}
album := resp.GetAlbum()
artistName := ""
if len(album.GetArtists()) > 0 {
artistName = album.GetArtists()[0].GetArtist().GetName()
}
log.Debug().
Str("album_id", req.GetAlbumId()).
Str("title", album.GetTitle()).
Str("artist", artistName).
Msg("album found, monitoring")
query := album.GetTitle()
if artistName != "" {
query = artistName + " " + query
}
tracker := req.GetIndexerOptions().GetTracker()
if tracker == "" {
tracker = "all"
}
searchResult, err := service.indexer.Search(query, -1, tracker)
if err != nil {
log.Error().Err(err).Str("query", query).Msg("indexer search album failed")
return nil, err
}
log.Debug().Int("results", len(searchResult.GetResult())).Str("query", query).Msg("indexer search completed")
return &pb.MonitorAlbumResponse{}, nil
}