improved cid resolution

README.md

@@ -287,6 +287,10 @@ return a deprecation pointer at the `/api/watchers/*` routes
 
 ```bash
 POST   /api/admin/migrate-dual-roots  # reunify stranded midhash-rooted entries
+POST   /api/admin/sweep-literature-echoes[?apply=true]
+       # strip query echoes off literature cards: anilistid on site cards,
+       # workcid/chapterNumber/volumeNumber/chapterStart/chapterEnd on any card.
+       # Dry-run unless apply=true; returns {dryRun, planned, fixed, byKey}
 GET/PUT/DELETE /webdav/...            # mounted on /files
 ```
 

docs/meta-core-architecture.md

@@ -339,6 +339,9 @@ Watchers
 
 Admin
   POST /api/admin/migrate-dual-roots
+  POST /api/admin/sweep-literature-echoes   (dry-run unless ?apply=true;
+       strips query echoes off domain=literature fileType=card records —
+       see storage/literature_echo_sweep.go)
 
 WebDAV
   /webdav/...   (GET, PUT, DELETE, MKCOL, COPY, MOVE, PROPFIND)

internal/api/admin_migrate.go

@@ -2,6 +2,7 @@ package api
 
 import (
 	"net/http"
+	"strconv"
 )
 
 // handleMigrateDualRoots triggers the one-shot sweep that re-unifies
@@ -28,3 +29,73 @@ func (s *Server) handleMigrateDualRoots(w http.ResponseWriter, r *http.Request)
 		"fixed":   fixed,
 	})
 }
+
+// handleMigrateDomainScreen triggers the one-shot sweep that rewrites the
+// retired `domain=film|tv` values to `screen` and backfills `workForm` from
+// `contentKind` (METADATA_KEYS.md §14.17). See
+// storage.MigrateDomainScreen for why the vocabulary change is paid in the
+// data rather than with a read-side alias.
+//
+// Returns {fixed: N} — records changed, not records examined. Safe to call
+// repeatedly: a record already on the current vocabulary is skipped.
+func (s *Server) handleMigrateDomainScreen(w http.ResponseWriter, r *http.Request) {
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+
+	fixed, err := s.storage.MigrateDomainScreenWithTimeout()
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusOK, map[string]interface{}{
+		"success": true,
+		"fixed":   fixed,
+	})
+}
+
+// handleSweepLiteratureEchoes removes query echoes stored on literature cards:
+// `anilistid` on a site card, and `workcid` / position keys on any card
+// (meta-read docs/reading-model.md §7). See storage.SweepLiteratureEchoes for
+// why chapters and anchor cards' ids are deliberately left alone.
+//
+// Dry-run unless `?apply=true`: unlike the migrations above this one deletes
+// keys, so a bare call only reports what it would delete. An unparseable
+// `apply` is a 400 rather than a silent dry run, so a typo cannot pass for a
+// sweep that ran.
+//
+// Returns {success, dryRun, planned, fixed, byKey} — planned/fixed count
+// records, byKey counts fields per key. Safe to call repeatedly: a swept
+// record plans nothing on the next pass.
+func (s *Server) handleSweepLiteratureEchoes(w http.ResponseWriter, r *http.Request) {
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+
+	apply := false
+	if v := r.URL.Query().Get("apply"); v != "" {
+		parsed, err := strconv.ParseBool(v)
+		if err != nil {
+			writeError(w, http.StatusBadRequest, "apply must be true or false")
+			return
+		}
+		apply = parsed
+	}
+
+	res, err := s.storage.SweepLiteratureEchoesWithTimeout(apply)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusOK, map[string]interface{}{
+		"success": true,
+		"dryRun":  !apply,
+		"planned": res.Planned,
+		"fixed":   res.Fixed,
+		"byKey":   res.ByKey,
+	})
+}

internal/api/admin_sweep_test.go

@@ -0,0 +1,93 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/alicebob/miniredis/v2"
+
+	"github.com/metazla/meta-core/internal/storage"
+)
+
+// newSweepTestServer is a Server over miniredis with one literature card
+// carrying a copied `workcid`.
+func newSweepTestServer(t *testing.T) *Server {
+	t.Helper()
+	mr := miniredis.RunT(t)
+	c := storage.NewClient("")
+	if err := c.Connect("redis://" + mr.Addr()); err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	t.Cleanup(func() { _ = c.Close() })
+	for k, v := range map[string]string{
+		"domain":   "literature",
+		"fileType": "card",
+		"workcid":  "bagdsaaaua5qw42lmnfzxi3lbnztwcorrga2tonzy",
+	} {
+		if err := c.SetProperty("card-1", k, v); err != nil {
+			t.Fatalf("seed %s: %v", k, err)
+		}
+	}
+	return &Server{storage: c}
+}
+
+type sweepResponse struct {
+	Success bool           `json:"success"`
+	DryRun  bool           `json:"dryRun"`
+	Planned int            `json:"planned"`
+	Fixed   int            `json:"fixed"`
+	ByKey   map[string]int `json:"byKey"`
+}
+
+func postSweep(t *testing.T, s *Server, query string) (*httptest.ResponseRecorder, sweepResponse) {
+	t.Helper()
+	rr := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodPost, "/api/admin/sweep-literature-echoes"+query, nil)
+	s.handleSweepLiteratureEchoes(rr, req)
+	var body sweepResponse
+	if rr.Code == http.StatusOK {
+		if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil {
+			t.Fatalf("decode %s: %v", rr.Body.String(), err)
+		}
+	}
+	return rr, body
+}
+
+// A bare call must never delete: the default is the dry run.
+func TestSweepLiteratureEchoes_DefaultsToDryRun(t *testing.T) {
+	s := newSweepTestServer(t)
+
+	rr, body := postSweep(t, s, "")
+	if rr.Code != http.StatusOK {
+		t.Fatalf("status = %d, body %s", rr.Code, rr.Body.String())
+	}
+	if !body.Success || !body.DryRun || body.Planned != 1 || body.Fixed != 0 || body.ByKey["workcid"] != 1 {
+		t.Errorf("dry run = %+v", body)
+	}
+	if v, _ := s.storage.GetProperty("card-1", "workcid"); v == "" {
+		t.Errorf("dry run deleted workcid")
+	}
+
+	rr, body = postSweep(t, s, "?apply=true")
+	if rr.Code != http.StatusOK {
+		t.Fatalf("status = %d, body %s", rr.Code, rr.Body.String())
+	}
+	if body.DryRun || body.Planned != 1 || body.Fixed != 1 || body.ByKey["workcid"] != 1 {
+		t.Errorf("apply = %+v", body)
+	}
+	if v, _ := s.storage.GetProperty("card-1", "workcid"); v != "" {
+		t.Errorf("apply left workcid = %q", v)
+	}
+}
+
+// A typo must not pass for a sweep that ran.
+func TestSweepLiteratureEchoes_RejectsUnparseableApply(t *testing.T) {
+	s := newSweepTestServer(t)
+
+	rr, _ := postSweep(t, s, "?apply=yes-please")
+	if rr.Code != http.StatusBadRequest {
+		t.Errorf("status = %d, want 400", rr.Code)
+	}
+}

internal/api/files_register.go

@@ -0,0 +1,132 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+	"os"
+	"path"
+	"path/filepath"
+	"strings"
+
+	"github.com/metazla/meta-core/internal/watcher"
+)
+
+// registerFileRequest is the body of POST /api/files/register.
+type registerFileRequest struct {
+	// Path relative to FILES_PATH, e.g. "plugin/subtitles/ab12….eng.srt".
+	Path string `json:"path"`
+}
+
+type registerFileResponse struct {
+	Root    string `json:"root"`
+	CID     string `json:"cid"`
+	Created bool   `json:"created"`
+	Path    string `json:"path"`
+	Size    int64  `json:"size"`
+}
+
+// pluginRelPath normalises a FILES_PATH-relative path and refuses anything
+// outside plugin/. Plugin output is the only tree a service may register this
+// way: /files/watch is the user's library and belongs to the watcher.
+func pluginRelPath(raw string) (string, bool) {
+	p := strings.TrimPrefix(filepath.ToSlash(strings.TrimSpace(raw)), "/")
+	if p == "" {
+		return "", false
+	}
+	for _, seg := range strings.Split(p, "/") {
+		if seg == ".." {
+			return "", false
+		}
+	}
+	clean := path.Clean(p)
+	if !strings.HasPrefix(clean, "plugin/") || clean == "plugin/" {
+		return "", false
+	}
+	return clean, true
+}
+
+// handleRegisterFile handles POST /api/files/register.
+//
+// Does for ONE file what the watcher does for every file it scans: midhash256
+// it, mint a root carrying filePath/sizeByte/mtimeNano, register the
+// cid:<midhash> alias, and append a file:events entry so meta-share's
+// ipfs_seed chunks, seeds and provides it.
+//
+// It exists for cores that run with ENABLE_FILE_WATCHER=false (a meta-watch
+// client box). Without it, bytes written under /files/plugin over WebDAV are
+// never resolvable by CID and never reach the swarm. Idempotent: the same
+// content at the same path returns the existing root with created=false.
+func (s *Server) handleRegisterFile(w http.ResponseWriter, r *http.Request) {
+	if s.storage == nil || !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+
+	var req registerFileRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid JSON body")
+		return
+	}
+	rel, ok := pluginRelPath(req.Path)
+	if !ok {
+		writeError(w, http.StatusBadRequest, "path must be a file under plugin/")
+		return
+	}
+
+	fullPath := filepath.Join(s.config.FilesPath, filepath.FromSlash(rel))
+	absFiles, _ := filepath.Abs(filepath.Join(s.config.FilesPath, "plugin"))
+	absFull, _ := filepath.Abs(fullPath)
+	if !strings.HasPrefix(absFull, absFiles+string(filepath.Separator)) {
+		writeError(w, http.StatusBadRequest, "path must be a file under plugin/")
+		return
+	}
+
+	info, err := os.Stat(fullPath)
+	if err != nil {
+		writeError(w, http.StatusNotFound, "file not found")
+		return
+	}
+	if info.IsDir() {
+		writeError(w, http.StatusBadRequest, "path is a directory, not a file")
+		return
+	}
+
+	midhash, err := watcher.ComputeMidHash256(fullPath, info.Size())
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, "failed to hash file")
+		return
+	}
+
+	root, created, err := s.storage.RegisterFileTuple(fullPath, info.Size(), info.ModTime().UnixNano(), midhash)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	if created {
+		ev := watcher.FileEvent{
+			Type:       watcher.EventTypeAdd,
+			Path:       rel,
+			Size:       info.Size(),
+			MidHash256: midhash,
+			Timestamp:  watcher.NowMS(),
+		}
+		if err := watcher.PublishEvent(s.storage, ev); err != nil {
+			// The record is written and resolvable; only seeding is late.
+			writeError(w, http.StatusInternalServerError, "registered, but file event not published: "+err.Error())
+			return
+		}
+	}
+
+	status := http.StatusOK
+	if created {
+		status = http.StatusCreated
+	}
+	writeJSON(w, status, registerFileResponse{
+		Root:    root,
+		CID:     midhash,
+		Created: created,
+		Path:    rel,
+		Size:    info.Size(),
+	})
+}

internal/api/files_register_test.go

