minstrel/internal/library/scanrun.go

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{
					Scanned: lastStats.Scanned, Added: lastStats.Added,
					Updated: lastStats.Updated, Skipped: lastStats.Skipped,
					Errored: lastStats.Errored,
				})
				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{
					Processed: lastRes.Processed, Healed: lastRes.Healed,
					Skipped: lastRes.Skipped, Duplicates: lastRes.Duplicates,
				})
				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
}