package library import ( "context" "encoding/json" "fmt" "log/slog" "time" "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" "git.fabledsword.com/bvandeusen/minstrel/internal/coverart" "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" ) // LibraryStageTallies wires the scanner.Stats shape into the scan_runs jsonb // column. Field names match the DB-side jsonb schema referenced by the // admin SPA. Defined here (not on Stats directly) so the wire format // stays stable even if the scanner adds internal counters. type LibraryStageTallies struct { Scanned int `json:"scanned"` Added int `json:"added"` Updated int `json:"updated"` Skipped int `json:"skipped"` Errored int `json:"errored"` } // MBIDBackfillStageTallies wires BackfillMBIDsResult into the scan_runs jsonb column. type MBIDBackfillStageTallies struct { Processed int `json:"processed"` Healed int `json:"healed"` Skipped int `json:"skipped"` Duplicates int `json:"duplicates"` } // CoverEnrichStageTallies wires EnrichBatch tallies into the scan_runs jsonb column. type CoverEnrichStageTallies struct { Processed int `json:"processed"` Succeeded int `json:"succeeded"` Failed int `json:"failed"` } // ArtistArtEnrichStageTallies wires EnrichArtistBatch tallies into // the scan_runs jsonb column. type ArtistArtEnrichStageTallies struct { Processed int `json:"processed"` Succeeded int `json:"succeeded"` Failed int `json:"failed"` } // RunScanConfig assembles the knobs for the orchestrator. BackfillCap is // the album/artist-MBID-backfill batch cap (5000); EnrichCap / // ArtistEnrichCap gate cover/artist-art enrichment (0 = off, <0 = // unbounded — #388 removed the global cover-art cap; remote providers // self-throttle per provider). type RunScanConfig struct { BackfillCap int EnrichCap int ArtistEnrichCap int // M7 cover-sources: cap for the artist-art enrich stage DataDir string // M7 cover-sources: where to write artist-art files } // RunScan is the unified scan-chain orchestrator. Creates a scan_runs row, // runs file walk → MBID backfill → cover enrich in sequence, persists per- // stage tallies as each completes, sets finished_at on the row. // // Errors at any stage are logged + recorded in error_message but do NOT // halt subsequent stages — a scanner failure shouldn't prevent the cover // enricher from picking up rows that earlier scans imported. // // Returns the scan_runs row id so the manual-trigger endpoint can return it. func RunScan( ctx context.Context, pool *pgxpool.Pool, scanner *Scanner, enricher *coverart.Enricher, logger *slog.Logger, cfg RunScanConfig, ) (pgtype.UUID, error) { q := dbq.New(pool) row, err := q.InsertScanRun(ctx) if err != nil { return pgtype.UUID{}, fmt.Errorf("insert scan_run: %w", err) } logger.Info("scan run started", "id", row.ID) publishScanEvent("scan.run_started", row.ID, nil) var firstErr error captureErr := func(stage string, stageErr error) { logger.Warn("scan run stage failed", "stage", stage, "id", row.ID, "err", stageErr) if firstErr == nil { firstErr = fmt.Errorf("%s: %w", stage, stageErr) } } // Stage 1: library file-walk scan. if scanner != nil { var lastStats Stats libPub := newStagePublisher(ctx, 500, time.Second, func() []byte { blob, _ := json.Marshal(LibraryStageTallies(lastStats)) return blob }, func(blob []byte) error { return q.UpdateScanRunLibrary(ctx, dbq.UpdateScanRunLibraryParams{ ID: row.ID, Library: blob, }) }, ) stats, serr := scanner.Scan(ctx, func(s Stats) { lastStats = s libPub.Tick() }) lastStats = stats // ensure final flush sees post-loop value if err := libPub.Flush(); err != nil { captureErr("library_persist", err) } if serr != nil { captureErr("library", serr) } } // Stage 2: MBID backfill. if cfg.BackfillCap != 0 { var lastRes BackfillMBIDsResult bfPub := newStagePublisher(ctx, 500, time.Second, func() []byte { blob, _ := json.Marshal(MBIDBackfillStageTallies(lastRes)) return blob }, func(blob []byte) error { return q.UpdateScanRunMbidBackfill(ctx, dbq.UpdateScanRunMbidBackfillParams{ ID: row.ID, MbidBackfill: blob, }) }, ) bfRes, bferr := BackfillMBIDs(ctx, pool, logger, cfg.BackfillCap, func(r BackfillMBIDsResult) { lastRes = r bfPub.Tick() }) lastRes = bfRes if err := bfPub.Flush(); err != nil { captureErr("mbid_backfill_persist", err) } if bferr != nil { captureErr("mbid_backfill", bferr) } } // Stage 2b: track recording-MBID backfill. Unblocks the ListenBrainz // similarity pipeline (gated on tracks.mbid IS NOT NULL). Log-only // progress — no scan_runs jsonb column to avoid a schema addition. // // Unbounded (-1), unlike the album backfill's 5000 staged cap: this // is a one-time whole-library heal with no progress UI, and a cap // just means tracks.mbid stays partially NULL until N future scans // catch up (re-reading untagged files wastefully each pass). One // uncapped pass converges; subsequent scans only re-read the // remaining NULL (genuinely untagged) rows, which is cheap. if cfg.BackfillCap != 0 { if _, tberr := BackfillTrackMBIDs(ctx, pool, logger, -1, nil); tberr != nil { captureErr("track_mbid_backfill", tberr) } } // Stage 3: cover enrichment. if enricher != nil && cfg.EnrichCap != 0 { var lastP, lastS, lastF int coverPub := newStagePublisher(ctx, 500, time.Second, func() []byte { blob, _ := json.Marshal(CoverEnrichStageTallies{ Processed: lastP, Succeeded: lastS, Failed: lastF, }) return blob }, func(blob []byte) error { return q.UpdateScanRunCoverEnrich(ctx, dbq.UpdateScanRunCoverEnrichParams{ ID: row.ID, CoverEnrich: blob, }) }, ) processed, succeeded, failed, ceerr := enricher.EnrichBatch(ctx, cfg.EnrichCap, func(p, s, f int) { lastP, lastS, lastF = p, s, f coverPub.Tick() }) lastP, lastS, lastF = processed, succeeded, failed if err := coverPub.Flush(); err != nil { captureErr("cover_enrich_persist", err) } if ceerr != nil { captureErr("cover_enrich", ceerr) } } // Stage 4: artist art enrichment. if enricher != nil && cfg.ArtistEnrichCap != 0 && cfg.DataDir != "" { var lastP, lastS, lastF int aaPub := newStagePublisher(ctx, 500, time.Second, func() []byte { blob, _ := json.Marshal(ArtistArtEnrichStageTallies{ Processed: lastP, Succeeded: lastS, Failed: lastF, }) return blob }, func(blob []byte) error { return q.UpdateScanRunArtistArtEnrich(ctx, dbq.UpdateScanRunArtistArtEnrichParams{ ID: row.ID, ArtistArtEnrich: blob, }) }, ) processed, succeeded, failed, ceerr := enricher.EnrichArtistBatch(ctx, cfg.ArtistEnrichCap, cfg.DataDir, func(p, s, f int) { lastP, lastS, lastF = p, s, f aaPub.Tick() }) lastP, lastS, lastF = processed, succeeded, failed if err := aaPub.Flush(); err != nil { captureErr("artist_art_enrich_persist", err) } if ceerr != nil { captureErr("artist_art_enrich", ceerr) } } // Finalize: set finished_at + error message (or empty string if all stages succeeded). errMsg := "" if firstErr != nil { errMsg = firstErr.Error() } if ferr := q.FinishScanRun(ctx, dbq.FinishScanRunParams{ ID: row.ID, Column2: errMsg, }); ferr != nil { // Even the finalize failed; log loudly, return the original error if any. logger.Error("scan run finalize failed", "id", row.ID, "err", ferr) if firstErr != nil { return row.ID, firstErr } return row.ID, ferr } logger.Info("scan run complete", "id", row.ID, "error", errMsg) publishScanEvent("scan.run_finished", row.ID, map[string]any{ "error_message": errMsg, }) return row.ID, firstErr }