cid vector

README.md

@@ -209,6 +209,7 @@ the `/file/{cid}` reverse-lookup path.)
 
 ```bash
 GET    /api/metadata/hash-ids
+GET    /api/files/tuples
 GET    /api/metadata/list
 POST   /api/metadata/search
 POST   /api/metadata/batch
@@ -287,6 +288,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

@@ -301,6 +301,7 @@ CID-addressed (public, auth-bypassed at the perimeter)
 
 Editor / KV / schema / snapshot
   GET /api/metadata/hash-ids
+  GET /api/files/tuples
   GET /api/metadata/list
   POST /api/metadata/search
   POST /api/metadata/batch
@@ -339,6 +340,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
@@ -54,3 +55,47 @@ func (s *Server) handleMigrateDomainScreen(w http.ResponseWriter, r *http.Reques
 		"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/files_tuples.go

@@ -0,0 +1,74 @@
+package api
+
+import "net/http"
+
+// fileTupleJSON is one file-backed root on the wire.
+type fileTupleJSON struct {
+	HashID    string `json:"hashId"`
+	FilePath  string `json:"filePath"`
+	SizeByte  int64  `json:"sizeByte"`
+	MtimeNano int64  `json:"mtimeNano"`
+}
+
+// FileTuplesResponse is the body of GET /api/files/tuples.
+//
+// Files is a pointer so the summary form can omit it while the full form
+// always carries it — an empty library is `"files": []`, never an absent key,
+// which a client could mistake for a meta-core that lacks the rows.
+type FileTuplesResponse struct {
+	Files     *[]fileTupleJSON `json:"files,omitempty"`
+	Count     int              `json:"count"`
+	TotalSize int64            `json:"totalSize"`
+}
+
+// handleGetFileTuples handles GET /api/files/tuples.
+//
+// Lists the FILE-BACKED roots — records carrying filePath, sizeByte and
+// mtimeNano, i.e. what the watcher or /api/files/register minted — with their
+// sizes, plus the count and total. Gateway / cid-rooted records carry no file
+// and are not included; neither is a locator record that only merged a cache
+// file's metadata (it has no mtimeNano).
+//
+// It exists so a client never walks every hashId to answer "how many files,
+// how big": GetTuplesForAllFiles reads the three fields for every root in
+// chunked MGETs server-side (~250 round-trips for 82k roots, no SCAN), where
+// the client-side equivalent was one HTTP request per record.
+//
+// ?summary=1 omits the rows and returns only {count, totalSize}.
+func (s *Server) handleGetFileTuples(w http.ResponseWriter, r *http.Request) {
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+
+	tuples, err := s.storage.GetTuplesForAllFiles()
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	summary := r.URL.Query().Get("summary")
+	withRows := summary == "" || summary == "0" || summary == "false"
+
+	resp := FileTuplesResponse{Count: len(tuples)}
+	var rows []fileTupleJSON
+	if withRows {
+		rows = make([]fileTupleJSON, 0, len(tuples))
+	}
+	for _, t := range tuples {
+		resp.TotalSize += t.Size
+		if withRows {
+			rows = append(rows, fileTupleJSON{
+				HashID:    t.HashID,
+				FilePath:  t.FilePath,
+				SizeByte:  t.Size,
+				MtimeNano: t.MtimeNano,
+			})
+		}
+	}
+	if withRows {
+		resp.Files = &rows
+	}
+
+	writeJSON(w, http.StatusOK, resp)
+}

internal/api/files_tuples_test.go

@@ -0,0 +1,114 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"sort"
+	"testing"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/metazla/meta-core/internal/storage"
+)
+
+func newFileTuplesTestServer(t *testing.T) (*Server, *storage.Client) {
+	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() })
+	return &Server{storage: c}, c
+}
+
+func getFileTuples(t *testing.T, s *Server, query string) (int, FileTuplesResponse) {
+	t.Helper()
+	req := httptest.NewRequest(http.MethodGet, "/api/files/tuples"+query, nil)
+	rr := httptest.NewRecorder()
+	s.handleGetFileTuples(rr, req)
+	var resp FileTuplesResponse
+	_ = json.Unmarshal(rr.Body.Bytes(), &resp)
+	return rr.Code, resp
+}
+
+func mustMint(t *testing.T, c *storage.Client, path string, size, mtime int64) string {
+	t.Helper()
+	id, err := c.Mint(path, size, mtime)
+	if err != nil {
+		t.Fatalf("mint %s: %v", path, err)
+	}
+	return id
+}
+
+// Only records that are actually files count: a gateway record and a record
+// that merely carries a path + size (no mtimeNano) are both left out.
+func TestFileTuples_OnlyFileBackedRoots(t *testing.T) {
+	s, c := newFileTuplesTestServer(t)
+	a := mustMint(t, c, "/files/watch/a.mkv", 1000, 1)
+	b := mustMint(t, c, "/files/watch/b.mkv", 2500, 2)
+	if err := c.SetMetadataFlat("bagcsgatewayrecord", map[string]string{"title": "Some Release"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := c.SetMetadataFlat("01partialroot", map[string]string{"filePath": "/files/cache/x.mkv", "sizeByte": "999"}); err != nil {
+		t.Fatal(err)
+	}
+
+	code, resp := getFileTuples(t, s, "")
+	if code != http.StatusOK {
+		t.Fatalf("want 200, got %d", code)
+	}
+	if resp.Count != 2 || resp.TotalSize != 3500 {
+		t.Fatalf("want count 2 / totalSize 3500, got %d / %d", resp.Count, resp.TotalSize)
+	}
+	if resp.Files == nil || len(*resp.Files) != 2 {
+		t.Fatalf("want 2 rows, got %+v", resp.Files)
+	}
+	got := []string{(*resp.Files)[0].HashID, (*resp.Files)[1].HashID}
+	want := []string{a, b}
+	sort.Strings(got)
+	sort.Strings(want)
+	if got[0] != want[0] || got[1] != want[1] {
+		t.Fatalf("want roots %v, got %v", want, got)
+	}
+}
+
+func TestFileTuples_SummaryOmitsRows(t *testing.T) {
+	s, c := newFileTuplesTestServer(t)
+	mustMint(t, c, "/files/watch/a.mkv", 1000, 1)
+	mustMint(t, c, "/files/watch/b.mkv", 2500, 2)
+
+	code, resp := getFileTuples(t, s, "?summary=1")
+	if code != http.StatusOK {
+		t.Fatalf("want 200, got %d", code)
+	}
+	if resp.Files != nil {
+		t.Fatalf("summary must omit rows, got %d", len(*resp.Files))
+	}
+	if resp.Count != 2 || resp.TotalSize != 3500 {
+		t.Fatalf("want count 2 / totalSize 3500, got %d / %d", resp.Count, resp.TotalSize)
+	}
+}
+
+// An empty library still carries "files": [] — clients treat an absent key as
+// "this meta-core cannot list files", so it must not disappear when empty.
+func TestFileTuples_EmptyLibraryKeepsFilesKey(t *testing.T) {
+	s, _ := newFileTuplesTestServer(t)
+
+	code, resp := getFileTuples(t, s, "")
+	if code != http.StatusOK {
+		t.Fatalf("want 200, got %d", code)
+	}
+	if resp.Files == nil || len(*resp.Files) != 0 || resp.Count != 0 {
+		t.Fatalf("want an empty files array and count 0, got %+v", resp)
+	}
+}
+
+func TestFileTuples_StorageNotConnected(t *testing.T) {
+	s := &Server{storage: storage.NewClient("")}
+
+	code, _ := getFileTuples(t, s, "")
+	if code != http.StatusServiceUnavailable {
+		t.Fatalf("want 503, got %d", code)
+	}
+}

internal/api/server.go

@@ -243,6 +243,14 @@ 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")
+	// File-backed roots (filePath+sizeByte+mtimeNano) with sizes, batched
+	// server-side, so clients answer "how many files, how big" without one
+	// read per hashId. ?summary=1 returns only {count, totalSize}.
+	s.router.HandleFunc("/api/files/tuples", s.handleGetFileTuples).Methods("GET")
 
 	// Public metadata-by-CID resolver. Reverse-index lookup → full document.
 	// Auth-bypassed alongside /api/file/{cid} so meta-share peers can query
@@ -256,6 +264,9 @@ func (s *Server) setupRoutes() {
 	// 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/cid/cid-rank-vectors.json

@@ -128,6 +128,12 @@
       "rank": 5,
       "note": "codec 0x1009 + IDENTITY multihash 0x00, digest = the verbatim UTF-8 URL. WIRE-IDENTICAL to url (0x1006) apart from the codec, and that difference is the entire contract: 0x1006 means 'fetch these bytes once and seed them', 0x1009 means 'open this in a browser, there are no bytes for us'. A decoder that matched on SHAPE rather than on the codec slot would see an identity multihash wrapping an http(s) URL, take the 0x1006 path, fetch a watch/Bandcamp page, chunk it, seed the HTML and link that content CID onto the record AS THE FILE — the MSR1 poisoning shape, except permanent and replicated. Same tier/reasoning as the rest of the locator family."
     },
+    {
+      "name": "provider-file-locator",
+      "cid": "bagfcaaa2bvxxazloon2we5djorwgk43gnfwgkorxga3dcobtgq",
+      "rank": 5,
+      "note": "codec 0x100A + IDENTITY multihash 0x00, digest = varint(len(source)) || source || id - here (\"opensubtitles\",\"file:7061834\"). The GENERIC fetchable locator: one codec for every provider whose bytes are fetched the same way; the provider is inside the digest. Framing byte-identical to card (0x1007), but the byte path RESOLVES it (via source's fetch tier) instead of 404ing. Same codec-slot rule as the rest of the family: match on the CODEC, never the mh (identity 0x00 would fall through and score 0). Tier 5: once the fetched bytes' sha2-256 is linked onto the record it wins the canonical election with no addressing change."
+    },
     {
       "name": "sha1",
       "cid": "baeircff3eqvxxaqqot2lt242stkdtk3gq4c2vra",

internal/cid/rank.go

@@ -75,6 +75,14 @@ const (
 	CodeYtVideo = 0x1008 // custom — identity-multihash CID wrapping a YouTube (kind, id)
 	CodeExtPlay = 0x1009 // custom — identity-multihash CID wrapping a URL to OPEN, never fetch
 
+	// CodeProviderFile is the FETCHABLE twin of CodeExtPlay: a provider-held
+	// file addressed as varint(len(source))‖source‖id — ("opensubtitles",
+	// "file:7061834"). Framing is byte-identical to CodeCard; only the byte
+	// path differs (it redeems the id through that source's feeder instead of
+	// 404ing). Same codec-slot rule and same tier: once the fetched bytes'
+	// sha2-256 is linked onto the record it wins with no addressing change.
+	CodeProviderFile = 0x100A // custom — identity-multihash CID wrapping a (source, id) provider file
+
 	// CodeNzbPosting is the INVERSE trap of the two above: its multihash is a
 	// REAL sha2-256 digest (over the normalised, sorted article Message-ID
 	// set of a Usenet posting we scanned ourselves — never a locator, no
@@ -183,7 +191,7 @@ func Rank(cidStr string) int {
 	// Opaque locators. Codec-only: their multihash is identity (0x00), so
 	// there is nothing to match on the mh side.
 	if codec == CodeNzbRelease || codec == CodeURL || codec == CodeCard ||
-		codec == CodeYtVideo || codec == CodeExtPlay {
+		codec == CodeYtVideo || codec == CodeExtPlay || codec == CodeProviderFile {
 		return RankLocator
 	}
 	// nzb-posting: codec-only, checked BEFORE the mh switch below — its

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/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 {
-		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)
 	}
 }