package lidarrrequests import ( "context" "fmt" "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/lidarr" "git.fabledsword.com/bvandeusen/minstrel/internal/lidarrconfig" ) // Reconciler is a background worker that periodically scans approved // lidarr_requests. For rows whose Lidarr add hasn't been confirmed it // (re)sends the add idempotently until it sticks; for confirmed rows it // 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 clientFn func() *lidarr.Client logger *slog.Logger bus *eventbus.Bus tick time.Duration batch int32 } // NewReconciler constructs a Reconciler with production defaults: // 5-minute tick, batch size 50. clientFn is the per-tick Lidarr client // factory (same pattern as Service) so config changes apply without a // restart; pass nil to disable the add-retry path. 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, clientFn func() *lidarr.Client, logger *slog.Logger, bus *eventbus.Bus) *Reconciler { if clientFn == nil { clientFn = func() *lidarr.Client { return nil } } return &Reconciler{ pool: pool, lidarrCfg: cfg, clientFn: clientFn, 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 "" } const hex = "0123456789abcdef" out := make([]byte, 36) pos := 0 for i, b := range u.Bytes { if i == 4 || i == 6 || i == 8 || i == 10 { out[pos] = '-' pos++ } out[pos] = hex[b>>4] out[pos+1] = hex[b&0xf] pos += 2 } return string(out) } // 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, cfg, row); err != nil { r.logger.Warn("lidarrrequests: reconciler row failed", "request_id", row.ID, "kind", row.Kind, "err", err, ) } } return nil } // reconcileRow drives a request forward by one step. If Lidarr hasn't // confirmed the add yet it (re)sends it and returns — the import can't // land before Lidarr has the add, so there's nothing to match this tick. // Once confirmed, it checks the library for a match and, if found, calls // CompleteLidarrRequest. Returns an error only for unexpected failures // (the add retry's transient error included, so tickOnce logs WARN and // it's retried next tick — "Lidarr just keeps trying"); a no-match is // not an error. func (r *Reconciler) reconcileRow(ctx context.Context, q *dbq.Queries, cfg lidarrconfig.Config, row dbq.LidarrRequest) error { if !row.LidarrAddConfirmedAt.Valid { if err := r.ensureLidarrAdd(ctx, q, cfg, row); err != nil { return err } // ensureLidarrAdd either confirmed the add or no-op'd (Lidarr // client unavailable). Fall through to import-matching so an // item already present in the library still completes this tick. } 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 } } // ensureLidarrAdd (re)sends the Lidarr add for an approved request whose // add hasn't been confirmed. On success (including ErrAlreadyExists, // handled inside sendLidarrAdd) it stamps lidarr_add_confirmed_at so this // stops re-sending and the next tick moves to import-matching. A failure // is returned so tickOnce logs WARN and the row is retried next tick; // there is intentionally no failed-state or attempt cap — per product // decision Lidarr just keeps trying and the operator monitors Lidarr. func (r *Reconciler) ensureLidarrAdd(ctx context.Context, q *dbq.Queries, cfg lidarrconfig.Config, row dbq.LidarrRequest) error { client := r.clientFn() if client == nil { // Lidarr disabled mid-flight; tickOnce already gates on // cfg.Enabled, so this is just defensive — nothing to do. return nil } if err := sendLidarrAdd(ctx, client, cfg, row); err != nil { return fmt.Errorf("lidarr add retry: %w", err) } return q.MarkLidarrRequestAddConfirmed(ctx, row.ID) } 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 }