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/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

@@ -422,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/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/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 {
+	return err
-		log.Printf("[Dispatcher] Failed to publish to stream: %v", 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 := w.storage.RegisterFileTuple(absPath, size, mtimeNano, midhash); err != nil {
-	if err != nil {
+		log.Printf("[Watcher] register %s: %v", absPath, err)
-		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)
 	}
 }