new meta-core api, now redis is not exposed as an "API"

internal/api/admin_migrate.go

@@ -0,0 +1,30 @@
+package api
+
+import (
+	"net/http"
+)
+
+// handleMigrateDualRoots triggers the one-shot sweep that re-unifies
+// stranded midhash-rooted entries into their UUID roots. See
+// storage.MigrateDualRoots for the data-flow detail and
+// docs/api-mediated-access.md for the bug background.
+//
+// Returns {fixed: N} on success. Safe to call repeatedly — entries already
+// migrated are skipped on the next pass.
+func (s *Server) handleMigrateDualRoots(w http.ResponseWriter, r *http.Request) {
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+
+	fixed, err := s.storage.MigrateDualRootsWithTimeout()
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusOK, map[string]interface{}{
+		"success": true,
+		"fixed":   fixed,
+	})
+}

internal/api/cid_resolve.go

@@ -0,0 +1,68 @@
+package api
+
+import (
+	"net/http"
+
+	"github.com/gorilla/mux"
+)
+
+// handleGetMetaByCID handles GET /api/meta/{cid}.
+//
+// Resolves any known CID — midhash256, sha256, IPFS, btih, or anything else
+// in the reverse index — to the full metadata document for that file.
+// Returns:
+//   - 200 with the document on a hit
+//   - 404 if the CID isn't registered
+//   - 503 if storage is down
+//
+// CIDs in the URL must be in prefixed-token form ("midhash256:bafk…",
+// "sha256:bafk…", "ipfs:bafy…"). Bare CIDs without an algorithm prefix
+// won't resolve — callers should pass the token they got from a previous
+// metadata document, /api/file/{cid}/info, or a meta-share federated query.
+//
+// This endpoint is the public read API for the CID-resolution layer and
+// is auth-bypassed alongside /api/file/{cid} so that meta-share peers and
+// IPFS-style clients can use it without going through Authelia.
+func (s *Server) handleGetMetaByCID(w http.ResponseWriter, r *http.Request) {
+	cid := mux.Vars(r)["cid"]
+	if cid == "" {
+		writeError(w, http.StatusBadRequest, "cid is required")
+		return
+	}
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+
+	uuid, err := s.storage.GetByCID(cid)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	if uuid == "" {
+		writeErrorSlug(w, http.StatusNotFound, ErrUnknownCID, "no metadata for this CID", false)
+		return
+	}
+
+	doc, err := s.storage.GetMetadataDocument(uuid)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	if doc == nil {
+		// Reverse index pointed at a UUID with no remaining keys — stale
+		// alias, probably mid-delete race. Treat as not-found.
+		writeErrorSlug(w, http.StatusNotFound, ErrUnknownCID, "no metadata for this CID", false)
+		return
+	}
+
+	// Response envelope mirrors GetMetadataDocument JSON tags (metadata /
+	// cids / duplicates), with cid added so the caller can correlate when
+	// they fan out N lookups.
+	writeJSON(w, http.StatusOK, map[string]interface{}{
+		"cid":        cid,
+		"metadata":   doc.Flat,
+		"cids":       doc.CIDs,
+		"duplicates": doc.Duplicates,
+	})
+}

internal/api/handlers.go

@@ -63,10 +63,48 @@ type DataPathResponse struct {
 	Exists bool   `json:"exists"`
 }
 
-// ErrorResponse is the response for errors
+// ErrorResponse is the typed error envelope returned on every 4xx/5xx.
+//
+// Shape and slug vocabulary documented in docs/api-mediated-access.md
+// ("Error envelope"). Stable slug strings let clients switch on the failure
+// category without parsing the message; `retryable` tells callers whether
+// a backoff-and-retry has any chance of succeeding.
 type ErrorResponse struct {
-	Error   string `json:"error"`
-	Message string `json:"message,omitempty"`
+	Error     string `json:"error"`
+	Message   string `json:"message,omitempty"`
+	Retryable bool   `json:"retryable"`
+}
+
+// Stable error-slug strings. Pinned in the doc; consumers may match on
+// these. Do not rename without a deprecation cycle.
+const (
+	ErrAliasCollision    = "alias_collision"
+	ErrUnknownRoot       = "unknown_root"
+	ErrUnknownCID        = "unknown_cid"
+	ErrSchemaViolation   = "schema_violation"
+	ErrStorageUnavailable = "storage_unavailable"
+	ErrInternal          = "internal"
+)
+
+// defaultSlugFor maps an HTTP status code to the most likely error slug.
+// Handlers that want to override (e.g. /api/meta/{cid} → "unknown_cid"
+// instead of "unknown_root" on 404) call writeErrorSlug directly.
+func defaultSlugFor(status int) (slug string, retryable bool) {
+	switch status {
+	case http.StatusConflict:
+		return ErrAliasCollision, false
+	case http.StatusNotFound:
+		return ErrUnknownRoot, false
+	case http.StatusBadRequest:
+		return ErrSchemaViolation, false
+	case http.StatusServiceUnavailable:
+		return ErrStorageUnavailable, true
+	case http.StatusInternalServerError:
+		return ErrInternal, true
+	default:
+		// 4xx → not retryable, 5xx → retryable. Conservative default.
+		return ErrInternal, status >= 500
+	}
 }
 
 var startTime = time.Now()
