package lidarrrequests import ( "context" "log/slog" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" "git.fabledsword.com/bvandeusen/minstrel/internal/eventbus" "git.fabledsword.com/bvandeusen/minstrel/internal/lidarrconfig" ) // Reconciler is a background worker that periodically scans approved // lidarr_requests and transitions them to completed when their target // track/album/artist has appeared in the local library (matched by MBID). // Errors per-row are logged at WARN and do not abort the tick. type Reconciler struct { pool *pgxpool.Pool lidarrCfg *lidarrconfig.Service logger *slog.Logger bus *eventbus.Bus tick time.Duration batch int32 } // NewReconciler constructs a Reconciler with production defaults: // 5-minute tick, batch size 50. Pass nil for bus when no SSE publishing // is desired (tests, or environments without the event stream enabled). func NewReconciler(pool *pgxpool.Pool, cfg *lidarrconfig.Service, logger *slog.Logger, bus *eventbus.Bus) *Reconciler { return &Reconciler{ pool: pool, lidarrCfg: cfg, logger: logger, bus: bus, tick: 5 * time.Minute, batch: 50, } } // publishCompleted broadcasts a request.status_changed event scoped to // the original requester. No-op when bus is nil. func (r *Reconciler) publishCompleted(row dbq.LidarrRequest) { if r.bus == nil { return } r.bus.Publish(eventbus.Event{ Kind: "request.status_changed", UserID: formatUUIDForBus(row.UserID), Data: map[string]any{ "request_id": formatUUIDForBus(row.ID), "status": "completed", "kind": string(row.Kind), }, }) } // formatUUIDForBus renders a pgtype.UUID as the canonical 8-4-4-4-12 hex // string. Mirrors helpers in internal/api/convert.go and other packages; // inlined here to avoid a back-edge dependency from lidarrrequests onto // the api layer. func formatUUIDForBus(u pgtype.UUID) string { if !u.Valid { return "" } return string([]byte{ hex(u.Bytes[0]>>4), hex(u.Bytes[0]&0xf), hex(u.Bytes[1]>>4), hex(u.Bytes[1]&0xf), hex(u.Bytes[2]>>4), hex(u.Bytes[2]&0xf), hex(u.Bytes[3]>>4), hex(u.Bytes[3]&0xf), '-', hex(u.Bytes[4]>>4), hex(u.Bytes[4]&0xf), hex(u.Bytes[5]>>4), hex(u.Bytes[5]&0xf), '-', hex(u.Bytes[6]>>4), hex(u.Bytes[6]&0xf), hex(u.Bytes[7]>>4), hex(u.Bytes[7]&0xf), '-', hex(u.Bytes[8]>>4), hex(u.Bytes[8]&0xf), hex(u.Bytes[9]>>4), hex(u.Bytes[9]&0xf), '-', hex(u.Bytes[10]>>4), hex(u.Bytes[10]&0xf), hex(u.Bytes[11]>>4), hex(u.Bytes[11]&0xf), hex(u.Bytes[12]>>4), hex(u.Bytes[12]&0xf), hex(u.Bytes[13]>>4), hex(u.Bytes[13]&0xf), hex(u.Bytes[14]>>4), hex(u.Bytes[14]&0xf), hex(u.Bytes[15]>>4), hex(u.Bytes[15]&0xf), }) } func hex(b byte) byte { if b < 10 { return '0' + b } return 'a' + (b - 10) } // Run blocks until ctx is cancelled, ticking every r.tick. Errors from // tickOnce are logged at WARN and never propagated. func (r *Reconciler) Run(ctx context.Context) { t := time.NewTicker(r.tick) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: if err := r.tickOnce(ctx); err != nil { r.logger.Warn("lidarrrequests: reconciler tick failed", "err", err) } } } } // tickOnce loads approved requests (oldest first, up to r.batch) and // transitions any whose target MBID is now present in the local library. // It is exported-for-tests via the lowercase name (package-internal). func (r *Reconciler) tickOnce(ctx context.Context) error { cfg, err := r.lidarrCfg.Get(ctx) if err != nil { return err } if !cfg.Enabled { r.logger.Debug("lidarrrequests: reconciler skipping tick — lidarr disabled") return nil } q := dbq.New(r.pool) rows, err := q.ListApprovedLidarrRequestsForReconcile(ctx, r.batch) if err != nil { return err } for _, row := range rows { if err := r.reconcileRow(ctx, q, row); err != nil { r.logger.Warn("lidarrrequests: reconciler row failed", "request_id", row.ID, "kind", row.Kind, "err", err, ) } } return nil } // reconcileRow checks the library for a match and, if found, calls // CompleteLidarrRequest. It returns an error only for unexpected DB // failures; a no-match is not an error. func (r *Reconciler) reconcileRow(ctx context.Context, q *dbq.Queries, row dbq.LidarrRequest) error { switch row.Kind { case dbq.LidarrRequestKindArtist: return r.reconcileArtist(ctx, q, row) case dbq.LidarrRequestKindAlbum: return r.reconcileAlbum(ctx, q, row) case dbq.LidarrRequestKindTrack: return r.reconcileTrack(ctx, q, row) default: r.logger.Warn("lidarrrequests: reconciler unknown kind", "kind", row.Kind) return nil } } func (r *Reconciler) reconcileArtist(ctx context.Context, q *dbq.Queries, row dbq.LidarrRequest) error { var artistID pgtype.UUID err := r.pool.QueryRow(ctx, "SELECT id FROM artists WHERE mbid = $1", row.LidarrArtistMbid, ).Scan(&artistID) if err != nil { if isNoRows(err) { return nil // not in library yet } return err } completed, err := q.CompleteLidarrRequest(ctx, dbq.CompleteLidarrRequestParams{ ID: row.ID, MatchedArtistID: artistID, MatchedAlbumID: pgtype.UUID{}, MatchedTrackID: pgtype.UUID{}, }) if err != nil { return err } r.publishCompleted(completed) return nil } func (r *Reconciler) reconcileAlbum(ctx context.Context, q *dbq.Queries, row dbq.LidarrRequest) error { if row.LidarrAlbumMbid == nil { return nil } var albumID pgtype.UUID err := r.pool.QueryRow(ctx, "SELECT id FROM albums WHERE mbid = $1", *row.LidarrAlbumMbid, ).Scan(&albumID) if err != nil { if isNoRows(err) { return nil } return err } completed, err := q.CompleteLidarrRequest(ctx, dbq.CompleteLidarrRequestParams{ ID: row.ID, MatchedAlbumID: albumID, MatchedArtistID: pgtype.UUID{}, MatchedTrackID: pgtype.UUID{}, }) if err != nil { return err } r.publishCompleted(completed) return nil } func (r *Reconciler) reconcileTrack(ctx context.Context, q *dbq.Queries, row dbq.LidarrRequest) error { if row.LidarrAlbumMbid == nil { return nil } // Track-kind requests match via their parent album's MBID, not track.mbid. var albumID pgtype.UUID err := r.pool.QueryRow(ctx, "SELECT id FROM albums WHERE mbid = $1", *row.LidarrAlbumMbid, ).Scan(&albumID) if err != nil { if isNoRows(err) { return nil } return err } // Load any track from that album to set matched_track_id. var trackID pgtype.UUID err = r.pool.QueryRow(ctx, "SELECT id FROM tracks WHERE album_id = $1 ORDER BY id LIMIT 1", albumID, ).Scan(&trackID) if err != nil { if isNoRows(err) { // Album is in the library but no tracks ingested yet — try again next tick. return nil } return err } completed, err := q.CompleteLidarrRequest(ctx, dbq.CompleteLidarrRequestParams{ ID: row.ID, MatchedAlbumID: albumID, MatchedTrackID: trackID, MatchedArtistID: pgtype.UUID{}, }) if err != nil { return err } r.publishCompleted(completed) return nil } func isNoRows(err error) bool { return err == pgx.ErrNoRows }