diff --git a/cmd/meta-core/main.go b/cmd/meta-core/main.go index 0ed0c39..0079876 100644 --- a/cmd/meta-core/main.go +++ b/cmd/meta-core/main.go @@ -51,6 +51,15 @@ func main() { } log.Println("[meta-core] Connected to Redis") + // Schema-version sentinel: alpha clean-wipe gate. Refuses to boot when + // the existing Redis store predates the UUID-rooted schema, so legacy + // data can't silently leak into the new code paths. See + // docs/uuid-rooted-metadata.md. + if err := storageClient.EnsureSchemaVersion(storage.SchemaVersion); err != nil { + log.Fatalf("[meta-core] %v", err) + } + log.Printf("[meta-core] Redis schema version OK (v%d)", storage.SchemaVersion) + // Create and start service discovery disc := discovery.NewService(cfg) if err := disc.Start(); err != nil { diff --git a/docs/api-mediated-access.md b/docs/api-mediated-access.md new file mode 100644 index 0000000..ae3009b --- /dev/null +++ b/docs/api-mediated-access.md @@ -0,0 +1,625 @@ +# API-Mediated Access (Redis Lockdown) + +## Status + +**Proposed** — design accepted; implementation pending. Builds on +[uuid-rooted-metadata.md](uuid-rooted-metadata.md), which assumes meta-core +is the single writer to the metadata keyspace. That assumption is currently +violated by meta-sort, and this doc is the plan to put it back on solid +ground. + +## Motivation + +meta-core was designed as the sidecar that mediates every metadata read and +write — a single point that owns the Redis schema, enforces invariants +(reverse index, canonical_cid reconciliation, schema-version sentinel), +and presents a stable HTTP contract. Other services were meant to talk to +meta-core, not to Redis. + +In practice, meta-sort opens its own Redis connection and writes +`file:/` keys with raw pipeline SETs +(`packages/meta-sort/packages/meta-sort-core/src/kv/RedisClient.ts:269-296`). +It picked up this shortcut early, before meta-core had a complete write +API, and the shortcut stuck. + +This bypass is a real problem in three ways: + +1. **Schema invariants leak.** When meta-core landed the UUID-rooted schema + ([uuid-rooted-metadata.md](uuid-rooted-metadata.md)), meta-sort kept + writing midhash-rooted entries. The result was *dual roots per file* — + one set of keys at the watcher's UUID, another at meta-sort's midhash. + `file:__index__` doubled in size. The bug was invisible to meta-sort + (its writes succeeded) but broke the reverse-index resolution + downstream services rely on. + +2. **Schema evolution is impossible.** Changing the Redis layout requires + coordinating across every service that touches it. The UUID-root + migration would have been a single-PR change if writes already routed + through the API. + +3. **Self-hosted distributed deployment widens the threat model.** Peers + are run by different operators. Plugin authors may ship their own + services. We can't expect every binary in the network to honour + undocumented conventions. A buggy or careless service that connects + to Redis and writes the wrong key shape can corrupt the schema for + everybody on that node. + +The third point is the one that flipped the design from "soft lock by +convention" to "hard lock by network topology." We're not protecting +against malicious services — that's an out-of-scope threat. We're +protecting against *accidental* corruption from honest mistakes in code +we don't fully control. + +## The architecture in one picture + +``` + meta-core (only Redis writer + reader) + ┌──────────────────────────────────────┐ + │ HTTP API surface: │ + │ POST/PUT/PATCH /api/metadata/* │ ◄── writes + │ GET /api/meta/{cid} │ ◄── reads (by CID) + │ GET /api/metadata/* │ ◄── reads (by id) + │ GET /api/file/{cid} │ ◄── content + │ GET /api/events/files │ ◄── SSE stream + │ GET /api/events/meta │ ◄── SSE stream + │ │ │ + │ ▼ │ + │ Redis (file:*, cid:*, streams) │ + │ ◄── container-local, no exposed │ + │ port, no docker-network alias │ + └──────────────────────────────────────┘ + ▲ ▲ + │ HTTP │ HTTP SSE + │ │ + ┌───────────┴────┐ ┌──────┴─────────┐ + │ meta-sort │ │ meta-fuse │ + │ meta-stremio │ │ meta-share │ + │ meta-dup │ │ (any future) │ + └────────────────┘ └────────────────┘ +``` + +**The single-writer rule.** meta-core has exclusive write authority over +the `file:*`, `cid:*`, `file:__index__`, and `meta-core:*` keyspaces. No +other service can write — not because of an honoured convention, but +because they can't reach Redis at all. + +**Stream reads also mediated.** Events go out through SSE endpoints on +meta-core's HTTP surface. Consumers don't see the underlying Redis +Streams. This is the Kubernetes / Consul pattern (see "Industry context" +below). It costs an extra HTTP hop versus the current direct XREAD; the +benefit is that the storage layer is free to evolve (or be replaced) +without breaking every consumer. + +**Redis is private.** No host port mapping, no docker-network alias, no +URL in `kv-leader.info`. Discovery returns only meta-core's HTTP base. + +## Eventing — HTTP Server-Sent Events + +SSE is the chosen transport for cross-service event consumption. The +mapping from Redis Streams onto SSE is direct enough that no eventing +feature we currently use is lost. + +### Feature mapping (what we keep, how we keep it) + +| Redis Streams | SSE equivalent | +|---|---| +| Fan-out to N consumers | meta-core holds N SSE connections, broadcasts events to each | +| Replay from a cursor | `Last-Event-ID` header on reconnect — built into the SSE spec | +| Retention / MAXLEN | Underlying Redis Stream still does the trimming; SSE is a thin shim | +| Backpressure on slow consumers | meta-core's SSE handler blocks on the TCP write; events stay in Redis until trimmed | +| Multiple streams (file:events, meta:events) | Separate endpoints: `/api/events/files`, `/api/events/meta` | +| At-least-once via XACK | Implicit — client persists last-received `id`, sends it on reconnect (see "Cursor semantics and consumer-group migration" below) | + +The one Streams feature we don't get is **consumer groups for +horizontally-scaled replicas of the same logical consumer.** We don't use +this today and don't plan to — meta-sort, meta-fuse, etc. each run one +instance. If horizontal scale ever happens, it'll be by sharding hash-IDs +across replicas, not by splitting an event stream. SSE stays adequate. + +### Cursor semantics and consumer-group migration + +Today three consumers use `XREADGROUP` against named consumer groups, not +bare `XREAD`: + +- meta-sort: group `meta-sort-processor` + (`packages/meta-sort/.../events/FileEventConsumer.ts:23,66-74`) +- meta-dup: group on `file:events` + (`packages/meta-dup/.../MetaEventConsumer.ts:102-104`) +- meta-fuse: dual path — `XREADGROUP` on one consumer, plain `XREAD` on + another (`packages/meta-fuse/.../kv/RedisClient.ts:826,907`) + +`XREADGROUP` keeps server-side state per group: a Pending Entries List +(PEL) and a `last-delivered-id` cursor. The load-bearing use today is +meta-sort's `processPendingEntries(idleMs=30000)`, which exists to claim +work from a crashed consumer's PEL on restart. + +SSE has no equivalent server-side state. After the migration the contract +shifts to: **each consumer persists its own `Last-Event-ID` and resumes +from it on reconnect.** Implications: + +- PEL-based crash recovery is gone. A crashed consumer that hadn't + persisted its cursor since entry X resumes from X (its last *durably + persisted* cursor), not from "whatever was in flight." Acceptable for + every current consumer — none run multiple instances per group — but + call out in the PR C migration of meta-sort that the recovery primitive + is changing. +- Each consumer must own its cursor persistence. Cheapest place is the + consumer's own state (a file, a Redis key written under its own + prefix, an in-memory variable if the consumer is OK with at-most-once + on its own crash). +- No server-side "ACK" call. The cursor *is* the ack — when the consumer + successfully writes Last-Event-ID = N to its own state, entries ≤ N + are considered handled. The SSE handler does nothing. + +### Endpoint spec + +``` +GET /api/events/files + Request headers: + Last-Event-ID: (optional; resume from this cursor) + Accept: text/event-stream + Response: + Content-Type: text/event-stream + Cache-Control: no-cache + Connection: keep-alive + X-Accel-Buffering: no (disable nginx response buffering) +``` + +Same shape for `/api/events/meta`. + +### Event format + +One SSE event per Redis Stream entry. Format: + +``` +id: 1747999999999-0 +event: add +data: {"path":"/watch/foo.mkv","size":4500000000,"midhash256":"bafk…","timestamp":1747999999999} + +id: 1748000000123-0 +event: change +data: {"path":"/watch/foo.mkv","size":4500000000,"midhash256":"bafk…","timestamp":1748000000123} +``` + +- `id` — **opaque** cursor. Today it's the Redis Stream entry ID + (`-`); consumers MUST treat it as a black-box string they + echo back via `Last-Event-ID`. This contract lets meta-core swap the + backing store later without breaking every consumer's cursor handling. +- `event` — the `type` field from the underlying entry (`add` | `change` + | `delete` | `rename` | `reset` for `file:events`; `set` | `del` | + `expire` for `meta:events`). +- `data` — JSON-encoded rest of the entry. Keys match the existing stream + payload exactly so downstream consumers don't change their parser. + +### Backing-store retention and lifecycle + +**Contract: the SSE wire is a faithful mirror of the underlying Redis +Streams.** Consumers rely on `Last-Event-ID` resumption working across +*every* kind of meta-core hiccup, including a restart — that's the entire +point of having a cursor protocol. The backing store must therefore +preserve entries across meta-core process restart and bound its growth +explicitly, not implicitly. + +Today's behaviour is wrong on both counts: + +| Stream | Restart behaviour | Retention | +|---|---|---| +| `meta:events` | **Cleared on every MetaPublisher.Start** (`events/meta_publisher.go:50-58`) | No `MAXLEN` (unbounded) | +| `file:events` | Persists across restart | No `MAXLEN` (unbounded), cleared only on deliberate reset (`watcher/dispatcher.go:82-95`) | + +The `meta:events` clear-on-restart is the load-bearing problem. With it +in place, every meta-core restart silently invalidates every consumer's +persisted cursor — the SSE `gap` event would fire on the first reconnect +after every restart, and the consumer would have to re-bootstrap. That +turns "restart meta-core to apply a config change" into "every downstream +service does a full re-sync," which is exactly the regression SSE is +supposed to prevent. + +PR A must fix both streams: + +1. **Remove the unconditional `Del(MetaEventsStream)` in + `MetaPublisher.Start`.** Entries persist across restart. The republish + path (`RepublishAllMetadata`) stays as a one-shot bootstrap for + missing entries, but it does not pre-truncate the stream. +2. **Add `MAXLEN ~ N` to every `XAdd` on both streams** so growth is + bounded. Approximate trim (`~`) is cheap. Default `N = 100_000` per + stream is fine for the foreseeable workload (a few KB per entry, + bounded total ≈ a few hundred MB) and easy to tune via config. +3. **Keep the deliberate `EmitReset → ClearStream` path on + `file:events`.** That clear is a feature: a reset is the consumer + contract for "throw out everything you knew, re-bootstrap." Consumers + detect it through the `event: reset` payload that's already emitted + as the first entry after the clear (`dispatcher.go:99-`). The SSE + `gap` event handles the race where a consumer's cursor was trimmed + *between* the clear and its reconnect. + +After PR A, the cursor contract is **"durable across meta-core restart +and bounded by `MAXLEN ~ N`"** — matching what consumers already assume +when they persist a Last-Event-ID. + +### Heartbeats + +Server writes an SSE comment line (`:keep-alive\n\n`) every 30 seconds +when no events are flowing. Keeps reverse proxies and load balancers +from killing the connection as idle. The client ignores comment lines +per the SSE spec. + +### Reconnect semantics + +Per the SSE spec the client (or `EventSource`) auto-reconnects after a +network error, sending `Last-Event-ID`. The server resumes from the next +entry after that ID by translating to Redis Stream `XREAD … STREAMS file:events `. + +**Gap handling.** If the requested ID has been trimmed out of the Redis +Stream by retention, the server emits one synthetic event before resuming +from the oldest available entry: + +``` +event: gap +data: {"requested":"","resumeFrom":"","reason":"retention"} +``` + +Consumers that care about gap-free delivery (currently none — the +schema indexer tolerates gaps; ingest tasks rebuild from scratch) can +react. Consumers that don't care ignore it. + +### Error handling + +- **Redis unreachable on connect**: 503 with `Retry-After: 5`. +- **Redis fails mid-stream**: close the SSE connection. Client reconnects + with `Last-Event-ID`; the cycle resumes when Redis recovers. +- **Client disconnects**: handler goroutine exits cleanly; no per-client + state on meta-core. + +### Auth + +`/api/events/*` bypasses the hash-lock perimeter, same pattern as +`/api/meta/{cid}` and `/api/file/{cid}`. Event streams are internal +infrastructure and need to be reachable by peer services without going +through Authelia. + +**Bypass ≠ public.** "Hash-lock bypass" means the OIDC sidecar doesn't +gate the route; it does *not* automatically mean Caddy proxies it from +the public hostname. The two existing precedents differ on this: +`/api/file/{cid}` is publicly reachable (peer-to-peer content fetch is +its job), `/api/meta/{cid}` likewise. For the SSE endpoints, the choice +is deliberate: + +- If kept inside-only (no Caddy upstream for `/api/events/files` and + `/api/events/meta`), only sibling services on the docker network can + subscribe. This matches the rest of the "Redis is private" model and + is the recommended default. +- If exposed publicly, any browser can open an `EventSource` against + the stream without authentication — i.e. anyone can watch every file + event in the library. Only do this if there's a concrete consumer + that needs it and the visibility is acceptable. + +PR A's Caddy change (see PR A scope, item 4) makes this an explicit +choice rather than a default behaviour. The recommendation is +inside-only until a concrete external consumer appears. + +### Implementation footprint + +The SSE handler itself is ~80 lines of Go in meta-core: one handler per +stream, each running an `XREAD … BLOCK 30s` loop and writing +SSE-formatted lines to the response. Go's `net/http` natively supports +streaming responses with `http.Flusher.Flush()`. Standard SSE clients +exist for every language we use: Node `eventsource`, Rust +`eventsource-client` (or `reqwest` + `tokio_util::codec`), Python +`sseclient`, Go `r3labs/sse`. + +PR A as a whole is larger than the handler — see the PR A scope below +for the full list (backing-store fixes, Caddy wiring, poll deprecation) +that has to land alongside the handler for the cursor contract to hold. + +## Eventing — considered alternatives + +Recorded for posterity. None of these are being implemented; if any are +ever revisited it should be a deliberate reopen, not a drift. + +- **Long-polling.** Strictly worse than SSE — connection churn on every + event, latency floor at the poll interval, no built-in resumption. + Rejected. +- **Webhooks (meta-core POSTs to subscribers).** Symmetric coupling: + every consumer would need to run an HTTP server reachable from + meta-core. Out-of-order delivery, retry, dead-letter all need + reinventing. Right for cross-organization integrations, wrong for + in-cluster eventing. Rejected. +- **External broker (Kafka / NATS / RabbitMQ).** Industrial-grade + pub/sub. New dependency, new operational surface, overkill for the + scale (single-host dev, ≤20-peer POC). Revisit if MetaMesh ever needs + cross-WAN federation at high throughput. Rejected for now. +- **Two Redises (writes Redis private, events Redis exposed).** Strongest + separation. Doubles operational complexity for no benefit over the + SSE design — once events go through HTTP, the events-Redis isn't + exposed either. Rejected. +- **Redis ACLs (Option B from earlier brainstorm).** Allows direct Redis + access from clients while restricting commands by user. Adds + credential plumbing and ACL config to operate; lets a buggy service + still connect, just with limited verbs. The SSE design closes the + network entirely, which is strictly stronger. Rejected — but the + equivalent escape hatch (open Redis to other services again) stays + available if we ever need it. + +## Industry context — where this lands + +The pattern we're adopting is the **Kubernetes / etcd** pattern, and +that's the most informative comparison. + +Kubernetes is structurally identical to MetaMesh: +- A single source-of-truth store (etcd) holds all cluster state. +- A single service (kube-apiserver) is the sole writer to etcd. +- Many polyglot clients (kubectl, controllers, schedulers, kubelets, + operators) read and react to changes. +- Clients subscribe to *watch streams* for events. + +What Kubernetes does, point-for-point: +- **etcd is never reached by clients.** It runs on a separate port, + often a separate network. kube-apiserver is the only thing that holds + etcd credentials. The lock is by network topology, not authentication. +- **Watch is also mediated.** kube-apiserver exposes `/watch` as an HTTP + streaming endpoint (very similar to SSE in shape). Clients don't watch + etcd directly — even though etcd has its own native watch primitive. +- **Storage portability falls out for free.** etcd could be swapped for + another store without any client change. We don't plan to swap Redis, + but the optionality is real and free. + +Other systems with the same shape: +- **HashiCorp Consul.** Clients talk to a local agent over HTTP. The + agent talks to the cluster. KV writes and watches both flow through + the agent. Direct raft-store access is impossible. +- **Apache ZooKeeper.** Clients use the ZooKeeper client library which + speaks the ZooKeeper wire protocol; the data store is private. +- **Strapi / Sanity / Contentful** (headless CMS as analog). All + consumers — frontends, backend services, build pipelines — go through + the CMS HTTP API. Direct Postgres or content-store access is impossible + by design. + +Where MetaMesh sat *before*: closer to the bad-monolith pattern +(everyone connects directly to the shared Redis). Where it sits *after* +this design: aligned with the canonical microservices pattern, and +specifically with the Kubernetes flavour where streams are mediated too. + +### Advantages we're picking up + +- **Schema evolution.** One service can change its data layout without + coordinating releases across every other service. The UUID-root + migration is the cautionary tale of what happens without this. +- **Validation choke point.** Every write passes through one code path + that can enforce invariants (reverse index, canonical_cid, type + constraints, schema version). +- **Observability.** Audit logs, request metrics, slow-query detection + all live at one layer. +- **Threat-model fit.** Untrusted-service-in-the-network is now in + scope; network isolation makes the concern moot. +- **Storage portability.** Optionality only, but real. + +### Drawbacks (acknowledged) + +- **Latency.** One extra hop per write (TCP + HTTP parse + handler). + Local-host overhead is sub-millisecond; mitigations are batched writes + (PATCH the whole metadata blob per plugin completion, not per field) + and Keep-Alive / HTTP/2 connections. +- **SSE-handler plumbing.** ~80 lines per stream endpoint plus per-client + goroutines. Trivial at our scale, real at Kubernetes scale (which is + why they invested in chunked watch with bookmarks, resource versions, + etc.). We can defer the same optimisations forever. +- **API surface size.** Every operation needs a corresponding endpoint. + Most of it already exists; one new SSE endpoint per stream is the + only new surface area introduced here. + +## Enforcement — what "lock" actually means + +Three layers, in order of effectiveness: + +1. **Network isolation (hard).** Redis is on a container-private port. + No host port mapping. No docker-network alias. `kv-leader.info` does + not publish a Redis URL. Any service that wants to talk to Redis + needs container privileges, and at that point the lock isn't the + concern — operating-system isolation is. This is the actual lock. + +2. **Documentation (convention).** This file, plus a one-line note in + `CLAUDE.md` and in each consumer service's `CLAUDE.md`: "All metadata + I/O routes through meta-core's HTTP API. Redis is private to + metacore-app." Engineers reading the codebase find the rule on first + contact. This catches the case of a future meta-core PR that + accidentally exposes Redis again. + +3. **Code review (convention).** PRs that re-expose Redis (a port + mapping, an alias, a credential leak via leader-info) get rejected + with a pointer to this doc. + +No runtime canary or lint is proposed — the network isolation makes +violations physically impossible from outside, and inside meta-core the +small file count makes review sufficient. Layer 1 does the heavy +lifting. + +## Migration plan + +Four PRs, additive, ordered so the system stays working throughout. +Closing the network access (PR D) is the moment the lock becomes a +lock; everything before it is opt-in migration that consumers can adopt +at their own pace. + +### PR A — SSE endpoints in meta-core + +Adds `GET /api/events/files` and `GET /api/events/meta`. Implements the +`Last-Event-ID` semantics, gap handling, heartbeats, and the per-stream +XREAD loop. Redis Streams stay reachable directly; consumers can +migrate one at a time. + +Scope, in PR-order: + +1. Backing-store fixes (see "Backing-store retention and lifecycle" + above) — remove the `meta:events` clear-on-restart, add `MAXLEN ~ N` + to every `XAdd`. These are prerequisites for the SSE cursor contract + to hold; without them PR A is not deliverable. +2. SSE handlers for the two streams. +3. Deprecate `GET /api/events/poll` (`watcher/handlers.go:29`). It + exists today as a long-poll variant over the same `file:events` + stream and is described in code as "always empty in practice" + (`watcher/handlers.go:55-58`). After SSE lands it's redundant — keep + it responding 200 for one release with a `Deprecation` header, then + remove. The `/api/scan/*` deprecation pattern in the same file is + the template. +4. Caddy perimeter wiring (see "Auth" subsection above) — explicit + allow-list for `/api/events/files` and `/api/events/meta` matching + whatever the perimeter policy decides. Either inside-only via the + docker network, or public via Caddy with deliberate authorisation. + Do not let the routes default-leak past the hash-lock. + +Roughly 200 lines of Go + ~50 lines of test (covering reconnect from a +cursor, gap emission when the cursor is trimmed, heartbeat cadence, +durability across a meta-core restart). + +### PR B — meta-sort writes via HTTP + +Swap `RedisClient` for an `HttpKVClient` implementation of the existing +`IKVClient` interface in `packages/meta-sort-core`. All writes +(`setMetadataFlat`, `setMetadataProperty`, batch updates) become HTTP +calls against the **`/meta/{hash}` family** — the canonical write +surface for service-to-service traffic. Per-plugin write coalescing +uses `PATCH /meta/{hash}` (already calls `MergeMetadataFlat` +server-side with the partial-update semantics we need; see +`handlers.go:533-566`). + +Reads from Redis stay direct in this PR (no consumer migration yet); +only writes move. Streams stay on direct XREAD. + +**Pre-audit (already done):** meta-sort's production `kvClient.*` +surface is `setMetadataFlat` / `deleteMetadataFlat` / `getMetadataFlat` +/ `getMetadata` / `getAllHashIds` / `health` plus the `file:events` +stream consumer. No orphan keyspaces, no generic `kv.set(arbitrary_key)` +callers — the HTTP migration covers the entire write path with no +side bets. + +### PR C — Stream consumers migrate to SSE + +meta-sort's event consumer (`FileEventConsumer.ts`), meta-fuse's stream +reader, meta-stremio's event hook, meta-dup's monitor, meta-share's +ingest task — each swaps its `XREADGROUP` / `XREAD` loop for an SSE +client. Per-language standard libraries make this small (one HTTP +request, one event-handler closure). + +In the same PR, each consumer's `LeaderClient` (TypeScript x2, Python, +plus any new Rust binding for meta-share if added) drops its +*requirement* on the `redisUrl` field — i.e. it stops opening a Redis +connection. It can still tolerate the field being present (PR D removes +it from the publisher side). This ordering matters: PR D's loud-failure +guard cannot ship until every consumer has been demonstrated to boot +without `redisUrl`. + +The schema indexer is internal to meta-core and unaffected. + +### PR D — Close Redis network access + +PR D must land **after** every consumer in PR C has stopped requiring +`redisUrl` (verify by booting each service against a meta-core build +that already omits the field — staging environment or feature-flagged +release). Until that's true, this PR will deadlock the boot sequence. + +Then, in this order: + +1. Update consumer-side discovery (`LeaderClient` in each service) to + *log loudly* if it sees a `redisUrl` field — guards against + accidental rollback. (Loud log, not hard failure, while the field + is still being written — flip to hard failure once step 4 lands.) +2. Stop publishing `redisUrl` from `/urls` and from `kv-leader.info`. +3. Remove the `meta-core` docker-network alias on the metacore-app + service (other services no longer need DNS for the Redis port). +4. Remove `6380:6379` host port mapping in `dev/docker-compose.yml`. +5. Flip the LeaderClient guard from "warn" to "raise" — any future + consumer that tries to require the field fails loudly at boot. + +After PR D, attempting to connect to Redis from anywhere except +metacore-app produces a connection refused. The lock is real. + +### Out of scope for this work + +- **Redis ACLs / proxy.** Network isolation makes these unnecessary. + Available as an escape hatch if the threat model ever changes. +- **Replacing Redis Streams under the hood.** SSE is the externalised + contract; the backing store is an implementation detail meta-core can + change later. +- **External brokers.** Out of scope absent a concrete driver + (cross-WAN federation at high throughput). + +## Decisions recorded + +| Question | Decision | Notes | +|---|---|---| +| Writes from non-meta-core services? | Forbidden by network topology | Redis unreachable; not a convention | +| Reads from `file:*` direct? | Forbidden by network topology | Same lock, same reason | +| Event streams direct? | No — mediated through SSE | Kubernetes-style. Costs one HTTP hop; buys schema evolution + threat-model fit | +| Enforcement mechanism? | Network isolation (primary) + docs + code review (secondary) | No ACLs, no proxy, no lint | +| Backing store for streams? | Redis Streams (unchanged) | SSE wraps it; could be swapped later without consumer impact | +| Consumer groups / horizontal scale? | Not supported via SSE | Not needed today; revisit if scale ever requires it | +| Storage portability? | Not a goal but a free side-effect | Will not be exercised in alpha | +| **Canonical write API path?** | **`/meta/{hash}` family** | Has GET/PUT/**PATCH**/DELETE plus `/_add/{key}` and per-property `{key:.*}` operations. `PATCH` calls `MergeMetadataFlat` — matches the partial-update semantics PR B needs. `/api/metadata/{hashId}` stays as the editor-internal surface (no PATCH, no per-property routes) but is not the migration target. | +| **SSE event-id contract?** | **Opaque to consumers** | Today it's the Redis Stream entry ID (`-`); consumers MUST treat as opaque so the backing store can be swapped without breaking the wire contract. Documented in the SSE endpoint comment and in this doc. | +| **Typed error envelope?** | **`{error, message, retryable}` with stable `error` slug** | See "Error envelope" subsection below. | +| **Plugin direct-Redis risk?** | **Confirmed: no plugin uses Redis directly** | Grep across `packages/plugins/` for `ioredis` / `redis.createClient` / `new Redis()` returned empty. Plugins POST results back to meta-sort via `/api/plugins/callback`; meta-sort then writes. PR B picks up the plugin write path transparently. | +| **meta-sort non-`file:*` writes?** | **Confirmed: none in production** | Full audit of `kvClient.*` callers in `packages/meta-sort/packages/meta-sort-core/src/` shows only: `setMetadataFlat`, `deleteMetadataFlat`, `getMetadataFlat`, `getMetadata`, `getAllHashIds`, `health`, plus the `file:events` stream consumer (`initStreamConsumer` / `startStreamConsumer` / `processPendingEntries`). The generic `kv.set(key, value)` / `kv.setProperty(key, value)` methods on the interface exist for plugin sandboxes and tests but are not invoked by production code with non-`file:*` keys. PR B + PR C cover the full meta-sort surface. | + +## Error envelope + +Standard shape for all 4xx / 5xx responses from the metadata API: + +```json +{ + "error": "alias_collision", + "message": "CID midhash256:bafk… already resolves to UUID 01JKR…", + "retryable": false +} +``` + +Stable `error` slugs the client should switch on: + +| Slug | HTTP status | Retryable | Meaning | +|---|---|---|---| +| `alias_collision` | 409 | no | A different UUID already owns the CID the caller is trying to register. Caller bug; escalate. | +| `unknown_root` | 404 | no | The hashId / UUID doesn't exist. Caller probably has a stale reference. | +| `unknown_cid` | 404 | no | The CID isn't in the reverse index. No retry helps. | +| `schema_violation` | 400 | no | Body shape rejected (missing required field, invalid value). Caller bug. | +| `storage_unavailable` | 503 | yes | Redis transient. Caller should retry with backoff. | +| `internal` | 500 | yes (cautiously) | Unhandled server-side error. Retry once; escalate if persistent. | + +Today's handlers return `{error, message}` strings; PR A is the natural +place to introduce the typed shape (along with the new SSE endpoints +which will need their own error responses anyway). Going through a +typed envelope from the start avoids the retrofit cost. + +## Remaining open questions + +The audit closed everything that was answerable. What's left: + +- **Deprecation timeline for `/api/metadata/{hashId}`.** It stays during + the migration (editor uses it), but once we've consolidated on + `/meta/{hash}` for service-to-service writes the editor could migrate + too and we'd delete the duplicate surface. Not urgent; revisit + post-PR D when everything else is stable. + +- **What does `/api/events/files` emit during the watcher's initial + bootstrap scan?** When meta-core starts on a populated library, the + watcher discovers N files in quick succession and publishes N `add` + events. A consumer subscribing for the first time will receive that + whole flood. Probably fine — consumers are designed to handle bursts + — but worth a smoke test once PR A lands to confirm meta-share's + ingest doesn't trip on the cardinality. + +## Non-goals + +- **Authentication between internal services.** The hash-lock OIDC + perimeter handles external clients; internal service-to-service + trust is by network isolation (docker network for dev, equivalent + isolation in prod deployments). Adding service-to-service auth is a + separate concern. +- **Replacing the public HTTP API surface.** This proposal exercises + what already exists; the only new endpoints are the two SSE streams. +- **Introducing new transport, broker, proxy, or sidecar.** Everything + builds on what's deployed today. +- **Migrating meta-fuse / meta-stremio / meta-dup direct Redis reads + via XSCAN/GET** (separate from event-stream reads). These are + read-only and don't break schema invariants, but they will fail once + PR D closes Redis access. PR C is the migration window for them — + the same SSE adoption work covers it. diff --git a/docs/uuid-rooted-metadata.md b/docs/uuid-rooted-metadata.md new file mode 100644 index 0000000..3453764 --- /dev/null +++ b/docs/uuid-rooted-metadata.md @@ -0,0 +1,324 @@ +# UUID-Rooted Metadata with CID Aliasing + +## Status + +**Proposed** — design under discussion. Not yet implemented. Supersedes the +"midhash256 as primary key" model currently described in +[metadata-storage-structure.md](metadata-storage-structure.md). + +## Motivation + +Today, every metadata entry in Redis is keyed by a content hash, specifically +midhash256: + +``` +file:midhash256:abc.../filePath +file:midhash256:abc.../title +file:midhash256:abc.../fileinfo/duration +file:midhash256:abc.../tmdb/poster +... +``` + +Other digests (sha256, IPFS CIDs, BitTorrent infohashes) live as fields +*inside* that entry: + +``` +file:midhash256:abc.../cid_sha256 = sha256:def... +file:midhash256:abc.../cid_ipfs = bafy... +``` + +Two problems with this: + +1. **midhash256 is privileged by accident.** It IS the key, while every other + CID is a secondary field. There's nothing intrinsically special about + midhash256 — it's just the first hash computed. Hash-algorithm changes, or + new digest types added by plugins, force rewrites of every root. The + "promotion" workarounds people reach for (re-keying entries when a stronger + hash arrives) solve the wrong problem; the privilege itself is the issue. + +2. **CID lookup is O(n).** Resolving "given any CID, return that file's + metadata" currently scans every indexed entry (see + `storage/client.go::LookupPathByCID`, which acknowledges this in a + comment). Acceptable for admin tooling, unacceptable for a public API + surface — especially when peers on meta-share may query by any CID variant. + +The product requirement is clear: **any CID a client knows about must resolve +to that file's metadata in O(1).** IPFS-compatible CIDs are the externally +meaningful identity for content retrieval; the internal storage layout should +not bias toward any one digest type. + +## Goal + +A single externally-meaningful contract: + +``` +GET /api/meta/ → full metadata document for that file +``` + +Where `` may be any supported type (midhash256, sha256, IPFS, btih, +sha3, future additions). All CID types are treated symmetrically. No CID type +is "the key." + +## Design + +### Two-tier key structure + +The root key becomes an opaque internal UUID; CIDs are aliases that point to +it. + +``` +file:/ ← canonical storage (UUIDv7, never disclosed externally) +cid: → ← reverse index (one Redis key per known CID) +``` + +The metadata keyspace under `file:/` remains open and flat, exactly as +today — anything that wrote `file:/fileinfo/duration` now writes +`file:/fileinfo/duration`. Only the root identifier changes. + +**Example layout:** + +``` +file:01JKR8XW5T4QMV7AHNJ2DEFGHK/filePath = /watch/Inception.mkv +file:01JKR8XW5T4QMV7AHNJ2DEFGHK/size = 4500000000 +file:01JKR8XW5T4QMV7AHNJ2DEFGHK/mtime = 1729872000000000000 +file:01JKR8XW5T4QMV7AHNJ2DEFGHK/cids = SET{midhash256:abc, sha256:def, ipfs:bafy...} +file:01JKR8XW5T4QMV7AHNJ2DEFGHK/duplicates = SET{/watch/dup1.mkv, /watch/dup2.mkv} +file:01JKR8XW5T4QMV7AHNJ2DEFGHK/canonical_cid = ipfs:bafy... +file:01JKR8XW5T4QMV7AHNJ2DEFGHK/title = "Inception" +file:01JKR8XW5T4QMV7AHNJ2DEFGHK/fileinfo/duration = 7200.5 +file:01JKR8XW5T4QMV7AHNJ2DEFGHK/tmdb/poster = bafy... +file:01JKR8XW5T4QMV7AHNJ2DEFGHK/subtitle/eng = ... + +cid:midhash256:abc → 01JKR8XW5T4QMV7AHNJ2DEFGHK +cid:sha256:def → 01JKR8XW5T4QMV7AHNJ2DEFGHK +cid:ipfs:bafy... → 01JKR8XW5T4QMV7AHNJ2DEFGHK +``` + +### UUID format: UUIDv7 + +UUIDv7 chosen over UUIDv4 or 64-bit random: + +- **128 bits**: collision-free under distributed minting; no central allocator. +- **Time-sortable prefix**: `KEYS file:01JKR8*` retrieves entries minted in a + given window. Restores some of the debuggability lost by going opaque. +- **Monotonic-ish**: adjacent UUIDs cluster on disk; modestly better SCAN + locality. + +Encoded as a 26-character Crockford Base32 string (ULID-compatible) for +compactness and case-insensitive matching. + +### New root fields owned by meta-core + +The metadata schema is open — any writer can put any key under +`file:/`. This proposal does not change that. It adds three keys whose +sole writer is meta-core itself: + +| Key | Type | Purpose | +|-----|------|---------| +| `cids` | Redis SET | All CIDs known for this file. Mirror of the `cid:*` reverse index. Source of truth for delete-time GC. | +| `duplicates` | Redis SET | Additional file paths whose content matches this entry. One UUID = one set of metadata; multiple physical files may share that content. Consumed by meta-dup. | +| `canonical_cid` | STRING | The CID this entry should be presented as externally (meta-share announce, UI display). Selected by the rank function (below). | + +Existing meta-core-owned fields (`filePath`, `size`, `mtime`, `midhash256`) +keep their names and writers. Plugin-written fields (`tmdb/*`, `fileinfo/*`, +`subtitle/`, …) are unaffected. + +Redis SET is a departure from the all-strings convention in the current +storage doc, but `file:__index__` is already a SET — so the precedent exists +for using sets where they're the right structure. + +### Canonical CID selection + +`canonical_cid` is a label, not a key. Updating it costs one write — no +rename, no race, no broken external references. + +Selection rule, highest-wins: + +1. IPFS CID (any variant) +2. SHA-256 / SHA-3 family +3. BitTorrent infohash (v2 preferred over v1) +4. midhash256 + +Ranked by **utility for external retrieval**, not by cryptographic strength. +midhash256 and sha256 are both 256-bit, but only sha256/IPFS CIDs are +externally meaningful for content addressing. btih ranks above midhash because +it enables swarm interop even if it's weaker as a digest. + +Reconciler runs on the leader, recomputes `canonical_cid` whenever +`AddAlias` registers a new CID for an entry. + +## Internal operations + +The meta-core storage interface adds five operations. None are exposed +externally — they're internal to meta-core. External callers see only the +CID-based API. + +```go +// Mint creates a new root with no aliases. Called when the watcher +// computes a midhash for a previously-unseen file. Returns the new UUID. +Mint(filePath string, size int64, mtime int64) (uuid string, err error) + +// AddAlias registers a CID as resolving to this UUID. Updates both the +// cid: reverse index and the file:/cids set atomically. +// Triggers canonical_cid reconciliation. +AddAlias(uuid string, cid string) error + +// GetByCID resolves any known CID to its UUID. Returns ("", nil) if unknown. +GetByCID(cid string) (uuid string, err error) + +// MergeIfAliasExists is called when the watcher mints a fresh UUID for a +// file path but, on hashing, discovers the resulting CID already belongs +// to a different UUID. Merges the new (mostly-empty) UUID into the +// existing one, transferring filePath into the duplicates set, and +// returns the canonical UUID. +MergeIfAliasExists(newUUID, cid string) (canonicalUUID string, merged bool, err error) + +// DeleteRoot removes the entire entry, including every cid:* alias and +// every file:/* key. Used for file deletion events. +DeleteRoot(uuid string) error +``` + +### Lifecycle: file arrival + +``` +1. Watcher sees /watch/foo.mkv created. +2. Watcher computes midhash256 → "midhash256:abc...". +3. Watcher calls GetByCID("midhash256:abc..."). + → uuid_X (existing entry) or "" (new file). + +4a. If existing (uuid_X): + - Add "/watch/foo.mkv" to file:/duplicates. + - No new root created. + - meta-dup is notified via the existing keyspace event channel. + +4b. If new: + - Mint("/watch/foo.mkv", size, mtime) → uuid_Y. + - AddAlias(uuid_Y, "midhash256:abc..."). + - Plugins now process the file, referenced by midhash256:abc... + (UUID is never disclosed to them.) + +5. Later, the full-hash plugin computes sha256. + - Plugin reports back: "for CID midhash256:abc..., sha256 is sha256:def...". + - meta-core resolves midhash256:abc → uuid_Y via GetByCID. + - meta-core calls AddAlias(uuid_Y, "sha256:def..."). + - Reconciler updates canonical_cid if sha256 outranks midhash256. +``` + +### Lifecycle: file deletion + +``` +1. Watcher sees /watch/foo.mkv deleted. +2. Resolve which entry owns this path: + a. If /watch/foo.mkv is in some entry's duplicates set: + - Remove from set. Done. + b. If /watch/foo.mkv is the filePath of some entry: + - If duplicates is non-empty: promote one duplicate to filePath. + - If duplicates is empty: DeleteRoot(uuid). + - Iterates the cids set, deletes each cid: reverse-index key. + - Deletes every file:/* key. + - Removes uuid from file:__index__. +``` + +## Public API + +``` +GET /api/meta/ + → 200 { "filePath": "...", "canonical_cid": "...", + "cids": [...], "title": "...", "tmdb": {...}, ... } + → 404 if no such CID + +GET /api/file/ (already exists — implementation now O(1)) + → streams file content +``` + +The UUID is **not** included in the response. Clients address everything by +CID; meta-core's choice of internal identifier remains a private +implementation detail and may change. + +Auth: `/api/meta/` follows the same pattern as `/api/file/` and +bypasses the auth perimeter. It's the metadata twin of the content endpoint, +and meta-share calls it from federated search responses. + +## Cross-service impact + +This is a meta-core schema change. Other services interact only through CIDs, +so the impact is contained but real. + +- **meta-sort**: writers must resolve plugin callbacks to a UUID before + writing. The existing `LeaderClient` gains a `getByCID` cache. +- **meta-fuse**: VFS rebuild already iterates `file:__index__`; only the + ID format changes. +- **meta-stremio**: queries by CID; needs no logic change beyond its storage + client. +- **meta-dup**: gains a cleaner data source — duplicates are now explicit in + the `duplicates` set, not implicit in colliding writes. +- **meta-share**: announces `canonical_cid` rather than midhash. Federated + search responses include all CIDs from the `cids` set so any peer querying + by any digest gets a hit. +- **Plugins**: no change. They speak CIDs only. + +## Migration + +One-time sweep, idempotent, runs on first boot after upgrade: + +``` +For each hashID in file:__index__ (current shape: midhash256:abc, sha256:def, …): + 1. Mint a fresh uuid_Y. + 2. AddAlias(uuid_Y, hashID). The existing primary becomes an alias. + 3. For each cid_* field on the old entry, AddAlias(uuid_Y, that value). + 4. RENAME file:/ → file:/ for every + property (Lua script, one entry at a time; non-atomic across entries, + but each entry stays internally consistent). + 5. Reconcile canonical_cid. + 6. Remove hashID from file:__index__, add uuid_Y. +``` + +Crash-recoverable: re-running the sweep finds entries with both old and new +roots and resumes by detecting which step completed (presence of +`file:/cids` indicates step 3 finished; absence of any +`file:/*` keys indicates step 4 finished). + +## Decisions recorded + +| Question | Decision | Notes | +|---|---|---| +| When is the UUID minted? | At first-hash | Pragmatic for the initial migration. Long-term target: at discovery, fully decoupled from hashing. Tracked as an open question below. | +| UUID format | UUIDv7, 26-char Crockford Base32 | Time-sortable, 128-bit, distributed-safe. | +| UUID visible externally? | No | Pure meta-core internal. Clients address by CID only. | +| Merge semantics | One UUID per content; additional paths into `duplicates` | One file (logical) = one metadata set. Multiple physical paths supported via `duplicates`. | +| Canonical CID rank | IPFS > sha-family > btih > midhash | Utility-weighted, not strength-weighted. | +| `cids` and `duplicates` field type | Redis SET | Departure from "all-strings" convention but consistent with the existing `file:__index__` precedent. | +| Reverse-index format | Flat `cid:` STRING keys | Simple, scan-friendly, no hot-key contention. | + +## Open questions + +- **Discovery-time UUID minting.** The cleaner long-term model: watcher mints + a UUID at discovery (before hashing), then `AddAlias` runs when the midhash + arrives. Requires an explicit "reconcile if alias already exists" step + because two paths with the same content would each get their own UUID until + hashing reveals the collision. Worth doing eventually; not required for + this iteration. + +- **meta-share announce strategy.** Announce only `canonical_cid`, or every + entry in `cids`? Maximum discoverability vs. DHT footprint. Decide as part + of the meta-share work, not here. + +- **IPFS gateway endpoint.** Should meta-core expose `/ipfs/` as a + content-streaming endpoint for external IPFS clients? Adjacent to this + design but not part of it. + +- **Keyspace-notification consumers.** Today's parsers (`schema/indexer.go`) + handle `file::/...` with a special split rule for the colon + inside `midhash256:abc`. With UUIDs (no colons), `parseFieldPath` simplifies + — but any external consumer subscribed by hashID pattern needs updating. + Inventory those before flipping the migration. + +## Non-goals + +- **Not** a content-replication or IPFS-pinning system. Registering an IPFS + CID alias does not pin the file anywhere — it only states "if somebody + hashes this file with IPFS's chunker, this is the CID they'd get." +- **Not** a versioning scheme. Metadata is a snapshot of current state; + historical versions are out of scope. +- **Not** a strong-consistency system. Reverse-index writes happen under the + leader lock; reads from followers may briefly miss a freshly-added alias. diff --git a/internal/api/admin_migrate.go b/internal/api/admin_migrate.go new file mode 100644 index 0000000..e27f127 --- /dev/null +++ b/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, + }) +} diff --git a/internal/api/cid_resolve.go b/internal/api/cid_resolve.go new file mode 100644 index 0000000..1676423 --- /dev/null +++ b/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, + }) +} diff --git a/internal/api/handlers.go b/internal/api/handlers.go index 9d7aa40..421d75b 100644 --- a/internal/api/handlers.go +++ b/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) diff --git a/internal/api/server.go b/internal/api/server.go index 475124e..d4b87c7 100644 --- a/internal/api/server.go +++ b/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") diff --git a/internal/api/sse_events.go b/internal/api/sse_events.go new file mode 100644 index 0000000..8a72dcd --- /dev/null +++ b/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 (`-`); 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 "-". +// 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") +} diff --git a/internal/cid/rank.go b/internal/cid/rank.go new file mode 100644 index 0000000..ec270a3 --- /dev/null +++ b/internal/cid/rank.go @@ -0,0 +1,91 @@ +// Package cid centralizes CID-format helpers: ranking by utility for +// external presentation, normalizing prefixed CID tokens, and the rules for +// picking which CID is the canonical externally-visible identifier of a +// file. +// +// A "CID token" in this codebase is the prefixed form ":", +// e.g. "midhash256:bafkr4ih...", "sha256:bafkrei...", "ipfs:bafy...". The +// prefix makes the type recoverable without parsing the multicodec, which is +// what the reverse index (cid: → uuid), the per-root cids set, and +// the canonical_cid field all rely on. +package cid + +import "strings" + +// Rank returns a score for a CID token. Higher = more preferred as the +// public-facing "canonical" identifier for a file. Unknown algorithms rank +// below all known ones (0) so a plugin that invents a new digest type +// can't accidentally outrank an established one until it's registered here. +// +// Ordering rationale (utility-weighted, NOT cryptographic-strength): +// - IPFS CIDs win because the network around them — gateways, swarm, +// content-addressing tooling — has the broadest reach. A cryptographically +// equivalent sha256 in a non-CID form has less downstream value. +// - sha256 / sha3 follow: standard, widely recognized, no platform lock-in. +// - BitTorrent infohashes (btih) outrank midhash256 not for strength +// (v1 is sha1, weaker than midhash256's sha256 inner) but for swarm +// interop — a btih makes a file discoverable on existing torrent +// infrastructure that knows nothing about MetaMesh. +// - midhash256 is the floor. Internally meaningful, externally invisible. +func Rank(token string) int { + switch AlgorithmOf(token) { + case "ipfs": + return 40 + case "sha256", "sha3", "sha3-256", "sha3-384", "sha3-512": + return 30 + case "btih", "btih-v2": + return 20 + case "midhash256": + return 10 + default: + return 0 + } +} + +// AlgorithmOf returns the algorithm prefix of a CID token, or "" if the +// token is malformed (no colon, or empty prefix). +func AlgorithmOf(token string) string { + colon := strings.IndexByte(token, ':') + if colon <= 0 { + return "" + } + return token[:colon] +} + +// Token assembles a CID token from its algorithm and raw value. Mirror of +// AlgorithmOf — callers writing to the reverse index, cids set, or +// canonical_cid field route through this so the format stays consistent. +func Token(algorithm, raw string) string { + if algorithm == "" || raw == "" { + return "" + } + return algorithm + ":" + raw +} + +// ValueOf returns the raw value portion of a CID token (everything after +// the first colon). Returns "" for malformed tokens. +func ValueOf(token string) string { + colon := strings.IndexByte(token, ':') + if colon <= 0 || colon == len(token)-1 { + return "" + } + return token[colon+1:] +} + +// Better picks the higher-ranked of two tokens. On ties, returns the +// lexicographically smaller one — gives the reconciler a deterministic +// answer so an unchanged canonical_cid doesn't churn when peers add the +// same digest under a different ordering. +func Better(a, b string) string { + ra, rb := Rank(a), Rank(b) + if ra > rb { + return a + } + if rb > ra { + return b + } + if a <= b { + return a + } + return b +} diff --git a/internal/cid/rank_test.go b/internal/cid/rank_test.go new file mode 100644 index 0000000..ce2cf61 --- /dev/null +++ b/internal/cid/rank_test.go @@ -0,0 +1,85 @@ +package cid + +import "testing" + +func TestRank_Ordering(t *testing.T) { + cases := []struct { + token string + want int + }{ + {"ipfs:bafy", 40}, + {"sha256:bafk", 30}, + {"sha3-256:abc", 30}, + {"btih:def", 20}, + {"btih-v2:def", 20}, + {"midhash256:bafk", 10}, + {"unknown:xyz", 0}, + {"", 0}, + {"bare-no-colon", 0}, + } + for _, tc := range cases { + if got := Rank(tc.token); got != tc.want { + t.Errorf("Rank(%q) = %d, want %d", tc.token, got, tc.want) + } + } +} + +func TestRank_StrictOrdering(t *testing.T) { + // The exact numbers don't matter, but the inequality chain does — this + // is what the canonical_cid reconciler relies on. + ipfs := Rank("ipfs:x") + sha := Rank("sha256:x") + btih := Rank("btih:x") + mid := Rank("midhash256:x") + unknown := Rank("unknown:x") + + if !(ipfs > sha && sha > btih && btih > mid && mid > unknown) { + t.Fatalf("rank ordering broken: ipfs=%d sha=%d btih=%d mid=%d unknown=%d", + ipfs, sha, btih, mid, unknown) + } +} + +func TestAlgorithmOf(t *testing.T) { + cases := map[string]string{ + "ipfs:bafy": "ipfs", + "sha256:bafk": "sha256", + "midhash256:abc": "midhash256", + "": "", + "no-colon": "", + ":leading-colon": "", // empty algo not allowed + } + for in, want := range cases { + if got := AlgorithmOf(in); got != want { + t.Errorf("AlgorithmOf(%q) = %q, want %q", in, got, want) + } + } +} + +func TestToken(t *testing.T) { + if got := Token("sha256", "abc"); got != "sha256:abc" { + t.Errorf("Token(sha256, abc) = %q", got) + } + if got := Token("", "abc"); got != "" { + t.Errorf("Token with empty algo should return empty, got %q", got) + } + if got := Token("sha256", ""); got != "" { + t.Errorf("Token with empty value should return empty, got %q", got) + } +} + +func TestBetter(t *testing.T) { + // IPFS beats sha256 + if got := Better("ipfs:a", "sha256:b"); got != "ipfs:a" { + t.Errorf("Better(ipfs, sha256) = %q, want ipfs:a", got) + } + if got := Better("sha256:b", "ipfs:a"); got != "ipfs:a" { + t.Errorf("Better(sha256, ipfs) = %q, want ipfs:a (order shouldn't matter)", got) + } + // Tie → lexicographically smaller wins, deterministic + if got := Better("ipfs:b", "ipfs:a"); got != "ipfs:a" { + t.Errorf("Better(ipfs:b, ipfs:a) = %q, want ipfs:a", got) + } + if got := Better("ipfs:a", "ipfs:b"); got != "ipfs:a" { + t.Errorf("Better(ipfs:a, ipfs:b) = %q, want ipfs:a", got) + } +} diff --git a/internal/events/meta_publisher.go b/internal/events/meta_publisher.go index 9fa8d71..4967bad 100644 --- a/internal/events/meta_publisher.go +++ b/internal/events/meta_publisher.go @@ -14,6 +14,12 @@ import ( const ( // MetaEventsStream is the Redis stream name for metadata events MetaEventsStream = "meta:events" + + // MetaEventsMaxLen is the approximate cap on stream length. Bounded + // retention is a contract requirement for the SSE wire — see + // docs/api-mediated-access.md "Backing-store retention and lifecycle". + // Cheap (~) trim keeps growth bounded without serialising writes. + MetaEventsMaxLen = 100_000 ) // MetaPublisher subscribes to Redis keyspace notifications @@ -37,7 +43,12 @@ func NewMetaPublisher(client *redis.Client) *MetaPublisher { } } -// Start begins subscribing to keyspace notifications +// Start begins subscribing to keyspace notifications. +// +// The stream is NOT truncated on startup — the SSE wire contract requires +// Last-Event-ID resumption to survive a meta-core restart, which means +// entries must persist across restart. Growth is bounded instead by +// MAXLEN ~ MetaEventsMaxLen on every XAdd (see publishEvent). func (p *MetaPublisher) Start() error { p.mu.Lock() if p.running { @@ -47,16 +58,6 @@ func (p *MetaPublisher) Start() error { p.running = true p.mu.Unlock() - // Clear the stream on startup - consumers will re-bootstrap from Redis keys - // This ensures the stream only contains events from the current session - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - if err := p.client.Del(ctx, MetaEventsStream).Err(); err != nil { - log.Printf("[MetaPublisher] Warning: failed to clear stream on startup: %v", err) - } else { - log.Printf("[MetaPublisher] Cleared %s stream for fresh start", MetaEventsStream) - } - p.wg.Add(1) go p.subscribeLoop() @@ -133,9 +134,13 @@ func (p *MetaPublisher) publishEvent(channel, operation string) { "ts": fmt.Sprintf("%d", time.Now().UnixMilli()), } - // Add to stream (no MaxLen - stream is cleared on meta-core restart) + // Approximate MAXLEN trim keeps the stream bounded across restarts + // (see Start). Cap is intentionally generous; most consumers reconnect + // well within retention. _, err := p.client.XAdd(p.ctx, &redis.XAddArgs{ Stream: MetaEventsStream, + MaxLen: MetaEventsMaxLen, + Approx: true, Values: fields, }).Result() @@ -154,6 +159,8 @@ func (p *MetaPublisher) publishEventDirect(operation, key string) { _, err := p.client.XAdd(p.ctx, &redis.XAddArgs{ Stream: MetaEventsStream, + MaxLen: MetaEventsMaxLen, + Approx: true, Values: fields, }).Result() diff --git a/internal/leader/election.go b/internal/leader/election.go index b693969..774322d 100644 --- a/internal/leader/election.go +++ b/internal/leader/election.go @@ -75,7 +75,11 @@ func (p *LeaderInfoProvider) buildLeaderInfo() *LeaderLockInfo { Hostname: hostname, BaseUrl: baseUrl, ApiUrl: fmt.Sprintf("http://%s:%d", ip, p.config.HTTPPort), - RedisUrl: fmt.Sprintf("redis://%s:%d", ip, p.config.RedisPort), + // RedisUrl deliberately empty — api-mediated-access PR D removes + // direct Redis exposure. Consumers route metadata I/O through + // meta-core's HTTP API (/meta/{hash}*) and event streams through + // SSE (/api/events/{files,meta}). + RedisUrl: "", WebdavUrl: webdavUrl, WebdavUrlInternal: webdavUrlInternal, Timestamp: time.Now().UnixMilli(), diff --git a/internal/schema/indexer.go b/internal/schema/indexer.go index 8804e7d..c2732d7 100644 --- a/internal/schema/indexer.go +++ b/internal/schema/indexer.go @@ -193,9 +193,10 @@ func (i *Indexer) processEvent(values map[string]any) { } } -// parseFieldPath turns "file:/" into the logical field -// path used as the map key. hashID may itself contain ':' (e.g. -// "midhash256:abc123"); the split is on the first '/'. +// parseFieldPath turns "file:/" into the logical field +// path used as the map key. rootID is opaque to the indexer — currently a +// UUIDv7, historically a midhash256:abc token; both shapes parse identically +// because the split is on the first '/' after "file:". // // When the last segment of propertyPath matches ^[a-z]{2,3}$ it is treated as // a language code: the field name collapses to the prefix and key_hint is set. diff --git a/internal/snapshot/import.go b/internal/snapshot/import.go index 03d1329..50f2523 100644 --- a/internal/snapshot/import.go +++ b/internal/snapshot/import.go @@ -108,6 +108,10 @@ func (i *Importer) Import(r io.ReaderAt, size int64, opts ImportOptions) (*Impor return nil, fmt.Errorf("snapshot schemaVersion %d exceeds supported %d", manifest.SchemaVersion, ManifestSchemaVersion) } + if manifest.SchemaVersion < ManifestSchemaVersion { + return nil, fmt.Errorf("snapshot schemaVersion %d is no longer supported (current: %d); v1 used content-hash root keys which are incompatible with the UUID-rooted schema", + manifest.SchemaVersion, ManifestSchemaVersion) + } result := &ImportResult{ Mode: opts.Mode, diff --git a/internal/snapshot/manifest.go b/internal/snapshot/manifest.go index 9877316..3e6e95d 100644 --- a/internal/snapshot/manifest.go +++ b/internal/snapshot/manifest.go @@ -1,15 +1,26 @@ // Package snapshot implements export/import/wipe of MetaMesh metadata. // // The on-disk format is a ZIP archive containing one nested-JSON file per -// known CID (cid_midhash256). Schema is defined by ManifestSchemaVersion; -// readers must reject archives whose schemaVersion exceeds what they know. +// indexed root. Schema is defined by ManifestSchemaVersion; readers must +// reject archives whose schemaVersion exceeds what they know. +// +// v2 (current): root identifiers are UUIDv7 (Crockford Base32). Entries +// under metadata/.json carry the flat string properties; the cids +// and duplicates Redis SETs are NOT serialized yet — they are rebuilt +// transparently on import because SetMetadataFlat / MergeMetadataFlat +// auto-register reverse-index aliases for any cid_*/midhash256 field they +// see (see internal/storage/cid_resolution.go). +// +// v1 (legacy): roots were content-hash-typed tokens like "midhash256:abc". +// v1 archives can no longer be imported — they would resurrect the +// privileged-hashID schema this layer was built to retire. package snapshot import "time" // ManifestSchemaVersion is the on-disk format version. Bump when the layout -// of metadata/{cid}.json or the manifest changes in a non-additive way. -const ManifestSchemaVersion = 1 +// of metadata/{root}.json or the manifest changes in a non-additive way. +const ManifestSchemaVersion = 2 // MetadataDir is the path inside the ZIP that holds per-CID JSON files. const MetadataDir = "metadata/" diff --git a/internal/storage/cid_resolution.go b/internal/storage/cid_resolution.go new file mode 100644 index 0000000..faef938 --- /dev/null +++ b/internal/storage/cid_resolution.go @@ -0,0 +1,500 @@ +package storage + +import ( + "context" + "fmt" + "strconv" + "strings" + "time" + + "github.com/metazla/meta-core/internal/cid" + "github.com/redis/go-redis/v9" +) + +// This file implements the CID-resolution layer that lives on top of the +// flat metadata storage in client.go. The shape is described in detail in +// docs/uuid-rooted-metadata.md; the short version: +// +// - Roots are opaque UUIDv7s — file:/ for every property +// - Every known CID for a file is registered as a reverse-index entry: +// cid:: → (Redis STRING) +// and mirrored in a per-root set for fast enumeration: +// file:/cids (Redis SET) +// - A canonical_cid field tracks the best (rank-wise) CID to advertise +// externally; updated by reconcileCanonicalCIDLocked whenever the set +// changes +// +// Auto-alias hooks in SetProperty/SetMetadataFlat/MergeMetadataFlat catch +// any write of a cid_* or midhash256 field and register the alias, so +// plugins that just write their result fields don't need to know about +// the reverse index at all. The hook helpers are *Locked — they assume +// c.mu is already held by the caller, avoiding nested RLock (which can +// deadlock against a waiting writer). + +// CIDIndexPrefix is the Redis key prefix for the reverse index. Combined +// with a CID token, it forms the full key: "cid:midhash256:bafk…". +const CIDIndexPrefix = "cid:" + +// CIDsField is the per-root SET key that mirrors all CID aliases for a file. +// Stored at file:/cids — read by reconcileCanonicalCIDLocked and by +// DeleteRoot to find every reverse-index entry that needs to be cleaned up. +const CIDsField = "cids" + +// CanonicalCIDField is the per-root STRING key that stores the externally +// preferred CID for this file. Picked by cid.Better over the cids set. +const CanonicalCIDField = "canonical_cid" + +// DuplicatesField is the per-root SET key holding additional file paths +// whose content hashes to the same CIDs as this root. The watcher writes +// here when it sees a new path for content that's already indexed; meta-dup +// reads from it. +const DuplicatesField = "duplicates" + +// buildCIDIndexKey returns the absolute Redis key for a reverse-index entry. +// token is a CID in prefixed form ("midhash256:bafk…", "sha256:…", etc.). +func (c *Client) buildCIDIndexKey(token string) string { + return c.buildKey(CIDIndexPrefix + token) +} + +// Mint creates a new root with filePath/sizeByte/mtimeNano and registers it +// in the file index. Returns the new UUID. The caller is expected to follow +// up with AddAlias for at least one CID — a root without any alias is +// reachable only via the index, which works but defeats the point. +func (c *Client) Mint(filePath string, size, mtimeNano int64) (string, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return "", fmt.Errorf("not connected") + } + + uuid, err := NewUUID() + if err != nil { + return "", fmt.Errorf("uuid: %w", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + prefix := c.buildKeyPrefix(uuid) + pipe := c.client.Pipeline() + pipe.Set(ctx, prefix+"filePath", filePath, 0) + pipe.Set(ctx, prefix+"sizeByte", strconv.FormatInt(size, 10), 0) + pipe.Set(ctx, prefix+"mtimeNano", strconv.FormatInt(mtimeNano, 10), 0) + pipe.SAdd(ctx, c.buildIndexKey(), uuid) + if _, err := pipe.Exec(ctx); err != nil { + return "", fmt.Errorf("mint pipeline: %w", err) + } + return uuid, nil +} + +// AddAlias registers a CID token as resolving to uuid. Idempotent: writing +// the same alias twice is a no-op. After the alias is registered, the +// canonical_cid field is reconciled — picks the highest-ranked token in +// the cids set and writes it to canonical_cid if it changed. +func (c *Client) AddAlias(uuid, token string) error { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + return c.addAliasLocked(ctx, uuid, token) +} + +// addAliasLocked is the lock-free body of AddAlias. The caller must already +// hold c.mu (read or write). Splits out so the alias-maintenance hooks +// invoked from SetProperty / SetMetadataFlat / MergeMetadataFlat can reuse +// the work without re-entering the lock — nested RLock can deadlock +// against a waiting writer. +func (c *Client) addAliasLocked(ctx context.Context, uuid, token string) error { + if uuid == "" || token == "" { + return fmt.Errorf("uuid and token are required") + } + if cid.AlgorithmOf(token) == "" { + return fmt.Errorf("token %q is not in : form", token) + } + + // Self-pointing alias guard. If the caller is trying to register + // `cid::` → `` (i.e. the root IS the cid value), + // that's the dual-root failure mode in disguise — it means somebody + // wrote /meta/ without first resolving via ResolveRoot. Refuse + // the alias write; leave the existing alias (if any) alone. Still + // add to this root's cids set so the field-level write succeeds. + if uuid == cid.ValueOf(token) { + if _, err := c.client.SAdd(ctx, c.buildKeyPrefix(uuid)+CIDsField, token).Result(); err != nil { + return fmt.Errorf("sadd cids set: %w", err) + } + return c.reconcileCanonicalCIDLocked(ctx, uuid) + } + + // Guard against the dual-root pattern: if `cid:` already points + // at a DIFFERENT uuid, don't silently overwrite. The historical bug: + // meta-sort writing cid_midhash256= to root "" caused the hook + // to register cid:midhash256: → , severing the watcher's prior + // cid:midhash256: → alias. With write-path resolution + // now in place callers shouldn't hit this anymore, but the guard is + // cheap and prevents regression if a new caller bypasses ResolveRoot. + indexKey := c.buildCIDIndexKey(token) + existing, err := c.client.Get(ctx, indexKey).Result() + if err != nil && err != redis.Nil { + return fmt.Errorf("read existing alias: %w", err) + } + if existing != "" && existing != uuid { + // Existing alias points elsewhere. Leave it alone and just add the + // token to this root's cids set so it's still discoverable via the + // root → cids enumeration. + if _, err := c.client.SAdd(ctx, c.buildKeyPrefix(uuid)+CIDsField, token).Result(); err != nil { + return fmt.Errorf("sadd cids set: %w", err) + } + return c.reconcileCanonicalCIDLocked(ctx, uuid) + } + + prefix := c.buildKeyPrefix(uuid) + pipe := c.client.Pipeline() + pipe.Set(ctx, indexKey, uuid, 0) + pipe.SAdd(ctx, prefix+CIDsField, token) + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("add alias pipeline: %w", err) + } + return c.reconcileCanonicalCIDLocked(ctx, uuid) +} + +// ResolveRoot maps a hash supplied by a caller (a meta-sort write, an +// editor lookup, an SSE consumer) to the actual storage root key. +// +// Callers historically send a bare midhash256 value (e.g. +// "bagacba…") because that's what the watcher publishes on file:events. +// The UUID-rooted-metadata design (see docs/uuid-rooted-metadata.md) +// keeps roots opaque and aliases CIDs into the reverse index: +// +// cid:midhash256:bagacba… → 01KSBHM* +// +// Without resolution, every /meta/ write creates a parallel +// midhash-rooted entry that the watcher's UUID never sees. The editor +// then shows two entries per file: the (rich) midhash root and the +// (empty) UUID root containing only `duplicates`. +// +// Behaviour: +// - empty hash → empty (caller's problem) +// - hash matches cid:midhash256: in the reverse index → return UUID +// - otherwise return hash unchanged (legacy / explicit-UUID writes) +// +// Errors are converted to "no resolution" — if Redis is down or the +// lookup fails we let the caller proceed with the bare hash and surface +// the Redis failure on the actual operation. +func (c *Client) ResolveRoot(hash string) string { + if hash == "" { + return hash + } + uuid, err := c.GetByCID("midhash256:" + hash) + if err != nil || uuid == "" { + return hash + } + return uuid +} + +// GetByCID resolves a CID token to the UUID of its root. Returns ("", nil) +// when the token is unknown; the empty string is the "not found" signal +// rather than a typed error so callers can branch without errors.As. +func (c *Client) GetByCID(token string) (string, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return "", fmt.Errorf("not connected") + } + if token == "" { + return "", nil + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + uuid, err := c.client.Get(ctx, c.buildCIDIndexKey(token)).Result() + if err == redis.Nil { + return "", nil + } + if err != nil { + return "", fmt.Errorf("get cid index: %w", err) + } + return uuid, nil +} + +// AddDuplicatePath records that path is another physical location whose +// content matches uuid. Returns true if path was newly added (not already +// in the set). The watcher calls this when it sees a file whose midhash +// matches an existing root but at a different filePath. +func (c *Client) AddDuplicatePath(uuid, path string) (bool, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return false, fmt.Errorf("not connected") + } + if uuid == "" || path == "" { + return false, nil + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + added, err := c.client.SAdd(ctx, c.buildKeyPrefix(uuid)+DuplicatesField, path).Result() + if err != nil { + return false, fmt.Errorf("sadd duplicates: %w", err) + } + return added > 0, nil +} + +// DeleteRoot removes everything keyed off uuid: every file:/* property, +// every reverse-index entry listed in the cids set, and the index membership. +// Idempotent — a uuid with no remaining keys is a successful no-op. +func (c *Client) DeleteRoot(uuid string) error { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return fmt.Errorf("not connected") + } + if uuid == "" { + return nil + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + prefix := c.buildKeyPrefix(uuid) + + // 1. Read the cids set first — once we delete file:/* we lose the + // list of reverse-index entries to clean up. Tolerate a missing set + // (empty list = nothing to unmap). + tokens, err := c.client.SMembers(ctx, prefix+CIDsField).Result() + if err != nil && err != redis.Nil { + return fmt.Errorf("smembers cids: %w", err) + } + + // 2. Delete every reverse-index entry. Build the keys upfront, then DEL + // in one shot so we don't pay per-RTT. + if len(tokens) > 0 { + keys := make([]string, 0, len(tokens)) + for _, t := range tokens { + keys = append(keys, c.buildCIDIndexKey(t)) + } + if err := c.client.Del(ctx, keys...).Err(); err != nil { + return fmt.Errorf("del cid index: %w", err) + } + } + + // 3. Scan and delete every file:/* key. + var cursor uint64 + for { + keys, next, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result() + if err != nil { + return fmt.Errorf("scan root keys: %w", err) + } + if len(keys) > 0 { + if err := c.client.Del(ctx, keys...).Err(); err != nil { + return fmt.Errorf("del root keys: %w", err) + } + } + cursor = next + if cursor == 0 { + break + } + } + + // 4. Remove from the file index set. + if err := c.client.SRem(ctx, c.buildIndexKey(), uuid).Err(); err != nil { + return fmt.Errorf("srem index: %w", err) + } + return nil +} + +// reconcileCanonicalCIDLocked reads the cids set, picks the highest-ranked +// token, and updates the canonical_cid field if it changed. Called from +// addAliasLocked after a new alias is registered. Caller must hold c.mu. +func (c *Client) reconcileCanonicalCIDLocked(ctx context.Context, uuid string) error { + prefix := c.buildKeyPrefix(uuid) + + tokens, err := c.client.SMembers(ctx, prefix+CIDsField).Result() + if err != nil { + return fmt.Errorf("smembers cids: %w", err) + } + if len(tokens) == 0 { + return nil + } + + best := tokens[0] + for _, t := range tokens[1:] { + best = cid.Better(best, t) + } + + canonicalKey := prefix + CanonicalCIDField + current, err := c.client.Get(ctx, canonicalKey).Result() + if err != nil && err != redis.Nil { + return fmt.Errorf("get canonical_cid: %w", err) + } + if current == best { + return nil + } + if err := c.client.Set(ctx, canonicalKey, best, 0).Err(); err != nil { + return fmt.Errorf("set canonical_cid: %w", err) + } + return nil +} + +// maybeAddAliasFromFieldLocked inspects a (field, value) pair written into +// a root. If it looks like a CID field — name is "midhash256" or starts +// with "cid_" — it registers the implied CID token as an alias. Plugins +// write CID fields without knowing about the reverse index; this hook +// keeps the index in sync transparently. +// +// Caller must hold c.mu. Errors are intentionally swallowed at the call +// site (logged, not returned): an out-of-sync reverse index is recoverable +// via a sweep, while a failed property write is not. +func (c *Client) maybeAddAliasFromFieldLocked(ctx context.Context, uuid, field, value string) error { + token := cidTokenFromField(field, value) + if token == "" { + return nil + } + return c.addAliasLocked(ctx, uuid, token) +} + +// maybeAddAliasesFromMetadataLocked bulk-applies maybeAddAliasFromFieldLocked +// over every entry in a metadata map. Used by SetMetadataFlat / +// MergeMetadataFlat after the underlying pipeline write succeeds. Caller +// must hold c.mu. +func (c *Client) maybeAddAliasesFromMetadataLocked(ctx context.Context, uuid string, metadata map[string]string) { + for k, v := range metadata { + _ = c.maybeAddAliasFromFieldLocked(ctx, uuid, k, v) + } +} + +// cidTokenFromField turns a (field-name, raw-value) pair into a normalized +// CID token ":". Returns "" for non-CID fields. +// +// Acceptable inputs: +// - field="midhash256", value="bafk…" → "midhash256:bafk…" +// - field="cid_sha256", value="bafk…" → "sha256:bafk…" +// - field="cid_ipfs", value="ipfs:bafy…" → "ipfs:bafy…" (trusted) +// +// Plugins that already use the prefixed form get their value passed through; +// plugins that store the bare CID get prefixed from the field name. The +// detection heuristic is just "value contains a colon" — base32lower CIDs +// never contain colons, so this is unambiguous. +func cidTokenFromField(field, value string) string { + if value == "" { + return "" + } + var algo string + switch { + case field == "midhash256": + algo = "midhash256" + case strings.HasPrefix(field, "cid_"): + algo = strings.TrimPrefix(field, "cid_") + default: + return "" + } + if algo == "" { + return "" + } + if strings.IndexByte(value, ':') > 0 { + return value + } + return cid.Token(algo, value) +} + +// MetadataDocument is the complete state of one file as the public API +// returns it: every flat string field, plus the cids and duplicates sets +// hoisted into typed arrays. Distinct from GetMetadataFlat (which only +// returns string keys) — clients querying /api/meta/ want the SETs +// expanded too, since "what other CIDs identify this file" and "what other +// paths point at this content" are core to the response. +type MetadataDocument struct { + Flat map[string]string `json:"metadata"` + CIDs []string `json:"cids"` + Duplicates []string `json:"duplicates"` +} + +// GetMetadataDocument returns the full document for a root: every string +// property under file:/* plus the cids and duplicates SETs. Returns +// nil if the root has no keys at all (deleted or never existed). +func (c *Client) GetMetadataDocument(uuid string) (*MetadataDocument, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return nil, fmt.Errorf("not connected") + } + if uuid == "" { + return nil, nil + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + flat, err := c.getMetadataFlatInternal(ctx, uuid) + if err != nil { + return nil, err + } + prefix := c.buildKeyPrefix(uuid) + + cids, err := c.client.SMembers(ctx, prefix+CIDsField).Result() + if err != nil && err != redis.Nil { + return nil, fmt.Errorf("smembers cids: %w", err) + } + dups, err := c.client.SMembers(ctx, prefix+DuplicatesField).Result() + if err != nil && err != redis.Nil { + return nil, fmt.Errorf("smembers duplicates: %w", err) + } + + if len(flat) == 0 && len(cids) == 0 && len(dups) == 0 { + return nil, nil + } + return &MetadataDocument{ + Flat: flat, + CIDs: cids, + Duplicates: dups, + }, nil +} + +// lookupSidecarPathByCID is the fallback that LookupPathByCID falls into +// when the token isn't a known root alias. Used to resolve poster/backdrop +// CIDs, which are stored as values pointing at sidecar files (posterPath / +// backdropPath) rather than registered as aliases of the file itself. +// +// O(n) — walks every root. Acceptable because (a) sidecar lookups are rare +// vs. file CID lookups and (b) the alternative is registering sidecars in +// the reverse index, which conflates "this CID is the file" with "this CID +// is an image associated with the file." +func (c *Client) lookupSidecarPathByCID(token string) (string, error) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + hashIDs, err := c.getAllHashIDsInternal(ctx) + if err != nil { + return "", err + } + + for _, hashID := range hashIDs { + meta, err := c.getMetadataFlatInternal(ctx, hashID) + if err != nil || len(meta) == 0 { + continue + } + if meta["poster"] == token { + if p := meta["posterPath"]; p != "" { + return normalizeFilesRelativePath(p), nil + } + } + if meta["backdrop"] == token { + if p := meta["backdropPath"]; p != "" { + return normalizeFilesRelativePath(p), nil + } + } + } + return "", nil +} diff --git a/internal/storage/client.go b/internal/storage/client.go index 666f5a9..f378a4b 100644 --- a/internal/storage/client.go +++ b/internal/storage/client.go @@ -192,6 +192,10 @@ func (c *Client) SetMetadataFlat(hashID string, metadata map[string]string) erro return fmt.Errorf("sadd index failed: %w", err) } + // Auto-register reverse-index aliases for any cid_* or midhash256 fields + // in the payload. See cid_resolution.go for the rules. + c.maybeAddAliasesFromMetadataLocked(ctx, hashID, metadata) + return nil } @@ -263,11 +267,17 @@ func (c *Client) SetProperty(hashID, property, value string) error { return fmt.Errorf("sadd index failed: %w", err) } + // Auto-register reverse-index alias if this is a CID field. + _ = c.maybeAddAliasFromFieldLocked(ctx, hashID, property, value) + return nil } // DeleteMetadata deletes all metadata for a file // Uses SCAN + DEL for flat keys: file:{hashId}/* +// Also unmaps every reverse-index entry registered for this root, so a +// subsequent GET /api/meta/ correctly returns 404 instead of +// pointing at a UUID with no keys. func (c *Client) DeleteMetadata(hashID string) (int64, error) { c.mu.RLock() defer c.mu.RUnlock() @@ -281,6 +291,22 @@ func (c *Client) DeleteMetadata(hashID string) (int64, error) { prefix := c.buildKeyPrefix(hashID) + // Read the cids set first — once file:/* is gone we've lost the + // list of reverse-index keys that need cleanup. + tokens, err := c.client.SMembers(ctx, prefix+CIDsField).Result() + if err != nil && err != redis.Nil { + return 0, fmt.Errorf("smembers cids: %w", err) + } + if len(tokens) > 0 { + idxKeys := make([]string, 0, len(tokens)) + for _, t := range tokens { + idxKeys = append(idxKeys, c.buildCIDIndexKey(t)) + } + if err := c.client.Del(ctx, idxKeys...).Err(); err != nil { + return 0, fmt.Errorf("del cid index: %w", err) + } + } + // Scan and delete all keys with this prefix var deletedCount int64 var cursor uint64 @@ -323,72 +349,31 @@ func (c *Client) CountFiles() (int, error) { // LookupPathByCID resolves a CID to its file path on disk. // -// Resolution covers three cases: -// - midhash256 hashID itself: the CID identifies a file's content, return its filePath. -// - poster/backdrop fields: the CID is for the image asset, return posterPath/backdropPath. -// - any cid_* field: the CID is another digest of the same file, return its filePath. +// Primary path: the reverse index (cid: → uuid). O(1) lookup followed +// by a single GET of file:/filePath. // -// Walks every indexed hash; acceptable for admin/UI use, but would benefit from -// a reverse index (cid → hashID/field) if call volume grows. -func (c *Client) LookupPathByCID(cid string) (string, error) { - c.mu.RLock() - defer c.mu.RUnlock() - - if c.client == nil { - return "", fmt.Errorf("not connected") - } - - if cid == "" { - return "", nil - } - - ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) - defer cancel() - - hashIDs, err := c.getAllHashIDsInternal(ctx) +// Fallback path: poster/backdrop CIDs aren't registered as aliases (they +// refer to sidecar images, not the main file content), so unknown tokens +// fall through to a scan of poster/backdrop fields. O(n) but rare in +// practice — sidecar lookups are infrequent compared to file CID lookups. +// +// Accepts CID tokens in prefixed form ("midhash256:bafk…", "sha256:…", +// "ipfs:bafy…"). Bare CIDs without an algorithm prefix won't resolve via +// the reverse index — callers should pass the prefixed form. +func (c *Client) LookupPathByCID(token string) (string, error) { + uuid, err := c.GetByCID(token) if err != nil { return "", err } - - for _, hashID := range hashIDs { - meta, err := c.getMetadataFlatInternal(ctx, hashID) - if err != nil || len(meta) == 0 { - continue - } - - // midhash256 == hashID: the file itself. - if hashID == cid { - if fp := meta["filePath"]; fp != "" { - return normalizeFilesRelativePath(fp), nil - } - } - - // Poster / backdrop refer to a sidecar image asset. - if meta["poster"] == cid { - if p := meta["posterPath"]; p != "" { - return normalizeFilesRelativePath(p), nil - } - } - if meta["backdrop"] == cid { - if p := meta["backdropPath"]; p != "" { - return normalizeFilesRelativePath(p), nil - } - } - - // Any other cid_* field is an additional digest of the file itself. - for field, value := range meta { - if value != cid { - continue - } - if strings.HasPrefix(field, "cid_") { - if fp := meta["filePath"]; fp != "" { - return normalizeFilesRelativePath(fp), nil - } - } + if uuid != "" { + fp, err := c.GetProperty(uuid, "filePath") + if err != nil { + return "", err } + return normalizeFilesRelativePath(fp), nil } - - return "", nil + // Not a file alias — try poster/backdrop sidecars. + return c.lookupSidecarPathByCID(token) } // normalizeFilesRelativePath strips a leading "/files/" prefix so handlers can @@ -477,6 +462,10 @@ func (c *Client) MergeMetadataFlat(hashID string, metadata map[string]string) (i return len(metadata), fmt.Errorf("sadd index failed: %w", err) } + // Auto-register reverse-index aliases for any cid_* or midhash256 fields + // in the payload. See cid_resolution.go for the rules. + c.maybeAddAliasesFromMetadataLocked(ctx, hashID, metadata) + return len(metadata), nil } @@ -599,19 +588,27 @@ func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) { return true, c.client.Set(ctx, key, newValue, 0).Err() } -// FileTuple holds the (path, size, mtime) record for one file. Used by the -// watcher hydrator at startup to avoid re-mid-hashing files whose tuple -// matches what's already in Redis. +// FileTuple holds the (uuid, path, size, mtime, midhash) record for one +// file. Used by the watcher hydrator at startup to avoid re-mid-hashing +// files whose tuple matches what's already in Redis. +// +// HashID is the root identifier (UUIDv7 in the current model); MidHash256 +// is the actual midhash CID for the file, present iff the watcher already +// computed and stored it. Hydration uses HashID for the state-registry +// key and MidHash256 for the in-memory "I already know this hash" flag. type FileTuple struct { - HashID string - FilePath string - Size int64 - MtimeNano int64 + HashID string + FilePath string + Size int64 + MtimeNano int64 + MidHash256 string } -// GetTuplesForAllFiles returns one FileTuple per indexed CID, fetched in -// batched MGETs (no SCAN per CID). Skips files missing any of the three -// fields. Sub-second on local Redis for ~10K files. +// GetTuplesForAllFiles returns one FileTuple per indexed root, fetched in +// batched MGETs (no SCAN per root). Skips files missing any of filePath, +// sizeByte, or mtimeNano. midhash256 is optional — files that haven't +// been hashed yet still appear in the result with an empty MidHash256. +// Sub-second on local Redis for ~10K files. func (c *Client) GetTuplesForAllFiles() ([]FileTuple, error) { c.mu.RLock() defer c.mu.RUnlock() @@ -631,11 +628,11 @@ func (c *Client) GetTuplesForAllFiles() ([]FileTuple, error) { return nil, nil } - // Build the key list: 3 keys per CID, in a fixed order so we can decode + // Build the key list: 4 keys per root, in a fixed order so we can decode // the MGET response by position. fields[i] for hashIDs[j] sits at - // keys[j*3+i]. - const fieldsPerFile = 3 - fields := [fieldsPerFile]string{"filePath", "sizeByte", "mtimeNano"} + // keys[j*4+i]. + const fieldsPerFile = 4 + fields := [fieldsPerFile]string{"filePath", "sizeByte", "mtimeNano", "midhash256"} keys := make([]string, 0, len(hashIDs)*fieldsPerFile) for _, h := range hashIDs { prefix := c.buildKeyPrefix(h) @@ -665,6 +662,7 @@ func (c *Client) GetTuplesForAllFiles() ([]FileTuple, error) { filePath, _ := values[base].(string) sizeStr, _ := values[base+1].(string) mtimeStr, _ := values[base+2].(string) + midhash, _ := values[base+3].(string) if filePath == "" || sizeStr == "" || mtimeStr == "" { continue } @@ -677,10 +675,11 @@ func (c *Client) GetTuplesForAllFiles() ([]FileTuple, error) { continue } out = append(out, FileTuple{ - HashID: h, - FilePath: filePath, - Size: size, - MtimeNano: mtime, + HashID: h, + FilePath: filePath, + Size: size, + MtimeNano: mtime, + MidHash256: midhash, }) } return out, nil diff --git a/internal/storage/dual_root_migration.go b/internal/storage/dual_root_migration.go new file mode 100644 index 0000000..2a118d2 --- /dev/null +++ b/internal/storage/dual_root_migration.go @@ -0,0 +1,162 @@ +package storage + +import ( + "context" + "fmt" + "log" + "strings" + "time" +) + +// MigrateDualRoots scans file:__index__ for hashIds that look like raw +// midhash256 CID values (i.e. NOT UUIDs/ULIDs) and have a matching alias +// `cid:midhash256:` pointing to a different UUID. For each such +// "stranded" root we: +// +// 1. Copy every file:/ into file:/ +// using MergeMetadataFlat (preserves anything the UUID root already +// holds, e.g. duplicates and watcher-owned fields). +// 2. Delete the entire file:/* subtree. +// 3. Drop the midhash hashId from file:__index__ (kept on the UUID). +// +// This is the one-shot fix for the dual-root pattern described in +// docs/api-mediated-access.md. The pattern was caused by the (now-fixed) +// /meta/{hash} write handlers not resolving CID → UUID before writing. +// +// Returns the number of stranded roots fixed. +func (c *Client) MigrateDualRoots(ctx context.Context) (int, error) { + c.mu.RLock() + if c.client == nil { + c.mu.RUnlock() + return 0, fmt.Errorf("not connected") + } + hashIDs, err := c.getAllHashIDsInternal(ctx) + c.mu.RUnlock() + if err != nil { + return 0, fmt.Errorf("get hash ids: %w", err) + } + + // Build the orphaned-UUID index: for every UUID-rooted entry, record + // midhash256 → UUID. This lets us find the "real" target when an alias + // has been overwritten to self-point (the failure mode the in-flight + // fix prevents going forward, but a one-shot still needs to repair + // existing dual-rooted data). + midhashToUUID := make(map[string]string) + for _, hashID := range hashIDs { + if !looksLikeULID(hashID) { + continue + } + mh, err := c.GetProperty(hashID, "midhash256") + if err != nil || mh == "" { + continue + } + midhashToUUID[mh] = hashID + } + + fixed := 0 + for _, hashID := range hashIDs { + // Skip likely-UUID hashIds. ULIDs are 26 chars, uppercase Crockford + // Base32. We treat anything that's not all-uppercase-alphanumeric + // of length ~26 as a CID-shaped hashId worth checking. + if looksLikeULID(hashID) { + continue + } + + // Is there an alias pointing to a different uuid? + token := "midhash256:" + hashID + uuid, err := c.GetByCID(token) + if err != nil { + log.Printf("[MigrateDualRoots] GetByCID(%s): %v (skipping)", token, err) + continue + } + // Three cases: + // - uuid empty → no alias, this midhash root is standalone (skip) + // - uuid == hashID → self-pointing alias; check the orphan index + // for a UUID that claims this midhash via its midhash256 field + // - uuid != hashID and != "" → proper alias, migrate to that UUID + if uuid == "" || uuid == hashID { + if orphan, ok := midhashToUUID[hashID]; ok && orphan != "" { + uuid = orphan + } else { + continue + } + } + + // Pull every key under file:/* and merge into file:/. + flat, err := c.GetMetadataFlat(hashID) + if err != nil { + log.Printf("[MigrateDualRoots] GetMetadataFlat(%s): %v (skipping)", hashID, err) + continue + } + if len(flat) > 0 { + if _, err := c.MergeMetadataFlat(uuid, flat); err != nil { + log.Printf("[MigrateDualRoots] MergeMetadataFlat(%s ← %s): %v", uuid, hashID, err) + continue + } + } + + // Delete the stranded root entirely. + if _, err := c.DeleteMetadata(hashID); err != nil { + log.Printf("[MigrateDualRoots] DeleteMetadata(%s): %v", hashID, err) + continue + } + + // Re-point the alias to the surviving UUID. DeleteMetadata wipes + // the cid:* entries for the deleted root, so the alias for this + // midhash may have been removed. Restore it pointing at the + // canonical UUID so future ResolveRoot calls land correctly. + if err := c.AddAlias(uuid, "midhash256:"+hashID); err != nil { + log.Printf("[MigrateDualRoots] AddAlias(%s → midhash256:%s): %v", uuid, hashID, err) + // Not fatal — data is migrated; the alias mismatch can be + // re-fixed by another sweep. + } + + fixed++ + } + + if fixed > 0 { + log.Printf("[MigrateDualRoots] Reunited %d stranded midhash roots into their UUID counterparts", fixed) + } + return fixed, nil +} + +// looksLikeULID is a cheap heuristic for "this hashId is opaque internal +// UUID, not a CID we should resolve." ULIDs are 26-char Crockford Base32; +// CIDs in this codebase are longer and lowercase. Tolerates dashes for +// future-proofing. +func looksLikeULID(s string) bool { + if len(s) != 26 { + return false + } + for _, r := range s { + if r >= 'A' && r <= 'Z' { + continue + } + if r >= '0' && r <= '9' { + continue + } + return false + } + return true +} + +// MigrateDualRootsWithTimeout is a convenience wrapper used by the API +// handler and the startup hook. Default to a generous 5-minute deadline — +// the per-entry work is small but the index can be large. +func (c *Client) MigrateDualRootsWithTimeout() (int, error) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + return c.MigrateDualRoots(ctx) +} + +// MigrateDualRootsReport returns a human-readable summary of what +// MigrateDualRoots would do or did. Useful for the admin endpoint. +func (c *Client) MigrateDualRootsReport(ctx context.Context) (string, int, error) { + fixed, err := c.MigrateDualRoots(ctx) + if err != nil { + return "", fixed, err + } + parts := []string{} + parts = append(parts, fmt.Sprintf("merged %d stranded midhash roots into UUID roots", fixed)) + return strings.Join(parts, "; "), fixed, nil +} diff --git a/internal/storage/schema_version.go b/internal/storage/schema_version.go new file mode 100644 index 0000000..f64c292 --- /dev/null +++ b/internal/storage/schema_version.go @@ -0,0 +1,83 @@ +package storage + +import ( + "context" + "fmt" + "strconv" + "time" + + "github.com/redis/go-redis/v9" +) + +// SchemaVersion is the Redis-layout version this build of meta-core expects. +// Distinct from the snapshot manifest version (which is on-disk format); +// this number describes the in-memory Redis schema — primarily, "are roots +// UUIDs or content-hash tokens, and do reverse-index keys cid:* exist?" +// +// Bump only when a Redis-layout change can't be detected from the data +// itself or healed by the auto-alias hooks. Each bump is a flag day in +// alpha: operators wipe and restart. +const SchemaVersion = 2 + +// SchemaVersionKey is the sentinel under which we record SchemaVersion in +// Redis. Checked on every meta-core boot. +const SchemaVersionKey = "meta-core:schema-version" + +// EnsureSchemaVersion verifies the Redis store matches the expected schema +// version. Called once at startup, before any other reads/writes. +// +// Behavior: +// - sentinel present and matches expected → OK +// - sentinel missing AND file:__index__ empty → fresh store, stamp sentinel +// - sentinel missing AND file:__index__ non-empty → legacy data, refuse +// - sentinel present but mismatched → refuse, point operator at clean:all +// +// Refusal is hard-fail; the caller is expected to log.Fatalf and exit. +// The error message names the wipe remediation because in alpha there +// is no automated migration — operators bring up a clean store. +func (c *Client) EnsureSchemaVersion(expected int) error { + c.mu.Lock() + defer c.mu.Unlock() + + if c.client == nil { + return fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + key := c.buildKey(SchemaVersionKey) + have, err := c.client.Get(ctx, key).Result() + if err != nil && err != redis.Nil { + return fmt.Errorf("get schema version: %w", err) + } + + if err == redis.Nil { + // Sentinel absent. Fresh store iff there's no file:__index__ membership. + n, sErr := c.client.SCard(ctx, c.buildIndexKey()).Result() + if sErr != nil && sErr != redis.Nil { + return fmt.Errorf("scard file:__index__: %w", sErr) + } + if n > 0 { + return fmt.Errorf( + "redis has %d indexed roots but no %q sentinel — this is alpha "+ + "data from before the UUID-rooted schema landed. Run "+ + "`pnpm run clean:all` (or `docker exec metacore-app redis-cli FLUSHDB`) "+ + "to wipe and restart, or downgrade meta-core to a compatible build", + n, SchemaVersionKey) + } + if err := c.client.Set(ctx, key, strconv.Itoa(expected), 0).Err(); err != nil { + return fmt.Errorf("set schema version: %w", err) + } + return nil + } + + if have == strconv.Itoa(expected) { + return nil + } + return fmt.Errorf( + "redis schema version is %q, this build of meta-core expects %d. "+ + "Run `pnpm run clean:all` to wipe and restart, or use a meta-core "+ + "build that matches the existing data", + have, expected) +} diff --git a/internal/storage/uuid.go b/internal/storage/uuid.go new file mode 100644 index 0000000..10e7c7a --- /dev/null +++ b/internal/storage/uuid.go @@ -0,0 +1,62 @@ +package storage + +import ( + "github.com/google/uuid" +) + +// NewUUID returns a new UUIDv7 encoded as 26-char Crockford Base32 (ULID +// layout). Used as the root key for file metadata entries. +// +// Why this format over canonical 8-4-4-4-12 hex: +// - 26 chars vs 36 — shorter Redis keys, less prefix space waste +// - Time-sortable prefix — KEYS file:01JKR* returns entries minted in a +// given time window, restoring some of the debug-ability lost vs the +// previous midhash256:abc rooting +// - Case-insensitive — no ambiguity when copying from logs +// - Crockford alphabet skips I/L/O/U, so eyeballed inspection of a key +// can't be confused for an ambient digit/letter typo +func NewUUID() (string, error) { + u, err := uuid.NewV7() + if err != nil { + return "", err + } + return encodeCrockford(u[:]), nil +} + +const crockfordAlphabet = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" + +// encodeCrockford encodes a 16-byte value as 26 Crockford Base32 chars. +// 128 bits packs into 130 bits (26 × 5); the top 2 bits of the first char +// carry the high bits of byte 0, the remaining 3 are implicit zero. This +// matches the ULID encoding so the output is interchangeable with ULIDs in +// any tool that accepts them. +func encodeCrockford(b []byte) string { + out := make([]byte, 26) + out[0] = crockfordAlphabet[(b[0]&0xE0)>>5] + out[1] = crockfordAlphabet[b[0]&0x1F] + out[2] = crockfordAlphabet[(b[1]&0xF8)>>3] + out[3] = crockfordAlphabet[((b[1]&0x07)<<2)|((b[2]&0xC0)>>6)] + out[4] = crockfordAlphabet[(b[2]&0x3E)>>1] + out[5] = crockfordAlphabet[((b[2]&0x01)<<4)|((b[3]&0xF0)>>4)] + out[6] = crockfordAlphabet[((b[3]&0x0F)<<1)|((b[4]&0x80)>>7)] + out[7] = crockfordAlphabet[(b[4]&0x7C)>>2] + out[8] = crockfordAlphabet[((b[4]&0x03)<<3)|((b[5]&0xE0)>>5)] + out[9] = crockfordAlphabet[b[5]&0x1F] + out[10] = crockfordAlphabet[(b[6]&0xF8)>>3] + out[11] = crockfordAlphabet[((b[6]&0x07)<<2)|((b[7]&0xC0)>>6)] + out[12] = crockfordAlphabet[(b[7]&0x3E)>>1] + out[13] = crockfordAlphabet[((b[7]&0x01)<<4)|((b[8]&0xF0)>>4)] + out[14] = crockfordAlphabet[((b[8]&0x0F)<<1)|((b[9]&0x80)>>7)] + out[15] = crockfordAlphabet[(b[9]&0x7C)>>2] + out[16] = crockfordAlphabet[((b[9]&0x03)<<3)|((b[10]&0xE0)>>5)] + out[17] = crockfordAlphabet[b[10]&0x1F] + out[18] = crockfordAlphabet[(b[11]&0xF8)>>3] + out[19] = crockfordAlphabet[((b[11]&0x07)<<2)|((b[12]&0xC0)>>6)] + out[20] = crockfordAlphabet[(b[12]&0x3E)>>1] + out[21] = crockfordAlphabet[((b[12]&0x01)<<4)|((b[13]&0xF0)>>4)] + out[22] = crockfordAlphabet[((b[13]&0x0F)<<1)|((b[14]&0x80)>>7)] + out[23] = crockfordAlphabet[(b[14]&0x7C)>>2] + out[24] = crockfordAlphabet[((b[14]&0x03)<<3)|((b[15]&0xE0)>>5)] + out[25] = crockfordAlphabet[b[15]&0x1F] + return string(out) +} diff --git a/internal/storage/uuid_test.go b/internal/storage/uuid_test.go new file mode 100644 index 0000000..1d364d5 --- /dev/null +++ b/internal/storage/uuid_test.go @@ -0,0 +1,77 @@ +package storage + +import ( + "strings" + "testing" + "time" +) + +func TestNewUUID_Shape(t *testing.T) { + id, err := NewUUID() + if err != nil { + t.Fatalf("NewUUID: %v", err) + } + if len(id) != 26 { + t.Errorf("expected 26 chars, got %d (%q)", len(id), id) + } + const alphabet = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" + for i, c := range id { + if !strings.ContainsRune(alphabet, c) { + t.Errorf("char %d (%q) not in Crockford alphabet (uuid=%q)", i, c, id) + } + } +} + +func TestNewUUID_Unique(t *testing.T) { + const n = 1000 + seen := make(map[string]struct{}, n) + for i := 0; i < n; i++ { + id, err := NewUUID() + if err != nil { + t.Fatalf("NewUUID at i=%d: %v", i, err) + } + if _, dup := seen[id]; dup { + t.Fatalf("collision after %d mints: %q", i+1, id) + } + seen[id] = struct{}{} + } +} + +func TestNewUUID_TimeSortable(t *testing.T) { + // UUIDv7 has a millisecond timestamp prefix; consecutive mints in the + // same millisecond may not be strictly ordered (random tail), but + // mints across milliseconds are. Just check that an early ID + // lexicographically sorts before a later one when there's a real + // time gap. Doesn't test the no-gap case — the v7 spec only + // guarantees ordering across millisecond boundaries. + a, _ := NewUUID() + time.Sleep(2 * time.Millisecond) + b, _ := NewUUID() + if a >= b { + t.Errorf("expected %q < %q (UUIDv7 should be roughly time-sortable)", a, b) + } +} + +func TestCIDTokenFromField(t *testing.T) { + cases := []struct { + field, value string + want string + }{ + {"midhash256", "bafk", "midhash256:bafk"}, + {"cid_sha256", "bafk", "sha256:bafk"}, + {"cid_ipfs", "bafy", "ipfs:bafy"}, + {"cid_sha256", "sha256:bafk", "sha256:bafk"}, // already prefixed + {"cid_ipfs", "ipfs:bafy", "ipfs:bafy"}, // already prefixed + {"title", "Inception", ""}, // non-CID field + {"tmdb/title", "Inception", ""}, // non-CID nested + {"cid_", "bafk", ""}, // empty algo + {"midhash256", "", ""}, // empty value + {"cid_sha256", "", ""}, // empty value + } + for _, tc := range cases { + if got := cidTokenFromField(tc.field, tc.value); got != tc.want { + t.Errorf("cidTokenFromField(%q, %q) = %q, want %q", + tc.field, tc.value, got, tc.want) + } + } +} diff --git a/internal/watcher/dispatcher.go b/internal/watcher/dispatcher.go index 429a629..7c8013c 100644 --- a/internal/watcher/dispatcher.go +++ b/internal/watcher/dispatcher.go @@ -13,6 +13,11 @@ import ( const ( // EventsStream is the Redis Stream name for file events EventsStream = "file:events" + + // EventsStreamMaxLen is the approximate cap on stream length. Bounded + // retention is a contract requirement for the SSE wire — see + // docs/api-mediated-access.md "Backing-store retention and lifecycle". + EventsStreamMaxLen = 100_000 ) // Dispatcher sends file events to Redis Stream @@ -70,8 +75,10 @@ func (d *Dispatcher) dispatchToStream(event FileEvent) { fields["watcherId"] = event.WatcherID } - // Publish to stream (no MaxLen - stream is cleared on reset) - _, err := client.XAdd(EventsStream, 0, fields) + // Approximate MAXLEN trim keeps the stream bounded. The deliberate + // `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) } diff --git a/internal/watcher/handlers.go b/internal/watcher/handlers.go index 5e71218..dcf2cb7 100644 --- a/internal/watcher/handlers.go +++ b/internal/watcher/handlers.go @@ -33,8 +33,16 @@ func (h *Handlers) RegisterRoutes(r *mux.Router) { r.HandleFunc("/api/scan/status", h.handleScanStatusDeprecated).Methods("GET") } -// handlePoll handles GET /api/events/poll +// handlePoll handles GET /api/events/poll. +// +// DEPRECATED. The long-poll variant over file:events has been superseded by +// the SSE endpoint at /api/events/files. New consumers should use SSE. This +// handler stays for one release with a Deprecation header before removal — +// see docs/api-mediated-access.md PR A scope. func (h *Handlers) handlePoll(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Deprecation", "true") + w.Header().Set("Link", `; rel="successor-version"`) + w.Header().Set("Sunset", "after next release") // Get since parameter sinceStr := r.URL.Query().Get("since") sinceMS := int64(0) diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go index 6ba4d36..5ffa730 100644 --- a/internal/watcher/watcher.go +++ b/internal/watcher/watcher.go @@ -36,20 +36,60 @@ func (w *Watcher) SetStorageClient(s *storage.Client) { // writeFileTuple persists path/size/mtime/midhash for a freshly-hashed file. // Caller already has all four values from its existing os.Stat / FileInfo — -// no extra IO is performed here beyond the Redis write itself. +// no extra IO is performed here beyond the Redis writes themselves. // -// This is the inverse-index that lets a fresh meta-core boot skip -// re-mid-hashing files whose (path, size, mtime) match what's in Redis. +// Flow (see docs/uuid-rooted-metadata.md): +// - Look up the midhash in the reverse index (cid:midhash256:… → uuid) +// - If unknown, Mint a fresh UUID root and write the midhash256 field +// (the SetProperty hook auto-registers the alias) +// - If known, this is a duplicate physical file for the same content — +// record the new path in the duplicates set and refresh size/mtime +// +// This is also the inverse-index that lets a fresh meta-core boot skip +// re-mid-hashing files whose (path, size, mtime) match what's in Redis, +// via HydrateStateFromStorage below. func (w *Watcher) writeFileTuple(absPath string, size, mtimeNano int64, midhash string) { if w.storage == nil || midhash == "" { return } - if _, err := w.storage.MergeMetadataFlat(midhash, map[string]string{ - "filePath": absPath, + token := "midhash256:" + midhash + + uuid, err := w.storage.GetByCID(token) + if err != nil { + log.Printf("[Watcher] GetByCID(%s): %v", token, err) + return + } + + if uuid == "" { + // New content. Mint sets filePath/sizeByte/mtimeNano; SetProperty + // then writes midhash256 — and the cid_resolution.go hook on + // SetProperty 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, "midhash256", midhash); err != nil { + log.Printf("[Watcher] SetProperty midhash256 for %s: %v", 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] failed to write tuple for %s: %v", midhash, err) + log.Printf("[Watcher] refresh tuple for %s: %v", uuid, err) } } @@ -319,7 +359,7 @@ func (w *Watcher) HydrateStateFromStorage(roots []WatcherRoot) (loaded, skipped ws.Set(rel, &FileState{ Size: t.Size, MtimeNano: t.MtimeNano, - MidHash: t.HashID, + MidHash: t.MidHash256, }) loaded++ matched = true