Compare commits

...
4 Commits
Author SHA1 Message Date
Worph 40b553016c cid vector
Build and Push Docker Image / build (push) Failing after 13s
2026-09-13 14:36:50 +02:00
Worph 4894810fd4 new file tuple API
Build and Push Docker Image / build (push) Failing after 13s
2026-09-13 12:01:33 +02:00
Worph ac21315255 improved cid resolution
Build and Push Docker Image / build (push) Failing after 13s
2026-09-12 21:37:38 +02:00
Worph ec1b227631 improved search 2026-09-12 16:37:51 +02:00
18 changed files with 1436 additions and 42 deletions
+5
View File
@@ -209,6 +209,7 @@ the `/file/{cid}` reverse-lookup path.)
```bash ```bash
GET /api/metadata/hash-ids GET /api/metadata/hash-ids
GET /api/files/tuples
GET /api/metadata/list GET /api/metadata/list
POST /api/metadata/search POST /api/metadata/search
POST /api/metadata/batch POST /api/metadata/batch
@@ -287,6 +288,10 @@ return a deprecation pointer at the `/api/watchers/*` routes
```bash ```bash
POST /api/admin/migrate-dual-roots # reunify stranded midhash-rooted entries 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 GET/PUT/DELETE /webdav/... # mounted on /files
``` ```
+4
View File
@@ -301,6 +301,7 @@ CID-addressed (public, auth-bypassed at the perimeter)
Editor / KV / schema / snapshot Editor / KV / schema / snapshot
GET /api/metadata/hash-ids GET /api/metadata/hash-ids
GET /api/files/tuples
GET /api/metadata/list GET /api/metadata/list
POST /api/metadata/search POST /api/metadata/search
POST /api/metadata/batch POST /api/metadata/batch
@@ -339,6 +340,9 @@ Watchers
Admin Admin
POST /api/admin/migrate-dual-roots 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
/webdav/... (GET, PUT, DELETE, MKCOL, COPY, MOVE, PROPFIND) /webdav/... (GET, PUT, DELETE, MKCOL, COPY, MOVE, PROPFIND)
+45
View File
@@ -2,6 +2,7 @@ package api
import ( import (
"net/http" "net/http"
"strconv"
) )
// handleMigrateDualRoots triggers the one-shot sweep that re-unifies // 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, "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,
})
}
+93
View File
@@ -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)
}
}
+132
View File
@@ -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(),
})
}
+144
View File
@@ -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)
}
}
+74
View File
@@ -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)
}
+114
View File
@@ -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)
}
}
+11
View File
@@ -243,6 +243,14 @@ func (s *Server) setupRoutes() {
s.router.HandleFunc("/api/file/{cid}/info", s.handleGetFileByCIDInfo).Methods("GET") 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.handleGetFileByCID).Methods("GET")
s.router.HandleFunc("/file/cid", s.handleComputeFileCID).Methods("POST") 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. // Public metadata-by-CID resolver. Reverse-index lookup → full document.
// Auth-bypassed alongside /api/file/{cid} so meta-share peers can query // 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 // One-shot vocabulary sweep: domain film|tv → screen, workForm backfill
// (METADATA_KEYS.md §14.17). Idempotent. // (METADATA_KEYS.md §14.17). Idempotent.
s.router.HandleFunc("/api/admin/migrate-domain-screen", s.handleMigrateDomainScreen).Methods("POST") 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 // SSE event streams. Mediated mirror of the Redis Streams; consumers
// reach them over HTTP instead of touching Redis directly. Auth-bypass // reach them over HTTP instead of touching Redis directly. Auth-bypass
+92
View File
@@ -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
}
+71
View File
@@ -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)
}
}
}
+6
View File
@@ -128,6 +128,12 @@
"rank": 5, "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." "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", "name": "sha1",
"cid": "baeircff3eqvxxaqqot2lt242stkdtk3gq4c2vra", "cid": "baeircff3eqvxxaqqot2lt242stkdtk3gq4c2vra",
+9 -1
View File
@@ -75,6 +75,14 @@ const (
CodeYtVideo = 0x1008 // custom — identity-multihash CID wrapping a YouTube (kind, id) CodeYtVideo = 0x1008 // custom — identity-multihash CID wrapping a YouTube (kind, id)
CodeExtPlay = 0x1009 // custom — identity-multihash CID wrapping a URL to OPEN, never fetch 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 // CodeNzbPosting is the INVERSE trap of the two above: its multihash is a
// REAL sha2-256 digest (over the normalised, sorted article Message-ID // REAL sha2-256 digest (over the normalised, sorted article Message-ID
// set of a Usenet posting we scanned ourselves — never a locator, no // 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 // Opaque locators. Codec-only: their multihash is identity (0x00), so
// there is nothing to match on the mh side. // there is nothing to match on the mh side.
if codec == CodeNzbRelease || codec == CodeURL || codec == CodeCard || if codec == CodeNzbRelease || codec == CodeURL || codec == CodeCard ||
codec == CodeYtVideo || codec == CodeExtPlay { codec == CodeYtVideo || codec == CodeExtPlay || codec == CodeProviderFile {
return RankLocator return RankLocator
} }
// nzb-posting: codec-only, checked BEFORE the mh switch below — its // nzb-posting: codec-only, checked BEFORE the mh switch below — its
+47
View File
@@ -422,3 +422,50 @@ func (c *Client) GetMetadataDocument(uuid string) (*MetadataDocument, error) {
Duplicates: dups, Duplicates: dups,
}, nil }, 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
}
+326
View File
@@ -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)
}
@@ -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)
}
}
+12 -4
View File
@@ -50,8 +50,18 @@ func (d *Dispatcher) dispatchToStream(event FileEvent) {
client := d.storageClient client := d.storageClient
d.mu.RUnlock() 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() { if client == nil || !client.IsConnected() {
return return fmt.Errorf("storage client not connected")
} }
// Build stream fields // Build stream fields
@@ -79,9 +89,7 @@ func (d *Dispatcher) dispatchToStream(event FileEvent) {
// `EmitReset → ClearStream` path is still the only way the stream gets // `EmitReset → ClearStream` path is still the only way the stream gets
// truncated to zero (signalled to consumers via the `reset` event). // truncated to zero (signalled to consumers via the `reset` event).
_, err := client.XAdd(EventsStream, EventsStreamMaxLen, fields) _, 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 // EmitReset clears the stream and emits a reset event
+2 -37
View File
@@ -4,7 +4,6 @@ import (
"log" "log"
"os" "os"
"path/filepath" "path/filepath"
"strconv"
"strings" "strings"
"sync" "sync"
@@ -54,42 +53,8 @@ func (w *Watcher) writeFileTuple(absPath string, size, mtimeNano int64, midhash
return 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)
} }
} }