diff --git a/cmd/silo/main.go b/cmd/silo/main.go index c1e36143..d58c242d 100644 --- a/cmd/silo/main.go +++ b/cmd/silo/main.go @@ -2242,6 +2242,14 @@ func main() { if compatSearchService != nil { compatSearchService.StartCoverageRefresh(appCtx) compatDeps.CatalogSearchProvider = compatSearchService.Provider() + // Latch the resolved provider for the package-level enqueue + // helpers (idempotent with the API router's latch; this also + // covers modes that wire jellycompat without the router). + activeSearchProvider := catalog.SearchProviderPostgres + if _, ok := compatSearchService.Provider().(*catalog.MeilisearchSearchProvider); ok { + activeSearchProvider = catalog.SearchProviderMeilisearch + } + catalog.SetActiveSearchIndexProvider(activeSearchProvider) } if deps.S3Public != nil { diff --git a/internal/api/router.go b/internal/api/router.go index 1e517320..dd0382e1 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -466,6 +466,9 @@ func NewRouter(deps Dependencies) chi.Router { activeSearchProvider = catalog.SearchProviderMeilisearch } searchIndexEvents.WithActiveProvider(activeSearchProvider) + // Latch the provider for the package-level enqueue helpers used by + // metadata/scanner/etc. so they skip the per-call settings lookup. + catalog.SetActiveSearchIndexProvider(activeSearchProvider) itemRepo.WithSearchIndexEvents(searchIndexEvents) episodeRepo = catalog.NewEpisodeRepository(deps.DB) providerIDRepo = catalog.NewProviderIDRepository(deps.DB) diff --git a/internal/catalog/search_index_repo.go b/internal/catalog/search_index_repo.go index 0a5bec2a..1c7fff28 100644 --- a/internal/catalog/search_index_repo.go +++ b/internal/catalog/search_index_repo.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "strings" + "sync/atomic" "time" "github.com/jackc/pgx/v5" @@ -69,6 +70,25 @@ func (r *SearchIndexEventRepository) disabledByActiveProvider() bool { return r != nil && r.activeProviderKnown && r.activeProvider != SearchProviderMeilisearch } +// globalActiveSearchProvider latches the resolved catalog search provider for +// the process. catalog.search.provider is restart-keyed, so the value cannot +// change while the process runs; latching it lets the package-level +// EnqueueSearchIndex* helpers (used from metadata, scanner, catalogseed, ...) +// decide enqueue-or-skip without re-querying server_settings inside every +// write transaction. Unset (early bootstrap, unit tests) falls back to the +// settings query. +var globalActiveSearchProvider atomic.Value // string + +// SetActiveSearchIndexProvider records the provider resolved at startup. +// Called from server wiring after the catalog search service is constructed. +func SetActiveSearchIndexProvider(provider string) { + provider = normalizeCatalogSearchProvider(provider) + if provider == "" { + provider = SearchProviderPostgres + } + globalActiveSearchProvider.Store(provider) +} + func (r *SearchIndexEventRepository) EnqueueUpsert(ctx context.Context, execer itemExecer, contentID string) error { return r.enqueue(ctx, execer, SearchProviderMeilisearch, SearchIndexEventUpsert, contentID, "") } @@ -145,6 +165,9 @@ func (r *SearchIndexEventRepository) shouldEnqueue(ctx context.Context, execer i if r.activeProviderKnown { return r.activeProvider == SearchProviderMeilisearch, nil } + if provider, ok := globalActiveSearchProvider.Load().(string); ok { + return provider == SearchProviderMeilisearch, nil + } return searchIndexProviderEnabled(ctx, execer) } @@ -319,6 +342,30 @@ func (r *SearchIndexEventRepository) PendingCount(ctx context.Context, provider return count, err } +// DeadLetterCount reports how many outbox events exhausted their retry budget +// and were dead-lettered (MarkFailed sets processed_at with the final error +// once attempts reach searchIndexEventMaxAttempts). Each one is an item whose +// index document stays silently stale until the next rebuild, so the admin +// status surfaces this count. +func (r *SearchIndexEventRepository) DeadLetterCount(ctx context.Context, provider string) (int, error) { + if r == nil || r.pool == nil { + return 0, nil + } + var count int + err := r.pool.QueryRow(ctx, ` + SELECT COUNT(*) + FROM catalog_search_index_events + WHERE provider = $1 + AND processed_at IS NOT NULL + AND attempts >= $2 + AND last_error <> '' + `, provider, searchIndexEventMaxAttempts).Scan(&count) + if isSearchIndexSchemaUnavailable(err) { + return 0, nil + } + return count, err +} + func (r *SearchIndexEventRepository) MarkProcessed(ctx context.Context, ids []int64) error { if r == nil || r.pool == nil || len(ids) == 0 { return nil diff --git a/internal/catalog/search_indexer.go b/internal/catalog/search_indexer.go index f98859b5..bf564f64 100644 --- a/internal/catalog/search_indexer.go +++ b/internal/catalog/search_indexer.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "log/slog" "math" "strings" "time" @@ -39,6 +40,7 @@ type CatalogSearchIndexRebuildStats struct { ActiveIndexUID string `json:"active_index_uid,omitempty"` DocumentCount int `json:"document_count"` VectorDocCount int `json:"vector_document_count"` + RemovedIndexes int `json:"removed_indexes"` } type queuedMeilisearchTask struct { @@ -49,6 +51,21 @@ type queuedMeilisearchTask struct { const meilisearchMaxDocumentPayloadBytes = 80 * 1024 * 1024 +// meilisearchIndexingTimeout bounds a single indexing HTTP call (document +// batches run up to meilisearchMaxDocumentPayloadBytes). It is deliberately +// independent of catalog.search.meilisearch.timeout_ms: that setting protects +// the interactive search hot path and defaults to 800ms, which is far too +// tight to upload a multi-megabyte rebuild batch to a non-loopback +// Meilisearch — and raising it to make indexing work would loosen search +// fallback latency at the same time. +const meilisearchIndexingTimeout = 2 * time.Minute + +// catalogSearchExcludeMangaChaptersSQL excludes per-chapter manga rows from +// catalog search documents and semantic coverage counts; chapters are reached +// through their parent series and would otherwise flood the index. The +// predicate expects media_items to be aliased as `mi`. +const catalogSearchExcludeMangaChaptersSQL = `NOT EXISTS (SELECT 1 FROM manga_chapters mc WHERE mc.chapter_content_id = mi.content_id)` + type CatalogSearchIndexer struct { pool *pgxpool.Pool settingsStore SettingsStore @@ -247,6 +264,14 @@ func (i *CatalogSearchIndexer) Rebuild(ctx context.Context, progress SearchIndex if err != nil { return stats, err } + priorState, err := i.events.GetState(ctx, SearchProviderMeilisearch) + if err != nil { + return stats, err + } + totalDocs, err := countCatalogSearchEligibleDocuments(ctx, i.pool, settings.IndexTypes) + if err != nil { + return stats, err + } buildIndexUID := fmt.Sprintf("%s_rebuild_%d", settings.MeilisearchIndex, time.Now().Unix()) stats.ActiveIndexUID = buildIndexUID @@ -286,9 +311,10 @@ func (i *CatalogSearchIndexer) Rebuild(ctx context.Context, progress SearchIndex docCount: len(batch), vecCount: catalogSearchVectorDocumentCount(batch), }) - reportSearchIndexProgress(progress, 25, fmt.Sprintf("Submitted %d catalog items", stats.DocumentCount+queuedDocumentCount(queuedTasks))) + submitted := stats.DocumentCount + queuedDocumentCount(queuedTasks) + reportSearchIndexProgress(progress, rebuildIndexingPercent(submitted, totalDocs), fmt.Sprintf("Submitted %d of %d catalog items", submitted, totalDocs)) if len(queuedTasks) >= settings.RebuildQueueDepth { - if err := waitNextMeilisearchTask(ctx, client, &queuedTasks, &stats, progress); err != nil { + if err := waitNextMeilisearchTask(ctx, client, &queuedTasks, &stats, progress, totalDocs); err != nil { return stats, err } } @@ -296,7 +322,7 @@ func (i *CatalogSearchIndexer) Rebuild(ctx context.Context, progress SearchIndex lastID = docs[len(docs)-1].ContentID } for len(queuedTasks) > 0 { - if err := waitNextMeilisearchTask(ctx, client, &queuedTasks, &stats, progress); err != nil { + if err := waitNextMeilisearchTask(ctx, client, &queuedTasks, &stats, progress, totalDocs); err != nil { return stats, err } } @@ -308,23 +334,91 @@ func (i *CatalogSearchIndexer) Rebuild(ctx context.Context, progress SearchIndex if vectorCount, err := countCatalogSearchVectorDocuments(ctx, i.pool, settings.IndexTypes, ""); err == nil { stats.VectorDocCount = vectorCount } + // Swap the state pointer BEFORE marking events processed. If the process + // dies between the two, still-pending events simply replay into the new + // active index as idempotent upserts on the next sync. The reverse order + // would mark events processed while the old index is still active, losing + // those changes from the served index until the next rebuild. + if err := i.events.UpdateStateAfterRebuild(ctx, SearchProviderMeilisearch, buildIndexUID, catalogSearchMeilisearchSchemaVersion(settings.Embedder, settings.IndexTypes, settings.SemanticEnabled), docCount, rebuildEventHighWater); err != nil { + return stats, err + } if err := i.events.MarkProcessedThrough(ctx, SearchProviderMeilisearch, rebuildEventHighWater); err != nil { return stats, err } - if err := i.events.UpdateStateAfterRebuild(ctx, SearchProviderMeilisearch, buildIndexUID, catalogSearchMeilisearchSchemaVersion(settings.Embedder, settings.IndexTypes, settings.SemanticEnabled), docCount, rebuildEventHighWater); err != nil { - return stats, err + reportSearchIndexProgress(progress, 95, "Removing superseded catalog search indexes") + removed, err := cleanupSupersededMeilisearchIndexes(ctx, client, settings.MeilisearchIndex, buildIndexUID, priorState.ActiveIndexUID) + stats.RemovedIndexes = removed + if err != nil { + // The new index is already active; a failed cleanup costs disk on the + // Meilisearch instance, not correctness, and the next rebuild retries. + slog.Warn("catalog search: failed to remove superseded meilisearch indexes", "err", err, "removed", removed) } setSearchIndexTaskResult(progress, stats) reportSearchIndexProgress(progress, 100, fmt.Sprintf("Rebuilt catalog search index with %d documents", docCount)) return stats, nil } +// rebuildIndexingPercent maps rebuild document progress onto the 5-90% band of +// the task's progress bar (index creation sits below, finalization above). +func rebuildIndexingPercent(done, total int) float64 { + if total <= 0 { + return 50 + } + if done > total { + done = total + } + return 5 + 85*float64(done)/float64(total) +} + +// cleanupSupersededMeilisearchIndexes deletes indexes this rebuild has made +// unreachable: the previously active index and any `_rebuild_*` +// leftovers from failed or superseded runs. Without it, every rebuild leaks a +// full copy of the catalog on the Meilisearch instance. +func cleanupSupersededMeilisearchIndexes(ctx context.Context, client *meilisearchClient, indexPrefix, activeUID, previousActiveUID string) (int, error) { + uids, err := client.ListIndexUIDs(ctx) + if err != nil { + return 0, err + } + removed := 0 + for _, uid := range staleCatalogSearchIndexUIDs(uids, indexPrefix, activeUID, previousActiveUID) { + task, err := client.DeleteIndex(ctx, uid) + if err != nil { + return removed, err + } + if err := client.WaitTask(ctx, task); err != nil { + return removed, err + } + removed++ + } + return removed, nil +} + +// staleCatalogSearchIndexUIDs selects which index uids a finished rebuild +// should delete: every `_rebuild_` index except the newly active one, +// plus the previously active index (which may predate the rebuild naming +// scheme). Indexes outside the prefix are never touched, so a shared +// Meilisearch instance stays safe. +func staleCatalogSearchIndexUIDs(uids []string, indexPrefix, activeUID, previousActiveUID string) []string { + rebuildPrefix := indexPrefix + "_rebuild_" + var stale []string + for _, uid := range uids { + if uid == "" || uid == activeUID { + continue + } + if strings.HasPrefix(uid, rebuildPrefix) || uid == previousActiveUID { + stale = append(stale, uid) + } + } + return stale +} + func waitNextMeilisearchTask( ctx context.Context, client *meilisearchClient, queue *[]queuedMeilisearchTask, stats *CatalogSearchIndexRebuildStats, progress SearchIndexProgressReporter, + totalDocs int, ) error { if len(*queue) == 0 { return nil @@ -336,7 +430,7 @@ func waitNextMeilisearchTask( } stats.DocumentCount += next.docCount stats.VectorDocCount += next.vecCount - reportSearchIndexProgress(progress, 25, fmt.Sprintf("Indexed %d catalog items", stats.DocumentCount)) + reportSearchIndexProgress(progress, rebuildIndexingPercent(stats.DocumentCount, totalDocs), fmt.Sprintf("Indexed %d of %d catalog items", stats.DocumentCount, totalDocs)) return nil } @@ -409,7 +503,7 @@ func (i *CatalogSearchIndexer) loadClient(ctx context.Context) (CatalogSearchSet if err != nil || !ok { return settings, nil, ok, err } - client, err := newMeilisearchClient(settings.MeilisearchURL, settings.MeilisearchAPIKey, settings.Timeout) + client, err := newMeilisearchClient(settings.MeilisearchURL, settings.MeilisearchAPIKey, meilisearchIndexingTimeout) if err != nil { return settings, nil, false, err } @@ -456,7 +550,7 @@ func catalogSearchMeilisearchSettings(embedder string, semanticEnabled bool) map "tagline", }, "pagination": map[string]any{ - "maxTotalHits": meilisearchDefaultCandidateScanCap, + "maxTotalHits": meilisearchCandidateScanCap, }, } if semanticEnabled { @@ -540,10 +634,7 @@ func (i *CatalogSearchIndexer) LoadDocumentsAfter(ctx context.Context, afterCont typeFilter := normalizeCatalogSearchItemTypes(itemTypes) whereClause := ` WHERE ($1::text = '' OR mi.content_id > $1) - AND NOT EXISTS ( - SELECT 1 FROM manga_chapters mc - WHERE mc.chapter_content_id = mi.content_id - )` + AND ` + catalogSearchExcludeMangaChaptersSQL args := []any{afterContentID, limit} if len(typeFilter) > 0 { whereClause += ` @@ -577,10 +668,7 @@ func (i *CatalogSearchIndexer) LoadDocumentsByIDs(ctx context.Context, contentID typeFilter := normalizeCatalogSearchItemTypes(itemTypes) whereClause := ` WHERE mi.content_id = ANY($1) - AND NOT EXISTS ( - SELECT 1 FROM manga_chapters mc - WHERE mc.chapter_content_id = mi.content_id - )` + AND ` + catalogSearchExcludeMangaChaptersSQL args := []any{contentIDs} if len(typeFilter) > 0 { whereClause += ` @@ -779,7 +867,7 @@ func countCatalogSearchVectorDocuments(ctx context.Context, q coverageQuerier, i SELECT COUNT(*) FROM media_item_embeddings e JOIN media_items mi ON mi.content_id = e.media_item_id - WHERE NOT EXISTS (SELECT 1 FROM manga_chapters mc WHERE mc.chapter_content_id = mi.content_id) + WHERE `+catalogSearchExcludeMangaChaptersSQL+` AND ($1::text[] IS NULL OR mi.type = ANY($1)) AND (mi.status = 'matched' OR mi.type IN ('audiobook','ebook')) AND ($2 = '' OR e.model = $2) @@ -789,6 +877,30 @@ func countCatalogSearchVectorDocuments(ctx context.Context, q coverageQuerier, i return count, nil } +// countCatalogSearchEligibleDocuments counts the items a rebuild will index +// (nil/empty itemTypes => all types). It applies the same predicate as the +// document loaders — NOT the vector-eligibility predicate — so the total is an +// exact denominator for rebuild progress reporting. +func countCatalogSearchEligibleDocuments(ctx context.Context, q coverageQuerier, itemTypes []string) (int, error) { + if q == nil { + return 0, nil + } + var typeArg any + if typeFilter := normalizeCatalogSearchItemTypes(itemTypes); len(typeFilter) > 0 { + typeArg = typeFilter + } + var count int + if err := q.QueryRow(ctx, ` + SELECT COUNT(*) + FROM media_items mi + WHERE `+catalogSearchExcludeMangaChaptersSQL+` + AND ($1::text[] IS NULL OR mi.type = ANY($1)) + `, typeArg).Scan(&count); err != nil { + return 0, fmt.Errorf("count catalog search eligible documents: %w", err) + } + return count, nil +} + func catalogSearchTitleVariants(doc catalogSearchDocument) []string { return compactNonEmptyStrings([]string{ doc.Title, diff --git a/internal/catalog/search_indexer_test.go b/internal/catalog/search_indexer_test.go index 6b0a7d88..4519e6f1 100644 --- a/internal/catalog/search_indexer_test.go +++ b/internal/catalog/search_indexer_test.go @@ -4,12 +4,127 @@ import ( "context" "fmt" "os" + "slices" + "sort" "testing" "time" "github.com/jackc/pgx/v5/pgxpool" ) +func TestCoalesceSearchIndexEvents(t *testing.T) { + sortedCopy := func(values []string) []string { + out := append([]string(nil), values...) + sort.Strings(out) + return out + } + cases := []struct { + name string + events []SearchIndexEvent + wantUpserts []string + wantDeletes []string + }{ + { + name: "later delete wins over earlier upsert", + events: []SearchIndexEvent{ + {Action: SearchIndexEventUpsert, ContentID: "x"}, + {Action: SearchIndexEventDelete, ContentID: "x"}, + }, + wantDeletes: []string{"x"}, + }, + { + name: "later upsert resurrects earlier delete", + events: []SearchIndexEvent{ + {Action: SearchIndexEventDelete, ContentID: "x"}, + {Action: SearchIndexEventUpsert, ContentID: "x"}, + }, + wantUpserts: []string{"x"}, + }, + { + name: "rename deletes previous id and upserts new id", + events: []SearchIndexEvent{ + {Action: SearchIndexEventRename, ContentID: "new", PreviousContentID: "old"}, + }, + wantUpserts: []string{"new"}, + wantDeletes: []string{"old"}, + }, + { + name: "delete after rename removes the renamed id", + events: []SearchIndexEvent{ + {Action: SearchIndexEventRename, ContentID: "new", PreviousContentID: "old"}, + {Action: SearchIndexEventDelete, ContentID: "new"}, + }, + wantDeletes: []string{"new", "old"}, + }, + { + name: "upsert after rename resurrects the previous id", + events: []SearchIndexEvent{ + {Action: SearchIndexEventRename, ContentID: "new", PreviousContentID: "old"}, + {Action: SearchIndexEventUpsert, ContentID: "old"}, + }, + wantUpserts: []string{"new", "old"}, + }, + { + name: "blank ids are dropped", + events: []SearchIndexEvent{ + {Action: SearchIndexEventUpsert, ContentID: " "}, + {Action: SearchIndexEventRename, ContentID: "new", PreviousContentID: ""}, + }, + wantUpserts: []string{"new"}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + upserts, deletes := coalesceSearchIndexEvents(tc.events) + if got, want := sortedCopy(upserts), sortedCopy(tc.wantUpserts); !slices.Equal(got, want) { + t.Fatalf("upserts = %v, want %v", got, want) + } + if got, want := sortedCopy(deletes), sortedCopy(tc.wantDeletes); !slices.Equal(got, want) { + t.Fatalf("deletes = %v, want %v", got, want) + } + }) + } +} + +func TestStaleCatalogSearchIndexUIDs(t *testing.T) { + uids := []string{ + "silo_media_items_rebuild_100", // superseded rebuild + "silo_media_items_rebuild_200", // newly active + "silo_media_items", // legacy previously-active index + "other_app_index", // unrelated index on a shared instance + "silo_media_items_rebuild_50", // failed-run leftover + "", + } + stale := staleCatalogSearchIndexUIDs(uids, "silo_media_items", "silo_media_items_rebuild_200", "silo_media_items") + sort.Strings(stale) + want := []string{"silo_media_items", "silo_media_items_rebuild_100", "silo_media_items_rebuild_50"} + if !slices.Equal(stale, want) { + t.Fatalf("stale = %v, want %v", stale, want) + } + + // The newly active index must survive even when it is also the previous + // active uid (re-running a rebuild that already swapped). + if got := staleCatalogSearchIndexUIDs([]string{"idx_rebuild_1"}, "idx", "idx_rebuild_1", "idx_rebuild_1"); len(got) != 0 { + t.Fatalf("active index must never be deleted, got %v", got) + } +} + +func TestRebuildIndexingPercent(t *testing.T) { + if got := rebuildIndexingPercent(0, 0); got != 50 { + t.Fatalf("unknown total should report midpoint, got %v", got) + } + if got := rebuildIndexingPercent(0, 10); got != 5 { + t.Fatalf("start of band = %v, want 5", got) + } + if got := rebuildIndexingPercent(10, 10); got != 90 { + t.Fatalf("end of band = %v, want 90", got) + } + // Documents created after the total was counted must not push past the band. + if got := rebuildIndexingPercent(15, 10); got != 90 { + t.Fatalf("overshoot should clamp to 90, got %v", got) + } +} + func TestAttachDocumentVectorsSkipsWhenSemanticDisabled(t *testing.T) { docs := []catalogSearchDocument{{ContentID: "movie-1", Title: "Movie"}} indexer := &CatalogSearchIndexer{pool: new(pgxpool.Pool)} diff --git a/internal/catalog/search_meilisearch_client.go b/internal/catalog/search_meilisearch_client.go index 14b91cab..2af1c4a6 100644 --- a/internal/catalog/search_meilisearch_client.go +++ b/internal/catalog/search_meilisearch_client.go @@ -228,6 +228,52 @@ func (c *meilisearchClient) DeleteDocuments(ctx context.Context, uid string, ids return newMeilisearchTaskRef(task.TaskUID), nil } +// DeleteIndex removes an index. A missing index is not an error — cleanup of +// superseded rebuild indexes must be idempotent across crashes and retries. +func (c *meilisearchClient) DeleteIndex(ctx context.Context, uid string) (meilisearchTaskRef, error) { + var task meilisearchTask + err := c.do(ctx, http.MethodDelete, "/indexes/"+url.PathEscape(uid), nil, &task) + if err != nil { + var httpErr *meilisearchHTTPError + if errors.As(err, &httpErr) && (httpErr.StatusCode == http.StatusNotFound || httpErr.Code == "index_not_found") { + return meilisearchTaskRef{}, nil + } + return meilisearchTaskRef{}, err + } + return newMeilisearchTaskRef(task.TaskUID), nil +} + +type meilisearchIndexListResponse struct { + Results []struct { + UID string `json:"uid"` + } `json:"results"` + Offset int `json:"offset"` + Limit int `json:"limit"` + Total int `json:"total"` +} + +// ListIndexUIDs pages through GET /indexes and returns every index uid on the +// instance. Meilisearch caps the page size, so this loops until the reported +// total is reached (or a page comes back short/empty). +func (c *meilisearchClient) ListIndexUIDs(ctx context.Context) ([]string, error) { + const pageLimit = 100 + var uids []string + for offset := 0; ; { + var out meilisearchIndexListResponse + endpoint := fmt.Sprintf("/indexes?limit=%d&offset=%d", pageLimit, offset) + if err := c.do(ctx, http.MethodGet, endpoint, nil, &out); err != nil { + return nil, err + } + for _, result := range out.Results { + uids = append(uids, result.UID) + } + offset += len(out.Results) + if len(out.Results) == 0 || offset >= out.Total { + return uids, nil + } + } +} + func (c *meilisearchClient) Stats(ctx context.Context, uid string) (int, error) { var out meilisearchStatsResponse err := c.do(ctx, http.MethodGet, "/indexes/"+url.PathEscape(uid)+"/stats", nil, &out) @@ -273,8 +319,13 @@ func (c *meilisearchClient) do(ctx context.Context, method, endpoint string, bod return fmt.Errorf("meilisearch client is not configured") } reqURL := *c.baseURL - reqURL.Path = path.Join(c.baseURL.Path, endpoint) - if strings.HasSuffix(endpoint, "/") && !strings.HasSuffix(reqURL.Path, "/") { + endpointPath := endpoint + if idx := strings.IndexByte(endpoint, '?'); idx >= 0 { + endpointPath = endpoint[:idx] + reqURL.RawQuery = endpoint[idx+1:] + } + reqURL.Path = path.Join(c.baseURL.Path, endpointPath) + if strings.HasSuffix(endpointPath, "/") && !strings.HasSuffix(reqURL.Path, "/") { reqURL.Path += "/" } diff --git a/internal/catalog/search_meilisearch_client_test.go b/internal/catalog/search_meilisearch_client_test.go index 1c6b693d..97299552 100644 --- a/internal/catalog/search_meilisearch_client_test.go +++ b/internal/catalog/search_meilisearch_client_test.go @@ -2,8 +2,11 @@ package catalog import ( "context" + "errors" + "fmt" "net/http" "net/http/httptest" + "strings" "testing" "time" ) @@ -34,3 +37,146 @@ func TestMeilisearchWaitTaskNoopsWithoutTask(t *testing.T) { t.Fatalf("WaitTask(no task) returned error: %v", err) } } + +func TestMeilisearchWaitTaskSurfacesTaskFailure(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"taskUid":7,"status":"failed","error":{"message":"document is malformed","code":"invalid_document"}}`)) + })) + defer server.Close() + + client, err := newMeilisearchClient(server.URL, "", time.Second) + if err != nil { + t.Fatalf("newMeilisearchClient: %v", err) + } + err = client.WaitTask(context.Background(), newMeilisearchTaskRef(7)) + if err == nil || !strings.Contains(err.Error(), "document is malformed") { + t.Fatalf("WaitTask error = %v, want task failure with meilisearch message", err) + } +} + +func TestMeilisearchClientJoinsBasePathAndSendsAuth(t *testing.T) { + var gotPath, gotAuth string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + gotAuth = r.Header.Get("Authorization") + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"status":"available"}`)) + })) + defer server.Close() + + client, err := newMeilisearchClient(server.URL+"/meili", "secret-key", time.Second) + if err != nil { + t.Fatalf("newMeilisearchClient: %v", err) + } + if err := client.Health(context.Background()); err != nil { + t.Fatalf("Health: %v", err) + } + if gotPath != "/meili/health" { + t.Fatalf("request path = %q, want /meili/health", gotPath) + } + if gotAuth != "Bearer secret-key" { + t.Fatalf("Authorization = %q, want Bearer secret-key", gotAuth) + } +} + +func TestMeilisearchClientDecodesJSONErrorBody(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"message":"invalid filter expression","code":"invalid_search_filter"}`)) + })) + defer server.Close() + + client, err := newMeilisearchClient(server.URL, "", time.Second) + if err != nil { + t.Fatalf("newMeilisearchClient: %v", err) + } + _, err = client.Search(context.Background(), "idx", meilisearchSearchRequest{}) + var httpErr *meilisearchHTTPError + if !errors.As(err, &httpErr) { + t.Fatalf("Search error = %v, want *meilisearchHTTPError", err) + } + if httpErr.StatusCode != http.StatusBadRequest || httpErr.Message != "invalid filter expression" || httpErr.Code != "invalid_search_filter" { + t.Fatalf("decoded error = %+v, want status 400 with message and code", httpErr) + } +} + +func TestMeilisearchClientKeepsNonJSONErrorBody(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte(" upstream proxy exploded\n")) + })) + defer server.Close() + + client, err := newMeilisearchClient(server.URL, "", time.Second) + if err != nil { + t.Fatalf("newMeilisearchClient: %v", err) + } + _, err = client.Stats(context.Background(), "idx") + var httpErr *meilisearchHTTPError + if !errors.As(err, &httpErr) { + t.Fatalf("Stats error = %v, want *meilisearchHTTPError", err) + } + if httpErr.StatusCode != http.StatusInternalServerError || httpErr.Message != "upstream proxy exploded" { + t.Fatalf("decoded error = %+v, want trimmed plain-text message", httpErr) + } +} + +func TestMeilisearchDeleteIndexToleratesMissingIndex(t *testing.T) { + var gotMethod, gotPath string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotMethod = r.Method + gotPath = r.URL.Path + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`{"message":"Index gone not found.","code":"index_not_found"}`)) + })) + defer server.Close() + + client, err := newMeilisearchClient(server.URL, "", time.Second) + if err != nil { + t.Fatalf("newMeilisearchClient: %v", err) + } + ref, err := client.DeleteIndex(context.Background(), "gone") + if err != nil { + t.Fatalf("DeleteIndex of a missing index should be a no-op, got %v", err) + } + if ref.hasTask { + t.Fatalf("missing index should not yield a task, got %+v", ref) + } + if gotMethod != http.MethodDelete || gotPath != "/indexes/gone" { + t.Fatalf("unexpected request %s %s, want DELETE /indexes/gone", gotMethod, gotPath) + } +} + +func TestMeilisearchListIndexUIDsPaginates(t *testing.T) { + var offsets []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + offset := r.URL.Query().Get("offset") + offsets = append(offsets, offset) + w.Header().Set("Content-Type", "application/json") + switch offset { + case "0": + _, _ = w.Write([]byte(`{"results":[{"uid":"a"},{"uid":"b"}],"offset":0,"limit":2,"total":3}`)) + default: + _, _ = fmt.Fprintf(w, `{"results":[{"uid":"c"}],"offset":%s,"limit":2,"total":3}`, offset) + } + })) + defer server.Close() + + client, err := newMeilisearchClient(server.URL, "", time.Second) + if err != nil { + t.Fatalf("newMeilisearchClient: %v", err) + } + uids, err := client.ListIndexUIDs(context.Background()) + if err != nil { + t.Fatalf("ListIndexUIDs: %v", err) + } + if len(uids) != 3 || uids[0] != "a" || uids[1] != "b" || uids[2] != "c" { + t.Fatalf("uids = %v, want [a b c]", uids) + } + if len(offsets) != 2 || offsets[0] != "0" || offsets[1] != "2" { + t.Fatalf("request offsets = %v, want [0 2]", offsets) + } +} diff --git a/internal/catalog/search_meilisearch_provider.go b/internal/catalog/search_meilisearch_provider.go index 046e40d3..06a3598c 100644 --- a/internal/catalog/search_meilisearch_provider.go +++ b/internal/catalog/search_meilisearch_provider.go @@ -16,15 +16,22 @@ import ( ) const ( - meilisearchDefaultBatchSize = 100 - meilisearchDefaultCandidateScanCap = 1000 - meilisearchDefaultDeepOffsetLimit = 500 + meilisearchSearchBatchSize = 100 + meilisearchCandidateScanCap = 1000 + meilisearchDeepOffsetLimit = 500 meilisearchShortHybridMinTerms = 2 meilisearchStrictMatchingTermCount = 2 meilisearchTitleOnlyTermCount = 2 meilisearchCircuitCooldown = 30 * time.Second meilisearchQueryVectorCacheTTL = 15 * time.Minute meilisearchQueryVectorCacheMax = 1024 + // meilisearchIndexStateCacheTTL bounds how long the provider serves the + // cached catalog_search_index_state row (and the informational pending + // count) before refetching. The state only changes on rebuild/sync, so this + // keeps two Postgres round trips off every search request; a failed search + // invalidates the cache immediately so a swapped-away index is repaired on + // the next request rather than after the TTL. + meilisearchIndexStateCacheTTL = 3 * time.Second // semanticCapabilityProbeTTL rate-limits the embedder capability check // (settings fetch + hybrid probe) so a healthy index is validated at most // once per window. The probe is advisory only and never trips the circuit. @@ -49,10 +56,6 @@ type MeilisearchProviderConfig struct { Index string Timeout time.Duration MatchingStrategy string - BatchSize int - CandidateScanCap int - DeepOffsetLimit int - CircuitCooldown time.Duration IndexTypes []string SemanticEnabled bool SemanticRatio float64 @@ -72,8 +75,20 @@ type MeilisearchSearchProvider struct { unhealthyUntil time.Time unhealthyReason string lastFallback string - vectorCache map[string]cachedCatalogSearchQueryVector - vectorCacheSeq int64 + + // vecMu guards the query-vector cache. It is separate from mu so cache + // reads/writes on the semantic path never contend with the circuit-breaker + // state that mu protects. + vecMu sync.Mutex + vectorCache map[string]cachedCatalogSearchQueryVector + vectorCacheSeq int64 + + // stateMu guards the cached index state + pending count (see + // meilisearchIndexStateCacheTTL). + stateMu sync.Mutex + cachedState SearchIndexState + cachedPending int + stateCachedAt time.Time // capMu guards the rate-limited semantic capability cache. It is separate // from mu so a capability probe can never contend with or mutate the @@ -107,18 +122,6 @@ func NewMeilisearchSearchProvider( if config.MatchingStrategy == "" { config.MatchingStrategy = DefaultMeilisearchMatchingStrategy } - if config.BatchSize <= 0 { - config.BatchSize = meilisearchDefaultBatchSize - } - if config.CandidateScanCap <= 0 { - config.CandidateScanCap = meilisearchDefaultCandidateScanCap - } - if config.DeepOffsetLimit <= 0 { - config.DeepOffsetLimit = meilisearchDefaultDeepOffsetLimit - } - if config.CircuitCooldown <= 0 { - config.CircuitCooldown = meilisearchCircuitCooldown - } config.IndexTypes = normalizeCatalogSearchItemTypes(config.IndexTypes) if config.SemanticRatio < 0 || config.SemanticRatio > 1 { config.SemanticRatio = DefaultMeilisearchSemanticRatio @@ -159,10 +162,10 @@ func (p *MeilisearchSearchProvider) Search(ctx context.Context, req CatalogSearc if !p.indexCoversRequest(req.ItemTypes) { return p.fallbackSearch(ctx, req, "meilisearch index does not cover requested media scope") } - if req.Offset > p.config.DeepOffsetLimit { + if req.Offset > meilisearchDeepOffsetLimit { return p.fallbackSearch(ctx, req, "deep offset exceeds meilisearch scan policy") } - state, err := p.stateRepo.GetState(ctx, SearchProviderMeilisearch) + state, pending, err := p.indexState(ctx) if err != nil { p.markFallback("index state unavailable") return p.fallback.Search(ctx, req) @@ -173,13 +176,13 @@ func (p *MeilisearchSearchProvider) Search(ctx context.Context, req CatalogSearc if state.SchemaVersion != catalogSearchMeilisearchSchemaVersion(p.config.Embedder, p.config.IndexTypes, p.config.SemanticEnabled) { return p.fallbackSearch(ctx, req, "meilisearch index schema mismatch") } - pending := 0 - if count, err := p.stateRepo.PendingCount(ctx, SearchProviderMeilisearch); err == nil && count > 0 { - pending = count - } result, err := p.searchMeilisearch(ctx, req, state.ActiveIndexUID) if err != nil { + // The cached state may point at an index a rebuild just swapped away + // and deleted; drop it so the next request refetches instead of + // failing for the rest of the TTL. + p.invalidateIndexState() if p.shouldTripCircuit(err) { p.tripCircuit(err) } @@ -214,18 +217,51 @@ func (p *MeilisearchSearchProvider) indexCoversRequest(itemTypes []string) bool return true } +// indexState returns the active index state and pending outbox depth, cached +// for meilisearchIndexStateCacheTTL so the search hot path is not charged two +// Postgres round trips per request. Errors are never cached — a failed refresh +// falls through to the caller and the next request retries immediately. +func (p *MeilisearchSearchProvider) indexState(ctx context.Context) (SearchIndexState, int, error) { + p.stateMu.Lock() + if !p.stateCachedAt.IsZero() && time.Since(p.stateCachedAt) < meilisearchIndexStateCacheTTL { + state, pending := p.cachedState, p.cachedPending + p.stateMu.Unlock() + return state, pending, nil + } + p.stateMu.Unlock() + + state, err := p.stateRepo.GetState(ctx, SearchProviderMeilisearch) + if err != nil { + return SearchIndexState{}, 0, err + } + pending := 0 + if count, err := p.stateRepo.PendingCount(ctx, SearchProviderMeilisearch); err == nil && count > 0 { + pending = count + } + + p.stateMu.Lock() + p.cachedState = state + p.cachedPending = pending + p.stateCachedAt = time.Now() + p.stateMu.Unlock() + return state, pending, nil +} + +func (p *MeilisearchSearchProvider) invalidateIndexState() { + p.stateMu.Lock() + p.stateCachedAt = time.Time{} + p.stateMu.Unlock() +} + func (p *MeilisearchSearchProvider) searchMeilisearch(ctx context.Context, req CatalogSearchRequest, indexUID string) (*CatalogSearchResult, error) { target := req.Offset + req.Limit + 1 if target <= 0 { target = 1 } - batchSize := p.config.BatchSize + batchSize := meilisearchSearchBatchSize if batchSize < req.Limit+1 { batchSize = req.Limit + 1 } - if batchSize <= 0 { - batchSize = meilisearchDefaultBatchSize - } var accessible []*models.MediaItem meiliOffset := 0 @@ -235,11 +271,11 @@ func (p *MeilisearchSearchProvider) searchMeilisearch(ctx context.Context, req C baseSearchReq, semanticFallback := p.buildMeilisearchSearchRequest(ctx, req) for len(accessible) < target && !exhausted { - if scanned >= p.config.CandidateScanCap { + if scanned >= meilisearchCandidateScanCap { return nil, fmt.Errorf("meilisearch candidate scan cap reached") } nextLimit := batchSize - if remaining := p.config.CandidateScanCap - scanned; remaining < nextLimit { + if remaining := meilisearchCandidateScanCap - scanned; remaining < nextLimit { nextLimit = remaining } searchReq := baseSearchReq @@ -285,7 +321,10 @@ func (p *MeilisearchSearchProvider) searchMeilisearch(ctx context.Context, req C } accessible = append(accessible, orderItemsByIDPosition(hydrated, position)...) - if meiliOffset >= estimatedTotalHits || len(resp.Hits) < nextLimit { + // A short page is the only reliable end-of-results signal. + // estimatedTotalHits is an estimate and may undercount; treating it as + // authoritative could stop pagination early and drop real results. + if len(resp.Hits) < nextLimit { exhausted = true } } @@ -308,7 +347,6 @@ func (p *MeilisearchSearchProvider) searchMeilisearch(ctx context.Context, req C } else if estimatedTotalHits == 0 { total = 0 } - p.markFallback(semanticFallback) // Derive Mode/SemanticUsed from the POST-downgrade request: the hybrid // downgrade above nils baseSearchReq.Hybrid on error, so a hybrid request // that fell back to keyword correctly reports keyword / semantic_used=false. @@ -416,13 +454,13 @@ func (p *MeilisearchSearchProvider) cachedQueryVector(ctx context.Context, query cacheKey := strings.ToLower(normalized) now := time.Now() - p.mu.Lock() + p.vecMu.Lock() if cached, ok := p.vectorCache[cacheKey]; ok && now.Before(cached.expiresAt) { vector := cloneFloat32Slice(cached.vector) - p.mu.Unlock() + p.vecMu.Unlock() return vector, nil } - p.mu.Unlock() + p.vecMu.Unlock() vector, err := p.config.Vectorizer.EmbedSearchQuery(ctx, normalized) if err != nil { @@ -433,8 +471,8 @@ func (p *MeilisearchSearchProvider) cachedQueryVector(ctx context.Context, query return nil, err } - p.mu.Lock() - defer p.mu.Unlock() + p.vecMu.Lock() + defer p.vecMu.Unlock() if p.vectorCache == nil { p.vectorCache = make(map[string]cachedCatalogSearchQueryVector) } @@ -496,7 +534,7 @@ func (p *MeilisearchSearchProvider) circuitBlocked(now time.Time) (string, bool) func (p *MeilisearchSearchProvider) tripCircuit(cause error) { p.mu.Lock() defer p.mu.Unlock() - p.unhealthyUntil = time.Now().Add(p.config.CircuitCooldown) + p.unhealthyUntil = time.Now().Add(meilisearchCircuitCooldown) p.unhealthyReason = cause.Error() p.lastFallback = cause.Error() } @@ -534,10 +572,7 @@ func (p *MeilisearchSearchProvider) shouldTripCircuit(err error) bool { httpErr.StatusCode >= http.StatusInternalServerError } var decodeErr *meilisearchDecodeError - if errors.As(err, &decodeErr) { - return true - } - return false + return errors.As(err, &decodeErr) } func (p *MeilisearchSearchProvider) Status() CatalogSearchMeiliStatus { diff --git a/internal/catalog/search_provider.go b/internal/catalog/search_provider.go index deab4071..5a79f7c6 100644 --- a/internal/catalog/search_provider.go +++ b/internal/catalog/search_provider.go @@ -404,15 +404,19 @@ type CatalogSearchMeiliStatus struct { } type CatalogSearchIndexStateStatus struct { - ActiveIndexUID string `json:"active_index_uid"` - SchemaVersion int `json:"schema_version"` - ExpectedSchemaVersion int `json:"expected_schema_version"` - DocumentCount int `json:"document_count"` - VectorDocumentCount int `json:"vector_document_count"` - PendingEvents int `json:"pending_events"` - LastRebuildAt *time.Time `json:"last_rebuild_at,omitempty"` - LastSyncAt *time.Time `json:"last_sync_at,omitempty"` - LastProcessedEventID int64 `json:"last_processed_event_id"` + ActiveIndexUID string `json:"active_index_uid"` + SchemaVersion int `json:"schema_version"` + ExpectedSchemaVersion int `json:"expected_schema_version"` + DocumentCount int `json:"document_count"` + VectorDocumentCount int `json:"vector_document_count"` + PendingEvents int `json:"pending_events"` + // DeadLetteredEvents counts outbox events that exhausted their retries and + // were dropped; each is an item whose index document is stale until the + // next rebuild. + DeadLetteredEvents int `json:"dead_lettered_events"` + LastRebuildAt *time.Time `json:"last_rebuild_at,omitempty"` + LastSyncAt *time.Time `json:"last_sync_at,omitempty"` + LastProcessedEventID int64 `json:"last_processed_event_id"` } type CatalogSearchTaskLink struct { diff --git a/internal/catalog/search_provider_test.go b/internal/catalog/search_provider_test.go index 88cfcdd7..612892bd 100644 --- a/internal/catalog/search_provider_test.go +++ b/internal/catalog/search_provider_test.go @@ -481,6 +481,111 @@ func (f fakeMeilisearchIndexStateStore) PendingCount(context.Context, string) (i return f.pending, nil } +type countingMeilisearchIndexStateStore struct { + state SearchIndexState + pending int + getStateCalls int + pendingCalls int +} + +func (f *countingMeilisearchIndexStateStore) GetState(context.Context, string) (SearchIndexState, error) { + f.getStateCalls++ + return f.state, nil +} + +func (f *countingMeilisearchIndexStateStore) PendingCount(context.Context, string) (int, error) { + f.pendingCalls++ + return f.pending, nil +} + +func TestMeilisearchProviderCachesIndexStateAcrossRequests(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"hits":[],"estimatedTotalHits":0}`)) + })) + defer server.Close() + + client, err := newMeilisearchClient(server.URL, "", time.Second) + if err != nil { + t.Fatalf("newMeilisearchClient: %v", err) + } + store := &countingMeilisearchIndexStateStore{ + state: SearchIndexState{ + ActiveIndexUID: "search-index", + SchemaVersion: catalogSearchMeilisearchSchemaVersion(DefaultMeilisearchEmbedder, nil, false), + }, + pending: 3, + } + provider := &MeilisearchSearchProvider{ + stateRepo: store, + fallback: &PostgresSearchProvider{}, + client: client, + config: MeilisearchProviderConfig{ + MatchingStrategy: DefaultMeilisearchMatchingStrategy, + Embedder: DefaultMeilisearchEmbedder, + }, + } + + for i := 0; i < 3; i++ { + result, err := provider.Search(context.Background(), CatalogSearchRequest{Query: "sponge", Limit: 10}) + if err != nil { + t.Fatalf("Search %d returned error: %v", i, err) + } + if result.IndexPendingEvents != 3 { + t.Fatalf("Search %d IndexPendingEvents = %d, want cached 3", i, result.IndexPendingEvents) + } + } + if store.getStateCalls != 1 || store.pendingCalls != 1 { + t.Fatalf("state store calls = %d/%d, want 1/1 (state cached within TTL)", store.getStateCalls, store.pendingCalls) + } +} + +func TestMeilisearchProviderInvalidatesStateCacheOnSearchFailure(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + // 404 models the cached index having been deleted by a rebuild's + // cleanup; it must not trip the circuit, only the state cache. + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`{"message":"Index search-index not found.","code":"index_not_found"}`)) + })) + defer server.Close() + + client, err := newMeilisearchClient(server.URL, "", time.Second) + if err != nil { + t.Fatalf("newMeilisearchClient: %v", err) + } + store := &countingMeilisearchIndexStateStore{ + state: SearchIndexState{ + ActiveIndexUID: "search-index", + SchemaVersion: catalogSearchMeilisearchSchemaVersion(DefaultMeilisearchEmbedder, nil, false), + }, + } + provider := &MeilisearchSearchProvider{ + stateRepo: store, + fallback: &PostgresSearchProvider{}, + client: client, + config: MeilisearchProviderConfig{ + MatchingStrategy: DefaultMeilisearchMatchingStrategy, + Embedder: DefaultMeilisearchEmbedder, + }, + } + + // Both searches fail (the nil-repo postgres fallback errors too); what + // matters is that the failed first search dropped the cached state so the + // second one refetched instead of reusing it for the rest of the TTL. + for i := 0; i < 2; i++ { + if _, err := provider.Search(context.Background(), CatalogSearchRequest{Query: "sponge", Limit: 10}); err == nil { + t.Fatalf("Search %d should surface the fallback error in this setup", i) + } + } + if store.getStateCalls != 2 { + t.Fatalf("getStateCalls = %d, want 2 (cache invalidated after failed search)", store.getStateCalls) + } + if reason, blocked := provider.circuitBlocked(time.Now()); blocked { + t.Fatalf("HTTP 404 must not trip the circuit, got open circuit: %s", reason) + } +} + func TestMeilisearchProviderUsesActiveIndexWhenPendingUpdatesExist(t *testing.T) { requests := 0 var gotMethod, gotPath string @@ -509,9 +614,6 @@ func TestMeilisearchProviderUsesActiveIndexWhenPendingUpdatesExist(t *testing.T) fallback: &PostgresSearchProvider{}, client: client, config: MeilisearchProviderConfig{ - BatchSize: meilisearchDefaultBatchSize, - CandidateScanCap: meilisearchDefaultCandidateScanCap, - DeepOffsetLimit: meilisearchDefaultDeepOffsetLimit, MatchingStrategy: DefaultMeilisearchMatchingStrategy, Embedder: DefaultMeilisearchEmbedder, }, diff --git a/internal/catalog/search_service.go b/internal/catalog/search_service.go index 9bb39d54..8eed7be9 100644 --- a/internal/catalog/search_service.go +++ b/internal/catalog/search_service.go @@ -175,6 +175,9 @@ func (s *CatalogSearchService) Status(ctx context.Context) CatalogSearchRuntimeS if pending, err := s.state.PendingCount(ctx, SearchProviderMeilisearch); err == nil { status.Index.PendingEvents = pending } + if deadLettered, err := s.state.DeadLetterCount(ctx, SearchProviderMeilisearch); err == nil { + status.Index.DeadLetteredEvents = deadLettered + } } if s.itemRepo != nil && s.itemRepo.pool != nil { if vectorCount, err := countCatalogSearchVectorDocuments(ctx, s.itemRepo.pool, settings.IndexTypes, ""); err == nil { diff --git a/internal/catalog/semantic_coverage.go b/internal/catalog/semantic_coverage.go index 4b32d5eb..09683e85 100644 --- a/internal/catalog/semantic_coverage.go +++ b/internal/catalog/semantic_coverage.go @@ -58,7 +58,7 @@ type catalogTypeCoverage struct { const semanticCoverageEligibleByTypeSQL = ` SELECT mi.type, COUNT(*) AS eligible FROM media_items mi -WHERE NOT EXISTS (SELECT 1 FROM manga_chapters mc WHERE mc.chapter_content_id = mi.content_id) +WHERE ` + catalogSearchExcludeMangaChaptersSQL + ` AND ($1::text[] IS NULL OR mi.type = ANY($1)) AND (mi.status = 'matched' OR mi.type IN ('audiobook','ebook')) GROUP BY mi.type` @@ -72,7 +72,7 @@ const semanticCoverageVectorizedByTypeSQL = ` SELECT mi.type, COUNT(*) AS vectorized FROM media_item_embeddings e JOIN media_items mi ON mi.content_id = e.media_item_id -WHERE NOT EXISTS (SELECT 1 FROM manga_chapters mc WHERE mc.chapter_content_id = mi.content_id) +WHERE ` + catalogSearchExcludeMangaChaptersSQL + ` AND ($1::text[] IS NULL OR mi.type = ANY($1)) AND (mi.status = 'matched' OR mi.type IN ('audiobook','ebook')) AND ($2 = '' OR e.model = $2) diff --git a/web/src/hooks/queries/admin/settings.ts b/web/src/hooks/queries/admin/settings.ts index 4199ab33..b185a329 100644 --- a/web/src/hooks/queries/admin/settings.ts +++ b/web/src/hooks/queries/admin/settings.ts @@ -42,6 +42,7 @@ export interface CatalogSearchStatus { document_count: number; vector_document_count: number; pending_events: number; + dead_lettered_events: number; last_rebuild_at?: string; last_sync_at?: string; last_processed_event_id: number; diff --git a/web/src/pages/admin-settings/SearchSettings.tsx b/web/src/pages/admin-settings/SearchSettings.tsx index 505c4f2b..452ddf2b 100644 --- a/web/src/pages/admin-settings/SearchSettings.tsx +++ b/web/src/pages/admin-settings/SearchSettings.tsx @@ -289,6 +289,13 @@ export default function SearchSettings() { )} + {status.index.dead_lettered_events > 0 && ( + + )}