package requests import ( "context" "errors" "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/notifications" "git.fabledsword.com/bvandeusen/minstrel/internal/wanted" ) // Reconciler is a background worker that completes approved requests once // the library holds what they asked for: an artist request when the artist // is in the library, an album or track request when the want it recorded is // held (internal/wanted settles wants). Errors per row are logged at WARN and // do not abort the tick. type Reconciler struct { pool *pgxpool.Pool logger *slog.Logger bus *eventbus.Bus notifier *notifications.Notifier // nil: no inbox notifications 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. func NewReconciler(pool *pgxpool.Pool, logger *slog.Logger, bus *eventbus.Bus) *Reconciler { return &Reconciler{pool: pool, logger: logger, bus: bus, tick: 5 * time.Minute, batch: 50} } // SetNotifier makes completions land in the requester's notifications inbox // (M489). Without one, completions are only broadcast on the bus. func (r *Reconciler) SetNotifier(n *notifications.Notifier) { r.notifier = n } // publishCompleted tells the requester their request arrived: a // request.status_changed event for open screens, and a notification that // outlives the connection. func (r *Reconciler) publishCompleted(ctx context.Context, row dbq.Request) { p := notifications.Payload{ RequestID: formatUUIDForBus(row.ID), RequestKind: string(row.Kind), Name: DisplayName(row), Artist: row.ArtistName, Title: requestedTitle(row), } if row.MatchedAlbumID.Valid { p.AlbumID = formatUUIDForBus(row.MatchedAlbumID) } else if row.MatchedArtistID.Valid { p.ArtistID = formatUUIDForBus(row.MatchedArtistID) } r.notifier.NotifyLogged(ctx, notifications.KindRequestCompleted, notifications.ToUser(row.UserID), p.Map()) 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 requests 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("requests: reconciler tick failed", "err", err) } } } } // tickOnce loads approved requests (oldest first, up to r.batch) and // completes those the library now holds. func (r *Reconciler) tickOnce(ctx context.Context) error { q := dbq.New(r.pool) rows, err := q.ListApprovedRequestsForReconcile(ctx, r.batch) if err != nil { return err } for _, row := range rows { if err := r.reconcileRow(ctx, q, row); err != nil { r.logger.Warn("requests: reconciler row failed", "request_id", row.ID, "kind", row.Kind, "err", err) } } return nil } // reconcileRow completes one request if the library holds it. A request // not yet held is not an error. func (r *Reconciler) reconcileRow(ctx context.Context, q *dbq.Queries, row dbq.Request) error { if row.Kind == dbq.RequestKindArtist { return r.reconcileArtist(ctx, q, row) } return r.reconcileAlbum(ctx, q, row) } func (r *Reconciler) reconcileArtist(ctx context.Context, q *dbq.Queries, row dbq.Request) error { var artistID pgtype.UUID err := r.pool.QueryRow(ctx, "SELECT id FROM artists WHERE mbid = $1", row.ArtistMbid).Scan(&artistID) if errors.Is(err, pgx.ErrNoRows) { return nil // not in library yet } if err != nil { return err } completed, err := q.CompleteRequest(ctx, dbq.CompleteRequestParams{ID: row.ID, MatchedArtistID: artistID}) if err != nil { return err } r.publishCompleted(ctx, completed) return nil } // reconcileAlbum completes an album or track request once its want is held, // matched to the album that answered it. A track request is matched to the // track it named when that recording arrived, else the newest arrival. func (r *Reconciler) reconcileAlbum(ctx context.Context, q *dbq.Queries, row dbq.Request) error { if row.AlbumMbid == nil || *row.AlbumMbid == "" { return nil } w, err := q.GetWantByReleaseGroup(ctx, *row.AlbumMbid) if errors.Is(err, pgx.ErrNoRows) { // Removed from the wanted list while the request waited. Wanting it // again keeps the approval meaningful; the admin who removed it can // reject a request instead. _, err = wanted.WantWith(ctx, q, wantFor(row)) return err } if err != nil { return err } if !w.FulfilledAt.Valid || !w.AlbumID.Valid { return nil } params := dbq.CompleteRequestParams{ID: row.ID, MatchedAlbumID: w.AlbumID} if row.Kind == dbq.RequestKindTrack { var trackID pgtype.UUID err := r.pool.QueryRow(ctx, ` SELECT id FROM tracks WHERE album_id = $1 AND missing_since IS NULL ORDER BY (mbid IS NOT DISTINCT FROM $2) DESC, added_at DESC, id LIMIT 1`, w.AlbumID, row.TrackMbid).Scan(&trackID) if err != nil && !errors.Is(err, pgx.ErrNoRows) { return err } params.MatchedTrackID = trackID } completed, err := q.CompleteRequest(ctx, params) if err != nil { return err } r.publishCompleted(ctx, completed) return nil }