From ed6c084c68080f09bb2243ec4fcea4e1fb83dca5 Mon Sep 17 00:00:00 2001 From: Quick <31828688+Quick104@users.noreply.github.com> Date: Fri, 26 Jun 2026 08:34:04 -0400 Subject: [PATCH] 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 --- cmd/silo/main.go | 9 ++- internal/api/handlers/items.go | 14 ++-- .../handlers/items_catalog_response_test.go | 17 +++++ internal/catalog/item_repo.go | 14 ++++ internal/catalog/search_index_repo.go | 64 +++++++++++++++++-- internal/catalog/search_index_repo_test.go | 11 ++++ .../catalog/search_meilisearch_client_test.go | 17 +++++ .../recommendations/search_query_vector.go | 20 +++++- .../search_query_vector_test.go | 22 +++++++ internal/scanner/scanner.go | 7 ++ 10 files changed, 180 insertions(+), 15 deletions(-) create mode 100644 internal/catalog/search_meilisearch_client_test.go create mode 100644 internal/recommendations/search_query_vector_test.go diff --git a/cmd/silo/main.go b/cmd/silo/main.go index fe8cf9d9..873360d4 100644 --- a/cmd/silo/main.go +++ b/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) diff --git a/internal/api/handlers/items.go b/internal/api/handlers/items.go index 2fb0c4eb..75a4d5c3 100644 --- a/internal/api/handlers/items.go +++ b/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 } diff --git a/internal/api/handlers/items_catalog_response_test.go b/internal/api/handlers/items_catalog_response_test.go index 737393c6..7430476b 100644 --- a/internal/api/handlers/items_catalog_response_test.go +++ b/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 diff --git a/internal/catalog/item_repo.go b/internal/catalog/item_repo.go index e460fe45..739c901d 100644 --- a/internal/catalog/item_repo.go +++ b/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) diff --git a/internal/catalog/search_index_repo.go b/internal/catalog/search_index_repo.go index c99039d5..0a5bec2a 100644 --- a/internal/catalog/search_index_repo.go +++ b/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 } diff --git a/internal/catalog/search_index_repo_test.go b/internal/catalog/search_index_repo_test.go index 71be926a..c0662192 100644 --- a/internal/catalog/search_index_repo_test.go +++ b/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") + } +} diff --git a/internal/catalog/search_meilisearch_client_test.go b/internal/catalog/search_meilisearch_client_test.go new file mode 100644 index 00000000..c5c4cfab --- /dev/null +++ b/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) + } +} diff --git a/internal/recommendations/search_query_vector.go b/internal/recommendations/search_query_vector.go index 72cb299a..10859cb0 100644 --- a/internal/recommendations/search_query_vector.go +++ b/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) +} diff --git a/internal/recommendations/search_query_vector_test.go b/internal/recommendations/search_query_vector_test.go new file mode 100644 index 00000000..57ad6210 --- /dev/null +++ b/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") + } +} diff --git a/internal/scanner/scanner.go b/internal/scanner/scanner.go index 23dbae6b..612706b9 100644 --- a/internal/scanner/scanner.go +++ b/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 {