diff --git a/internal/api/metadata.go b/internal/api/metadata.go index 46b3bd8..ee83d8d 100644 --- a/internal/api/metadata.go +++ b/internal/api/metadata.go @@ -395,7 +395,18 @@ func (s *Server) handleSearchMetadata(w http.ResponseWriter, r *http.Request) { return } - // Get all hash IDs + // Free-text query: served from the in-memory search index (search_index.go) + // so we match in memory instead of doing a SCAN+MGET Redis round-trip per + // record (that was 40s+ over ~72k records). Full metadata is fetched only + // for the <=limit matches. + if req.Query != "" && req.Property == "" { + s.serveIndexedSearch(w, req) + return + } + + // Property filter (rare) or empty query (returns the first `limit`): the + // legacy scan. Empty query only fetches `limit` records; property search is + // low-volume and not on the hot search path. allHashIDs, err := s.storage.GetAllHashIDs() if err != nil { writeError(w, http.StatusInternalServerError, err.Error()) @@ -405,9 +416,6 @@ func (s *Server) handleSearchMetadata(w http.ResponseWriter, r *http.Request) { var results []MetadataSearchResult total := len(allHashIDs) - // Search fields for query - searchFields := []string{"title", "fileName", "originaltitle", "showtitle", "filePath"} - for _, hashID := range allHashIDs { if len(results) >= req.Limit { break @@ -428,28 +436,6 @@ func (s *Server) handleSearchMetadata(w http.ResponseWriter, r *http.Request) { } } - // General query search - if req.Query != "" && req.Property == "" { - queryLower := strings.ToLower(req.Query) - - // Check if query matches hashId - if strings.Contains(strings.ToLower(hashID), queryLower) { - matches = true - } - - // Check search fields - if !matches { - for _, field := range searchFields { - if value, ok := metadata[field]; ok { - if strings.Contains(strings.ToLower(value), queryLower) { - matches = true - break - } - } - } - } - } - // If no query or property filter, return all if req.Query == "" && req.Property == "" { matches = true diff --git a/internal/api/search_index.go b/internal/api/search_index.go new file mode 100644 index 0000000..8b26a4b --- /dev/null +++ b/internal/api/search_index.go @@ -0,0 +1,110 @@ +package api + +import ( + "log" + "net/http" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/metazla/meta-core/internal/storage" +) + +// searchIndexRefreshInterval is how often the in-memory search index is rebuilt +// from Redis in the background. Records ingest continuously, so a newly stored +// record becomes searchable within one interval; the rebuild is a single +// batched bulk read (storage.GetSearchIndexTuples), off the request path. +const searchIndexRefreshInterval = 60 * time.Second + +// searchIndex is an in-memory snapshot of every record's search haystack +// (hashID + lowercased searchable fields), rebuilt periodically from Redis. It +// exists so POST /api/metadata/search matches in memory (string ops over a +// slice) instead of doing a SCAN+MGET Redis round-trip per record — the latter +// took 40s+ over ~72k records and made a live parallel search path unusable. +type searchIndex struct { + mu sync.RWMutex + tuples []storage.SearchTuple + ready atomic.Bool +} + +// snapshot returns the current tuples slice and whether the index has built at +// least once. The refresh swaps in a fresh slice under the write lock (never +// mutates in place), so the returned slice is safe to read lock-free. +func (idx *searchIndex) snapshot() ([]storage.SearchTuple, bool) { + idx.mu.RLock() + defer idx.mu.RUnlock() + return idx.tuples, idx.ready.Load() +} + +// refresh rebuilds the index from Redis once (a batched bulk read). +func (idx *searchIndex) refresh(stor *storage.Client) { + if stor == nil || !stor.IsConnected() { + return + } + tuples, err := stor.GetSearchIndexTuples() + if err != nil { + log.Printf("[API] search index refresh failed: %v", err) + return + } + idx.mu.Lock() + idx.tuples = tuples + idx.mu.Unlock() + idx.ready.Store(true) +} + +// runRefreshLoop builds the index immediately, then rebuilds on a ticker for the +// process lifetime. +func (idx *searchIndex) runRefreshLoop(stor *storage.Client) { + idx.refresh(stor) + ticker := time.NewTicker(searchIndexRefreshInterval) + defer ticker.Stop() + for range ticker.C { + idx.refresh(stor) + } +} + +// serveIndexedSearch answers a free-text (non-property) query from the in-memory +// index: token-AND match against each record's haystack (or a whole-query hashID +// substring, preserving the legacy behaviour), then fetch full metadata for only +// the <=limit matches. No Redis round-trip per scanned record. +func (s *Server) serveIndexedSearch(w http.ResponseWriter, req MetadataSearchRequest) { + tuples, ready := s.search.snapshot() + if !ready { + // First call before the background build finished: warm inline with the + // efficient batched read (seconds), never the per-record SCAN fallback. + s.search.refresh(s.storage) + tuples, _ = s.search.snapshot() + } + tokens := strings.Fields(strings.ToLower(req.Query)) + queryLower := strings.ToLower(req.Query) + results := make([]MetadataSearchResult, 0, req.Limit) + for _, t := range tuples { + if len(results) >= req.Limit { + break + } + if !matchAllTokens(t.Haystack, tokens) && + !strings.Contains(strings.ToLower(t.HashID), queryLower) { + continue + } + // Full metadata is cached in the index — return it straight from memory, + // no Redis round-trip per result. + results = append(results, MetadataSearchResult{HashID: t.HashID, Metadata: t.Fields}) + } + writeJSON(w, http.StatusOK, MetadataSearchResponse{ + Results: results, + Count: len(results), + Total: len(tuples), + }) +} + +// matchAllTokens reports whether haystack contains every token (AND semantics). +// An empty token list matches everything. +func matchAllTokens(haystack string, tokens []string) bool { + for _, t := range tokens { + if !strings.Contains(haystack, t) { + return false + } + } + return true +} diff --git a/internal/api/server.go b/internal/api/server.go index 53ef89d..fa7f638 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -41,6 +41,7 @@ type Server struct { metaPublisher *events.MetaPublisher schemaIndexer *schema.Indexer webdavHandler *webdav.Handler + search *searchIndex router *mux.Router server *http.Server } @@ -59,6 +60,7 @@ func NewServer( discovery: disc, cleaner: cleaner, storage: stor, + search: &searchIndex{}, router: mux.NewRouter(), } @@ -314,6 +316,13 @@ func (s *Server) Start() error { s.mountStatsPoller.Start() } + // Build + periodically refresh the in-memory search index so + // POST /api/metadata/search matches in memory instead of a Redis + // round-trip per record (see search_index.go). + if s.storage != nil { + go s.search.runRefreshLoop(s.storage) + } + // Start meta publisher (if storage is connected) // Publishes keyspace notifications to meta:events stream if s.storage != nil && s.storage.IsConnected() { diff --git a/internal/storage/client.go b/internal/storage/client.go index f8953fd..127fd83 100644 --- a/internal/storage/client.go +++ b/internal/storage/client.go @@ -683,6 +683,115 @@ func (c *Client) GetTuplesForAllFiles() ([]FileTuple, error) { return out, nil } +// SearchTuple is one record's search state: its hashID, a lowercased haystack of +// its searchable text fields (for token/substring match), and the full flat +// metadata map (so a search hit is returned straight from memory — no per-result +// Redis fetch). +type SearchTuple struct { + HashID string + Haystack string + Fields map[string]string +} + +// searchIndexFields are the metadata fields folded into a record's search +// haystack — must stay in sync with the fields the search handler documents. +var searchIndexFields = []string{"title", "fileName", "originaltitle", "showtitle", "filePath"} + +// GetSearchIndexTuples reads EVERY record's full metadata in a single keyspace +// SCAN pass over the flat `file:{hashID}/{field}` keys plus batched MGETs, +// grouped by record — one bulk pass instead of a SCAN+MGET per record (which is +// O(records) Redis round-trips, ~40s over 72k field keys). It is the bulk source +// the API folds into its in-memory search index, so per-request search does zero +// Redis round-trips (match AND result metadata both come from memory). +func (c *Client) GetSearchIndexTuples() ([]SearchTuple, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return nil, fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) + defer cancel() + + filePrefix := c.buildKey("file:") // key layout: {filePrefix}{hashID}/{field} + indexKey := c.buildIndexKey() // {filePrefix}__index__ (a SET, skip it) + pattern := filePrefix + "*" + + records := make(map[string]map[string]string) + keyBatch := make([]string, 0, 1000) + + flush := func() error { + if len(keyBatch) == 0 { + return nil + } + values, err := c.client.MGet(ctx, keyBatch...).Result() + if err != nil { + return fmt.Errorf("mget: %w", err) + } + for i, k := range keyBatch { + v, ok := values[i].(string) + if !ok { + continue + } + // {filePrefix}{hashID}/{field}; hashID has no '/', field may. + rest := strings.TrimPrefix(k, filePrefix) + slash := strings.IndexByte(rest, '/') + if slash < 0 { + continue + } + hashID, field := rest[:slash], rest[slash+1:] + m := records[hashID] + if m == nil { + m = make(map[string]string) + records[hashID] = m + } + m[field] = v + } + keyBatch = keyBatch[:0] + return nil + } + + var cursor uint64 + for { + keys, next, err := c.client.Scan(ctx, cursor, pattern, 1000).Result() + if err != nil { + return nil, fmt.Errorf("scan: %w", err) + } + for _, k := range keys { + if k == indexKey { + continue + } + keyBatch = append(keyBatch, k) + if len(keyBatch) >= 1000 { + if err := flush(); err != nil { + return nil, err + } + } + } + cursor = next + if cursor == 0 { + break + } + } + if err := flush(); err != nil { + return nil, err + } + + out := make([]SearchTuple, 0, len(records)) + for hashID, fields := range records { + var sb strings.Builder + for _, f := range searchIndexFields { + if v, ok := fields[f]; ok && v != "" { + sb.WriteString(strings.ToLower(v)) + sb.WriteByte('\n') + } + } + out = append(out, SearchTuple{HashID: hashID, Haystack: sb.String(), Fields: fields}) + } + return out, nil +} + // GetMemoryInfo returns Redis memory usage information func (c *Client) GetMemoryInfo() (string, error) { c.mu.RLock()