@@ -0,0 +1,144 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/metazla/meta-core/internal/config"
+	"github.com/metazla/meta-core/internal/storage"
+)
+
+// newRegisterTestServer wires handleRegisterFile to an in-memory Redis and a
+// temp FILES_PATH. Built directly, not through NewServer, so no watcher runs —
+// which is exactly the configuration this route exists for.
+func newRegisterTestServer(t *testing.T) (*Server, string, *miniredis.Miniredis) {
+	t.Helper()
+	mr := miniredis.RunT(t)
+	c := storage.NewClient("")
+	if err := c.Connect("redis://" + mr.Addr()); err != nil {
+		t.Fatalf("connect to miniredis: %v", err)
+	}
+	t.Cleanup(func() { _ = c.Close() })
+	files := t.TempDir()
+	return &Server{storage: c, config: &config.Config{FilesPath: files}}, files, mr
+}
+
+func writeFixture(t *testing.T, files, rel, body string) string {
+	t.Helper()
+	full := filepath.Join(files, filepath.FromSlash(rel))
+	if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(full, []byte(body), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	return full
+}
+
+func postRegister(t *testing.T, s *Server, path string) (int, registerFileResponse) {
+	t.Helper()
+	body, _ := json.Marshal(registerFileRequest{Path: path})
+	req := httptest.NewRequest(http.MethodPost, "/api/files/register", strings.NewReader(string(body)))
+	rr := httptest.NewRecorder()
+	s.handleRegisterFile(rr, req)
+	var resp registerFileResponse
+	_ = json.Unmarshal(rr.Body.Bytes(), &resp)
+	return rr.Code, resp
+}
+
+func streamLen(t *testing.T, mr *miniredis.Miniredis) int {
+	t.Helper()
+	entries, err := mr.Stream("file:events")
+	if err != nil {
+		return 0
+	}
+	return len(entries)
+}
+
+// The whole point of the route: with the watcher off, a file written under
+// plugin/ becomes a record resolvable by its midhash, and meta-share hears
+// about it on file:events.
+func TestRegisterFile_MintsAliasAndPublishesEvent(t *testing.T) {
+	s, files, mr := newRegisterTestServer(t)
+	rel := "plugin/subtitles/ab12cd34.eng.srt"
+	full := writeFixture(t, files, rel, "1\n00:00:01,000 --> 00:00:02,000\nHello\n")
+
+	code, resp := postRegister(t, s, rel)
+	if code != http.StatusCreated || !resp.Created {
+		t.Fatalf("want 201 created, got %d %+v", code, resp)
+	}
+	if !strings.HasPrefix(resp.CID, "b") || resp.Root == "" {
+		t.Fatalf("want a multibase cid and a root, got %+v", resp)
+	}
+
+	root, err := s.storage.GetByCID(resp.CID)
+	if err != nil || root != resp.Root {
+		t.Fatalf("cid alias must resolve to the root: got %q err=%v, want %q", root, err, resp.Root)
+	}
+	if fp, _ := s.storage.GetProperty(resp.Root, "filePath"); fp != full {
+		t.Fatalf("filePath = %q, want %q", fp, full)
+	}
+
+	entries, err := mr.Stream("file:events")
+	if err != nil || len(entries) != 1 {
+		t.Fatalf("want exactly one file:events entry, got %d err=%v", len(entries), err)
+	}
+	vals := strings.Join(entries[0].Values, "|")
+	if !strings.Contains(vals, "midhash256|"+resp.CID) || !strings.Contains(vals, "path|"+rel) {
+		t.Fatalf("event must carry the midhash and the relative path, got %q", vals)
+	}
+}
+
+// Re-uploading the same subtitle is a no-op: same root, no second event (which
+// would make meta-share re-chunk and re-provide the file).
+func TestRegisterFile_IsIdempotent(t *testing.T) {
+	s, files, mr := newRegisterTestServer(t)
+	rel := "plugin/subtitles/same.eng.srt"
+	writeFixture(t, files, rel, "same bytes")
+
+	_, first := postRegister(t, s, rel)
+	code, second := postRegister(t, s, rel)
+	if code != http.StatusOK || second.Created {
+		t.Fatalf("second register must be 200 created=false, got %d %+v", code, second)
+	}
+	if second.Root != first.Root || second.CID != first.CID {
+		t.Fatalf("second register must return the same record: %+v vs %+v", second, first)
+	}
+	if n := streamLen(t, mr); n != 1 {
+		t.Fatalf("an idempotent re-register must not publish again, stream has %d", n)
+	}
+}
+
+// Only plugin output may be registered here. /files/watch is the user's
+// library and belongs to the watcher; traversal must not escape either.
+func TestRegisterFile_RejectsPathsOutsidePlugin(t *testing.T) {
+	s, files, _ := newRegisterTestServer(t)
+	writeFixture(t, files, "watch/movie.mkv", "x")
+	for _, p := range []string{
+		"", "plugin", "plugin/", "watch/movie.mkv", "/watch/movie.mkv",
+		"plugin/../watch/movie.mkv", "../etc/passwd", "plugin/../../etc/passwd",
+	} {
+		if code, _ := postRegister(t, s, p); code != http.StatusBadRequest {
+			t.Errorf("path %q: want 400, got %d", p, code)
+		}
+	}
+}
+
+func TestRegisterFile_MissingFileAndDirectory(t *testing.T) {
+	s, files, _ := newRegisterTestServer(t)
+	if code, _ := postRegister(t, s, "plugin/subtitles/nope.srt"); code != http.StatusNotFound {
+		t.Errorf("missing file: want 404, got %d", code)
+	}
+	if err := os.MkdirAll(filepath.Join(files, "plugin", "subtitles", "dir"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if code, _ := postRegister(t, s, "plugin/subtitles/dir"); code != http.StatusBadRequest {
+		t.Errorf("directory: want 400, got %d", code)
+	}
+}

internal/api/metadata.go

@@ -4,7 +4,6 @@ import (
 	"encoding/json"
 	"net/http"
 	"strconv"
-	"strings"
 
 	"github.com/gorilla/mux"
 )
@@ -39,6 +38,36 @@ type MetadataSearchRequest struct {
 	Property      string `json:"property,omitempty"`
 	PropertyValue string `json:"propertyValue,omitempty"`
 	Limit         int    `json:"limit,omitempty"`
+
+	// Filters is the structured form: AND across entries, OR within an
+	// entry's Values. Evaluated in memory against the search index alongside
+	// the free-text Query, so an id-anchored lookup costs a map probe per
+	// record rather than a Redis round-trip per record.
+	//
+	// Property/PropertyValue is the single-filter legacy shape and folds into
+	// this list (see normalizedFilters) with Exact=false, preserving its
+	// substring semantics byte-for-byte.
+	Filters []MetadataFilter `json:"filters,omitempty"`
+}
+
+// MetadataFilter is one structured constraint on a record.
+//
+// A record satisfies it when ANY of Values matches, and a search result must
+// satisfy EVERY filter (OR within, AND across). An entry with no usable value
+// constrains nothing — same rule the gateway's `genres_filter_matches` uses, so
+// a caller that lowers an empty filter doesn't accidentally empty the result.
+type MetadataFilter struct {
+	// Property is the field name, WITHOUT a key-set suffix: `genres`, not
+	// `genres/Action`. Both storage shapes are probed — see recordMatchesFilter.
+	Property string `json:"property"`
+	// Values are OR'd. Compared case-insensitively.
+	Values []string `json:"values,omitempty"`
+	// Exact selects CSV-exact comparison (the semantics meta-gateway's
+	// `any_csv_value_matches` applies at its own Layer-B re-filter) over
+	// substring containment. Callers doing identity lookups want this: with a
+	// result limit, a substring match on `tmdbid=1405` also admits `21405` and
+	// can crowd the real matches out of the page.
+	Exact bool `json:"exact,omitempty"`
 }
 
 // MetadataSearchResult is a single search result
@@ -415,65 +444,18 @@ func (s *Server) handleSearchMetadata(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// 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 == "" {
+	// Everything else — free text, structured filters, both, or neither
+	// (which returns the first `limit`) — is served from the in-memory search
+	// index (search_index.go). Matching and the returned metadata both come
+	// from memory, so a query costs zero Redis round-trips.
+	//
+	// There is deliberately no second path. The property branch used to fall
+	// through to a `GetAllHashIDs` + per-record `GetMetadataFlat` scan, which
+	// is what made an id-anchored lookup 3.63s over ~20k records (vs 10ms for
+	// the free-text branch on the same corpus) and kept meta-core out of the
+	// gateway's anchored search path entirely. The index already holds every
+	// field of every record, so that scan was re-reading data it had in hand.
 	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())
-		return
-	}
-
-	var results []MetadataSearchResult
-	total := len(allHashIDs)
-
-	for _, hashID := range allHashIDs {
-		if len(results) >= req.Limit {
-			break
-		}
-
-		metadata, err := s.storage.GetMetadataFlat(hashID)
-		if err != nil || metadata == nil {
-			continue
-		}
-
-		matches := false
-
-		// Property search
-		if req.Property != "" && req.PropertyValue != "" {
-			value, ok := metadata[req.Property]
-			if ok && strings.Contains(strings.ToLower(value), strings.ToLower(req.PropertyValue)) {
-				matches = true
-			}
-		}
-
-		// If no query or property filter, return all
-		if req.Query == "" && req.Property == "" {
-			matches = true
-		}
-
-		if matches {
-			results = append(results, MetadataSearchResult{
-				HashID:   hashID,
-				Metadata: metadata,
-			})
-		}
-	}
-
-	writeJSON(w, http.StatusOK, MetadataSearchResponse{
-		Results: results,
-		Count:   len(results),
-		Total:   total,
-	})
 }
 
 // handleBatchUpdate handles POST /api/metadata/batch

internal/api/search_filter_test.go

@@ -0,0 +1,211 @@
+package api
+
+import "testing"
+
+// Tests for the structured-filter matcher that lets meta-core answer an
+// id-anchored query from the in-memory search index (search_index.go).
+//
+// Why this matters beyond "filters work": meta-core is a *coarse pre-filter*
+// for meta-gateway, which re-evaluates every record it emits through its own
+// `query_eval::record_matches`. So the property that has to hold is that this
+// matcher stays a SUPERSET of that one — it may admit records the gateway will
+// drop, but it must never drop a record the gateway would have kept. Every
+// case below is written from that angle:
+//
+//   - key-set fields (`genres/Action`) must match a `genres:action` filter, or
+//     card-tier records vanish from a genre-scoped query;
+//   - a missing field fails CLOSED, matching record_matches;
+//   - an empty/unusable filter constrains NOTHING, so a caller that lowers a
+//     blank filter gets its results rather than an empty page.
+
+func f(property string, exact bool, values ...string) MetadataFilter {
+	return MetadataFilter{Property: property, Values: values, Exact: exact}
+}
+
+func TestRecordMatchesFilter(t *testing.T) {
+	fields := map[string]string{
+		"fileType":               "video",
+		"videoType":              "tv,anime",
+		"tmdbid":                 "1405",
+		"genres/Action":          "true",
+		"genres/Science Fiction": "true",
+		"source/gateway:nyaa.si": "true",
+		"categories/XXX":         "true",
+		"retracted":              "false",
+	}
+
+	cases := []struct {
+		name   string
+		filter MetadataFilter
+		want   bool
+	}{
+		// Flat scalar.
+		{"flat exact hit", f("fileType", true, "video"), true},
+		{"flat exact miss", f("fileType", true, "audio"), false},
+		{"flat exact is case-insensitive", f("fileType", true, "VIDEO"), true},
+
+		// csv-set field: each comma piece is compared independently.
+		{"csv member hit", f("videoType", true, "anime"), true},
+		{"csv member miss", f("videoType", true, "movie"), false},
+
+		// Exact vs contains — the reason Exact exists. A substring match on an
+		// identity field admits neighbours and, under a result limit, crowds
+		// the real matches off the page.
+		{"contains admits a superstring", f("tmdbid", false, "140"), true},
+		{"exact rejects a superstring", f("tmdbid", true, "140"), false},
+		{"exact accepts the whole value", f("tmdbid", true, "1405"), true},
+
+		// Key-set shape. The stored member is verbatim (`genres/Action`) while
+		// query values arrive lowercased, so the case-insensitive sweep is what
+		// makes a genre-scoped row work at all.
+		{"key-set exact casing", f("genres", true, "Action"), true},
+		{"key-set lowercased query", f("genres", true, "action"), true},
+		{"key-set with a space", f("genres", true, "science fiction"), true},
+		{"key-set miss", f("genres", true, "horror"), false},
+		{"key-set label containing a colon", f("source", true, "gateway:nyaa.si"), true},
+
+		// OR within one filter's values.
+		{"or hits on the second value", f("fileType", true, "audio", "video"), true},
+		{"or misses on all values", f("fileType", true, "audio", "image"), false},
+
+		// Fail-closed on an absent field — mirrors record_matches, and is what
+		// keeps a filtered query from returning unrelated records.
+		{"absent field fails closed", f("imdbid", true, "tt0903747"), false},
+
+		// A filter that constrains nothing must not empty the result.
+		{"no values constrains nothing", f("fileType", true), true},
+		{"blank value constrains nothing", f("fileType", true, "  "), true},
+		{"no property constrains nothing", f("", true, "video"), true},
+
+		// A FLAT field whose value happens to be "false" is still a value, and
+		// an exact filter for it matches. Only the key-set arm treats "false"
+		// as non-membership (pinned in the test below).
+		{"flat field valued false still matches", f("retracted", true, "false"), true},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := recordMatchesFilter(fields, &tc.filter); got != tc.want {
+				t.Fatalf("recordMatchesFilter(%+v) = %v, want %v", tc.filter, got, tc.want)
+			}
+		})
+	}
+}
+
+// `retracted = "false"` is a flat field, not a key-set member, so the flat arm
+// legitimately matches the literal string "false". The key-set arm is what must
+// treat it as non-membership — pinned separately so the two arms don't get
+// conflated if either is refactored.
+func TestFalseyKeySetMemberIsNotMembership(t *testing.T) {
+	fields := map[string]string{"genres/Action": "false", "genres/Drama": "0"}
+	for _, v := range []string{"Action", "Drama"} {
+		filter := f("genres", true, v)
+		if recordMatchesFilter(fields, &filter) {
+			t.Fatalf("genres/%s with a falsey value must not assert membership", v)
+		}
+	}
+}
+
+func TestRecordMatchesFiltersIsAndAcrossEntries(t *testing.T) {
+	fields := map[string]string{"fileType": "video", "season": "2"}
+
+	all := []MetadataFilter{f("fileType", true, "video"), f("season", true, "2")}
+	if !recordMatchesFilters(fields, all) {
+		t.Fatal("a record satisfying every filter must match")
+	}
+
+	one := []MetadataFilter{f("fileType", true, "video"), f("season", true, "3")}
+	if recordMatchesFilters(fields, one) {
+		t.Fatal("filters are AND across entries — one miss must reject the record")
+	}
+
+	if !recordMatchesFilters(fields, nil) {
+		t.Fatal("no filters must match everything")
+	}
+}
+
+// The legacy Property/PropertyValue pair keeps its substring semantics: it is
+// the shape existing callers already use, and tightening it silently would
+// change results for them. New callers pass Filters and opt into Exact.
+func TestNormalizedFiltersFoldsTheLegacyPair(t *testing.T) {
+	got := normalizedFilters(MetadataSearchRequest{Property: "tmdbid", PropertyValue: "1405"})
+	if len(got) != 1 {
+		t.Fatalf("legacy pair must fold into exactly one filter, got %d", len(got))
+	}
+	if got[0].Property != "tmdbid" || len(got[0].Values) != 1 || got[0].Values[0] != "1405" {
+		t.Fatalf("legacy pair folded incorrectly: %+v", got[0])
+	}
+	if got[0].Exact {
+		t.Fatal("the legacy pair must stay substring (Exact=false) for compatibility")
+	}
+
+	// A half-specified pair is not a filter — it must not become one that
+	// matches nothing and empties the response.
+	if n := len(normalizedFilters(MetadataSearchRequest{Property: "tmdbid"})); n != 0 {
+		t.Fatalf("a property with no value must not fold into a filter, got %d", n)
+	}
+	if n := len(normalizedFilters(MetadataSearchRequest{PropertyValue: "1405"})); n != 0 {
+		t.Fatalf("a value with no property must not fold into a filter, got %d", n)
+	}
+
+	both := normalizedFilters(MetadataSearchRequest{
+		Property:      "tmdbid",
+		PropertyValue: "1405",
+		Filters:       []MetadataFilter{f("fileType", true, "video")},
+	})
+	if len(both) != 2 {
+		t.Fatalf("the legacy pair must be appended to explicit filters, got %d", len(both))
+	}
+}
+
+// The acceptance bar for routing the anchored path through the index: a
+// full-miss filtered sweep over a realistic corpus must stay in the same class
+// as the free-text branch (measured 10 ms over 19 622 live records), not the
+// legacy per-record Redis scan it replaces (measured 3.63 s over the same
+// corpus). Full-miss is the worst case — the loop cannot short-circuit on the
+// result limit, so every record is evaluated.
+//
+// Run: go test ./internal/api/ -bench BenchmarkFilterSweep -benchtime 10x
+func BenchmarkFilterSweep(b *testing.B) {
+	const records = 20000
+	corpus := make([]map[string]string, records)
+	for i := range corpus {
+		corpus[i] = map[string]string{
+			"title":                  "Some Release",
+			"fileType":               "video",
+			"videoType":              "tv",
+			"tmdbid":                 "1405",
+			"season":                 "2",
+			"genres/Action":          "true",
+			"genres/Drama":           "true",
+			"source/gateway:nyaa.si": "true",
+			"categories/TV/Anime":    "true",
+			"indexer":                "nyaa.si",
+			"fileName":               "some.release.s02e05.mkv",
+			"description/eng":        "text",
+			"poster":                 "bafy",
+			"sizeByte":               "123456789",
+			"contentKind":            "episode",
+			"anchoredBy/tmdb:1405":   "true",
+		}
+	}
+	// A miss on the identity field: the shape an id-anchored gateway query
+	// takes, and the one that must not degrade.
+	filters := []MetadataFilter{
+		f("tmdbid", true, "999999"),
+		f("fileType", true, "video"),
+		f("genres", true, "action"),
+	}
+	b.ResetTimer()
+	for n := 0; n < b.N; n++ {
+		hits := 0
+		for _, fields := range corpus {
+			if recordMatchesFilters(fields, filters) {
+				hits++
+			}
+		}
+		if hits != 0 {
+			b.Fatalf("expected a full miss, got %d hits", hits)
+		}
+	}
+}

internal/api/search_index.go

