feat(server/m7-scan-progress): wire stagePublisher into 4 scan stages

Each of the 4 stage workers (library walk, MBID backfill, cover enrich, artist art enrich) gains a progressCb parameter that receives a snapshot of the current running state per item processed. The orchestrator stores the snapshot in a local var that the publisher's marshal closure reads at flush time. scanrun.go's 4 stage blocks are rewritten to construct one publisher per stage, pass a Tick-firing closure to the worker as progressCb, and call Flush() after the worker returns (routing any persist error to captureErr — preserves today's diagnostic behavior). The previous post-stage UpdateScanRun* calls are removed; the publisher's Flush handles the final write. Cadence: every 500 items OR every 1s, whichever fires first. Outside-orchestrator callers (scanner_test.go, enricher_test.go) pass nil for progressCb — no-op behavior preserved for tests that don't need progress observability. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-06 21:14:50 -04:00
parent 7b9ea131a2
commit 3e9e46f190
7 changed files with 141 additions and 45 deletions
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"log/slog"
+	"time"

 	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/jackc/pgx/v5/pgxpool"
@@ -93,68 +94,121 @@ func RunScan(

 	// Stage 1: library file-walk scan.
 	if scanner != nil {
-		stats, serr := scanner.Scan(ctx)
+		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)
 		}
-		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)
+		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)
 		}
-		blob, _ := json.Marshal(MBIDBackfillStageTallies{
-			Processed: bfRes.Processed, Healed: bfRes.Healed,
-			Skipped: bfRes.Skipped, Duplicates: bfRes.Duplicates,
-		})
-		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)
+		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)
 		}
-		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)
-		}
 	}

 	// Stage 4: artist art enrichment.
 	if enricher != nil && cfg.ArtistEnrichCap != 0 && cfg.DataDir != "" {
-		processed, succeeded, failed, ceerr := enricher.EnrichArtistBatch(ctx, cfg.ArtistEnrichCap, 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)
 		}
-		blob, _ := json.Marshal(ArtistArtEnrichStageTallies{
-			Processed: processed, Succeeded: succeeded, Failed: failed,
-		})
-		if uerr := q.UpdateScanRunArtistArtEnrich(ctx, dbq.UpdateScanRunArtistArtEnrichParams{
-			ID: row.ID, ArtistArtEnrich: blob,
-		}); uerr != nil {
-			captureErr("artist_art_enrich_persist", uerr)
-		}
 	}

 	// Finalize: set finished_at + error message (or empty string if all stages succeeded).