feat(server/m7-381): RunScan orchestrator chains library+backfill+enrich
Adds internal/library/scanrun.go with RunScan, which creates a scan_runs row and sequences file-walk → MBID backfill → cover enrich, persisting per-stage tallies as jsonb. Extends coverart.EnrichBatch to return (processed, succeeded, failed, err) so the orchestrator can classify outcomes. Replaces main.go's two separate boot goroutines with a single RunScan call; passes nil scanner in the no-startup-scan branch. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
+22
-20
@@ -83,32 +83,34 @@ func run() error {
|
|||||||
})
|
})
|
||||||
coverEnricher := coverart.NewEnricher(pool, logger.With("component", "coverart"), coverFetcher, cfg.Library.CoverArtFromMBCAA)
|
coverEnricher := coverart.NewEnricher(pool, logger.With("component", "coverart"), coverFetcher, cfg.Library.CoverArtFromMBCAA)
|
||||||
|
|
||||||
// One-shot MBID backfill: walks albums with NULL mbid, reads tags from
|
// One unified scan chain: library walk → MBID backfill → cover enrich.
|
||||||
// one track per album, persists the MBID. Filling in MBIDs unblocks the
|
// Boot-time scan and manual-trigger scans share this path; results land
|
||||||
// cover enricher for libraries that were imported before scanner-side
|
// in scan_runs for the admin overview.
|
||||||
// MBID extraction landed (M7 #380).
|
|
||||||
go func() {
|
|
||||||
if _, err := library.BackfillMBIDs(ctx, pool, logger.With("component", "mbid_backfill"), 5000); err != nil {
|
|
||||||
logger.Warn("mbid backfill failed", "err", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
if cfg.Library.ScanOnStartup && len(cfg.Library.ScanPaths) > 0 {
|
if cfg.Library.ScanOnStartup && len(cfg.Library.ScanPaths) > 0 {
|
||||||
go func() {
|
go func() {
|
||||||
if _, err := scanner.Scan(ctx); err != nil {
|
if _, err := library.RunScan(ctx, pool, scanner, coverEnricher,
|
||||||
logger.Error("startup scan failed", "err", err)
|
logger.With("component", "scan_run"),
|
||||||
}
|
library.RunScanConfig{
|
||||||
// Enrich any newly-imported albums (and resume any partial backlog).
|
BackfillCap: 5000,
|
||||||
if _, err := coverEnricher.EnrichBatch(ctx, cfg.Library.CoverArtBackfillCap); err != nil {
|
EnrichCap: cfg.Library.CoverArtBackfillCap,
|
||||||
logger.Warn("post-scan cover enrichment failed", "err", err)
|
},
|
||||||
|
); err != nil {
|
||||||
|
logger.Warn("startup scan run failed", "err", err)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
} else {
|
} else {
|
||||||
// No startup scan — still run the boot backfill once for any rows
|
// No startup file walk, but still run backfill + enrich so cover
|
||||||
// the previous run didn't get to.
|
// progress doesn't stall just because the operator disabled
|
||||||
|
// scan-on-startup.
|
||||||
go func() {
|
go func() {
|
||||||
if _, err := coverEnricher.EnrichBatch(ctx, cfg.Library.CoverArtBackfillCap); err != nil {
|
if _, err := library.RunScan(ctx, pool, nil, coverEnricher,
|
||||||
logger.Warn("boot cover enrichment failed", "err", err)
|
logger.With("component", "scan_run"),
|
||||||
|
library.RunScanConfig{
|
||||||
|
BackfillCap: 5000,
|
||||||
|
EnrichCap: cfg.Library.CoverArtBackfillCap,
|
||||||
|
},
|
||||||
|
); err != nil {
|
||||||
|
logger.Warn("boot-only scan run failed", "err", err)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -103,27 +103,45 @@ func (e *Enricher) EnrichAlbum(ctx context.Context, albumID pgtype.UUID) error {
|
|||||||
|
|
||||||
// EnrichBatch drains up to limit albums whose cover_art_source is NULL.
|
// EnrichBatch drains up to limit albums whose cover_art_source is NULL.
|
||||||
// Iterates serially — the fetcher's rate limiter prevents bursts.
|
// Iterates serially — the fetcher's rate limiter prevents bursts.
|
||||||
func (e *Enricher) EnrichBatch(ctx context.Context, limit int) (int, error) {
|
// Returns per-batch tallies for the orchestrator's scan-run record.
|
||||||
|
//
|
||||||
|
// "succeeded" means the album ended up with a non-NULL cover_art_source
|
||||||
|
// other than 'none' (i.e. we found art via sidecar or MBCAA). "failed"
|
||||||
|
// counts both EnrichAlbum errors and rows that ended at 'none'.
|
||||||
|
func (e *Enricher) EnrichBatch(ctx context.Context, limit int) (processed, succeeded, failed int, err error) {
|
||||||
if limit <= 0 {
|
if limit <= 0 {
|
||||||
return 0, nil
|
return 0, 0, 0, nil
|
||||||
}
|
}
|
||||||
q := dbq.New(e.pool)
|
q := dbq.New(e.pool)
|
||||||
rows, err := q.ListAlbumsMissingCover(ctx, int32(limit))
|
rows, qerr := q.ListAlbumsMissingCover(ctx, int32(limit))
|
||||||
if err != nil {
|
if qerr != nil {
|
||||||
return 0, fmt.Errorf("list missing: %w", err)
|
return 0, 0, 0, fmt.Errorf("list missing: %w", qerr)
|
||||||
}
|
}
|
||||||
processed := 0
|
|
||||||
for _, r := range rows {
|
for _, r := range rows {
|
||||||
if ctx.Err() != nil {
|
if ctx.Err() != nil {
|
||||||
return processed, ctx.Err()
|
return processed, succeeded, failed, ctx.Err()
|
||||||
}
|
|
||||||
if err := e.EnrichAlbum(ctx, r.ID); err != nil {
|
|
||||||
e.logger.Warn("coverart: batch entry failed",
|
|
||||||
"album_id", uuidString(r.ID), "err", err)
|
|
||||||
}
|
}
|
||||||
processed++
|
processed++
|
||||||
|
if eerr := e.EnrichAlbum(ctx, r.ID); eerr != nil {
|
||||||
|
e.logger.Warn("coverart: batch entry failed",
|
||||||
|
"album_id", uuidString(r.ID), "err", eerr)
|
||||||
|
failed++
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
// Re-read the source to classify the outcome.
|
||||||
|
row, rerr := q.GetAlbumWithFirstTrackPath(ctx, r.ID)
|
||||||
|
if rerr != nil {
|
||||||
|
failed++
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if row.CoverArtSource != nil &&
|
||||||
|
(*row.CoverArtSource == "sidecar" || *row.CoverArtSource == "mbcaa" || *row.CoverArtSource == "embedded") {
|
||||||
|
succeeded++
|
||||||
|
} else {
|
||||||
|
failed++
|
||||||
|
}
|
||||||
}
|
}
|
||||||
return processed, nil
|
return processed, succeeded, failed, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// EnrichRetryMissing drains both NULL and 'none' rows. Used by the admin
|
// EnrichRetryMissing drains both NULL and 'none' rows. Used by the admin
|
||||||
|
|||||||
@@ -219,7 +219,7 @@ func TestEnrichBatch_DrainsNullSourceOnly(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
e := NewEnricher(pool, discardLogger(), nil, false)
|
e := NewEnricher(pool, discardLogger(), nil, false)
|
||||||
processed, err := e.EnrichBatch(context.Background(), 100)
|
processed, _, _, err := e.EnrichBatch(context.Background(), 100)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("batch: %v", err)
|
t.Fatalf("batch: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,150 @@
|
|||||||
|
package library
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"log/slog"
|
||||||
|
|
||||||
|
"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"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// RunScanConfig assembles the knobs for the orchestrator. BackfillCap and
|
||||||
|
// EnrichCap mirror the existing boot-time caps (5000 / coverArtBackfillCap)
|
||||||
|
// so manual triggers don't accidentally drain the whole library.
|
||||||
|
type RunScanConfig struct {
|
||||||
|
BackfillCap int
|
||||||
|
EnrichCap int
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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)
|
||||||
|
|
||||||
|
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 {
|
||||||
|
stats, serr := scanner.Scan(ctx)
|
||||||
|
if serr != nil {
|
||||||
|
captureErr("library", serr)
|
||||||
|
}
|
||||||
|
blob, _ := json.Marshal(LibraryStageTallies{
|
||||||
|
Scanned: stats.Scanned, Added: stats.Added, Updated: stats.Updated,
|
||||||
|
Skipped: stats.Skipped, Errored: stats.Errored,
|
||||||
|
})
|
||||||
|
if uerr := q.UpdateScanRunLibrary(ctx, dbq.UpdateScanRunLibraryParams{
|
||||||
|
ID: row.ID, Library: blob,
|
||||||
|
}); uerr != nil {
|
||||||
|
captureErr("library_persist", uerr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Stage 2: MBID backfill.
|
||||||
|
if cfg.BackfillCap != 0 {
|
||||||
|
bfRes, bferr := BackfillMBIDs(ctx, pool, logger, cfg.BackfillCap)
|
||||||
|
if bferr != nil {
|
||||||
|
captureErr("mbid_backfill", bferr)
|
||||||
|
}
|
||||||
|
blob, _ := json.Marshal(MBIDBackfillStageTallies{
|
||||||
|
Processed: bfRes.Processed, Healed: bfRes.Healed, Skipped: bfRes.Skipped,
|
||||||
|
})
|
||||||
|
if uerr := q.UpdateScanRunMbidBackfill(ctx, dbq.UpdateScanRunMbidBackfillParams{
|
||||||
|
ID: row.ID, MbidBackfill: blob,
|
||||||
|
}); uerr != nil {
|
||||||
|
captureErr("mbid_backfill_persist", uerr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Stage 3: cover enrichment.
|
||||||
|
if enricher != nil && cfg.EnrichCap != 0 {
|
||||||
|
processed, succeeded, failed, ceerr := enricher.EnrichBatch(ctx, cfg.EnrichCap)
|
||||||
|
if ceerr != nil {
|
||||||
|
captureErr("cover_enrich", ceerr)
|
||||||
|
}
|
||||||
|
blob, _ := json.Marshal(CoverEnrichStageTallies{
|
||||||
|
Processed: processed, Succeeded: succeeded, Failed: failed,
|
||||||
|
})
|
||||||
|
if uerr := q.UpdateScanRunCoverEnrich(ctx, dbq.UpdateScanRunCoverEnrichParams{
|
||||||
|
ID: row.ID, CoverEnrich: blob,
|
||||||
|
}); uerr != nil {
|
||||||
|
captureErr("cover_enrich_persist", uerr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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)
|
||||||
|
return row.ID, firstErr
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user