improved cid resolution
This commit is contained in:
@@ -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(),
|
||||||
|
})
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -243,6 +243,10 @@ 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")
|
||||||
|
|
||||||
// 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
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user