@@ -3,6 +3,7 @@ package api
 import (
 	"log"
 	"net/http"
+	"sort"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -17,19 +18,46 @@ import (
 // batched bulk read (storage.GetSearchIndexTuples), off the request path.
 const searchIndexRefreshInterval = 60 * time.Second
 
+// searchIndexReconcileEvery is how many refresh ticks pass between full
+// rebuilds. Every other tick is incremental. 30 ticks × 60 s = a full
+// reconcile every 30 minutes — see runRefreshLoop for why one is kept at all.
+const searchIndexReconcileEvery = 30
+
 // 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.
+// (hashID + lowercased searchable fields), kept current 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.
+//
+// # Two update paths, and why
+//
+// `refresh` rebuilds everything: a whole-keyspace SCAN, batched MGETs, and a
+// complete fresh copy. That is O(corpus) regardless of how much changed, and it
+// used to run every 60 s forever. Measured on a 19 622-record staging corpus
+// (2026-08-30): the loop accounted for ~45 % of ALL Redis CPU, and the Go side
+// held 783 MB for data Redis stores in 135 MB — largely the double copy.
+//
+// `refreshIncremental` re-reads only the records that changed, tracked by
+// storage's dirty set (search_dirty.go). Cost is O(change). It is the steady
+// state; the full rebuild runs once at startup (there is nothing to be
+// incremental against) and then only as an explicit repair.
+//
+// The distinction matters because meta-core is the platform's permanent,
+// additive store — every search hit on every gateway is retained — so a rebuild
+// priced against the whole corpus is a cost that only ever grows.
 type searchIndex struct {
 	mu sync.RWMutex
+	// byID is the authoritative content; tuples is a stable-order view of it
+	// rebuilt on each update. Map iteration order in Go is randomised, so
+	// serving a limited query straight off the map would return a different
+	// subset each call — the slice is what keeps results stable.
+	byID   map[string]storage.SearchTuple
 	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
+// least once. Every update 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()
@@ -37,7 +65,21 @@ func (idx *searchIndex) snapshot() ([]storage.SearchTuple, bool) {
 	return idx.tuples, idx.ready.Load()
 }
 
-// refresh rebuilds the index from Redis once (a batched bulk read).
+// rebuildViewLocked regenerates the stable-order slice from byID. Caller holds
+// the write lock.
+func (idx *searchIndex) rebuildViewLocked() {
+	view := make([]storage.SearchTuple, 0, len(idx.byID))
+	for _, t := range idx.byID {
+		view = append(view, t)
+	}
+	// Stable across rebuilds so a limited query returns a consistent page
+	// rather than a fresh random subset every interval.
+	sort.Slice(view, func(i, j int) bool { return view[i].HashID < view[j].HashID })
+	idx.tuples = view
+}
+
+// refresh rebuilds the whole index from Redis (a batched bulk read). Used for
+// the initial build and as an explicit repair — not the steady state.
 func (idx *searchIndex) refresh(stor *storage.Client) {
 	if stor == nil || !stor.IsConnected() {
 		return
@@ -47,27 +89,101 @@ func (idx *searchIndex) refresh(stor *storage.Client) {
 		log.Printf("[API] search index refresh failed: %v", err)
 		return
 	}
+	byID := make(map[string]storage.SearchTuple, len(tuples))
+	for _, t := range tuples {
+		byID[t.HashID] = t
+	}
 	idx.mu.Lock()
-	idx.tuples = tuples
+	idx.byID = byID
+	idx.rebuildViewLocked()
 	idx.mu.Unlock()
 	idx.ready.Store(true)
 }
 
-// runRefreshLoop builds the index immediately, then rebuilds on a ticker for the
-// process lifetime.
+// refreshIncremental drains the dirty set and patches only those records.
+// Returns whether more dirty entries remain, so a burst is caught up over
+// consecutive passes rather than one per interval.
+//
+// A record in the dirty set but absent from storage was deleted and is evicted
+// here — otherwise it would keep answering searches until the next full
+// rebuild. A batch that fails after the (destructive) drain is re-marked by
+// storage so it is retried rather than silently lost.
+func (idx *searchIndex) refreshIncremental(stor *storage.Client) (more bool) {
+	if stor == nil || !stor.IsConnected() {
+		return false
+	}
+	ids, more, err := stor.DrainDirty()
+	if err != nil {
+		log.Printf("[API] search index drain failed: %v", err)
+		return false
+	}
+	if len(ids) == 0 {
+		return false
+	}
+	tuples, gone, err := stor.GetSearchIndexTuplesFor(ids)
+	if err != nil {
+		stor.MarkDirty(ids...)
+		log.Printf("[API] incremental search index refresh failed: %v", err)
+		return false
+	}
+
+	idx.mu.Lock()
+	if idx.byID == nil {
+		idx.byID = make(map[string]storage.SearchTuple, len(tuples))
+	}
+	for _, t := range tuples {
+		idx.byID[t.HashID] = t
+	}
+	for _, id := range gone {
+		delete(idx.byID, id)
+	}
+	idx.rebuildViewLocked()
+	idx.mu.Unlock()
+
+	log.Printf("[API] search index: %d updated, %d evicted (%d records)", len(tuples), len(gone), len(idx.tuples))
+	return more
+}
+
+// runRefreshLoop builds the index once in full, then keeps it current from the
+// dirty set, with a periodic full reconcile as a safety net.
+//
+// The reconcile exists because a *missed* dirty mark is silent: no error, just
+// a record whose indexed copy quietly diverges from storage until something
+// forces a rebuild. Marking is spread across eight write paths plus anything
+// added later, and the cost of getting it wrong is invisible — so the index
+// should not depend on that coverage being perfect forever. Every
+// searchIndexReconcileEvery-th tick pays the old O(corpus) pass and repairs
+// whatever drifted.
+//
+// That still drops the full rebuild from every 60 s to every 30 min — a ~30×
+// reduction in the steady-state cost that measured at ~45 % of all Redis CPU —
+// while keeping the self-healing property that made the naive loop safe.
 func (idx *searchIndex) runRefreshLoop(stor *storage.Client) {
 	idx.refresh(stor)
 	ticker := time.NewTicker(searchIndexRefreshInterval)
 	defer ticker.Stop()
+	ticks := 0
 	for range ticker.C {
+		ticks++
+		if ticks%searchIndexReconcileEvery == 0 {
 			idx.refresh(stor)
+			continue
+		}
+		// Drain until caught up, so a burst larger than one batch does not take
+		// one interval per batch to land.
+		for idx.refreshIncremental(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.
+// serveIndexedSearch answers a search from the in-memory index: a token-AND
+// match against each record's haystack (or a whole-query hashID substring,
+// preserving the legacy behaviour) AND every structured filter, then returns
+// the full metadata for the <=limit survivors. No Redis round-trip per scanned
+// record — matching and results both come from memory.
+//
+// This is the ONLY search path. Free text, filters, both, or neither all land
+// here; only the `hashId` exact-match branch in handleSearchMetadata bypasses it.
 func (s *Server) serveIndexedSearch(w http.ResponseWriter, req MetadataSearchRequest) {
 	tuples, ready := s.search.snapshot()
 	if !ready {
@@ -78,15 +194,20 @@ func (s *Server) serveIndexedSearch(w http.ResponseWriter, req MetadataSearchReq
 	}
 	tokens := strings.Fields(strings.ToLower(req.Query))
 	queryLower := strings.ToLower(req.Query)
+	filters := normalizedFilters(req)
 	results := make([]MetadataSearchResult, 0, req.Limit)
 	for _, t := range tuples {
 		if len(results) >= req.Limit {
 			break
 		}
-		if !matchAllTokens(t.Haystack, tokens) &&
+		if req.Query != "" &&
+			!matchAllTokens(t.Haystack, tokens) &&
 			!strings.Contains(strings.ToLower(t.HashID), queryLower) {
 			continue
 		}
+		if !recordMatchesFilters(t.Fields, filters) {
+			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})
@@ -108,3 +229,114 @@ func matchAllTokens(haystack string, tokens []string) bool {
 	}
 	return true
 }
+
+// normalizedFilters returns the request's structured filters with the legacy
+// Property/PropertyValue pair folded in as one more entry.
+//
+// The legacy pair keeps Exact=false so its long-standing substring semantics
+// are unchanged for existing callers; new callers pass Filters and choose.
+func normalizedFilters(req MetadataSearchRequest) []MetadataFilter {
+	if req.Property == "" || req.PropertyValue == "" {
+		return req.Filters
+	}
+	out := make([]MetadataFilter, 0, len(req.Filters)+1)
+	out = append(out, req.Filters...)
+	return append(out, MetadataFilter{
+		Property: req.Property,
+		Values:   []string{req.PropertyValue},
+	})
+}
+
+// recordMatchesFilters reports whether fields satisfies every filter (AND).
+func recordMatchesFilters(fields map[string]string, filters []MetadataFilter) bool {
+	for i := range filters {
+		if !recordMatchesFilter(fields, &filters[i]) {
+			return false
+		}
+	}
+	return true
+}
+
+// recordMatchesFilter reports whether fields satisfies one filter (OR within
+// its values), probing BOTH storage shapes a field can take:
+//
+//   - a flat scalar/csv-set field — `fileType = "video"`, `videoType = "tv,anime"`;
+//   - a key-set member — `genres/Action = "true"`, `source/gateway:nyaa.si = "true"`
+//     (METADATA_KEYS "How the store is shaped").
+//
+// Probing both is what lets one filter shape serve the whole vocabulary. The
+// gateway needs a `genres:action` filter to find a record whose genre lives in
+// the key `genres/Action`, and a `fileType:video` filter to find one whose value
+// lives in the field `fileType` — without meta-core knowing which fields are
+// key-sets. It also means this stays a SUPERSET of the gateway's own
+// `record_matches`, which is the contract: meta-core pre-filters, the gateway
+// re-filters authoritatively.
+//
+// A filter with no usable value constrains nothing (matches). A filter whose
+// field is absent fails closed, matching `record_matches`.
+func recordMatchesFilter(fields map[string]string, f *MetadataFilter) bool {
+	values := make([]string, 0, len(f.Values))
+	for _, v := range f.Values {
+		if v = strings.TrimSpace(v); v != "" {
+			values = append(values, v)
+		}
+	}
+	if f.Property == "" || len(values) == 0 {
+		return true
+	}
+
+	// Key-set shape: `<property>/<value>` present and truthy. Try the exact
+	// key first — that hits whenever the caller's casing matches what was
+	// written, which is the common case and costs one map probe.
+	for _, v := range values {
+		if isTruthyMember(fields[f.Property+"/"+v]) {
+			return true
+		}
+	}
+	// Members are written verbatim (`genres/Action`) while query values arrive
+	// lowercased, so fall back to one case-insensitive sweep. Single pass over
+	// the record's fields for ALL values, not one pass per value.
+	wanted := make(map[string]struct{}, len(values))
+	for _, v := range values {
+		wanted[strings.ToLower(v)] = struct{}{}
+	}
+	prefix := strings.ToLower(f.Property) + "/"
+	for k, val := range fields {
+		if len(k) <= len(prefix) || !strings.EqualFold(k[:len(prefix)], prefix) {
+			continue
+		}
+		if _, ok := wanted[strings.ToLower(k[len(prefix):])]; ok && isTruthyMember(val) {
+			return true
+		}
+	}
+
+	// Flat shape: the field's own value, split on `,` for csv-set fields.
+	fieldValue, ok := fields[f.Property]
+	if !ok {
+		return false
+	}
+	for _, piece := range strings.Split(fieldValue, ",") {
+		piece = strings.TrimSpace(piece)
+		if piece == "" {
+			continue
+		}
+		for _, v := range values {
+			if f.Exact {
+				if strings.EqualFold(piece, v) {
+					return true
+				}
+			} else if strings.Contains(strings.ToLower(piece), strings.ToLower(v)) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// isTruthyMember reports whether a key-set member's value asserts membership.
+// Mirrors the gateway's `is_truthy_member`: presence with a non-empty,
+// non-"false"/"0" value.
+func isTruthyMember(v string) bool {
+	v = strings.TrimSpace(strings.ToLower(v))
+	return v != "" && v != "false" && v != "0"
+}

internal/api/server.go

@@ -243,6 +243,10 @@ func (s *Server) setupRoutes() {
 	s.router.HandleFunc("/api/file/{cid}/info", s.handleGetFileByCIDInfo).Methods("GET")
 	s.router.HandleFunc("/file/{cid}", s.handleGetFileByCID).Methods("GET")
 	s.router.HandleFunc("/file/cid", s.handleComputeFileCID).Methods("POST")
+	// Register one file under /files/plugin by midhash — the watcher's
+	// mint + alias + file:events path for a single file. Registered
+	// unconditionally: it exists for boxes whose watcher is OFF.
+	s.router.HandleFunc("/api/files/register", s.handleRegisterFile).Methods("POST")
 
 	// Public metadata-by-CID resolver. Reverse-index lookup → full document.
 	// Auth-bypassed alongside /api/file/{cid} so meta-share peers can query
@@ -253,6 +257,12 @@ func (s *Server) setupRoutes() {
 	// (the dual-root pattern caused by historical writes that bypassed CID
 	// resolution).
 	s.router.HandleFunc("/api/admin/migrate-dual-roots", s.handleMigrateDualRoots).Methods("POST")
+	// One-shot vocabulary sweep: domain film|tv → screen, workForm backfill
+	// (METADATA_KEYS.md §14.17). Idempotent.
+	s.router.HandleFunc("/api/admin/migrate-domain-screen", s.handleMigrateDomainScreen).Methods("POST")
+	// One-shot sweep of query echoes on literature cards (meta-read
+	// reading-model.md §7). Dry-run unless ?apply=true. Idempotent.
+	s.router.HandleFunc("/api/admin/sweep-literature-echoes", s.handleSweepLiteratureEchoes).Methods("POST")
 
 	// SSE event streams. Mediated mirror of the Redis Streams; consumers
 	// reach them over HTTP instead of touching Redis directly. Auth-bypass

internal/cid/card.go

@@ -0,0 +1,92 @@
+package cid
+
+import (
+	"encoding/base32"
+	"encoding/binary"
+	"fmt"
+	"strings"
+	"unicode/utf8"
+)
+
+// DecodeCard recovers the (source, id) pair a card locator (CodeCard, 0x1007)
+// wraps. Wire form:
+//
+//	0x01 ‖ varint(0x1007) ‖ 0x00 (identity mh) ‖ varint(len(digest)) ‖ digest
+//	digest = varint(len(source)) ‖ source ‖ id        (both UTF-8)
+//
+// Decode stops at the multihash code because ranking needs nothing past it;
+// reading *which work* a card addresses needs the digest too, so this is the
+// one place in Go that walks the whole thing.
+//
+// ⚠ Mirror of meta-read `card_cid::decode` and of meta-feeder-sdk
+// `compute_card_cid`. The vectors in card_test.go are pinned literals copied
+// from meta-read, not round-trips, so a drift on either side fails a test
+// instead of silently reading a different work out of the same CID.
+//
+// Identity multihash only: a 0x1007 CID carrying any other multihash is not
+// something anyone mints, and reading its digest as (source, id) would be
+// reading hash bytes as text. Bytes past the declared digest length are
+// ignored, as meta-read does.
+func DecodeCard(cidStr string) (source, id string, err error) {
+	if len(cidStr) < 2 || cidStr[0] != 'b' {
+		return "", "", fmt.Errorf("cid %q: not a multibase base32 CIDv1", cidStr)
+	}
+	raw, err := base32.StdEncoding.WithPadding(base32.NoPadding).DecodeString(strings.ToUpper(cidStr[1:]))
+	if err != nil {
+		return "", "", fmt.Errorf("cid %q: base32 decode: %w", cidStr, err)
+	}
+
+	next := func(buf []byte, what string) (uint64, []byte, error) {
+		v, n := binary.Uvarint(buf)
+		if n <= 0 {
+			return 0, nil, fmt.Errorf("cid %q: bad %s varint", cidStr, what)
+		}
+		return v, buf[n:], nil
+	}
+
+	version, raw, err := next(raw, "version")
+	if err != nil {
+		return "", "", err
+	}
+	if version != 1 {
+		return "", "", fmt.Errorf("cid %q: unsupported CID version %d", cidStr, version)
+	}
+	codec, raw, err := next(raw, "codec")
+	if err != nil {
+		return "", "", err
+	}
+	if codec != CodeCard {
+		return "", "", fmt.Errorf("cid %q: codec 0x%x is not a card locator", cidStr, codec)
+	}
+	mh, raw, err := next(raw, "multihash-code")
+	if err != nil {
+		return "", "", err
+	}
+	if mh != 0 {
+		return "", "", fmt.Errorf("cid %q: card locator with non-identity multihash 0x%x", cidStr, mh)
+	}
+	digestLen, raw, err := next(raw, "digest-length")
+	if err != nil {
+		return "", "", err
+	}
+	if digestLen > uint64(len(raw)) {
+		return "", "", fmt.Errorf("cid %q: digest truncated", cidStr)
+	}
+	digest := raw[:digestLen]
+
+	sourceLen, rest, err := next(digest, "source-length")
+	if err != nil {
+		return "", "", err
+	}
+	if sourceLen > uint64(len(rest)) {
+		return "", "", fmt.Errorf("cid %q: source truncated", cidStr)
+	}
+	source, id = string(rest[:sourceLen]), string(rest[sourceLen:])
+	if source == "" || id == "" {
+		return "", "", fmt.Errorf("cid %q: empty source or id", cidStr)
+	}
+	if !utf8.ValidString(source) || !utf8.ValidString(id) {
+		return "", "", fmt.Errorf("cid %q: source or id is not UTF-8", cidStr)
+	}
+	return source, id, nil
+}

internal/cid/card_test.go

@@ -0,0 +1,71 @@
+package cid
+
+import (
+	"encoding/base32"
+	"encoding/binary"
+	"strings"
+	"testing"
+)
+
+// Pinned from meta-read `src/card_cid.rs` / meta-feeder-sdk `compute_card_cid`
+// — literals, not round-trips (see DecodeCard).
+const (
+	goldenAnilistCard  = "bagdsaaaua5qw42lmnfzxi3lbnztwcorrga2tonzy"
+	goldenMangadexCard = "bagdsaabybrwwc3thmfsgk6bon5zgol3nmfxgoyjpmiygenzsgftgmlldgm4dqljugq4dmllbmeygmlldgjrdaytcgmzdcnjrgi"
+)
+
+// wireCID builds a CIDv1 from raw parts so the negative cases below can vary
+// one field at a time.
+func wireCID(codec, mh uint64, digest []byte) string {
+	buf := []byte{0x01}
+	buf = binary.AppendUvarint(buf, codec)
+	buf = binary.AppendUvarint(buf, mh)
+	buf = binary.AppendUvarint(buf, uint64(len(digest)))
+	buf = append(buf, digest...)
+	return "b" + strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(buf))
+}
+
+func cardDigest(source, id string) []byte {
+	d := binary.AppendUvarint(nil, uint64(len(source)))
+	d = append(d, source...)
+	return append(d, id...)
+}
+
+func TestDecodeCard_GoldenVectors(t *testing.T) {
+	for _, tc := range []struct {
+		cid, source, id string
+	}{
+		{goldenAnilistCard, "anilist", "manga:105778"},
+		{goldenMangadexCard, "mangadex.org", "/manga/b0b721ff-c388-4486-aa0f-c2b0bb321512"},
+	} {
+		source, id, err := DecodeCard(tc.cid)
+		if err != nil {
+			t.Fatalf("DecodeCard(%s): %v", tc.cid, err)
+		}
+		if source != tc.source || id != tc.id {
+			t.Errorf("DecodeCard(%s) = (%q, %q), want (%q, %q)", tc.cid, source, id, tc.source, tc.id)
+		}
+		// The test encoder must agree with the pinned literal, or the negative
+		// cases below are testing a different wire format.
+		if got := wireCID(CodeCard, 0, cardDigest(tc.source, tc.id)); got != tc.cid {
+			t.Errorf("wireCID(%q, %q) = %s, want %s", tc.source, tc.id, got, tc.cid)
+		}
+	}
+}
+
+func TestDecodeCard_RejectsEverythingElse(t *testing.T) {
+	for name, s := range map[string]string{
+		"url locator codec":      wireCID(CodeURL, 0, cardDigest("anilist", "1")),
+		"non-identity multihash": wireCID(CodeCard, CodeSha256, cardDigest("anilist", "1")),
+		"empty id":               wireCID(CodeCard, 0, cardDigest("anilist", "")),
+		"source past digest":     wireCID(CodeCard, 0, []byte{0x20, 'a'}),
+		"not a CID":              "title:berserk",
+		"CIDv0":                  "QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG",
+		"truncated":              goldenAnilistCard[:12],
+		"empty":                  "",
+	} {
+		if source, id, err := DecodeCard(s); err == nil {
+			t.Errorf("%s: DecodeCard(%q) = (%q, %q), want error", name, s, source, id)
+		}
+	}
+}

internal/storage/cid_resolution.go

@@ -309,6 +309,10 @@ func (c *Client) DeleteRoot(uuid string) error {
 	if err := c.client.SRem(ctx, c.buildIndexKey(), uuid).Err(); err != nil {
 		return fmt.Errorf("srem index: %w", err)
 	}
+	// Marked so the incremental search-index refresh reads it, finds it gone,
+	// and evicts it — otherwise a deleted root keeps answering searches until
+	// the next full rebuild.
+	c.markDirtyLocked(ctx, uuid)
 	return nil
 }
 
@@ -418,3 +422,50 @@ func (c *Client) GetMetadataDocument(uuid string) (*MetadataDocument, error) {
 		Duplicates: dups,
 	}, nil
 }
+
+// RegisterFileTuple records one physical file under its midhash256 CID, the
+// same way the file watcher does for every file it scans: mint a UUID root
+// (filePath/sizeByte/mtimeNano) and write `cids/<midhash>`, whose hook
+// registers the cid:<midhash> reverse alias. Content already known under
+// another path is recorded as a duplicate; the same path just refreshes
+// size/mtime.
+//
+// Returns the root and whether it was newly minted. Shared by the watcher and
+// POST /api/files/register, which is how a box running with the watcher off
+// (a meta-watch client core) still gets files it writes under /files/plugin
+// resolvable by CID.
+func (c *Client) RegisterFileTuple(absPath string, size, mtimeNano int64, midhash string) (string, bool, error) {
+	if midhash == "" {
+		return "", false, fmt.Errorf("midhash is required")
+	}
+	uuid, err := c.GetByCID(midhash)
+	if err != nil {
+		return "", false, fmt.Errorf("GetByCID(%s): %w", midhash, err)
+	}
+
+	if uuid == "" {
+		uuid, err = c.Mint(absPath, size, mtimeNano)
+		if err != nil {
+			return "", false, fmt.Errorf("Mint(%s): %w", absPath, err)
+		}
+		if err := c.SetProperty(uuid, CIDsKeyPrefix+midhash, "true"); err != nil {
+			return uuid, true, fmt.Errorf("SetProperty cids/%s for %s: %w", midhash, uuid, err)
+		}
+		return uuid, true, nil
+	}
+
+	existing, _ := c.GetProperty(uuid, "filePath")
+	if existing != absPath {
+		if _, err := c.AddDuplicatePath(uuid, absPath); err != nil {
+			return uuid, false, fmt.Errorf("AddDuplicatePath %s @ %s: %w", uuid, absPath, err)
+		}
+		return uuid, false, nil
+	}
+	if _, err := c.MergeMetadataFlat(uuid, map[string]string{
+		"sizeByte":  strconv.FormatInt(size, 10),
+		"mtimeNano": strconv.FormatInt(mtimeNano, 10),
+	}); err != nil {
+		return uuid, false, fmt.Errorf("refresh tuple for %s: %w", uuid, err)
+	}
+	return uuid, false, nil
+}

internal/storage/client.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"os"
 	"sort"
 	"strconv"
 	"strings"
@@ -173,6 +174,7 @@ func (c *Client) SetMetadataFlat(hashID string, metadata map[string]string) erro
 	if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
 		return fmt.Errorf("sadd index failed: %w", err)
 	}
+	c.markDirtyLocked(ctx, hashID)
 
 	// Auto-register reverse-index aliases for any cids/<cid> key-set members in
 	// the payload. That is the ONLY shape recognised — the legacy flat cid_* /
@@ -257,6 +259,7 @@ func (c *Client) SetProperty(hashID, property, value string) error {
 	if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
 		return fmt.Errorf("sadd index failed: %w", err)
 	}
+	c.markDirtyLocked(ctx, hashID)
 
 	// Auto-register reverse-index alias if this is a CID field.
 	_ = c.maybeAddAliasFromFieldLocked(ctx, hashID, property, value)
@@ -325,6 +328,10 @@ func (c *Client) DeleteMetadata(hashID string) (int64, error) {
 	if err := c.client.SRem(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
 		return deletedCount, fmt.Errorf("srem index failed: %w", err)
 	}
+	// Marked, not unmarked: the incremental refresh reads the record, finds it
+	// gone, and evicts it from the search index. Skipping this would leave a
+	// deleted record answering searches until the next full rebuild.
+	c.markDirtyLocked(ctx, hashID)
 
 	return deletedCount, nil
 }
@@ -441,6 +448,7 @@ func (c *Client) MergeMetadataFlat(hashID string, metadata map[string]string) (i
 	if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
 		return len(metadata), fmt.Errorf("sadd index failed: %w", err)
 	}
+	c.markDirtyLocked(ctx, hashID)
 
 	// Auto-register reverse-index aliases for any cids/<cid> key-set members in
 	// the payload. That is the ONLY shape recognised — the legacy flat cid_* /
@@ -468,6 +476,7 @@ func (c *Client) DeleteProperty(hashID, property string) error {
 		return err
 	}
 	c.removeFieldLocked(ctx, hashID, property)
+	c.markDirtyLocked(ctx, hashID)
 	return nil
 }
 
@@ -521,6 +530,7 @@ func (c *Client) AddToSet(hashID, property, value string) (bool, error) {
 	if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
 		return true, fmt.Errorf("sadd index failed: %w", err)
 	}
+	c.markDirtyLocked(ctx, hashID)
 
 	return true, nil
 }
@@ -574,11 +584,16 @@ func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) {
 		}
 		// The key is gone, so the field name must leave the index too.
 		c.removeFieldLocked(ctx, hashID, property)
+		c.markDirtyLocked(ctx, hashID)
 		return true, nil
 	}
 
 	newValue := strings.Join(newValues, "|")
-	return true, c.client.Set(ctx, key, newValue, 0).Err()
+	err = c.client.Set(ctx, key, newValue, 0).Err()
+	// The field survives with fewer members, so no field-index change — but the
+	// VALUE changed, which is exactly what the search index caches.
+	c.markDirtyLocked(ctx, hashID)
+	return true, err
 }
 
 // FileTuple holds the (uuid, path, size, mtime, midhash) record for one
@@ -691,6 +706,44 @@ type SearchTuple struct {
 // haystack — must stay in sync with the fields the search handler documents.
 var searchIndexFields = []string{"title", "fileName", "originaltitle", "showtitle", "filePath"}
 
+// defaultIndexExcludePrefixes are field-name prefixes kept OUT of the in-memory
+// search index. Redis still stores them and every per-record read still returns
+// them — this narrows the *index*, never the store.
+//
+// `categories/newznab/<code>` is the raw Newznab numeric category code, which
+// METADATA_KEYS documents as a mirror of the canonical `categories/<label>`
+// vocabulary that rides alongside it. Nothing in meta-gateway, meta-search,
+// meta-watch or meta-core reads it, and it is the single largest contributor to
+// the index: measured on a live 19.6k-record corpus, `categories/*` was 40% of
+// the whole keyspace and the `newznab` sub-vocabulary was 59% of that — ~24% of
+// all keys, for zero readers.
+//
+// ⚠ Excluding the WHOLE `categories/` family instead would be a security bug,
+// not an optimisation. `categories/XXX` (and `categories/XXX/<leaf>`) is the
+// server-side adult-content enforcement point — meta-search's
+// `query_eval::fields_are_adult` and meta-gateway's `cards::fields_are_adult`
+// both match on it, and the `exclude_adult` preference defaults to ON. A record
+// served from the index without its categories would read as non-adult and the
+// filter would fail OPEN. Only prefixes with no semantic readers belong here.
+var defaultIndexExcludePrefixes = []string{"categories/newznab/"}
+
+// indexExcludePrefixes resolves the exclusion list, allowing an operator to
+// override it with a comma-separated META_CORE_INDEX_EXCLUDE_PREFIXES. An
+// explicitly empty value disables exclusion entirely (index everything).
+func indexExcludePrefixes() []string {
+	raw, set := os.LookupEnv("META_CORE_INDEX_EXCLUDE_PREFIXES")
+	if !set {
+		return defaultIndexExcludePrefixes
+	}
+	out := make([]string, 0, 4)
+	for _, p := range strings.Split(raw, ",") {
+		if p = strings.TrimSpace(p); p != "" {
+			out = append(out, p)
+		}
+	}
+	return out
+}
+
 // 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
@@ -712,6 +765,7 @@ func (c *Client) GetSearchIndexTuples() ([]SearchTuple, error) {
 	indexKey := c.buildIndexKey()     // {filePrefix}__index__ (a SET, skip it)
 	pattern := filePrefix + "*"
 
+	excluded := indexExcludePrefixes()
 	records := make(map[string]map[string]string)
 	keyBatch := make([]string, 0, 1000)
 
@@ -762,6 +816,11 @@ func (c *Client) GetSearchIndexTuples() ([]SearchTuple, error) {
 			if strings.HasSuffix(k, "/"+FieldsField) {
 				continue
 			}
+			// Index-excluded fields are skipped here rather than after the
+			// MGET, so an excluded key costs neither the fetch nor the memory.
+			if _, field, ok := c.splitRecordKey(k); ok && hasAnyPrefix(field, excluded) {
+				continue
+			}
 			keyBatch = append(keyBatch, k)
 			if len(keyBatch) >= 1000 {
 				if err := flush(); err != nil {
@@ -780,18 +839,23 @@ func (c *Client) GetSearchIndexTuples() ([]SearchTuple, error) {
 
 	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})
+		// Shared with the incremental path (search_dirty.go) so the two can
+		// never disagree about what is searchable.
+		out = append(out, SearchTuple{HashID: hashID, Haystack: buildHaystack(fields), Fields: fields})
 	}
 	return out, nil
 }
 
+// hasAnyPrefix reports whether s starts with any of the given prefixes.
+func hasAnyPrefix(s string, prefixes []string) bool {
+	for _, p := range prefixes {
+		if strings.HasPrefix(s, p) {
+			return true
+		}
+	}
+	return false
+}
+
 // GetMemoryInfo returns Redis memory usage information
 func (c *Client) GetMemoryInfo() (string, error) {
 	c.mu.RLock()

internal/storage/domain_screen_migration.go

@@ -0,0 +1,193 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+	"time"
+)
+
+// The one-shot corpus sweep for METADATA_KEYS.md §14.17: the `film` + `tv`
+// domains merged into `screen`, and the standalone-vs-serial split they were
+// really carrying moved to the new `workForm` key.
+//
+// # Why a sweep and not a read-side alias
+//
+// `domain` is **stored, not derived on read** — that is what lets the
+// `contentKind` vocabulary stay open (a peer that has never heard of a kind
+// still routes the record correctly). The price of storing it is that a
+// vocabulary change has to be paid once, in the data: every consumer's filter
+// is an exact match, so a record left saying `film` is invisible to a
+// `domain:screen` wall, and teaching every consumer to accept both spellings
+// forever is exactly the alternation the merge exists to delete.
+//
+// So the values are rewritten at rest, here, and no alias vocabulary survives
+// in query filters. (meta-search additionally normalises on the way into its
+// index, because a record can still arrive from a peer that has not been swept
+// — that is a courtesy for un-swept peers, not a second source of truth.)
+//
+// # Contract
+//
+// Idempotent: a record already carrying `screen` and a `workForm` is skipped,
+// so the endpoint is safe to call repeatedly and safe to call on a fresh box.
+// Returns the number of records **changed**, not the number examined.
+
+// legacyDomainReplacements maps a retired `domain` value to its replacement.
+var legacyDomainReplacements = map[string]string{
+	"film": "screen",
+	"tv":   "screen",
+}
+
+// workFormByContentKind is the `contentKind` → `workForm` table from
+// METADATA_KEYS.md §1.
+//
+// ⚠ Mirror of the Rust table in `meta-feeder-sdk::domain` (and of
+// meta-search's copy). It exists here only to backfill the pre-merge corpus —
+// meta-core never classifies a record it is given, and nothing else in Go reads
+// it. If the tables ever disagree, the writers win: this one is a one-shot.
+var workFormByContentKind = map[string]string{
+	// screen
+	"movie":   "standalone",
+	"series":  "serial",
+	"episode": "serial",
+	// music — an album is a closed work, not an ongoing run.
+	"track":      "standalone",
+	"album":      "standalone",
+	"artist":     "standalone",
+	"musicVideo": "standalone",
+	"djMix":      "standalone",
+	"liveSet":    "standalone",
+	// literature
+	"book":      "standalone",
+	"audiobook": "standalone",
+	"comic":     "serial",
+	"manga":     "serial",
+	"magazine":  "serial",
+	// science
+	"paper": "standalone",
+	// ⚠ `pack` is deliberately absent: a season pack is `serial`, an album
+	// release `standalone`, and only the writer knows which. A pack that
+	// reaches this sweep without a `workForm` keeps none — inventing one would
+	// be a guess written permanently into the record.
+}
+
+// domainScreenPlan is one record's pending rewrite. Empty strings mean "leave
+// this field alone".
+type domainScreenPlan struct {
+	hashID   string
+	domain   string
+	workForm string
+}
+
+// MigrateDomainScreen rewrites `domain=film|tv` to `domain=screen` and
+// backfills `workForm` from `contentKind` across the whole corpus.
+//
+// Reads in two batched MGET passes (never a per-record SCAN — a single record
+// read costs a full keyspace sweep on this store), then writes only the records
+// that actually change.
+func (c *Client) MigrateDomainScreen(ctx context.Context) (int, error) {
+	plans, err := c.planDomainScreen(ctx)
+	if err != nil {
+		return 0, err
+	}
+	// Writes take the client lock themselves, so they happen after the read
+	// phase has released it.
+	fixed := 0
+	for _, p := range plans {
+		if p.domain != "" {
+			if err := c.SetProperty(p.hashID, "domain", p.domain); err != nil {
+				return fixed, fmt.Errorf("set domain on %s: %w", p.hashID, err)
+			}
+		}
+		if p.workForm != "" {
+			if err := c.SetProperty(p.hashID, "workForm", p.workForm); err != nil {
+				return fixed, fmt.Errorf("set workForm on %s: %w", p.hashID, err)
+			}
+		}
+		fixed++
+	}
+	return fixed, nil
+}
+
+// planDomainScreen is the read half: what would change, without changing it.
+func (c *Client) planDomainScreen(ctx context.Context) ([]domainScreenPlan, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	if c.client == nil {
+		return nil, fmt.Errorf("not connected")
+	}
+
+	hashIDs, err := c.client.SMembers(ctx, c.buildIndexKey()).Result()
+	if err != nil {
+		return nil, fmt.Errorf("smembers index: %w", err)
+	}
+	if len(hashIDs) == 0 {
+		return nil, nil
+	}
+
+	// Three keys per record, in a fixed order so the MGET response decodes by
+	// position: keys[i*3+n] belongs to hashIDs[i].
+	const fieldsPerFile = 3
+	fields := [fieldsPerFile]string{"domain", "workForm", "contentKind"}
+	keys := make([]string, 0, len(hashIDs)*fieldsPerFile)
+	for _, h := range hashIDs {
+		prefix := c.buildKeyPrefix(h)
+		for _, f := range fields {
+			keys = append(keys, prefix+f)
+		}
+	}
+
+	// Chunked so one MGET never carries the whole corpus.
+	const chunk = 3000
+	values := make([]interface{}, 0, len(keys))
+	for start := 0; start < len(keys); start += chunk {
+		end := start + chunk
+		if end > len(keys) {
+			end = len(keys)
+		}
+		got, err := c.client.MGet(ctx, keys[start:end]...).Result()
+		if err != nil {
+			return nil, fmt.Errorf("mget: %w", err)
+		}
+		values = append(values, got...)
+	}
+
+	str := func(i int) string {
+		if i >= len(values) {
+			return ""
+		}
+		s, _ := values[i].(string)
+		return s
+	}
+
+	plans := make([]domainScreenPlan, 0)
+	for i, h := range hashIDs {
+		base := i * fieldsPerFile
+		domain, workForm, contentKind := str(base), str(base+1), str(base+2)
+
+		plan := domainScreenPlan{hashID: h}
+		if replacement, ok := legacyDomainReplacements[domain]; ok {
+			plan.domain = replacement
+		}
+		// A stored `workForm` is authoritative — a `pack` writer knew something
+		// the table cannot. Only an absent one is filled in.
+		if workForm == "" {
+			if form, ok := workFormByContentKind[contentKind]; ok {
+				plan.workForm = form
+			}
+		}
+		if plan.domain != "" || plan.workForm != "" {
+			plans = append(plans, plan)
+		}
+	}
+	return plans, nil
+}
+
+// MigrateDomainScreenWithTimeout is the convenience wrapper used by the API
+// handler. Generous timeout: the read phase is two batched passes, but the
+// write phase is one round-trip per changed field.
+func (c *Client) MigrateDomainScreenWithTimeout() (int, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
+	defer cancel()
+	return c.MigrateDomainScreen(ctx)
+}

internal/storage/domain_screen_migration_test.go

@@ -0,0 +1,140 @@
+package storage
+
+import (
+	"context"
+	"testing"
+)
+
+// seedRecord writes one record's classification fields straight through
+// SetProperty, which is what a real writer uses (it maintains the field index
+// and the root index the sweep reads from).
+func seedRecord(t *testing.T, c *Client, hashID string, fields map[string]string) {
+	t.Helper()
+	for k, v := range fields {
+		if err := c.SetProperty(hashID, k, v); err != nil {
+			t.Fatalf("seed %s/%s: %v", hashID, k, err)
+		}
+	}
+}
+
+// The merge (METADATA_KEYS.md §14.17) is paid in the data: `domain` is stored,
+// not derived on read, so a record left saying `film` is invisible to a
+// `domain:screen` wall. This is the sweep that pays it.
+func TestMigrateDomainScreen_RewritesLegacyDomainsAndBackfillsWorkForm(t *testing.T) {
+	c, _ := newTestClient(t, "")
+
+	seedRecord(t, c, "rec-movie", map[string]string{
+		"domain":      "film",
+		"contentKind": "movie",
+	})
+	seedRecord(t, c, "rec-episode", map[string]string{
+		"domain":      "tv",
+		"contentKind": "episode",
+	})
+	// Already current, and already carries both keys: must be left alone.
+	seedRecord(t, c, "rec-current", map[string]string{
+		"domain":      "screen",
+		"contentKind": "series",
+		"workForm":    "serial",
+	})
+	// A different domain entirely — the sweep must not touch it, but it is
+	// still owed the `workForm` backfill.
+	seedRecord(t, c, "rec-book", map[string]string{
+		"domain":      "literature",
+		"contentKind": "book",
+	})
+
+	fixed, err := c.MigrateDomainScreen(context.Background())
+	if err != nil {
+		t.Fatalf("migrate: %v", err)
+	}
+	if fixed != 3 {
+		t.Errorf("fixed = %d, want 3 (the two legacy rows + the book's backfill)", fixed)
+	}
+
+	for _, tc := range []struct {
+		hashID   string
+		domain   string
+		workForm string
+	}{
+		{"rec-movie", "screen", "standalone"},
+		{"rec-episode", "screen", "serial"},
+		{"rec-current", "screen", "serial"},
+		{"rec-book", "literature", "standalone"},
+	} {
+		m, err := c.GetMetadataFlat(tc.hashID)
+		if err != nil {
+			t.Fatalf("read %s: %v", tc.hashID, err)
+		}
+		if m["domain"] != tc.domain {
+			t.Errorf("%s domain = %q, want %q", tc.hashID, m["domain"], tc.domain)
+		}
+		if m["workForm"] != tc.workForm {
+			t.Errorf("%s workForm = %q, want %q", tc.hashID, m["workForm"], tc.workForm)
+		}
+	}
+}
+
+// A second pass must find nothing to do — the endpoint is called by hand on
+// each box, and "run it twice" must not mean "write the whole corpus twice".
+func TestMigrateDomainScreen_IsIdempotent(t *testing.T) {
+	c, _ := newTestClient(t, "mm:")
+
+	seedRecord(t, c, "rec-1", map[string]string{
+		"domain":      "tv",
+		"contentKind": "episode",
+	})
+
+	if _, err := c.MigrateDomainScreen(context.Background()); err != nil {
+		t.Fatalf("first pass: %v", err)
+	}
+	fixed, err := c.MigrateDomainScreen(context.Background())
+	if err != nil {
+		t.Fatalf("second pass: %v", err)
+	}
+	if fixed != 0 {
+		t.Errorf("second pass fixed = %d, want 0", fixed)
+	}
+}
+
+// `pack` is the one kind neither routing table can decide — a season pack is
+// `serial`, an album release `standalone`. A pack that reaches the sweep
+// without a `workForm` must keep none: inventing one writes a guess into the
+// record permanently.
+func TestMigrateDomainScreen_NeverGuessesAPacksWorkForm(t *testing.T) {
+	c, _ := newTestClient(t, "")
+
+	seedRecord(t, c, "rec-pack", map[string]string{
+		"domain":      "tv",
+		"contentKind": "pack",
+	})
+	// A writer that DID know keeps what it wrote.
+	seedRecord(t, c, "rec-album", map[string]string{
+		"domain":      "music",
+		"contentKind": "pack",
+		"workForm":    "standalone",
+	})
+
+	if _, err := c.MigrateDomainScreen(context.Background()); err != nil {
+		t.Fatalf("migrate: %v", err)
+	}
+
+	m, err := c.GetMetadataFlat("rec-pack")
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	if m["domain"] != "screen" {
+		t.Errorf("domain = %q, want screen (the domain WAS knowable)", m["domain"])
+	}
+	if _, ok := m["workForm"]; ok {
+		t.Errorf("workForm = %q, want absent", m["workForm"])
+	}
+
+	a, err := c.GetMetadataFlat("rec-album")
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	if a["workForm"] != "standalone" {
+		t.Errorf("workForm = %q, want standalone (writer-supplied, never overwritten)", a["workForm"])
+	}
+}

internal/storage/literature_echo_sweep.go

@@ -0,0 +1,326 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+
+	"github.com/metazla/meta-core/internal/cid"
+)
+
+// The one-shot corpus sweep for meta-read `docs/reading-model.md` §7:
+// identity and position keys are resolved, never copied from a query, and a
+// card carries no position keys and no `workcid`.
+//
+// # Why a sweep
+//
+// A feeder answering a `workcid:` / `anilistid:` query could stamp the query's
+// own filter values onto what it returned. The gateway persists whatever passes
+// the filter, so the copy became a permanent binding — and stored records never
+// pick up the writer fix: a covered `(upstream, query)` never reaches the
+// feeder again, and the streaming dedup keeps meta-core's copy over a fresh
+// one. So the copies already at rest are removed here, once.
+//
+// What the copies cost: an `anilistid` on a MangaDex site card claims an
+// AniList binding nobody resolved, and meta-read reads a bare `anilistid` as an
+// anchor — the site card then competes with (and can hide) the real AniList
+// card for the same work. A `workcid` or `chapterNumber` on a card makes a work
+// join and sort like a position inside one.
+//
+// # What is swept, and what deliberately is not
+//
+//   - Only `domain=literature` + `fileType=card` records. Nothing else is read
+//     for writing, whatever keys it carries.
+//   - `workcid`, `chapterNumber`, `volumeNumber`, `chapterStart`, `chapterEnd`
+//     go from **every** such card: a card is a work, never a position in one,
+//     so no legitimate writer puts them there.
+//   - `anilistid` goes only from a **site card** — a card whose own `cids/`
+//     key-set holds a site-form card locator and no locator of any other form.
+//     On an anchor card the id *is* the card's identity (its locator is
+//     `("anilist", …)`). A card holding both forms (a cross-source merge) keeps
+//     it: which half the id belongs to is not recoverable at rest, and leaving
+//     a stamp is the safe direction.
+//   - **Chapters are never swept.** A bare `anilistid` on a chapter is graded
+//     Matched by rule, and a query-copied stamp is indistinguishable at rest
+//     from a title-matched one — deleting would drop real bindings with the
+//     fake ones.
+//
+// # Contract
+//
+// Dry-run unless `apply`: the sweep deletes keys, so the default call only
+// reports. Idempotent: a swept record plans nothing on the next pass, so the
+// endpoint is safe to call repeatedly and on a fresh box.
+
+// literatureEchoAnchorSources are the card-locator sources that name an id-bag
+// work (an anchor), never a site.
+//
+// ⚠ Mirror of meta-read `card_cid::is_host_form`. meta-read spells out only
+// `anilist` and `openlibrary`; the others carry no '.', so the '.' test already
+// excludes them and the two rules agree on every locator minted today. They are
+// listed so the rule reads on its own here.
+var literatureEchoAnchorSources = map[string]bool{
+	"anilist":     true,
+	"openlibrary": true,
+	"tmdb":        true,
+	"imdb":        true,
+	"musicbrainz": true,
+	"mal":         true,
+}
+
+// literatureCardEchoFields are removed from every literature card carrying
+// them — position keys and the work reference, none of which a card may hold.
+var literatureCardEchoFields = []string{
+	"workcid",
+	"chapterNumber",
+	"volumeNumber",
+	"chapterStart",
+	"chapterEnd",
+}
+
+// literatureEchoReadFields is the per-record MGET layout: the two gate fields,
+// then `anilistid`, then the echo fields in literatureCardEchoFields order.
+var literatureEchoReadFields = append(
+	[]string{"domain", "fileType", "anilistid"},
+	literatureCardEchoFields...,
+)
+
+// LiteratureEchoSweep is the result of one sweep pass. Planned and Fixed count
+// records; ByKey counts fields per key — removed on an apply pass, that would
+// be removed on a dry run. ByKey is never nil so it serialises as `{}`.
+type LiteratureEchoSweep struct {
+	Planned int
+	Fixed   int
+	ByKey   map[string]int
+}
+
+// literatureEchoPlan is one record's pending deletes, in a fixed order
+// (`anilistid` first, then literatureCardEchoFields order).
+type literatureEchoPlan struct {
+	hashID string
+	fields []string
+}
+
+// isSiteFormCardSource is the site-form rule on an already-decoded source: not
+// an anchor source, and shaped like a host.
+func isSiteFormCardSource(source string) bool {
+	return !literatureEchoAnchorSources[source] && strings.Contains(source, ".")
+}
+
+// isSiteFormCardLocator reports whether a bare CID is a card locator (0x1007)
+// whose source is a site (`mangadex.org`), not an anchor (`anilist`).
+func isSiteFormCardLocator(cidStr string) bool {
+	source, _, err := cid.DecodeCard(cidStr)
+	return err == nil && isSiteFormCardSource(source)
+}
+
+// isSiteCard reports whether a card's field names prove it a site card: at
+// least one site-form `cids/` card locator and no card locator of another form.
+// Non-card members (a content CID) say nothing about which work the card is and
+// are ignored. An empty field list — a record with no field-name index — is
+// never a site card, which keeps its `anilistid`; see planLiteratureEchoes.
+func isSiteCard(fields []string) bool {
+	site := false
+	for _, f := range fields {
+		member := cidFromKeysetField(f)
+		if member == "" {
+			continue
+		}
+		source, _, err := cid.DecodeCard(member)
+		if err != nil {
+			continue
+		}
+		if !isSiteFormCardSource(source) {
+			return false
+		}
+		site = true
+	}
+	return site
+}
+
+// SweepLiteratureEchoes removes query echoes stored on literature cards (see
+// the file comment). With apply=false it only plans.
+//
+// Reads in batched passes (never a per-record SCAN — a single record read
+// costs a full keyspace sweep on this store), then deletes only the planned
+// fields, through DeleteProperty so the field-name index and the search-dirty
+// set stay consistent.
+func (c *Client) SweepLiteratureEchoes(ctx context.Context, apply bool) (LiteratureEchoSweep, error) {
+	res := LiteratureEchoSweep{ByKey: map[string]int{}}
+
+	plans, err := c.planLiteratureEchoes(ctx)
+	if err != nil {
+		return res, err
+	}
+	res.Planned = len(plans)
+
+	if !apply {
+		for _, p := range plans {
+			for _, f := range p.fields {
+				res.ByKey[f]++
+			}
+		}
+		return res, nil
+	}
+
+	// Writes take the client lock themselves, so they happen after the read
+	// phase has released it. ByKey counts what was actually deleted, so a
+	// failure midway reports the true partial state.
+	for _, p := range plans {
+		// DeleteProperty carries its own per-call timeout; checking ctx here is
+		// what makes the wrapper's deadline bound the whole write phase.
+		if err := ctx.Err(); err != nil {
+			return res, err
+		}
+		for _, f := range p.fields {
+			if err := c.DeleteProperty(p.hashID, f); err != nil {
+				return res, fmt.Errorf("delete %s on %s: %w", f, p.hashID, err)
+			}
+			res.ByKey[f]++
+		}
+		res.Fixed++
+	}
+	return res, nil
+}
+
+// planLiteratureEchoes is the read half: what would be deleted, without
+// deleting it.
+func (c *Client) planLiteratureEchoes(ctx context.Context) ([]literatureEchoPlan, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	if c.client == nil {
+		return nil, fmt.Errorf("not connected")
+	}
+
+	hashIDs, err := c.client.SMembers(ctx, c.buildIndexKey()).Result()
+	if err != nil {
+		return nil, fmt.Errorf("smembers index: %w", err)
+	}
+	if len(hashIDs) == 0 {
+		return nil, nil
+	}
+
+	// Pass 1: the gate and echo fields for every record, in a fixed order so
+	// the MGET response decodes by position: keys[i*per+n] belongs to
+	// hashIDs[i].
+	per := len(literatureEchoReadFields)
+	keys := make([]string, 0, len(hashIDs)*per)
+	for _, h := range hashIDs {
+		prefix := c.buildKeyPrefix(h)
+		for _, f := range literatureEchoReadFields {
+			keys = append(keys, prefix+f)
+		}
+	}
+
+	// Chunked so one MGET never carries the whole corpus.
+	const chunk = 3000
+	values := make([]interface{}, 0, len(keys))
+	for start := 0; start < len(keys); start += chunk {
+		end := start + chunk
+		if end > len(keys) {
+			end = len(keys)
+		}
+		got, err := c.client.MGet(ctx, keys[start:end]...).Result()
+		if err != nil {
+			return nil, fmt.Errorf("mget: %w", err)
+		}
+		values = append(values, got...)
+	}
+
+	// A nil slot is an absent key. A present key holding "" is still an echo
+	// and still goes, so presence — not the value — decides.
+	get := func(i int) (string, bool) {
+		if i >= len(values) || values[i] == nil {
+			return "", false
+		}
+		s, ok := values[i].(string)
+		return s, ok
+	}
+
+	type candidate struct {
+		hashID     string
+		hasAnilist bool
+		echoes     []string
+	}
+	var cands []candidate
+	var probe []string // candidates whose anilistid needs their cids/ key-set
+	for i, h := range hashIDs {
+		base := i * per
+		domain, _ := get(base)
+		fileType, _ := get(base + 1)
+		if domain != "literature" || fileType != "card" {
+			continue
+		}
+		cand := candidate{hashID: h}
+		_, cand.hasAnilist = get(base + 2)
+		for j, f := range literatureCardEchoFields {
+			if _, ok := get(base + 3 + j); ok {
+				cand.echoes = append(cand.echoes, f)
+			}
+		}
+		if !cand.hasAnilist && len(cand.echoes) == 0 {
+			continue
+		}
+		if cand.hasAnilist {
+			probe = append(probe, h)
+		}
+		cands = append(cands, cand)
+	}
+
+	// Pass 2: the field-name index of only the cards carrying an `anilistid`
+	// — the `cids/` members are field NAMES, so the index is where they are
+	// listed. Pipelined per chunk; still no SCAN.
+	//
+	// A card with no index (never touched since before the index existed, and
+	// missed by the startup WarmFieldIndexes pass) reads as an empty set and
+	// keeps its `anilistid`: proving it a site card would need exactly the
+	// per-record SCAN this sweep refuses, and a stamp left in place is the
+	// safe direction.
+	siteCard := make(map[string]bool, len(probe))
+	for start := 0; start < len(probe); start += mgetBatch {
+		end := start + mgetBatch
+		if end > len(probe) {
+			end = len(probe)
+		}
+		pipe := c.client.Pipeline()
+		cmds := make([]*redis.StringSliceCmd, 0, end-start)
+		for _, h := range probe[start:end] {
+			cmds = append(cmds, pipe.SMembers(ctx, c.buildFieldsKey(h)))
+		}
+		if _, err := pipe.Exec(ctx); err != nil && err != redis.Nil {
+			return nil, fmt.Errorf("smembers fields: %w", err)
+		}
+		for k, cmd := range cmds {
+			fields, err := cmd.Result()
+			if err != nil && err != redis.Nil {
+				return nil, fmt.Errorf("smembers fields %s: %w", probe[start+k], err)
+			}
+			siteCard[probe[start+k]] = isSiteCard(fields)
+		}
+	}
+
+	plans := make([]literatureEchoPlan, 0, len(cands))
+	for _, cand := range cands {
+		var fields []string
+		if cand.hasAnilist && siteCard[cand.hashID] {
+			fields = append(fields, "anilistid")
+		}
+		fields = append(fields, cand.echoes...)
+		if len(fields) > 0 {
+			plans = append(plans, literatureEchoPlan{hashID: cand.hashID, fields: fields})
+		}
+	}
+	return plans, nil
+}
+
+// SweepLiteratureEchoesWithTimeout is the convenience wrapper used by the API
+// handler. Generous timeout: the read phase is batched, but the write phase is
+// one round-trip per deleted field.
+func (c *Client) SweepLiteratureEchoesWithTimeout(apply bool) (LiteratureEchoSweep, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
+	defer cancel()
+	return c.SweepLiteratureEchoes(ctx, apply)
+}

internal/storage/literature_echo_sweep_test.go

@@ -0,0 +1,249 @@
+package storage
+
+import (
+	"context"
+	"encoding/base32"
+	"encoding/binary"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+// Pinned from meta-read `src/card_cid.rs` — literals, not round-trips.
+const (
+	literatureAnchorCardCID = "bagdsaaaua5qw42lmnfzxi3lbnztwcorrga2tonzy" // ("anilist", "manga:105778")
+	literatureSiteCardCID   = "bagdsaabybrwwc3thmfsgk6bon5zgol3nmfxgoyjpmiygenzsgftgmlldgm4dqljugq4dmllbmeygmlldgjrdaytcgmzdcnjrgi"
+	// A non-card CID a card may also carry.
+	literatureContentCID = "bagacbabaec7v3fu2ygzh3e2sybg3fbzmisry2hbtpmck6vx3yftea6vzq35r4"
+)
+
+// mintCardCID encodes a card locator the way meta-feeder-sdk
+// `compute_card_cid` does. Asserted against the pinned literals before use.
+func mintCardCID(source, id string) string {
+	digest := binary.AppendUvarint(nil, uint64(len(source)))
+	digest = append(digest, source...)
+	digest = append(digest, id...)
+	wire := []byte{0x01}
+	wire = binary.AppendUvarint(wire, 0x1007)
+	wire = append(wire, 0x00)
+	wire = binary.AppendUvarint(wire, uint64(len(digest)))
+	wire = append(wire, digest...)
+	return "b" + strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(wire))
+}
+
+func TestIsSiteFormCardLocator(t *testing.T) {
+	if got := mintCardCID("anilist", "manga:105778"); got != literatureAnchorCardCID {
+		t.Fatalf("test encoder drifted: %s != %s", got, literatureAnchorCardCID)
+	}
+	if got := mintCardCID("mangadex.org", "/manga/b0b721ff-c388-4486-aa0f-c2b0bb321512"); got != literatureSiteCardCID {
+		t.Fatalf("test encoder drifted: %s != %s", got, literatureSiteCardCID)
+	}
+
+	for _, tc := range []struct {
+		name string
+		cid  string
+		want bool
+	}{
+		{"anilist anchor (golden)", literatureAnchorCardCID, false},
+		{"mangadex site (golden)", literatureSiteCardCID, true},
+		{"openlibrary anchor", mintCardCID("openlibrary", "OL45804W"), false},
+		{"another site", mintCardCID("weebcentral.com", "/series/01J76XY/chainsaw-man"), true},
+		// Not an anchor source, but not host-shaped either: meta-read reads it
+		// as an id-bag key, so it must not be treated as a site.
+		{"dotless unknown source", mintCardCID("suwayomi", "42"), false},
+		{"content CID", literatureContentCID, false},
+		{"title key", "title:chainsaw man", false},
+	} {
+		if got := isSiteFormCardLocator(tc.cid); got != tc.want {
+			t.Errorf("%s: isSiteFormCardLocator = %v, want %v", tc.name, got, tc.want)
+		}
+	}
+}
+
+// seedLiteratureCorpus writes one record per rule the sweep has to honour.
+func seedLiteratureCorpus(t *testing.T, c *Client) {
+	t.Helper()
+
+	// A site card carrying every echo: all of it goes.
+	seedRecord(t, c, "site-card", map[string]string{
+		"domain":                        "literature",
+		"fileType":                      "card",
+		"title":                         "Chainsaw Man",
+		"cids/" + literatureSiteCardCID: "true",
+		"anilistid":                     "105778",
+		"workcid":                       literatureSiteCardCID,
+		"chapterNumber":                 "12",
+		"volumeNumber":                  "2",
+		"chapterStart":                  "1",
+		"chapterEnd":                    "20",
+	})
+	// An anchor card: its anilistid is its identity and stays; the position
+	// echo and the workcid still go.
+	seedRecord(t, c, "anchor-card", map[string]string{
+		"domain":                          "literature",
+		"fileType":                        "card",
+		"title":                           "Chainsaw Man",
+		"cids/" + literatureAnchorCardCID: "true",
+		"anilistid":                       "105778",
+		"workcid":                         literatureAnchorCardCID,
+		"chapterNumber":                   "3",
+	})
+	// A site card that is already clean: nothing planned.
+	seedRecord(t, c, "clean-site-card", map[string]string{
+		"domain":                        "literature",
+		"fileType":                      "card",
+		"cids/" + literatureSiteCardCID: "true",
+	})
+	// Both locator forms: which one the id belongs to is unknowable, so it
+	// stays.
+	seedRecord(t, c, "merged-card", map[string]string{
+		"domain":                          "literature",
+		"fileType":                        "card",
+		"cids/" + literatureSiteCardCID:   "true",
+		"cids/" + literatureAnchorCardCID: "true",
+		"anilistid":                       "105778",
+	})
+	// No card locator at all: not provably a site card.
+	seedRecord(t, c, "unlocated-card", map[string]string{
+		"domain":                       "literature",
+		"fileType":                     "card",
+		"cids/" + literatureContentCID: "true",
+		"anilistid":                    "105778",
+	})
+	// A stored empty value is still an echo.
+	seedRecord(t, c, "blank-echo-card", map[string]string{
+		"domain":       "literature",
+		"fileType":     "card",
+		"volumeNumber": "",
+	})
+	// A chapter is never swept, whatever it carries.
+	seedRecord(t, c, "chapter", map[string]string{
+		"domain":                       "literature",
+		"fileType":                     "archive",
+		"cids/" + literatureContentCID: "true",
+		"anilistid":                    "105778",
+		"workcid":                      literatureSiteCardCID,
+		"chapterNumber":                "12",
+	})
+	// Not literature: out of scope even though it is a card.
+	seedRecord(t, c, "screen-card", map[string]string{
+		"domain":        "screen",
+		"fileType":      "card",
+		"workcid":       literatureSiteCardCID,
+		"chapterNumber": "1",
+	})
+}
+
+func TestPlanLiteratureEchoes_ScopesToLiteratureCards(t *testing.T) {
+	c, _ := newTestClient(t, "")
+	seedLiteratureCorpus(t, c)
+
+	plans, err := c.planLiteratureEchoes(context.Background())
+	if err != nil {
+		t.Fatalf("plan: %v", err)
+	}
+	got := map[string][]string{}
+	for _, p := range plans {
+		got[p.hashID] = p.fields
+	}
+	want := map[string][]string{
+		"site-card":       {"anilistid", "workcid", "chapterNumber", "volumeNumber", "chapterStart", "chapterEnd"},
+		"anchor-card":     {"workcid", "chapterNumber"},
+		"blank-echo-card": {"volumeNumber"},
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("plan =\n  %v\nwant\n  %v", got, want)
+	}
+}
+
+func TestSweepLiteratureEchoes_DryRunThenApply(t *testing.T) {
+	c, s := newTestClient(t, "mm:")
+	seedLiteratureCorpus(t, c)
+
+	wantByKey := map[string]int{
+		"anilistid":     1,
+		"workcid":       2,
+		"chapterNumber": 2,
+		"volumeNumber":  2,
+		"chapterStart":  1,
+		"chapterEnd":    1,
+	}
+
+	// Dry run: counts, and nothing written.
+	dry, err := c.SweepLiteratureEchoes(context.Background(), false)
+	if err != nil {
+		t.Fatalf("dry run: %v", err)
+	}
+	if dry.Planned != 3 || dry.Fixed != 0 {
+		t.Errorf("dry run planned/fixed = %d/%d, want 3/0", dry.Planned, dry.Fixed)
+	}
+	if !reflect.DeepEqual(dry.ByKey, wantByKey) {
+		t.Errorf("dry run byKey = %v, want %v", dry.ByKey, wantByKey)
+	}
+	if m, _ := c.GetMetadataFlat("site-card"); m["anilistid"] != "105778" {
+		t.Fatalf("dry run deleted anilistid: %v", m)
+	}
+
+	applied, err := c.SweepLiteratureEchoes(context.Background(), true)
+	if err != nil {
+		t.Fatalf("apply: %v", err)
+	}
+	if applied.Planned != 3 || applied.Fixed != 3 {
+		t.Errorf("apply planned/fixed = %d/%d, want 3/3", applied.Planned, applied.Fixed)
+	}
+	if !reflect.DeepEqual(applied.ByKey, wantByKey) {
+		t.Errorf("apply byKey = %v, want %v", applied.ByKey, wantByKey)
+	}
+
+	site, err := c.GetMetadataFlat("site-card")
+	if err != nil {
+		t.Fatalf("read site-card: %v", err)
+	}
+	for _, k := range []string{"anilistid", "workcid", "chapterNumber", "volumeNumber", "chapterStart", "chapterEnd"} {
+		if _, ok := site[k]; ok {
+			t.Errorf("site-card still carries %s", k)
+		}
+		// The flat key and its field-name index entry must both be gone, or the
+		// next read would MGET a ghost field.
+		if s.Exists("mm:file:site-card/" + k) {
+			t.Errorf("site-card flat key %s survived", k)
+		}
+		if member, _ := s.IsMember("mm:file:site-card/"+FieldsField, k); member {
+			t.Errorf("site-card field index still lists %s", k)
+		}
+	}
+	if site["title"] != "Chainsaw Man" || site["cids/"+literatureSiteCardCID] != "true" {
+		t.Errorf("site-card lost fields it should keep: %v", site)
+	}
+
+	anchor, _ := c.GetMetadataFlat("anchor-card")
+	if anchor["anilistid"] != "105778" {
+		t.Errorf("anchor-card anilistid = %q, want kept", anchor["anilistid"])
+	}
+	if _, ok := anchor["workcid"]; ok {
+		t.Errorf("anchor-card still carries workcid")
+	}
+
+	chapter, _ := c.GetMetadataFlat("chapter")
+	if chapter["anilistid"] != "105778" || chapter["workcid"] != literatureSiteCardCID || chapter["chapterNumber"] != "12" {
+		t.Errorf("chapter was touched: %v", chapter)
+	}
+	screen, _ := c.GetMetadataFlat("screen-card")
+	if screen["workcid"] != literatureSiteCardCID {
+		t.Errorf("screen-card was touched: %v", screen)
+	}
+	merged, _ := c.GetMetadataFlat("merged-card")
+	if merged["anilistid"] != "105778" {
+		t.Errorf("merged-card anilistid = %q, want kept", merged["anilistid"])
+	}
+
+	// Idempotent: the endpoint is called by hand, and a second call must find
+	// nothing — with byKey still an empty map, not nil.
+	again, err := c.SweepLiteratureEchoes(context.Background(), true)
+	if err != nil {
+		t.Fatalf("second pass: %v", err)
+	}
+	if again.Planned != 0 || again.Fixed != 0 || again.ByKey == nil || len(again.ByKey) != 0 {
+		t.Errorf("second pass = %+v, want nothing planned and an empty byKey", again)
+	}
+}

internal/storage/search_dirty.go

@@ -0,0 +1,214 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"strings"
+	"time"
+)
+
+// Incremental search-index maintenance.
+//
+// # Why this exists
+//
+// The API's in-memory search index used to be rebuilt in full every 60 s:
+// GetSearchIndexTuples does a whole-keyspace SCAN plus batched MGETs, builds an
+// intermediate map of every record, then allocates a fresh []SearchTuple — all
+// while the previous index is still live serving reads. That cost is O(corpus),
+// not O(change): adding 200 records in a minute re-read every one of the other
+// ~620 000 field keys to notice them.
+//
+// Measured on a 19 622-record staging corpus (2026-08-30): SCAN was 57.8 % and
+// MGET 19.1 % of ALL Redis CPU, and ~90 % of the MGET calls plus ~48 % of the
+// SCAN calls were this loop — roughly 45 % of Redis's total work, spent
+// continuously whether or not anything changed. The Go side held 783 MB for
+// data Redis stores in 135 MB, largely because of the double copy.
+//
+// That was tolerable while the corpus was small and mostly static. It stops
+// being tolerable now that meta-core is the platform's permanent, additive
+// store: every search hit on every gateway is retained forever, so the rebuild
+// cost is a ratchet.
+//
+// # The shape
+//
+// Every write path marks its record in a dirty set; the refresh drains that set
+// and re-reads only those records. Cost becomes O(change).
+//
+// Marking rides along with the field-index maintenance every write path already
+// performs, so there is no new bookkeeping discipline to remember — if you are
+// adding a write path, the same line that keeps `__fields__` correct keeps this
+// correct.
+//
+// # Correctness
+//
+// The drain is a destructive SPOP, so a write that lands *during* a refresh is
+// either already in the drained batch or still in the set for the next tick —
+// never dropped. The cost of the race is one tick of staleness, which is well
+// inside the interval the index was already stale by.
+//
+// A refresh that fails after draining WOULD lose its batch, so drainDirty's
+// caller re-marks on error (see MarkDirty / the API's refreshIncremental).
+//
+// ⚠ **The destructive drain assumes ONE index consumer per Redis.** Two
+// meta-core processes sharing a keyspace would steal each other's marks, and
+// each would end up with an index missing whatever the other drained. That is
+// fine today — multi-instance leader election was deprecated and the dev/prod
+// stacks run a single meta-core — but it is the assumption to revisit first if
+// a second instance is ever pointed at the same Redis. A per-consumer dirty key
+// (`file:__dirty__:<consumer>`) is the shape that would fix it.
+//
+// The API's periodic full reconcile (searchIndexReconcileEvery) is the backstop
+// for both this and for a write path that forgets to mark: a missed mark costs
+// staleness until the next reconcile, never permanent divergence.
+
+// dirtyField is the reserved key holding the set of record ids whose metadata
+// changed since the last search-index refresh. It sits under the same `file:`
+// namespace as `__index__` and is likewise not a record.
+const dirtyField = "file:__dirty__"
+
+// dirtyDrainCap bounds one incremental refresh. A burst larger than this is
+// drained over several ticks rather than in one long-held pass — the point of
+// the change is to keep any single refresh cheap. Sized well above a normal
+// search-persist flush (which writes at most a few hundred records).
+const dirtyDrainCap = 5000
+
+// buildDirtyKey constructs the key for the dirty-record set.
+func (c *Client) buildDirtyKey() string {
+	return c.buildKey(dirtyField)
+}
+
+// markDirtyLocked records that hashID's metadata changed. Caller must hold c.mu.
+//
+// Best-effort by design: a failure here costs one record its incremental
+// refresh, and the periodic full rebuild still repairs it. Failing the caller's
+// write instead would trade a stale index entry for a lost write, which is
+// strictly worse.
+func (c *Client) markDirtyLocked(ctx context.Context, hashID string) {
+	if hashID == "" {
+		return
+	}
+	if err := c.client.SAdd(ctx, c.buildDirtyKey(), hashID).Err(); err != nil {
+		log.Printf("[STORAGE] dirty-mark failed for %s: %v (index repairs on the next full rebuild)", hashID, err)
+	}
+}
+
+// MarkDirty records that hashID needs re-indexing. Exported for the API layer,
+// which re-marks a batch whose refresh failed after the drain.
+func (c *Client) MarkDirty(hashIDs ...string) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	if c.client == nil || len(hashIDs) == 0 {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	members := make([]interface{}, 0, len(hashIDs))
+	for _, h := range hashIDs {
+		if h != "" {
+			members = append(members, h)
+		}
+	}
+	if len(members) == 0 {
+		return
+	}
+	if err := c.client.SAdd(ctx, c.buildDirtyKey(), members...).Err(); err != nil {
+		log.Printf("[STORAGE] dirty re-mark failed for %d records: %v", len(members), err)
+	}
+}
+
+// DrainDirty removes and returns up to dirtyDrainCap record ids from the dirty
+// set. `more` reports whether entries remain, so the caller can schedule
+// another pass instead of waiting a full interval to catch up on a burst.
+//
+// Destructive on purpose: SPOP makes concurrent refreshes disjoint and stops a
+// record being re-read every tick forever. The caller owns the drained ids
+// until it has indexed them — on failure it must MarkDirty them again.
+func (c *Client) DrainDirty() (ids []string, more bool, err error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	if c.client == nil {
+		return nil, false, fmt.Errorf("not connected")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	key := c.buildDirtyKey()
+	ids, err = c.client.SPopN(ctx, key, dirtyDrainCap).Result()
+	if err != nil {
+		return nil, false, fmt.Errorf("spop dirty: %w", err)
+	}
+	if len(ids) < dirtyDrainCap {
+		return ids, false, nil
+	}
+	remaining, err := c.client.SCard(ctx, key).Result()
+	if err != nil {
+		// We have the batch; not knowing the remainder just means the next
+		// tick picks it up on schedule.
+		return ids, false, nil
+	}
+	return ids, remaining > 0, nil
+}
+
+// GetSearchIndexTuplesFor reads the search tuples for a specific set of records
+// — the incremental counterpart to GetSearchIndexTuples.
+//
+// Reads go through the per-record field index (GetMetadataFlat), so this is two
+// round-trips per record rather than a keyspace scan. That trade is what makes
+// it O(change): it beats the full pass whenever the changed set is a small
+// fraction of the corpus, and the caller bounds the batch (dirtyDrainCap).
+//
+// A record present in the dirty set but absent from storage was deleted; it is
+// reported in `gone` so the caller can evict it from the index rather than
+// leaving a tombstone that still answers searches.
+func (c *Client) GetSearchIndexTuplesFor(hashIDs []string) (tuples []SearchTuple, gone []string, err error) {
+	if len(hashIDs) == 0 {
+		return nil, nil, nil
+	}
+
+	excluded := indexExcludePrefixes()
+	tuples = make([]SearchTuple, 0, len(hashIDs))
+
+	for _, hashID := range hashIDs {
+		fields, ferr := c.GetMetadataFlat(hashID)
+		if ferr != nil {
+			// Re-mark so the record is retried rather than silently missing
+			// from the index until the next full rebuild.
+			c.MarkDirty(hashID)
+			log.Printf("[STORAGE] incremental index read failed for %s: %v", hashID, ferr)
+			continue
+		}
+		if len(fields) == 0 {
+			gone = append(gone, hashID)
+			continue
+		}
+		for field := range fields {
+			if hasAnyPrefix(field, excluded) {
+				delete(fields, field)
+			}
+		}
+		tuples = append(tuples, SearchTuple{
+			HashID:   hashID,
+			Haystack: buildHaystack(fields),
+			Fields:   fields,
+		})
+	}
+	return tuples, gone, nil
+}
+
+// buildHaystack folds a record's searchable text fields into one lowercased
+// blob. Shared by the full and incremental index reads so the two can never
+// disagree about what is searchable — a drift there would make a record's
+// matchability depend on which path happened to index it.
+func buildHaystack(fields map[string]string) string {
+	var sb strings.Builder
+	for _, f := range searchIndexFields {
+		if v, ok := fields[f]; ok && v != "" {
+			sb.WriteString(strings.ToLower(v))
+			sb.WriteByte('\n')
+		}
+	}
+	return sb.String()
+}

internal/storage/search_dirty_test.go

@@ -0,0 +1,211 @@
+package storage
+
+import (
+	"context"
+	"sort"
+	"testing"
+)
+
+// Tests for incremental search-index maintenance (search_dirty.go).
+//
+// The dirty set is derived state on the hot write path, which makes its failure
+// mode the dangerous kind: a MISSING mark doesn't error, it just leaves the
+// search index quietly serving a stale copy of that record until someone
+// triggers a full rebuild. So the property worth pinning is coverage — every
+// path that changes what a search would match must mark — and that is what the
+// bulk of these tests assert, one write path at a time.
+//
+// The second property is that a delete is a mark too, not an unmark: the
+// refresh has to LEARN the record is gone in order to evict it. Marking on
+// delete is counter-intuitive enough to be worth a test of its own.
+
+func dirtyMembers(t *testing.T, c *Client) []string {
+	t.Helper()
+	got, err := c.client.SMembers(context.Background(), c.buildDirtyKey()).Result()
+	if err != nil {
+		t.Fatalf("smembers dirty: %v", err)
+	}
+	sort.Strings(got)
+	return got
+}
+
+func clearDirty(t *testing.T, c *Client) {
+	t.Helper()
+	if err := c.client.Del(context.Background(), c.buildDirtyKey()).Err(); err != nil {
+		t.Fatalf("del dirty: %v", err)
+	}
+}
+
+// Every write path must mark. A path that doesn't leaves the index serving a
+// stale record with no error anywhere.
+func TestDirty_EveryWritePathMarks(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		cases := []struct {
+			name  string
+			id    string
+			write func(id string) error
+		}{
+			{"SetMetadataFlat", "r-set", func(id string) error {
+				return c.SetMetadataFlat(id, map[string]string{"title": "A"})
+			}},
+			{"SetProperty", "r-prop", func(id string) error {
+				return c.SetProperty(id, "title", "B")
+			}},
+			{"MergeMetadataFlat", "r-merge", func(id string) error {
+				_, err := c.MergeMetadataFlat(id, map[string]string{"title": "C"})
+				return err
+			}},
+			{"AddToSet", "r-add", func(id string) error {
+				_, err := c.AddToSet(id, "tags", "x")
+				return err
+			}},
+		}
+		for _, tc := range cases {
+			t.Run(tc.name, func(t *testing.T) {
+				clearDirty(t, c)
+				if err := tc.write(tc.id); err != nil {
+					t.Fatalf("%s: %v", tc.name, err)
+				}
+				got := dirtyMembers(t, c)
+				if len(got) != 1 || got[0] != tc.id {
+					t.Fatalf("%s must mark %q dirty, got %v", tc.name, tc.id, got)
+				}
+			})
+		}
+	})
+}
+
+// RemoveFromSet's non-empty branch changes a field's VALUE without changing the
+// field-name index, so it is the one path that a mark riding on field-index
+// maintenance would have missed. Pinned explicitly for that reason.
+func TestDirty_RemoveFromSetMarksEvenWhenTheFieldSurvives(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		const id = "r-rem"
+		for _, v := range []string{"x", "y"} {
+			if _, err := c.AddToSet(id, "tags", v); err != nil {
+				t.Fatalf("AddToSet: %v", err)
+			}
+		}
+		clearDirty(t, c)
+
+		if _, err := c.RemoveFromSet(id, "tags", "x"); err != nil {
+			t.Fatalf("RemoveFromSet: %v", err)
+		}
+		// `tags` still exists (y remains) — no field-index change, but the
+		// searchable value changed.
+		if got := dirtyMembers(t, c); len(got) != 1 || got[0] != id {
+			t.Fatalf("a value-only change must still mark dirty, got %v", got)
+		}
+	})
+}
+
+// A delete MARKS rather than unmarks: the refresh must read the record, find it
+// gone, and evict it. Unmarking would leave a deleted record answering searches
+// until the next full rebuild.
+func TestDirty_DeletesMarkSoTheRefreshCanEvict(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		const id = "r-del"
+		if err := c.SetMetadataFlat(id, map[string]string{"title": "Doomed"}); err != nil {
+			t.Fatalf("SetMetadataFlat: %v", err)
+		}
+		clearDirty(t, c)
+
+		if _, err := c.DeleteMetadata(id); err != nil {
+			t.Fatalf("DeleteMetadata: %v", err)
+		}
+		if got := dirtyMembers(t, c); len(got) != 1 || got[0] != id {
+			t.Fatalf("a delete must mark dirty so the index evicts it, got %v", got)
+		}
+
+		// …and the incremental read reports it as gone rather than as an error
+		// or an empty-but-present tuple.
+		tuples, gone, err := c.GetSearchIndexTuplesFor([]string{id})
+		if err != nil {
+			t.Fatalf("GetSearchIndexTuplesFor: %v", err)
+		}
+		if len(tuples) != 0 {
+			t.Fatalf("a deleted record must produce no tuple, got %d", len(tuples))
+		}
+		if len(gone) != 1 || gone[0] != id {
+			t.Fatalf("a deleted record must be reported gone, got %v", gone)
+		}
+	})
+}
+
+// The drain is destructive so concurrent refreshes stay disjoint and a record
+// isn't re-read every tick forever. The caller owns what it drained.
+func TestDirty_DrainIsDestructive(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		clearDirty(t, c)
+		c.MarkDirty("a", "b", "c")
+
+		ids, more, err := c.DrainDirty()
+		if err != nil {
+			t.Fatalf("DrainDirty: %v", err)
+		}
+		sort.Strings(ids)
+		if len(ids) != 3 {
+			t.Fatalf("expected 3 drained ids, got %v", ids)
+		}
+		if more {
+			t.Fatal("a fully-drained set must not report more")
+		}
+		if got := dirtyMembers(t, c); len(got) != 0 {
+			t.Fatalf("drain must empty the set, got %v", got)
+		}
+
+		// A second drain is empty, not an error — the steady state when nothing
+		// has been written since the last tick.
+		ids, _, err = c.DrainDirty()
+		if err != nil || len(ids) != 0 {
+			t.Fatalf("draining an empty set must be a clean no-op, got %v / %v", ids, err)
+		}
+	})
+}
+
+// The incremental read must produce tuples indistinguishable from the full
+// rebuild's — same haystack, same fields, same exclusions. A drift here would
+// make a record's matchability depend on which path happened to index it, which
+// is invisible until someone searches for the one term that differs.
+func TestDirty_IncrementalTuplesMatchTheFullRebuild(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		const id = "r-parity"
+		if err := c.SetMetadataFlat(id, map[string]string{
+			"title":                   "Parity Release",
+			"fileName":                "parity.s01e01.mkv",
+			"fileType":                "video",
+			"categories/XXX":          "true",
+			"categories/newznab/5070": "true",
+		}); err != nil {
+			t.Fatalf("SetMetadataFlat: %v", err)
+		}
+
+		full := tuplesByHashID(t, c)[id]
+		incr, gone, err := c.GetSearchIndexTuplesFor([]string{id})
+		if err != nil {
+			t.Fatalf("GetSearchIndexTuplesFor: %v", err)
+		}
+		if len(gone) != 0 || len(incr) != 1 {
+			t.Fatalf("expected exactly one live tuple, got %d live / %d gone", len(incr), len(gone))
+		}
+
+		if incr[0].Haystack != full.Haystack {
+			t.Errorf("haystack drift:\n incremental %q\n full        %q", incr[0].Haystack, full.Haystack)
+		}
+		if len(incr[0].Fields) != len(full.Fields) {
+			t.Errorf("field-count drift: incremental %d, full %d", len(incr[0].Fields), len(full.Fields))
+		}
+		for k, v := range full.Fields {
+			if incr[0].Fields[k] != v {
+				t.Errorf("field %q drift: incremental %q, full %q", k, incr[0].Fields[k], v)
+			}
+		}
+		// The index projection must apply on both paths.
+		if _, ok := incr[0].Fields["categories/newznab/5070"]; ok {
+			t.Error("the incremental path must honour the index exclusion list too")
+		}
+		if _, ok := incr[0].Fields["categories/XXX"]; !ok {
+			t.Error("the incremental path must keep the adult-filter category")
+		}
+	})
+}

