From ac213152551a9685622ff14b38805904ef36592a Mon Sep 17 00:00:00 2001 From: Worph Date: Sat, 12 Sep 2026 21:37:38 +0200 Subject: [PATCH] improved cid resolution --- internal/api/files_register.go | 132 +++++++++++++++++++++++++ internal/api/files_register_test.go | 144 ++++++++++++++++++++++++++++ internal/api/server.go | 4 + internal/storage/cid_resolution.go | 47 +++++++++ internal/watcher/dispatcher.go | 16 +++- internal/watcher/watcher.go | 39 +------- 6 files changed, 341 insertions(+), 41 deletions(-) create mode 100644 internal/api/files_register.go create mode 100644 internal/api/files_register_test.go diff --git a/internal/api/files_register.go b/internal/api/files_register.go new file mode 100644 index 0000000..aeb9f18 --- /dev/null +++ b/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: 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(), + }) +} diff --git a/internal/api/files_register_test.go b/internal/api/files_register_test.go new file mode 100644 index 0000000..8817097 --- /dev/null +++ b/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) + } +} diff --git a/internal/api/server.go b/internal/api/server.go index f15e77c..6b0c75a 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -243,6 +243,10 @@ func (s *Server) setupRoutes() { s.router.HandleFunc("/api/file/{cid}/info", s.handleGetFileByCIDInfo).Methods("GET") s.router.HandleFunc("/file/{cid}", s.handleGetFileByCID).Methods("GET") s.router.HandleFunc("/file/cid", s.handleComputeFileCID).Methods("POST") + // Register one file under /files/plugin by midhash — the watcher's + // mint + alias + file:events path for a single file. Registered + // unconditionally: it exists for boxes whose watcher is OFF. + s.router.HandleFunc("/api/files/register", s.handleRegisterFile).Methods("POST") // Public metadata-by-CID resolver. Reverse-index lookup → full document. // Auth-bypassed alongside /api/file/{cid} so meta-share peers can query diff --git a/internal/storage/cid_resolution.go b/internal/storage/cid_resolution.go index eab6ec4..81efd25 100644 --- a/internal/storage/cid_resolution.go +++ b/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/`, whose hook +// registers the cid: 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 +} diff --git a/internal/watcher/dispatcher.go b/internal/watcher/dispatcher.go index 7c8013c..0f24264 100644 --- a/internal/watcher/dispatcher.go +++ b/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 diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go index 8b940d5..62f49fa 100644 --- a/internal/watcher/watcher.go +++ b/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/ 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) } }