feat(search): gate index events by active provider and harden rebuild reconcile

Completes the search-provider-interface wiring that the catalog hardening
commits already call into:

- Skip the transactional search-index-event write path when Meilisearch is
  not the active provider (ItemRepository.WithActiveSearchProvider /
  SearchIndexEventRepository.disabledByActiveProvider).
- Dead-letter catalog_search_index_events after 10 attempts instead of
  retrying forever.
- Track the rebuild high-water mark (MaxEventID / MarkProcessedThrough) and
  persist last_processed_event_id in UpdateStateAfterRebuild so a rebuild
  reconciles events enqueued during the rebuild.
- Validate (read-only) the embedding lock when embedding a search query
  instead of establishing/mutating it.
- Surface total_exact on the legacy /items browse response.
- Wire the active catalog search provider into the scanner and item repo at
  startup.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

cmd/silo/main.go

@@ -520,6 +520,12 @@ func main() {
 	// still encrypts/decrypts — no raw settings repo may escape into later wiring.
 	settingsRepo = catalog.NewEncryptedSettingsRepo(catalog.NewServerSettingsRepo(pool), dataCipher)
 	nodeID := resolveNodeIdentity()
+	catalogSearchStartupSettings, err := catalog.CatalogSearchSettingsFromMap(settings)
+	if err != nil {
+		slog.Warn("catalog search: failed to load settings for startup wiring; using postgres", "err", err)
+		catalogSearchStartupSettings = catalog.DefaultCatalogSearchSettings()
+	}
+	activeCatalogSearchProvider := catalog.ActiveCatalogSearchProvider(catalogSearchStartupSettings)
 
 	// Step 9: Validate
 	if err := cfg.Validate(); err != nil {
@@ -818,6 +824,7 @@ func main() {
 
 		ffprobePath := scanner.FFprobePathFromFFmpeg(cfg.Playback.FFmpegPath)
 		s := scanner.NewScanner(fileRepo, ffprobePath, deps.S3Public, cfg.Scanner.Workers, cfg.Scanner.EmptyTrashAfterScan)
+		s.SetSearchIndexProvider(activeCatalogSearchProvider)
 		configWatcher.OnChange(func(_, updated *config.Config) {
 			s.SetWorkers(updated.Scanner.Workers)
 		})
@@ -1064,7 +1071,7 @@ func main() {
 	if needsWorkers && deps.DB != nil && deps.FileRepo != nil {
 		chainRepo := metadata.NewChainRepository(deps.DB)
 		skippedRootRepo = metadata.NewSkippedRootRepository(deps.DB)
-		itemRepo = catalog.NewItemRepository(deps.DB)
+		itemRepo = catalog.NewItemRepository(deps.DB).WithActiveSearchProvider(activeCatalogSearchProvider)
 		episodeRepo = catalog.NewEpisodeRepository(deps.DB)
 		seasonRepo = catalog.NewSeasonRepository(deps.DB)
 		personRepo := catalog.NewPersonRepository(deps.DB)

internal/api/handlers/items.go

@@ -265,9 +265,10 @@ type itemListImageURLs struct {
 
 // browseResponse is the paginated response for the /items endpoint.
 type browseResponse struct {
-	Total   int                `json:"total"`
-	HasMore bool               `json:"has_more"`
-	Items   []itemListResponse `json:"items"`
+	Total      int                `json:"total"`
+	TotalExact bool               `json:"total_exact"`
+	HasMore    bool               `json:"has_more"`
+	Items      []itemListResponse `json:"items"`
 }
 
 type itemFiltersResponse struct {
@@ -600,9 +601,10 @@ func (h *ItemsHandler) writeCatalogBrowseResponse(w http.ResponseWriter, r *http
 	}
 
 	writeJSON(w, http.StatusOK, browseResponse{
-		Total:   result.Total,
-		HasMore: result.HasMore,
-		Items:   items,
+		Total:      result.Total,
+		TotalExact: result.TotalExact,
+		HasMore:    result.HasMore,
+		Items:      items,
 	})
 	return true
 }

internal/api/handlers/items_catalog_response_test.go

@@ -2,12 +2,29 @@ package handlers
 
 import (
 	"context"
+	"encoding/json"
+	"strings"
 	"testing"
 
 	"github.com/Silo-Server/silo-server/internal/catalog"
 	"github.com/Silo-Server/silo-server/internal/models"
 )
 
+func TestBrowseResponseIncludesTotalExact(t *testing.T) {
+	data, err := json.Marshal(browseResponse{
+		Total:      3,
+		TotalExact: true,
+		HasMore:    false,
+		Items:      []itemListResponse{},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(string(data), `"total_exact":true`) {
+		t.Fatalf("browse response missing total_exact: %s", data)
+	}
+}
+
 type countingItemListImageResolver struct {
 	singleCalls int
 	batchCalls  int

internal/catalog/item_repo.go

@@ -47,6 +47,17 @@ func (r *ItemRepository) WithSearchIndexEvents(events *SearchIndexEventRepositor
 	return r
 }
 
+func (r *ItemRepository) WithActiveSearchProvider(provider string) *ItemRepository {
+	if r == nil {
+		return nil
+	}
+	if r.searchIndexEvents == nil {
+		r.searchIndexEvents = NewSearchIndexEventRepository(r.pool)
+	}
+	r.searchIndexEvents.WithActiveProvider(provider)
+	return r
+}
+
 // GetPoster returns the current poster path and thumbhash for a media item.
 // Missing or NULL values are returned as empty strings.
 func (r *ItemRepository) GetPoster(ctx context.Context, contentID string) (posterPath string, posterThumbhash string, err error) {
@@ -397,6 +408,9 @@ func scanItemsWithTotal(rows pgx.Rows) ([]*models.MediaItem, int, error) {
 // Upsert inserts a new media item or updates all mutable fields if the
 // content_id already exists. The created_at timestamp is preserved on update.
 func (r *ItemRepository) Upsert(ctx context.Context, item *models.MediaItem) error {
+	if r.searchIndexEvents.disabledByActiveProvider() {
+		return r.upsert(ctx, r.pool, item)
+	}
 	tx, err := r.pool.Begin(ctx)
 	if err != nil {
 		return fmt.Errorf("begin media item upsert tx: %w", err)

internal/catalog/search_index_repo.go

@@ -18,6 +18,7 @@ const (
 	SearchIndexEventRename = "rename"
 
 	searchIndexMaintenanceLockID int64 = 0x53494c4f5345531
+	searchIndexEventMaxAttempts        = 10
 )
 
 type SearchIndexEvent struct {
@@ -64,6 +65,10 @@ func (r *SearchIndexEventRepository) WithActiveProvider(provider string) *Search
 	return r
 }
 
+func (r *SearchIndexEventRepository) disabledByActiveProvider() bool {
+	return r != nil && r.activeProviderKnown && r.activeProvider != SearchProviderMeilisearch
+}
+
 func (r *SearchIndexEventRepository) EnqueueUpsert(ctx context.Context, execer itemExecer, contentID string) error {
 	return r.enqueue(ctx, execer, SearchProviderMeilisearch, SearchIndexEventUpsert, contentID, "")
 }
@@ -330,6 +335,40 @@ func (r *SearchIndexEventRepository) MarkProcessed(ctx context.Context, ids []in
 	return err
 }
 
+func (r *SearchIndexEventRepository) MaxEventID(ctx context.Context, provider string) (int64, error) {
+	if r == nil || r.pool == nil {
+		return 0, nil
+	}
+	var maxID int64
+	err := r.pool.QueryRow(ctx, `
+		SELECT COALESCE(MAX(id), 0)
+		FROM catalog_search_index_events
+		WHERE provider = $1
+	`, provider).Scan(&maxID)
+	if isSearchIndexSchemaUnavailable(err) {
+		return 0, nil
+	}
+	return maxID, err
+}
+
+func (r *SearchIndexEventRepository) MarkProcessedThrough(ctx context.Context, provider string, maxID int64) error {
+	if r == nil || r.pool == nil || maxID <= 0 {
+		return nil
+	}
+	_, err := r.pool.Exec(ctx, `
+		UPDATE catalog_search_index_events
+		SET processed_at = NOW(),
+		    last_error = ''
+		WHERE provider = $1
+		  AND id <= $2
+		  AND processed_at IS NULL
+	`, provider, maxID)
+	if isSearchIndexSchemaUnavailable(err) {
+		return nil
+	}
+	return err
+}
+
 func (r *SearchIndexEventRepository) MarkFailed(ctx context.Context, ids []int64, cause error) error {
 	if r == nil || r.pool == nil || len(ids) == 0 {
 		return nil
@@ -341,10 +380,20 @@ func (r *SearchIndexEventRepository) MarkFailed(ctx context.Context, ids []int64
 	_, err := r.pool.Exec(ctx, `
 		UPDATE catalog_search_index_events
 		SET attempts = attempts + 1,
-		    available_at = NOW() + LEAST(((attempts + 1) * INTERVAL '30 seconds'), INTERVAL '15 minutes'),
-		    last_error = $2
+		    available_at = CASE
+		        WHEN attempts + 1 >= $3 THEN available_at
+		        ELSE NOW() + LEAST(((attempts + 1) * INTERVAL '30 seconds'), INTERVAL '15 minutes')
+		    END,
+		    processed_at = CASE
+		        WHEN attempts + 1 >= $3 THEN NOW()
+		        ELSE processed_at
+		    END,
+		    last_error = CASE
+		        WHEN attempts + 1 >= $3 THEN 'dead-lettered after ' || $3::text || ' attempts: ' || $2
+		        ELSE $2
+		    END
 		WHERE id = ANY($1)
-	`, ids, message)
+	`, ids, message, searchIndexEventMaxAttempts)
 	if isSearchIndexSchemaUnavailable(err) {
 		return nil
 	}
@@ -377,24 +426,25 @@ func (r *SearchIndexEventRepository) GetState(ctx context.Context, provider stri
 	return state, err
 }
 
-func (r *SearchIndexEventRepository) UpdateStateAfterRebuild(ctx context.Context, provider, activeIndexUID string, schemaVersion, documentCount int) error {
+func (r *SearchIndexEventRepository) UpdateStateAfterRebuild(ctx context.Context, provider, activeIndexUID string, schemaVersion, documentCount int, lastProcessedEventID int64) error {
 	if r == nil || r.pool == nil {
 		return nil
 	}
 	_, err := r.pool.Exec(ctx, `
 		INSERT INTO catalog_search_index_state (
 			provider, active_index_uid, schema_version, document_count,
-			last_rebuild_at, last_sync_at, updated_at
+			last_rebuild_at, last_sync_at, last_processed_event_id, updated_at
 		)
-		VALUES ($1, $2, $3, $4, NOW(), NOW(), NOW())
+		VALUES ($1, $2, $3, $4, NOW(), NOW(), $5, NOW())
 		ON CONFLICT (provider) DO UPDATE SET
 			active_index_uid = EXCLUDED.active_index_uid,
 			schema_version = EXCLUDED.schema_version,
 			document_count = EXCLUDED.document_count,
 			last_rebuild_at = EXCLUDED.last_rebuild_at,
 			last_sync_at = EXCLUDED.last_sync_at,
+			last_processed_event_id = GREATEST(catalog_search_index_state.last_processed_event_id, EXCLUDED.last_processed_event_id),
 			updated_at = NOW()
-	`, provider, activeIndexUID, schemaVersion, documentCount)
+	`, provider, activeIndexUID, schemaVersion, documentCount, lastProcessedEventID)
 	if isSearchIndexSchemaUnavailable(err) {
 		return nil
 	}

internal/catalog/search_index_repo_test.go

@@ -91,3 +91,14 @@ func TestEnqueueSearchIndexUpsertRunsWhenProviderIsMeilisearch(t *testing.T) {
 		t.Fatalf("expected one Exec call when provider is meilisearch, got %d", execer.calls)
 	}
 }
+
+func TestItemRepositoryActiveProviderDisablesSearchIndexEvents(t *testing.T) {
+	repo := (&ItemRepository{}).WithActiveSearchProvider(SearchProviderPostgres)
+
+	if repo.searchIndexEvents == nil {
+		t.Fatal("searchIndexEvents is nil")
+	}
+	if !repo.searchIndexEvents.disabledByActiveProvider() {
+		t.Fatal("postgres active provider should disable search index event work")
+	}
+}

internal/catalog/search_meilisearch_client_test.go

@@ -0,0 +1,17 @@
+package catalog
+
+import (
+	"context"
+	"strings"
+	"testing"
+)
+
+func TestMeilisearchWaitTaskRejectsZeroTaskUID(t *testing.T) {
+	err := (&meilisearchClient{}).WaitTask(context.Background(), 0)
+	if err == nil {
+		t.Fatal("WaitTask(0) returned nil")
+	}
+	if !strings.Contains(err.Error(), "task UID is required") {
+		t.Fatalf("WaitTask(0) error = %v", err)
+	}
+}

internal/recommendations/search_query_vector.go

@@ -32,8 +32,26 @@ func (e *Engine) EmbedSearchQuery(ctx context.Context, query string) ([]float32,
 	if len(vectors) == 0 || len(vectors[0]) == 0 {
 		return nil, fmt.Errorf("embedding API returned no query vector")
 	}
-	if err := e.ensureEmbeddingLock(ctx, vectors[0]); err != nil {
+	if err := e.validateQueryEmbeddingLock(ctx, vectors[0]); err != nil {
 		return nil, err
 	}
 	return ensureCanonicalDimensions(vectors[0])
 }
+
+func (e *Engine) validateQueryEmbeddingLock(ctx context.Context, vector []float32) error {
+	lock, err := e.repo.GetEmbeddingLock(ctx)
+	if err != nil {
+		return fmt.Errorf("load embedding lock: %w", err)
+	}
+	if lock == nil {
+		return nil
+	}
+	return validateQueryEmbeddingLock(lock, e.cfg.EmbeddingBaseURL, e.cfg.EmbeddingModel, len(vector))
+}
+
+func validateQueryEmbeddingLock(lock *EmbeddingLock, baseURL, model string, sourceDimensions int) error {
+	if lock == nil {
+		return nil
+	}
+	return lock.Validate(baseURL, model, sourceDimensions)
+}

internal/recommendations/search_query_vector_test.go

@@ -0,0 +1,22 @@
+package recommendations
+
+import "testing"
+
+func TestValidateQueryEmbeddingLockAllowsMissingLock(t *testing.T) {
+	if err := validateQueryEmbeddingLock(nil, "http://embeddings", "model-a", 1536); err != nil {
+		t.Fatalf("missing query embedding lock returned error: %v", err)
+	}
+}
+
+func TestValidateQueryEmbeddingLockChecksExistingLock(t *testing.T) {
+	lock := &EmbeddingLock{
+		BaseURL:           "http://embeddings",
+		Model:             "model-a",
+		SourceDimensions:  384,
+		StorageDimensions: CanonicalEmbeddingDimensions,
+	}
+
+	if err := validateQueryEmbeddingLock(lock, "http://embeddings", "model-a", 1536); err == nil {
+		t.Fatal("dimension mismatch returned nil error")
+	}
+}

internal/scanner/scanner.go

@@ -215,6 +215,13 @@ func NewScanner(fileRepo *FileRepository, ffprobePath string, s3Client *s3client
 	return s
 }
 
+func (s *Scanner) SetSearchIndexProvider(provider string) {
+	if s == nil || s.itemRepo == nil {
+		return
+	}
+	s.itemRepo.WithActiveSearchProvider(provider)
+}
+
 // SetSeriesQueueSyncer installs the optional pending-series root queue synchronizer.
 func (s *Scanner) SetSeriesQueueSyncer(syncer SeriesQueueSyncer) {
 	if s == nil {