internal/storage/search_index_exclude_test.go

@@ -0,0 +1,116 @@
+package storage
+
+import (
+	"testing"
+)
+
+// Tests for the search-index field projection (GetSearchIndexTuples +
+// indexExcludePrefixes).
+//
+// The projection narrows the in-memory index only — Redis keeps every field and
+// every per-record read still returns it. What has to be pinned is the boundary
+// between the two, because getting it wrong in either direction is silent:
+// exclude too little and the index keeps paying for fields nothing reads;
+// exclude too much and a record served *from the index* comes back missing a
+// field a downstream consumer depends on.
+//
+// The specific hazard the default guards is adult-content filtering:
+// `categories/XXX` is the server-side enforcement point (meta-search's
+// `query_eval::fields_are_adult`, meta-gateway's `cards::fields_are_adult`),
+// and `exclude_adult` defaults to ON. Excluding the whole `categories/` family
+// would make every index-served record read as non-adult and the filter would
+// fail OPEN — so the default must keep `categories/XXX` while dropping the
+// reader-less `categories/newznab/` mirror.
+
+func tuplesByHashID(t *testing.T, c *Client) map[string]SearchTuple {
+	t.Helper()
+	got, err := c.GetSearchIndexTuples()
+	if err != nil {
+		t.Fatalf("GetSearchIndexTuples: %v", err)
+	}
+	out := make(map[string]SearchTuple, len(got))
+	for _, tu := range got {
+		out[tu.HashID] = tu
+	}
+	return out
+}
+
+func TestSearchIndex_ExcludesNewznabCategoriesButKeepsTheRest(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		const id = "root-1"
+		if err := c.SetMetadataFlat(id, map[string]string{
+			"title":                   "Some Release",
+			"fileType":                "video",
+			"categories/TV/Anime":     "true",
+			"categories/XXX":          "true",
+			"categories/newznab/5070": "true",
+			"categories/newznab/2040": "true",
+		}); err != nil {
+			t.Fatalf("SetMetadataFlat: %v", err)
+		}
+
+		fields := tuplesByHashID(t, c)[id].Fields
+		if fields == nil {
+			t.Fatal("record missing from the search index")
+		}
+
+		for _, excluded := range []string{"categories/newznab/5070", "categories/newznab/2040"} {
+			if _, ok := fields[excluded]; ok {
+				t.Errorf("%s must not be indexed — it is a reader-less mirror vocabulary", excluded)
+			}
+		}
+		// The load-bearing ones must survive. XXX especially: without it the
+		// adult filter fails open on every index-served record.
+		for _, kept := range []string{"categories/XXX", "categories/TV/Anime", "title", "fileType"} {
+			if _, ok := fields[kept]; !ok {
+				t.Errorf("%s must stay in the index", kept)
+			}
+		}
+
+		// The store itself is untouched — the projection is index-only.
+		stored, err := c.GetMetadataFlat(id)
+		if err != nil {
+			t.Fatalf("GetMetadataFlat: %v", err)
+		}
+		if _, ok := stored["categories/newznab/5070"]; !ok {
+			t.Error("an index-excluded field must still be stored and readable per-record")
+		}
+	})
+}
+
+func TestIndexExcludePrefixes_EnvOverride(t *testing.T) {
+	if got := indexExcludePrefixes(); len(got) != 1 || got[0] != "categories/newznab/" {
+		t.Fatalf("unset env must yield the default, got %v", got)
+	}
+
+	t.Setenv("META_CORE_INDEX_EXCLUDE_PREFIXES", "foo/, bar/ ,")
+	got := indexExcludePrefixes()
+	if len(got) != 2 || got[0] != "foo/" || got[1] != "bar/" {
+		t.Fatalf("override must trim and drop blanks, got %v", got)
+	}
+
+	// An explicitly empty value is the escape hatch: index everything.
+	t.Setenv("META_CORE_INDEX_EXCLUDE_PREFIXES", "")
+	if got := indexExcludePrefixes(); len(got) != 0 {
+		t.Fatalf("an empty override must disable exclusion, got %v", got)
+	}
+}
+
+func TestHasAnyPrefix(t *testing.T) {
+	prefixes := []string{"categories/newznab/", "tmp/"}
+	cases := map[string]bool{
+		"categories/newznab/5070": true,
+		"tmp/scratch":             true,
+		"categories/XXX":          false,
+		"categories/newznab":      false, // the family root is not the mirror
+		"title":                   false,
+	}
+	for field, want := range cases {
+		if got := hasAnyPrefix(field, prefixes); got != want {
+			t.Errorf("hasAnyPrefix(%q) = %v, want %v", field, got, want)
+		}
+	}
+	if hasAnyPrefix("anything", nil) {
+		t.Error("an empty prefix list must exclude nothing")
+	}
+}

