From 4e6e2153c513a98d35219960081cd9d68ef93098 Mon Sep 17 00:00:00 2001
From: RXWatcher <14085001+RXWatcher@users.noreply.github.com>
Date: Wed, 27 May 2026 09:24:52 +0200
Subject: [PATCH] perf(audiobooks): parallelize enricher sweep

Was: serial loop over claimed batch of 50.
Now: N workers fan out over the batch (default 4, configurable via
SILO_AUDIOBOOK_ENRICH_WORKERS).

Adds a runBatch seam used by Run so tests can stub the per-item
enrich function and assert worker fan-out without hitting providers.
---
 internal/audiobooks/enrichment.go      | 71 ++++++++++++++++++++------
 internal/audiobooks/enrichment_test.go | 62 ++++++++++++++++++++++
 2 files changed, 117 insertions(+), 16 deletions(-)
 create mode 100644 internal/audiobooks/enrichment_test.go

diff --git a/internal/audiobooks/enrichment.go b/internal/audiobooks/enrichment.go
index bf040ace..d1469e10 100644
--- a/internal/audiobooks/enrichment.go
+++ b/internal/audiobooks/enrichment.go
@@ -20,6 +20,8 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/jackc/pgx/v5/pgxpool"
@@ -148,23 +150,12 @@ func (e *Enricher) Run(ctx context.Context) (int, error) {
 		return 0, nil
 	}
 
-	slog.Info("audiobook enrichment: sweep started", "count", len(items))
+	slog.Info("audiobook enrichment: sweep started",
+		"count", len(items),
+		"workers", e.workers,
+	)
 
-	enriched := 0
-	for _, item := range items {
-		if ctx.Err() != nil {
-			break
-		}
-		if err := e.enrichItem(ctx, item); err != nil {
-			slog.Warn("audiobook enrichment: item failed",
-				"content_id", item.ContentID,
-				"title", item.Title,
-				"error", err,
-			)
-			continue
-		}
-		enriched++
-	}
+	enriched := e.runBatch(ctx, items, e.enrichItem)
 
 	slog.Info("audiobook enrichment: sweep complete",
 		"attempted", len(items),
@@ -173,6 +164,54 @@ func (e *Enricher) Run(ctx context.Context) (int, error) {
 	return enriched, nil
 }
 
+// runBatch processes items in parallel using e.workers goroutines. The
+// enrichFn parameter exists for testing: production calls pass e.enrichItem.
+// Returns the count of items where enrichFn returned nil.
+func (e *Enricher) runBatch(ctx context.Context, items []enrichmentItemRow, enrichFn func(context.Context, enrichmentItemRow) error) int {
+	workers := e.workers
+	if workers <= 0 {
+		workers = 1
+	}
+	if workers > len(items) {
+		workers = len(items)
+	}
+
+	ch := make(chan enrichmentItemRow, workers)
+	var (
+		wg       sync.WaitGroup
+		enriched int64
+	)
+	for i := 0; i < workers; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for item := range ch {
+				if ctx.Err() != nil {
+					continue // drain
+				}
+				if err := enrichFn(ctx, item); err != nil {
+					slog.Warn("audiobook enrichment: item failed",
+						"content_id", item.ContentID,
+						"title", item.Title,
+						"error", err,
+					)
+					continue
+				}
+				atomic.AddInt64(&enriched, 1)
+			}
+		}()
+	}
+	for _, item := range items {
+		if ctx.Err() != nil {
+			break
+		}
+		ch <- item
+	}
+	close(ch)
+	wg.Wait()
+	return int(enriched)
+}
+
 // claimBatch returns up to batchSize audiobook items that need enrichment.
 // "Needs enrichment" means poster_path IS NULL or empty AND last_refreshed IS NULL.
 // We skip items where last_refreshed IS NOT NULL — those have already had at
diff --git a/internal/audiobooks/enrichment_test.go b/internal/audiobooks/enrichment_test.go
new file mode 100644
index 00000000..6525b915
--- /dev/null
+++ b/internal/audiobooks/enrichment_test.go
@@ -0,0 +1,62 @@
+package audiobooks
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// TestEnricherRunFansOut verifies that runBatch processes a claimed batch with
+// multiple workers in flight at once. The test installs a fake enrichItem path
+// that blocks until N concurrent calls are observed; if runBatch is still
+// serial, the test will time out and fail the max-in-flight assertion.
+func TestEnricherRunFansOut(t *testing.T) {
+	const wantWorkers = 4
+	const itemCount = 16
+
+	items := make([]enrichmentItemRow, itemCount)
+	for i := range items {
+		items[i] = enrichmentItemRow{ContentID: "test", Title: "t"}
+	}
+
+	var inFlight int32
+	var maxInFlight int32
+	var signaled int32
+	var wg sync.WaitGroup
+	wg.Add(wantWorkers)
+	gate := make(chan struct{})
+
+	enrich := func(ctx context.Context, item enrichmentItemRow) error {
+		cur := atomic.AddInt32(&inFlight, 1)
+		defer atomic.AddInt32(&inFlight, -1)
+		for {
+			prev := atomic.LoadInt32(&maxInFlight)
+			if cur <= prev || atomic.CompareAndSwapInt32(&maxInFlight, prev, cur) {
+				break
+			}
+		}
+		// First wantWorkers callers release the gate; later ones don't block.
+		if atomic.AddInt32(&signaled, 1) <= wantWorkers {
+			wg.Done()
+		}
+		select {
+		case <-gate:
+		case <-time.After(2 * time.Second):
+		}
+		return nil
+	}
+
+	go func() {
+		wg.Wait()
+		close(gate)
+	}()
+
+	e := &Enricher{workers: wantWorkers, batchSize: itemCount}
+	e.runBatch(context.Background(), items, enrich)
+
+	if got := atomic.LoadInt32(&maxInFlight); got < wantWorkers {
+		t.Errorf("max in-flight = %d, want >= %d", got, wantWorkers)
+	}
+}