@@ -169,6 +207,10 @@ func (s *Server) handleGetMeta(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// Resolve CID → UUID via the reverse index so reads find data wherever
+	// the watcher minted the root, regardless of which CID the caller knows.
+	hashID = s.storage.ResolveRoot(hashID)
+
 	metadata, err := s.storage.GetMetadataFlat(hashID)
 	if err != nil {
 		writeError(w, http.StatusInternalServerError, err.Error())
@@ -203,6 +245,10 @@ func (s *Server) handlePutMeta(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// CID → UUID resolution so writes land on the canonical root, not on
+	// a parallel midhash-rooted entry the watcher's UUID never sees.
+	hashID = s.storage.ResolveRoot(hashID)
+
 	var metadata map[string]string
 	if err := json.NewDecoder(r.Body).Decode(&metadata); err != nil {
 		writeError(w, http.StatusBadRequest, "invalid JSON body")
@@ -235,6 +281,8 @@ func (s *Server) handleDeleteMeta(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	hashID = s.storage.ResolveRoot(hashID)
+
 	deleted, err := s.storage.DeleteMetadata(hashID)
 	if err != nil {
 		writeError(w, http.StatusInternalServerError, err.Error())
@@ -421,7 +469,7 @@ func (s *Server) handleGetFileByCID(w http.ResponseWriter, r *http.Request) {
 	}
 
 	if relPath == "" {
-		writeError(w, http.StatusNotFound, "file not found for CID")
+		writeErrorSlug(w, http.StatusNotFound, ErrUnknownCID, "file not found for CID", false)
 		return
 	}
 
@@ -431,7 +479,7 @@ func (s *Server) handleGetFileByCID(w http.ResponseWriter, r *http.Request) {
 	// Check if file exists
 	fileInfo, err := os.Stat(fullPath)
 	if err != nil {
-		writeError(w, http.StatusNotFound, "file does not exist on disk")
+		writeErrorSlug(w, http.StatusNotFound, ErrUnknownCID, "file does not exist on disk", false)
 		return
 	}
 
@@ -522,11 +570,26 @@ func writeJSON(w http.ResponseWriter, status int, data interface{}) {
 	json.NewEncoder(w).Encode(data)
 }
 
-// writeError writes an error response
+// writeError writes the typed error envelope. The slug + retryable bit
+// are inferred from the status (see defaultSlugFor); call writeErrorSlug
+// directly when a more specific slug applies.
 func writeError(w http.ResponseWriter, status int, message string) {
+	slug, retryable := defaultSlugFor(status)
 	writeJSON(w, status, ErrorResponse{
-		Error:   http.StatusText(status),
-		Message: message,
+		Error:     slug,
+		Message:   message,
+		Retryable: retryable,
+	})
+}
+
+// writeErrorSlug writes the typed envelope with an explicit slug and
+// retryable bit. Use this for endpoint-specific slugs (alias_collision,
+// unknown_cid) where the default mapping from HTTP status would be wrong.
+func writeErrorSlug(w http.ResponseWriter, status int, slug, message string, retryable bool) {
+	writeJSON(w, status, ErrorResponse{
+		Error:     slug,
+		Message:   message,
+		Retryable: retryable,
 	})
 }
 
@@ -546,6 +609,11 @@ func (s *Server) handlePatchMeta(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// PATCH is the hot write path for meta-sort. Resolve so plugin output
+	// merges into the watcher's UUID root rather than minting a parallel
+	// midhash-rooted entry.
+	hashID = s.storage.ResolveRoot(hashID)
+
 	var metadata map[string]string
 	if err := json.NewDecoder(r.Body).Decode(&metadata); err != nil {
 		writeError(w, http.StatusBadRequest, "invalid JSON body")
@@ -587,6 +655,8 @@ func (s *Server) handleGetProperty(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	hashID = s.storage.ResolveRoot(hashID)
+
 	value, err := s.storage.GetProperty(hashID, key)
 	if err != nil {
 		writeError(w, http.StatusInternalServerError, err.Error())
@@ -632,6 +702,8 @@ func (s *Server) handlePutProperty(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	hashID = s.storage.ResolveRoot(hashID)
+
 	// Read body
 	buf := make([]byte, 1024*1024) // 1MB max
 	n, err := r.Body.Read(buf)
@@ -686,6 +758,8 @@ func (s *Server) handleDeleteProperty(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	hashID = s.storage.ResolveRoot(hashID)
+
 	if err := s.storage.DeleteProperty(hashID, key); err != nil {
 		writeError(w, http.StatusInternalServerError, err.Error())
 		return
@@ -721,6 +795,8 @@ func (s *Server) handleAddToSet(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	hashID = s.storage.ResolveRoot(hashID)
+
 	// Read body
 	buf := make([]byte, 1024*1024) // 1MB max
 	n, err := r.Body.Read(buf)

internal/api/server.go

@@ -204,6 +204,23 @@ func (s *Server) setupRoutes() {
 	s.router.HandleFunc("/file/{cid}", s.handleGetFileByCID).Methods("GET")
 	s.router.HandleFunc("/file/cid", s.handleComputeFileCID).Methods("POST")
 
+	// Public metadata-by-CID resolver. Reverse-index lookup → full document.
+	// Auth-bypassed alongside /api/file/{cid} so meta-share peers can query
+	// without going through Authelia.
+	s.router.HandleFunc("/api/meta/{cid}", s.handleGetMetaByCID).Methods("GET")
+
+	// Admin: reunify stranded midhash-rooted entries into their UUID roots
+	// (the dual-root pattern caused by historical writes that bypassed CID
+	// resolution).
+	s.router.HandleFunc("/api/admin/migrate-dual-roots", s.handleMigrateDualRoots).Methods("POST")
+
+	// SSE event streams. Mediated mirror of the Redis Streams; consumers
+	// reach them over HTTP instead of touching Redis directly. Auth-bypass
+	// applies but exposure should be inside-only (do NOT proxy through
+	// Caddy publicly). See docs/api-mediated-access.md "Auth".
+	s.router.HandleFunc("/api/events/files", s.handleEventsFiles).Methods("GET")
+	s.router.HandleFunc("/api/events/meta", s.handleEventsMeta).Methods("GET")
+
 	// Service discovery
 	// Both /services and /api/services are supported for consistency with other services
 	s.router.HandleFunc("/services", s.handleListServices).Methods("GET")

internal/api/sse_events.go

@@ -0,0 +1,280 @@
+package api
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// SSE event stream endpoints for the file:events and meta:events Redis Streams.
+//
+// Wire contract (see docs/api-mediated-access.md):
+//   - One SSE event per Redis Stream entry.
+//   - `id:` is the opaque Redis Stream entry ID (`<ms>-<seq>`); consumers
+//     MUST treat it as a black-box string they echo back via Last-Event-ID.
+//   - `event:` is the `type` field from the underlying entry.
+//   - `data:` is the JSON-encoded rest of the entry (all fields except `type`).
+//   - Heartbeats are SSE comment lines (`:keep-alive\n\n`) every 30s of silence.
+//   - On reconnect with Last-Event-ID, resume from the next entry. If the
+//     cursor has been trimmed by retention, emit one `event: gap` payload
+//     before resuming from the oldest available entry.
+
+const (
+	sseHeartbeatInterval = 30 * time.Second
+
+	// sseReadBlock controls how long each XREAD blocks before returning.
+	// Much shorter than sseHeartbeatInterval so the handler stays responsive
+	// to client disconnects and the heartbeat check runs frequently enough
+	// that the keep-alive fires within a few seconds of its nominal cadence.
+	sseReadBlock = 2 * time.Second
+
+	// sseReadCount caps how many entries one XREAD returns. Small enough
+	// to keep per-flush memory low, large enough to drain bursts cheaply.
+	sseReadCount = 100
+)
+
+// handleEventsFiles streams the file:events Redis Stream as SSE.
+func (s *Server) handleEventsFiles(w http.ResponseWriter, r *http.Request) {
+	s.streamSSE(w, r, "file:events")
+}
+
+// handleEventsMeta streams the meta:events Redis Stream as SSE.
+func (s *Server) handleEventsMeta(w http.ResponseWriter, r *http.Request) {
+	s.streamSSE(w, r, "meta:events")
+}
+
+// streamSSE is the shared SSE pump shared by both endpoints. One handler per
+// stream — there is no fan-out; each subscriber holds its own XREAD loop.
+// That's fine at our scale (≤20 subscribers; one per consumer service).
+func (s *Server) streamSSE(w http.ResponseWriter, r *http.Request, stream string) {
+	if !s.storage.IsConnected() {
+		w.Header().Set("Retry-After", "5")
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		writeError(w, http.StatusInternalServerError, "streaming unsupported")
+		return
+	}
+
+	// The server-wide WriteTimeout (30s on Server.server in server.go) is
+	// fine for ordinary handlers but lethal for SSE — it would close every
+	// connection at exactly 30s, racing the heartbeat. Disable it
+	// per-request via http.NewResponseController. SSE endpoints never need
+	// a write deadline; the client disconnect path already cleans up.
+	rc := http.NewResponseController(w)
+	if err := rc.SetWriteDeadline(time.Time{}); err != nil {
+		// Older transports (or nested response writers) may not support
+		// this. Log and continue — the 30s cap will still apply, but the
+		// handler is still useful for short-lived subscriptions.
+		log.Printf("[SSE %s] could not disable write deadline: %v", stream, err)
+	}
+
+	// SSE response headers. X-Accel-Buffering disables nginx response
+	// buffering; the hash-lock sidecar uses nginx, so this matters.
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+	w.Header().Set("X-Accel-Buffering", "no")
+	w.WriteHeader(http.StatusOK)
+	flusher.Flush()
+
+	client := s.storage.GetRedisClient()
+	if client == nil {
+		// Storage was just disconnected — close cleanly.
+		return
+	}
+
+	ctx := r.Context()
+	cursor := resolveStartCursor(ctx, client, stream, r.Header.Get("Last-Event-ID"), w, flusher)
+
+	lastWrite := time.Now()
+	for {
+		if err := ctx.Err(); err != nil {
+			return
+		}
+
+		// Bounded BLOCK lets us re-check ctx and emit heartbeats. The
+		// context's deadline isn't propagated into XREAD's BLOCK arg —
+		// that's a poll-style timeout on Redis itself — so we keep BLOCK
+		// short.
+		readCtx, cancel := context.WithTimeout(ctx, sseReadBlock+2*time.Second)
+		result, err := client.XRead(readCtx, &redis.XReadArgs{
+			Streams: []string{stream, cursor},
+			Block:   sseReadBlock,
+			Count:   sseReadCount,
+		}).Result()
+		cancel()
+
+		if err != nil {
+			// redis.Nil = BLOCK timeout with no entries. Anything else
+			// while the client is still connected is a real failure.
+			if errors.Is(err, redis.Nil) || isContextTimeout(err) {
+				// fall through to heartbeat check
+			} else if errors.Is(err, context.Canceled) || ctx.Err() != nil {
+				return
+			} else {
+				log.Printf("[SSE %s] XREAD error: %v", stream, err)
+				return
+			}
+		}
+
+		for _, str := range result {
+			for _, msg := range str.Messages {
+				writeSSEEntry(w, flusher, msg)
+				cursor = msg.ID
+				lastWrite = time.Now()
+			}
+		}
+
+		// Heartbeat if quiet for >= the interval. Comment line; SSE
+		// clients ignore it per spec.
+		if time.Since(lastWrite) >= sseHeartbeatInterval {
+			if _, err := fmt.Fprint(w, ":keep-alive\n\n"); err != nil {
+				return
+			}
+			flusher.Flush()
+			lastWrite = time.Now()
+		}
+	}
+}
+
+// resolveStartCursor decides where to start reading.
+//   - No Last-Event-ID → start at "$" (only new entries).
+//   - Last-Event-ID provided AND still in stream → resume from it.
+//   - Last-Event-ID provided BUT trimmed → emit one synthetic `gap` event
+//     and resume from the oldest available entry's predecessor "0", so the
+//     next XREAD picks up everything still in the stream.
+//
+// The "still in stream" check is cheap: one XRANGE for the first entry.
+func resolveStartCursor(ctx context.Context, client *redis.Client, stream, lastID string, w http.ResponseWriter, flusher http.Flusher) string {
+	if lastID == "" {
+		return "$"
+	}
+
+	// One-entry XRANGE to find the oldest ID currently in the stream.
+	xCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
+	defer cancel()
+	entries, err := client.XRangeN(xCtx, stream, "-", "+", 1).Result()
+	if err != nil {
+		// Treat any error as "stream not available right now" and resume
+		// from the cursor — XREAD will either find it or block until new
+		// entries arrive.
+		log.Printf("[SSE %s] XRANGE probe failed: %v (resuming from cursor)", stream, err)
+		return lastID
+	}
+
+	if len(entries) == 0 {
+		// Stream is empty. Cursor may be from a previous incarnation; safest
+		// is to resume from the cursor — XREAD against an empty stream
+		// returns nothing and blocks until new entries arrive.
+		return lastID
+	}
+
+	oldest := entries[0].ID
+	if compareStreamIDs(lastID, oldest) >= 0 {
+		// Cursor is at or past the oldest — fine to resume.
+		return lastID
+	}
+
+	// Cursor was trimmed. Emit a gap event so consumers that care can react.
+	gap := map[string]string{
+		"requested":  lastID,
+		"resumeFrom": oldest,
+		"reason":     "retention",
+	}
+	payload, _ := json.Marshal(gap)
+	fmt.Fprintf(w, "event: gap\ndata: %s\n\n", string(payload))
+	flusher.Flush()
+
+	// Return "0" so the next XREAD picks up entries starting from the
+	// stream's true minimum — including `oldest` itself.
+	return "0"
+}
+
+// writeSSEEntry serialises one Redis Stream entry as an SSE event.
+//
+// `event:` comes from the entry's `type` field (no prefix munging). `data:`
+// is a JSON object containing all OTHER fields. Field values from the Redis
+// stream are interface{} (typically strings); we encode whatever the stream
+// holds.
+func writeSSEEntry(w http.ResponseWriter, flusher http.Flusher, msg redis.XMessage) {
+	eventType := "message"
+	data := make(map[string]interface{}, len(msg.Values))
+	for k, v := range msg.Values {
+		if k == "type" {
+			if s, ok := v.(string); ok {
+				eventType = s
+				continue
+			}
+		}
+		data[k] = v
+	}
+
+	payload, err := json.Marshal(data)
+	if err != nil {
+		log.Printf("[SSE] failed to marshal entry %s: %v", msg.ID, err)
+		return
+	}
+
+	// Spec: each event ends with a blank line. id/event/data lines are
+	// terminated with single \n.
+	fmt.Fprintf(w, "id: %s\n", msg.ID)
+	fmt.Fprintf(w, "event: %s\n", eventType)
+	fmt.Fprintf(w, "data: %s\n\n", string(payload))
+	flusher.Flush()
+}
+
+// compareStreamIDs compares two Redis Stream IDs of the form "<ms>-<seq>".
+// Returns -1/0/+1. Missing or malformed IDs sort as zero parts.
+func compareStreamIDs(a, b string) int {
+	aMs, aSeq := parseStreamID(a)
+	bMs, bSeq := parseStreamID(b)
+	switch {
+	case aMs < bMs:
+		return -1
+	case aMs > bMs:
+		return 1
+	case aSeq < bSeq:
+		return -1
+	case aSeq > bSeq:
+		return 1
+	default:
+		return 0
+	}
+}
+
+func parseStreamID(id string) (int64, int64) {
+	parts := strings.SplitN(id, "-", 2)
+	ms, _ := strconv.ParseInt(parts[0], 10, 64)
+	var seq int64
+	if len(parts) == 2 {
+		seq, _ = strconv.ParseInt(parts[1], 10, 64)
+	}
+	return ms, seq
+}
+
+// isContextTimeout returns true for the context deadline error wrapped by
+// go-redis when XREAD's read context fires. We treat it the same as
+// redis.Nil — i.e. an empty poll cycle, not a fatal error.
+func isContextTimeout(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, context.DeadlineExceeded) {
+		return true
+	}
+	// go-redis sometimes wraps this in its own error type. String match is
+	// a safety net.
+	return strings.Contains(err.Error(), "context deadline exceeded")
+}