internal/watcher/dispatcher.go

@@ -50,8 +50,18 @@ func (d *Dispatcher) dispatchToStream(event FileEvent) {
 	client := d.storageClient
 	d.mu.RUnlock()
 
+	if err := PublishEvent(client, event); err != nil {
+		log.Printf("[Dispatcher] Failed to publish to stream: %v", err)
+	}
+}
+
+// PublishEvent appends one file event to the file:events stream. Exported so a
+// caller with no Dispatcher (POST /api/files/register on a box whose watcher is
+// off) emits exactly the entry the watcher would — meta-share's ipfs_seed
+// consumes this stream to chunk, seed and provide new files.
+func PublishEvent(client *storage.Client, event FileEvent) error {
 	if client == nil || !client.IsConnected() {
-		return
+		return fmt.Errorf("storage client not connected")
 	}
 
 	// Build stream fields
@@ -79,9 +89,7 @@ func (d *Dispatcher) dispatchToStream(event FileEvent) {
 	// `EmitReset → ClearStream` path is still the only way the stream gets
 	// truncated to zero (signalled to consumers via the `reset` event).
 	_, err := client.XAdd(EventsStream, EventsStreamMaxLen, fields)
-	if err != nil {
-		log.Printf("[Dispatcher] Failed to publish to stream: %v", err)
-	}
+	return err
 }
 
 // EmitReset clears the stream and emits a reset event

internal/watcher/watcher.go

@@ -4,7 +4,6 @@ import (
 	"log"
 	"os"
 	"path/filepath"
-	"strconv"
 	"strings"
 	"sync"
 
@@ -54,42 +53,8 @@ func (w *Watcher) writeFileTuple(absPath string, size, mtimeNano int64, midhash
 		return
 	}
 
-	uuid, err := w.storage.GetByCID(midhash)
-	if err != nil {
-		log.Printf("[Watcher] GetByCID(%s): %v", midhash, err)
-		return
-	}
-
-	if uuid == "" {
-		// New content. Mint sets filePath/sizeByte/mtimeNano; SetProperty
-		// then writes the midhash as a cids/<cid> key-set member — and the
-		// cid_resolution.go hook registers the reverse-index alias for us.
-		uuid, err = w.storage.Mint(absPath, size, mtimeNano)
-		if err != nil {
-			log.Printf("[Watcher] Mint(%s): %v", absPath, err)
-			return
-		}
-		if err := w.storage.SetProperty(uuid, storage.CIDsKeyPrefix+midhash, "true"); err != nil {
-			log.Printf("[Watcher] SetProperty cids/%s for %s: %v", midhash, uuid, err)
-		}
-		return
-	}
-
-	// Known content. If this is a new physical path for the same content,
-	// record it as a duplicate. Otherwise refresh size/mtime — same content,
-	// same path, but the file may have been rewritten with a fresh mtime.
-	existing, _ := w.storage.GetProperty(uuid, "filePath")
-	if existing != absPath {
-		if _, err := w.storage.AddDuplicatePath(uuid, absPath); err != nil {
-			log.Printf("[Watcher] AddDuplicatePath %s @ %s: %v", uuid, absPath, err)
-		}
-		return
-	}
-	if _, err := w.storage.MergeMetadataFlat(uuid, map[string]string{
-		"sizeByte":  strconv.FormatInt(size, 10),
-		"mtimeNano": strconv.FormatInt(mtimeNano, 10),
-	}); err != nil {
-		log.Printf("[Watcher] refresh tuple for %s: %v", uuid, err)
+	if _, _, err := w.storage.RegisterFileTuple(absPath, size, mtimeNano, midhash); err != nil {
+		log.Printf("[Watcher] register %s: %v", absPath, err)
 	}
 }