Compare commits

..
4 Commits
Author SHA1 Message Date
Worph 76169920d6 improve meta core user management
Build and Push Docker Image / build (push) Failing after 13s
2026-07-18 17:30:17 +02:00
Worph df6a2d78f8 bug fix vector
Build and Push Docker Image / build (push) Failing after 14s
2026-07-14 16:02:26 +02:00
Worph 0ca03ac95a deprecate old cids fields 2026-07-13 22:02:29 +02:00
Worph 1798821c53 update metacore default 2026-07-12 14:47:21 +02:00
21 changed files with 1273 additions and 81 deletions
+6
View File
@@ -0,0 +1,6 @@
# FIX CRLF always when developer has not set
# Linux/Mac: git config --global core.autocrlf input
# Windows: git config --global core.autocrlf true
# Auto detect text files and perform LF normalization
* text=auto
* eol=lf
+1 -1
View File
@@ -138,7 +138,7 @@ Environment variables consumed by `internal/config/config.go`:
| `BASE_URL` | _empty_ | Overrides the constructed `baseUrl` if set (used for HTTPS perimeter). |
| `REDIS_PORT` | `6379` | Local Redis port. |
| `META_CORE_HTTP_PORT` | `9000` | Go HTTP+SSE+WebDAV port. |
| `META_CORE_HTTP_HOST` | `127.0.0.1` | HTTP bind. |
| `META_CORE_HTTP_HOST` | `0.0.0.0` | HTTP bind. |
| `HEALTH_CHECK_INTERVAL_MS` | `5000` | Internal health-loop cadence. |
| `HEARTBEAT_INTERVAL_MS` | `30000` | Service registry heartbeat. |
| `STALE_THRESHOLD_MS` | `60000` | Age past which a registry entry is marked stale. |
+6 -6
View File
@@ -18,7 +18,7 @@ out to be load-bearing).
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),
(reverse index, `cids/` key-set integrity, schema-version sentinel),
and presents a stable HTTP contract. Other services were meant to talk to
meta-core, not to Redis.
@@ -392,7 +392,7 @@ specifically with the Kubernetes flavour where streams are mediated too.
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
that can enforce invariants (reverse index, `cids/` key-set, type
constraints, schema version).
- **Observability.** Audit logs, request metrics, slow-query detection
all live at one layer.
@@ -697,11 +697,11 @@ contract violation."
**The fix.** Three layers in `internal/storage/cid_resolution.go`:
1. **`ResolveRoot(hash)`** — at the entry of every `/meta/{hash}`
read/write handler, look up `cid:midhash256:<hash>` in the reverse
index. If found, use the UUID as the storage root. Falls through to
the bare hash for direct-UUID writes / legacy entries.
read/write handler, look up `cid:<hash>` in the reverse index (bare
CID — no algorithm prefix). If found, use the UUID as the storage root.
Falls through to the bare hash for direct-UUID writes / legacy entries.
2. **Self-pointing alias guard in `addAliasLocked`** — refuse to
register `cid:<algo>:<v>` → `<v>`. The dual-root bug's chicken-and-
register `cid:<v>` → `<v>`. The dual-root bug's chicken-and-
egg is: meta-sort writes first, registers self-pointing alias,
watcher then can't recover. Refusing the self-write keeps the
watcher's legitimate alias intact.
+5 -4
View File
@@ -221,12 +221,13 @@ operators wipe with `pnpm run clean:all` and restart.
│ Root: opaque UUIDv7 (Crockford Base32; ULID layout, 26 chars) │
│ Storage: flat string keys │
│ file:<uuid>/<property> → string value │
│ file:<uuid>/cids → SET of CID tokens │
│ file:<uuid>/canonical_cid → preferred CID │
│ file:<uuid>/cids/<bareCid> → "true" (key-set) │
│ file:<uuid>/duplicates → SET of paths │
│ │
│ No stored canonical_cid — derived by rank on read. │
│ │
│ Aliases (reverse index, one Redis STRING each): │
│ cid:<algorithm>:<value> → <uuid> │
│ cid:<bareCid> → <uuid> │
│ │
│ Index: file:__index__ → SET of all UUIDs │
│ │
@@ -515,7 +516,7 @@ via `/health`; metadata I/O and SSE event streams stop.
| `BASE_URL` | _empty_ | Overrides constructed `baseUrl` (HTTPS perimeter). |
| `REDIS_PORT` | `6379` | Local Redis port. |
| `META_CORE_HTTP_PORT` | `9000` | Go HTTP+SSE+WebDAV port. |
| `META_CORE_HTTP_HOST` | `127.0.0.1` | HTTP bind address. |
| `META_CORE_HTTP_HOST` | `0.0.0.0` | HTTP bind address. |
| `HEALTH_CHECK_INTERVAL_MS` | `5000` | Internal health-loop cadence. |
| `HEARTBEAT_INTERVAL_MS` | `30000` | Service registry heartbeat. |
| `STALE_THRESHOLD_MS` | `60000` | Age past which a registry entry is stale. |
+49 -24
View File
@@ -19,10 +19,13 @@ For the external-access contract (HTTP-only, SSE for events), see
`01JKR8XW5T4QMV7AHNJ2DEFGHK`.
- **Properties are flat Redis STRING keys**, one per leaf:
`file:<uuid>/<property/path>` → string value. *Not* Redis Hashes.
- **CIDs are reverse-index aliases**: `cid:<algorithm>:<value>` → `<uuid>`
(a Redis STRING per known CID).
- **Per-root sets**: `file:<uuid>/cids` (every CID), `file:<uuid>/duplicates`
- **CIDs are reverse-index aliases**: `cid:<bareCid>` → `<uuid>`
(a Redis STRING per known CID; no algorithm prefix — the multicodec
already carries it).
- **Per-root sets**: `file:<uuid>/cids/<bareCid>` (a *key-set*: one STRING
`"true"` per known CID, midhash included), `file:<uuid>/duplicates`
(alternate paths with the same content), plus `file:__index__` (every UUID).
There is **no stored `canonical_cid`** — it is derived by rank on read.
- **A schema-version sentinel** (`meta-core:schema-version`) refuses to
boot against legacy data.
@@ -59,37 +62,54 @@ file:01JKR8XW5T4QMV7AHNJ2DEFGHK/tmdb/poster = "bafkreih...
### Reverse-index aliases (CIDs)
Every CID known for a file is registered as its own Redis STRING:
Every CID known for a file is registered as its own Redis STRING, keyed by
the **bare CID** — no algorithm prefix. A CIDv1 already encodes its hash
function in the multicodec, so naming it again in the key would be
redundant (and would let the same digest be indexed under two spellings):
```
cid:<algorithm>:<value> → <uuid>
cid:<bareCid> → <uuid>
```
Examples:
```
cid:midhash256:bafkreih5kapbjzq... → 01JKR8XW5T4QMV7AHNJ2DEFGHK
cid:sha256:abc... → 01JKR8XW5T4QMV7AHNJ2DEFGHK
cid:ipfs:bafy... → 01JKR8XW5T4QMV7AHNJ2DEFGHK
cid:bafkreih5kapbjzq... → 01JKR8XW5T4QMV7AHNJ2DEFGHK
cid:bagacbabaec7v3... → 01JKR8XW5T4QMV7AHNJ2DEFGHK
```
The reverse index makes `GET /api/meta/<cid>` an O(1) lookup for any
registered CID, independent of which hash algorithm "primary" — there is
no primary. The auto-alias hooks in `SetProperty` and `SetMetadataFlat`
(see `internal/storage/cid_resolution.go`) catch any write of a `cid_*`
or `midhash256` field and register the alias automatically, so plugins
don't need to know about the reverse index.
registered CID, independent of which hash algorithm — there is no primary.
The auto-alias hook (`maybeAddAliasFromFieldLocked` in
`internal/storage/cid_resolution.go`, called from `SetProperty`,
`SetMetadataFlat` and `MergeMetadataFlat`) registers the alias for any
field whose name starts with `cids/`, so writers emit key-set members
without knowing the reverse index exists.
> ⚠ The hook matches **only** the `cids/` prefix. A legacy `cid_*` or
> `midhash256` *named* field written to a root is stored verbatim and is
> **never indexed** — the record becomes unresolvable by that CID. Writers
> must collapse such fields to `cids/<cid>` members before the write. See
> `METADATA_KEYS.md` §14.13.
### Per-root sets
```
file:<uuid>/cids → Redis SET of "<algorithm>:<value>" tokens
file:<uuid>/canonical_cid → STRING — the highest-ranked CID for advertising externally
file:<uuid>/cids/<bareCid> → STRING "true" — one key per known CID (a key-set, not a Redis SET)
file:<uuid>/duplicates → Redis SET of alternate paths with same content
```
`canonical_cid` is reconciled by `reconcileCanonicalCIDLocked` whenever
the `cids` set changes; ranking is done by `internal/cid/rank.go`.
`cids` is a **key-set**: one flat `file:<uuid>/cids/<cid>` STRING per member,
which keeps it conflict-free to merge across peers (a SET would not be).
The record's own address (the midhash) is just the `0x1000` member — it has
no separate field.
There is **no stored `canonical_cid`**. The canonical CID is *derived on
read*: `GetMetadataDocument` (`internal/storage/cid_resolution.go`) ranks the
key-set's members through `internal/cid/rank.go` (`cid.Better`) and returns
the winner as the `canonical_cid` field of the JSON response only. Nothing
reconciles or persists it — an earlier `reconcileCanonicalCIDLocked` did, and
was removed with the flat `cid_*` keys.
### Global index
@@ -141,14 +161,15 @@ The Go API in `internal/storage/client.go`:
| Method | Behaviour |
|---|---|
| `GetMetadataFlat(uuid)` | `SCAN` for `file:<uuid>/*`, `MGET` the values, return `map[string]string`. |
| `SetMetadataFlat(uuid, m)` | Pipeline `SET` per field; `SADD file:__index__`; auto-register any `cid_*` aliases. |
| `SetMetadataFlat(uuid, m)` | Pipeline `SET` per field; `SADD file:__index__`; auto-register an alias for every `cids/<cid>` member. |
| `MergeMetadataFlat(uuid, m)` | Like `SetMetadataFlat` but doesn't pre-clear; used by `PATCH /meta/{hash}`. |
| `GetProperty(uuid, field)` | `GET file:<uuid>/<field>`. |
| `SetProperty(uuid, field, value)` | `SET file:<uuid>/<field>`; auto-register alias if the field is a CID. |
| `DeleteRoot(uuid)` | `SCAN` + `DEL` for `file:<uuid>/*`, unmap every reverse-index entry. |
| `Mint(filePath, size, mtimeNano)` | `NewUUID()` + write `filePath/sizeByte/mtimeNano` + add to index. |
| `AddAlias(uuid, "<algo>:<value>")` | Set `cid:<algo>:<value>` → `<uuid>`, add to `file:<uuid>/cids`, reconcile `canonical_cid`. |
| `ResolveCID("<algo>:<value>")` | `GET cid:<algo>:<value>` → uuid. |
| `AddAlias(uuid, "<bareCid>")` | Set `cid:<bareCid>` → `<uuid>` and the key-set member `file:<uuid>/cids/<bareCid>` = `"true"`. Nothing to reconcile — canonical is derived on read. |
| `ResolveRoot("<bareCid>")` | `GET cid:<bareCid>` → uuid (returns the input unchanged if it is already a UUID root). |
| `GetMetadataDocument(uuid)` | Flat fields + `cids` hoisted to an array + `canonical_cid` **derived** by `cid.Better` rank + `duplicates`. Backs `GET /api/meta/{cid}`. |
## Reconstruction (flat → nested)
@@ -237,10 +258,14 @@ docker exec metacore-app redis-cli GET 'file:01JKR8XW5T4QMV7AHNJ2DEFGHK/fileinfo
docker exec metacore-app redis-cli TYPE 'file:01JKR8XW5T4QMV7AHNJ2DEFGHK/filePath'
# string
# Reverse-index lookups
docker exec metacore-app redis-cli GET 'cid:midhash256:bafkreih...' # → uuid
docker exec metacore-app redis-cli SMEMBERS 'file:01JKR8XW5T4QMV7AHNJ2DEFGHK/cids'
docker exec metacore-app redis-cli GET 'file:01JKR8XW5T4QMV7AHNJ2DEFGHK/canonical_cid'
# Reverse-index lookup (bare CID — no algorithm prefix)
docker exec metacore-app redis-cli GET 'cid:bafkreih...' # → uuid
# The cids key-set (one STRING per member, NOT a Redis SET — SMEMBERS won't work)
docker exec metacore-app redis-cli --scan --pattern 'file:01JKR8XW5T4QMV7AHNJ2DEFGHK/cids/*'
# canonical_cid is NOT stored. Derive it via the API instead:
curl -k https://metacore-dev.localhost:8083/api/meta/<cid> | jq '.canonical_cid, .cids'
# Schema-version sentinel
docker exec metacore-app redis-cli GET 'meta-core:schema-version'
+65 -25
View File
@@ -4,6 +4,25 @@
**Status:** Implemented (2026 alpha clean-wipe gate landed). Kept for design rationale.
> ⚠ **Two parts of this proposal were superseded before shipping.** This doc
> is the design record; it is *not* the schema reference. For what is actually
> stored, see `METADATA_KEYS.md` §2/§14.13 and
> [metadata-storage-structure.md](./metadata-storage-structure.md).
>
> 1. **`canonical_cid` is not a stored key.** The proposed reconciler
> (`reconcileCanonicalCIDLocked`) was never kept — a stored scalar isn't
> conflict-free to merge across peers. The canonical CID is **derived on
> read** by ranking the `cids` members (`internal/cid/rank.go` →
> `cid.Better`), and appears only in the `GET /api/meta/{cid}` response.
> 2. **`cids` is a key-set, not a Redis SET, and holds bare CIDs, not
> `<algo>:<value>` tokens.** One flat STRING per member:
> `file:<uuid>/cids/<bareCid>` = `"true"`. The reverse index is likewise
> `cid:<bareCid>` → uuid. A CIDv1's multicodec already names its hash
> function, so the `<algo>:` prefix was redundant.
>
> Text below that references either shape describes the *proposal*, not the
> code. The rank ordering and everything else did ship as written.
## Motivation
Today, every metadata entry in Redis is keyed by a content hash, specifically
@@ -79,19 +98,21 @@ today — anything that wrote `file:<midhash>/fileinfo/duration` now writes
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/cids/bagacbabaec… = "true" ← midhash (record address)
file:01JKR8XW5T4QMV7AHNJ2DEFGHK/cids/bafybeigd… = "true" ← sha2-256
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
cid:bagacbabaec… → 01JKR8XW5T4QMV7AHNJ2DEFGHK
cid:bafybeigd… → 01JKR8XW5T4QMV7AHNJ2DEFGHK
```
*(As shipped — the proposal originally wrote `cids` as a Redis SET of
`<algo>:<value>` tokens plus a `canonical_cid` scalar. See the Status note.)*
### UUID format: UUIDv7
UUIDv7 chosen over UUIDv4 or 64-bit random:
@@ -111,26 +132,33 @@ The metadata schema is open — any writer can put any key under
`file:<uuid>/`. This proposal does not change that. It adds three keys whose
sole writer is meta-core itself:
| Key | Type | Purpose |
| Key | Type (as shipped) | Purpose |
|-----|------|---------|
| `cids` | Redis SET | All CIDs known for this file. Mirror of the `cid:*` reverse index. Source of truth for delete-time GC. |
| `cids/<bareCid>` | STRING `"true"` — a *key-set*, one key per member | All CIDs known for this file, midhash included. Mirror of the `cid:*` reverse index. Source of truth for delete-time GC. **Proposed as a Redis SET of `<algo>:<value>` tokens; shipped as a key-set of bare CIDs** — a key-set merges conflict-free across peers and the multicodec already names the algorithm. |
| `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). |
| ~~`canonical_cid`~~ | *(not stored)* | **Dropped.** Was to hold the externally-presented CID. Now derived on read by ranking the `cids` members; surfaced only on `GET /api/meta/{cid}`. |
Existing meta-core-owned fields (`filePath`, `size`, `mtime`, `midhash256`)
keep their names and writers. Plugin-written fields (`tmdb/*`, `fileinfo/*`,
`subtitle/<lang>`, …) are unaffected.
Redis SET is a departure from the all-strings convention in the current
~~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.
for using sets where they're the right structure.~~ **Reversed for `cids`.**
The all-strings convention won: `cids` ships as flat `cids/<cid>` STRING keys
so that two peers merging the same record can never disagree about set
membership. `duplicates` remained a Redis SET (it is leader-local and never
merged across peers).
### Canonical CID selection
`canonical_cid` is a label, not a key. Updating it costs one write — no
rename, no race, no broken external references.
The canonical CID is a label, not a key — so nothing external breaks when it
changes. **As shipped it is not written at all**: it is computed on every read
from the `cids` members, which makes the "update" cost zero and removes the
reconciler entirely.
Selection rule, highest-wins:
Selection rule, highest-wins (this part shipped as designed —
`internal/cid/rank.go`):
1. IPFS CID (any variant)
2. SHA-256 / SHA-3 family
@@ -142,8 +170,10 @@ 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.
~~Reconciler runs on the leader, recomputes `canonical_cid` whenever
`AddAlias` registers a new CID for an entry.~~ **Not shipped** — there is no
reconciler. `GetMetadataDocument` folds the members through `cid.Better` on
each read.
## Internal operations
@@ -157,8 +187,8 @@ CID-based API.
Mint(filePath string, size int64, mtime int64) (uuid string, err error)
// AddAlias registers a CID as resolving to this UUID. Updates both the
// cid:<cid> reverse index and the file:<uuid>/cids set atomically.
// Triggers canonical_cid reconciliation.
// cid:<cid> reverse index and the file:<uuid>/cids/<cid> key-set member.
// (As shipped: no reconciliation step — canonical is derived on read.)
AddAlias(uuid string, cid string) error
// GetByCID resolves any known CID to its UUID. Returns ("", nil) if unknown.
@@ -199,7 +229,8 @@ DeleteRoot(uuid string) error
- 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.
- Nothing else to do: the next read of this entry derives a canonical CID
of sha2-256, because it outranks the midhash.
```
### Lifecycle: file deletion
@@ -250,13 +281,20 @@ so the impact is contained but real.
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.
- **meta-search / meta-share**: announce the *derived* canonical CID rather
than the midhash — each mirrors the Go rank in `ingest::canonical_cid_from_metadata`
rather than reading a stored field. Federated search responses include all
CIDs from the `cids` key-set so any peer querying by any digest gets a hit.
(Written before the 2026-06 discovery/transport split; discovery is now
meta-search's job.)
- **Plugins**: no change. They speak CIDs only.
## Migration
> **Historical — this sweep was never run.** The alpha shipped behind a
> clean-wipe gate (Redis was flushed and the library re-ingested), so no
> in-place migration code exists. Kept for the record.
One-time sweep, idempotent, runs on first boot after upgrade:
```
@@ -285,7 +323,8 @@ roots and resumes by detecting which step completed (presence of
| 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. |
| ~~`cids` and `duplicates` field type~~ | ~~Redis SET~~ → **`cids` is a key-set of bare CIDs; `duplicates` stayed a SET** | Reversed before shipping. Flat `cids/<cid>` STRING keys merge conflict-free across peers; a SET does not. See §14.13 of `METADATA_KEYS.md`. |
| ~~Stored `canonical_cid`~~ | **Dropped — derived on read** | A stored scalar is not conflict-free to merge. `GetMetadataDocument` ranks the `cids` members per request. |
| Reverse-index format | Flat `cid:<cid>` STRING keys | Simple, scan-friendly, no hot-key contention. |
## Open questions
@@ -297,9 +336,10 @@ roots and resumes by detecting which step completed (presence of
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.
- ~~**meta-share announce strategy.**~~ **Settled.** The record carries *every*
member of `cids` on the wire (`Record.cids`) and addresses itself by the
single derived canonical (`Record.cid`), so a peer querying by any digest
gets a hit without the DHT footprint of announcing each one separately.
- **IPFS gateway endpoint.** Should meta-core expose `/ipfs/<cid>` as a
content-streaming endpoint for external IPFS clients? Adjacent to this
+3
View File
@@ -3,6 +3,7 @@ module github.com/metazla/meta-core
go 1.21
require (
github.com/alicebob/miniredis/v2 v2.31.1
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1
github.com/google/uuid v1.6.0
github.com/gorilla/mux v1.8.1
@@ -13,6 +14,8 @@ require (
)
require (
github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a // indirect
github.com/cespare/xxhash/v2 v2.2.0 // indirect
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
github.com/yuin/gopher-lua v1.1.0 // indirect
)
+12
View File
@@ -1,15 +1,24 @@
github.com/DmitriyVTitov/size v1.5.0/go.mod h1:le6rNI4CoLQV1b9gzp1+3d7hMAD/uu2QcJ+aYbNgiU0=
github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a h1:HbKu58rmZpUGpz5+4FfNmIU+FmZg2P3Xaj2v2bfNWmk=
github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a/go.mod h1:SGnFV6hVsYE877CKEZ6tDNTjaSXYUk6QqoIK6PrAtcc=
github.com/alicebob/miniredis/v2 v2.31.1 h1:7XAt0uUg3DtwEKW5ZAGa+K7FZV2DdKQo5K/6TTnfX8Y=
github.com/alicebob/miniredis/v2 v2.31.1/go.mod h1:UB/T2Uztp7MlFSDakaX1sTXUv5CASoprx0wulRT6HBg=
github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
github.com/decred/dcrd/crypto/blake256 v1.1.0 h1:zPMNGQCm0g4QTY27fOCorQW7EryeQ/U0x++OzVrdms8=
github.com/decred/dcrd/crypto/blake256 v1.1.0/go.mod h1:2OfgNZ5wDpcsFmHmCK5gZTPcCXqlm2ArzUIkw9czNJo=
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1 h1:5RVFMOWjMyRy8cARdy79nAmgYw3hK/4HUq48LQ6Wwqo=
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1/go.mod h1:ZXNYxsqcloTdSy/rNShjYzMhyjf0LaoftYK0p+A3h40=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY=
@@ -18,7 +27,10 @@ github.com/mr-tron/base58 v1.3.0 h1:K6Y13R2h+dku0wOqKtecgRnBUBPrZzLZy5aIj8lCcJI=
github.com/mr-tron/base58 v1.3.0/go.mod h1:2BuubE67DCSWwVfx37JWNG8emOC0sHEU4/HpcYgCLX8=
github.com/redis/go-redis/v9 v9.4.0 h1:Yzoz33UZw9I/mFhx4MNrB6Fk+XHO1VukNcCa1+lwyKk=
github.com/redis/go-redis/v9 v9.4.0/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
github.com/yuin/gopher-lua v1.1.0 h1:BojcDhfyDWgU2f2TOzYK/g5p2gxMrku8oupLDqlnSqE=
github.com/yuin/gopher-lua v1.1.0/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
golang.org/x/net v0.21.0 h1:AQyQV4dYCvJ7vGmJyKki9+PBdyvhkSd8EIx/qb0AYv4=
golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
golang.org/x/sys v0.0.0-20190204203706-41f3e6584952/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+86
View File
@@ -0,0 +1,86 @@
package api
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
)
// The per-algorithm CID fields (`cid_sha2-256`, `cid_midhash256`, …) and the
// stored `canonical_cid` scalar were removed (METADATA_KEYS.md §14.13). They
// must be rejected at the write boundary rather than stored.
//
// Why a hard 400 and not a silent rewrite: the reverse-index hook
// (maybeAddAliasFromFieldLocked) only registers an alias for a field named
// `cids/<cid>`. A `cid_*` field was therefore written verbatim and never
// indexed — the record silently became unresolvable by that CID. A loud
// failure is strictly better than a record that looks correct and cannot be
// found.
func TestRejectDeprecatedField(t *testing.T) {
rejected := []string{
"cid_midhash256",
"cid_sha2-256",
"cid_btih_v2",
"cid_crc32",
"midhash256",
"canonical_cid",
}
for _, f := range rejected {
if !rejectDeprecatedField(f) {
t.Errorf("rejectDeprecatedField(%q) = false, want true (deprecated shape)", f)
}
}
accepted := []string{
"cids/bagacbabaec7v3fu2ygzh3e2sybg3fbzmisry2hbtpmck6vx3yftea6vzq35r4",
"filePath",
"title",
"fileinfo/duration",
"tmdb/poster",
// Near-misses that must NOT be caught: the prefix is `cid_`, with an
// underscore. A field merely starting with "cid" is fine.
"cids",
"cidrank",
}
for _, f := range accepted {
if rejectDeprecatedField(f) {
t.Errorf("rejectDeprecatedField(%q) = true, want false (legitimate field)", f)
}
}
}
func TestValidateFields_Rejects400WithSlug(t *testing.T) {
rr := httptest.NewRecorder()
blocked := validateFields(rr, map[string]string{
"title": "Inception",
"cid_midhash256": "bagacbabaec7v3fu2ygzh3e2sybg3fbzmisry2hbtpmck6vx3yftea6vzq35r4",
})
if !blocked {
t.Fatal("validateFields returned false for a payload carrying cid_midhash256; the write would have been stored unindexed")
}
if rr.Code != http.StatusBadRequest {
t.Errorf("status = %d, want %d", rr.Code, http.StatusBadRequest)
}
var body map[string]interface{}
if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil {
t.Fatalf("response is not JSON: %v", err)
}
if got := body["error"]; got != ErrDeprecatedField {
t.Errorf("error slug = %v, want %q", got, ErrDeprecatedField)
}
}
func TestValidateFields_AllowsCidsKeyset(t *testing.T) {
rr := httptest.NewRecorder()
blocked := validateFields(rr, map[string]string{
"filePath": "/watch/Inception.mkv",
"cids/bagacbabaec7v3fu2ygzh3e2sybg3fbzmisry2hbtpmck6vx3yftea6vzq35r4": "true",
"cids/baejbeibku6ua2l6r5olnggx4pto2sii2s5jeeawgh4cbkhlbmp52gudbua": "true",
})
if blocked {
t.Fatalf("validateFields rejected a legitimate cids/ key-set write: %s", rr.Body.String())
}
}
+54
View File
@@ -4,6 +4,7 @@ import (
"crypto/sha256"
"encoding/base32"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
@@ -84,8 +85,45 @@ const (
ErrSchemaViolation = "schema_violation"
ErrStorageUnavailable = "storage_unavailable"
ErrInternal = "internal"
ErrDeprecatedField = "deprecated_field"
)
// rejectDeprecatedField reports whether a field name is one of the removed
// per-algorithm CID shapes (`cid_sha2-256`, `cid_midhash256`, `midhash256`, …).
//
// This is a tripwire, not a compatibility shim. Such a field used to be
// accepted and written verbatim — but the reverse-index hook
// (`maybeAddAliasFromFieldLocked`) only registers aliases for fields named
// `cids/<cid>`, so a `cid_*` field was stored and then NEVER INDEXED: the
// record silently became unresolvable by that CID. Failing the write loudly is
// strictly better than a record that looks fine and cannot be found.
//
// Writers must emit the bare-CID key-set instead: `cids/<bareCid>` = "true".
// See METADATA_KEYS.md §2/§14.13.
func rejectDeprecatedField(field string) bool {
return strings.HasPrefix(field, "cid_") ||
field == "midhash256" ||
field == "canonical_cid"
}
// validateFields rejects a write carrying any deprecated CID field. Returns
// true if the request was rejected (the handler must return immediately).
func validateFields(w http.ResponseWriter, metadata map[string]string) bool {
for field := range metadata {
if rejectDeprecatedField(field) {
writeErrorSlug(w, http.StatusBadRequest, ErrDeprecatedField,
fmt.Sprintf("field %q is a removed per-algorithm CID shape; "+
"write the bare-CID key-set instead: \"cids/<bareCid>\": \"true\" "+
"(a cid_* field would be stored but never indexed, leaving the "+
"record unresolvable by that CID). See METADATA_KEYS.md §2/§14.13.",
field),
false)
return true
}
}
return false
}
// 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.
@@ -255,6 +293,10 @@ func (s *Server) handlePutMeta(w http.ResponseWriter, r *http.Request) {
return
}
if validateFields(w, metadata) {
return
}
if err := s.storage.SetMetadataFlat(hashID, metadata); err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
@@ -620,6 +662,10 @@ func (s *Server) handlePatchMeta(w http.ResponseWriter, r *http.Request) {
return
}
if validateFields(w, metadata) {
return
}
updated, err := s.storage.MergeMetadataFlat(hashID, metadata)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
@@ -697,6 +743,10 @@ func (s *Server) handlePutProperty(w http.ResponseWriter, r *http.Request) {
return
}
if validateFields(w, map[string]string{key: ""}) {
return
}
if !s.storage.IsConnected() {
writeError(w, http.StatusServiceUnavailable, "storage not connected")
return
@@ -790,6 +840,10 @@ func (s *Server) handleAddToSet(w http.ResponseWriter, r *http.Request) {
return
}
if validateFields(w, map[string]string{key: ""}) {
return
}
if !s.storage.IsConnected() {
writeError(w, http.StatusServiceUnavailable, "storage not connected")
return
+20
View File
@@ -193,6 +193,9 @@ func (s *Server) handleGetMetadataByHashId(w http.ResponseWriter, r *http.Reques
return
}
// Same resolution as the PUT below: callers hold a CID, not the root.
hashID = s.storage.ResolveRoot(hashID)
metadata, err := s.storage.GetMetadataFlat(hashID)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
@@ -227,6 +230,17 @@ func (s *Server) handleUpdateMetadataByHashId(w http.ResponseWriter, r *http.Req
return
}
// The {hashId} in the URL is whatever CID the caller knows the file by —
// the gateway's auto-store keys this PUT by the hash it just computed
// (dispatch.rs → store_metadata_only), which for a usenet→IPFS fetch is
// the dag-pb root, not the nzb-release cid the record was minted under.
// Without resolving, the write lands on file:<thatCid>/* and forks a
// second root for a file that already has one: a full duplicate record,
// cross-linked and separately indexed, which every downstream ingest then
// sees as two files. Unknown CIDs pass through unchanged, so a genuinely
// new gateway record still mints its root at the hash.
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")
@@ -263,6 +277,12 @@ func (s *Server) handleDeleteMetadataByHashId(w http.ResponseWriter, r *http.Req
return
}
// meta-gateway retracts a dropped hit by whichever CID it holds
// (meta_core.rs `delete`, which documents this resolution as a
// precondition). Without it, deleting by a non-root CID is a silent
// no-op and the retracted record lingers.
hashID = s.storage.ResolveRoot(hashID)
deletedCount, err := s.storage.DeleteMetadata(hashID)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
+9
View File
@@ -201,6 +201,15 @@ func (s *Server) setupRoutes() {
s.router.HandleFunc("/api/kv/value", s.handleKVValueDelete).Methods("DELETE")
s.router.HandleFunc("/api/kv/key/{key:.*}", s.handleKVGetKey).Methods("GET")
// User Data Layer (UDL) routes — per-user signed records (My List,
// Continue Watching, ratings, profile). See internal/api/udl.go.
s.router.HandleFunc("/api/udl/record", s.handleUDLRecordGet).Methods("GET")
s.router.HandleFunc("/api/udl/record", s.handleUDLRecordPut).Methods("PUT")
s.router.HandleFunc("/api/udl/user/{uid}/key/{key}", s.handleUDLUserKey).Methods("GET")
s.router.HandleFunc("/api/udl/user/{uid}/cid/{cid}", s.handleUDLUserCid).Methods("GET")
s.router.HandleFunc("/api/udl/cid/{cid}/users", s.handleUDLCidUsers).Methods("GET")
s.router.HandleFunc("/api/udl/cid/{cid}/aggregate", s.handleUDLAggregate).Methods("GET")
// Metadata operations - base endpoints
s.router.HandleFunc("/meta/{hash}", s.handleGetMeta).Methods("GET")
s.router.HandleFunc("/meta/{hash}", s.handlePutMeta).Methods("PUT")
+239
View File
@@ -0,0 +1,239 @@
package api
// User Data Layer (UDL) HTTP surface.
//
// Backs meta-watch's per-user store (My List, Continue Watching, ratings,
// profile) now that the authoritative store lives in meta-core instead of
// meta-watch's local op-log. Records arrive already signed (and, for
// private-tier keys, already encrypted) as opaque base64 CBOR blobs; meta-core
// version-gates and indexes them but never inspects the crypto. See
// internal/storage/udl.go for the Redis model.
import (
"encoding/json"
"math"
"net/http"
"strings"
"github.com/gorilla/mux"
)
// UDLPutRequest is the body of PUT /api/udl/record.
type UDLPutRequest struct {
Uid string `json:"uid"`
Cid string `json:"cid"`
Key string `json:"key"`
Version int64 `json:"version"`
Ts int64 `json:"ts"`
Record string `json:"record"` // base64 CBOR, opaque + signed
Value *json.RawMessage `json:"value,omitempty"`
}
// UDLPutResponse is the response of PUT /api/udl/record.
type UDLPutResponse struct {
Accepted bool `json:"accepted"`
}
// UDLRecordResponse is the response of GET /api/udl/record.
type UDLRecordResponse struct {
Record string `json:"record"`
Version int64 `json:"version"`
Ts int64 `json:"ts"`
Exists bool `json:"exists"`
}
// UDLAggregate mirrors meta-watch's Aggregate wire shape exactly (snake_case
// field names; `avg` is null for boolean keys like `like`), so the meta-watch
// client deserializes it into the same struct it serializes to the UI.
type UDLAggregate struct {
Key string `json:"key"`
Count int `json:"count"`
Avg *float64 `json:"avg"`
Trues int `json:"trues"`
SampleSize int `json:"sample_size"`
}
// handleUDLRecordGet handles GET /api/udl/record?uid&cid&key
func (s *Server) handleUDLRecordGet(w http.ResponseWriter, r *http.Request) {
if !s.storage.IsConnected() {
writeError(w, http.StatusServiceUnavailable, "storage not connected")
return
}
q := r.URL.Query()
uid, cid, key := q.Get("uid"), q.Get("cid"), q.Get("key")
if uid == "" || cid == "" || key == "" {
writeError(w, http.StatusBadRequest, "uid, cid and key are required")
return
}
cell, exists, err := s.storage.UDLGetCell(uid, cid, key)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, UDLRecordResponse{
Record: cell.Record,
Version: cell.Version,
Ts: cell.Ts,
Exists: exists,
})
}
// handleUDLRecordPut handles PUT /api/udl/record. Version-gated: a stale write
// (version <= stored) returns 409 stale_version so the client can re-read and
// retry with a higher version.
func (s *Server) handleUDLRecordPut(w http.ResponseWriter, r *http.Request) {
if !s.storage.IsConnected() {
writeError(w, http.StatusServiceUnavailable, "storage not connected")
return
}
var req UDLPutRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid JSON body")
return
}
if req.Uid == "" || req.Cid == "" || req.Key == "" || req.Record == "" {
writeError(w, http.StatusBadRequest, "uid, cid, key and record are required")
return
}
publicValue := ""
hasValue := false
if req.Value != nil {
publicValue = string(*req.Value)
hasValue = true
}
accepted, err := s.storage.UDLUpsertIfNewer(req.Uid, req.Cid, req.Key, req.Version, req.Ts, req.Record, publicValue, hasValue)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
if !accepted {
writeErrorSlug(w, http.StatusConflict, "stale_version", "a record with an equal or newer version already exists", true)
return
}
writeJSON(w, http.StatusOK, UDLPutResponse{Accepted: true})
}
// handleUDLUserKey handles GET /api/udl/user/{uid}/key/{key}
// (active_for_user_key — My List, Continue Watching).
func (s *Server) handleUDLUserKey(w http.ResponseWriter, r *http.Request) {
if !s.storage.IsConnected() {
writeError(w, http.StatusServiceUnavailable, "storage not connected")
return
}
vars := mux.Vars(r)
uid, key := vars["uid"], vars["key"]
if uid == "" || key == "" {
writeError(w, http.StatusBadRequest, "uid and key are required")
return
}
entries, err := s.storage.UDLListUserKey(uid, key)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]interface{}{"entries": entries})
}
// handleUDLUserCid handles GET /api/udl/user/{uid}/cid/{cid}
// (list_for_user_cid).
func (s *Server) handleUDLUserCid(w http.ResponseWriter, r *http.Request) {
if !s.storage.IsConnected() {
writeError(w, http.StatusServiceUnavailable, "storage not connected")
return
}
vars := mux.Vars(r)
uid, cid := vars["uid"], vars["cid"]
if uid == "" || cid == "" {
writeError(w, http.StatusBadRequest, "uid and cid are required")
return
}
entries, err := s.storage.UDLListUserCid(uid, cid)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]interface{}{"entries": entries})
}
// handleUDLCidUsers handles GET /api/udl/cid/{cid}/users (uids_for_cid).
func (s *Server) handleUDLCidUsers(w http.ResponseWriter, r *http.Request) {
if !s.storage.IsConnected() {
writeError(w, http.StatusServiceUnavailable, "storage not connected")
return
}
cid := mux.Vars(r)["cid"]
if cid == "" {
writeError(w, http.StatusBadRequest, "cid is required")
return
}
users, err := s.storage.UDLCidUsers(cid)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]interface{}{"users": users})
}
// handleUDLAggregate handles GET /api/udl/cid/{cid}/aggregate?keys=like,rating.
// Computes per-key population stats from the stored public value fields
// (absorbs meta-watch's aggregate.rs). Reflects only records this core holds.
func (s *Server) handleUDLAggregate(w http.ResponseWriter, r *http.Request) {
if !s.storage.IsConnected() {
writeError(w, http.StatusServiceUnavailable, "storage not connected")
return
}
cid := mux.Vars(r)["cid"]
if cid == "" {
writeError(w, http.StatusBadRequest, "cid is required")
return
}
keysParam := r.URL.Query().Get("keys")
if keysParam == "" {
keysParam = "like,rating"
}
aggregates := make([]UDLAggregate, 0)
for _, key := range strings.Split(keysParam, ",") {
key = strings.TrimSpace(key)
if key == "" {
continue
}
values, err := s.storage.UDLPublicValues(cid, key)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
aggregates = append(aggregates, aggregatePublicValues(key, values))
}
writeJSON(w, http.StatusOK, map[string]interface{}{"aggregates": aggregates})
}
// aggregatePublicValues folds raw JSON scalar values into an UDLAggregate:
// counts booleans as trues and averages numeric values.
func aggregatePublicValues(key string, values []string) UDLAggregate {
agg := UDLAggregate{Key: key, SampleSize: len(values), Count: len(values)}
var sum float64
var numeric int
for _, raw := range values {
var v interface{}
if err := json.Unmarshal([]byte(raw), &v); err != nil {
continue
}
switch t := v.(type) {
case bool:
if t {
agg.Trues++
}
case float64:
sum += t
numeric++
}
}
if numeric > 0 {
avg := math.Round((sum/float64(numeric))*100) / 100
agg.Avg = &avg
}
return agg
}
+153
View File
@@ -0,0 +1,153 @@
{
"$schema": "cid-rank-vectors/v1",
"$comment": [
"CANONICAL golden vectors for the CID utility rank. THE single source of truth.",
"The rank decides which of a record's several CIDs is elected canonical - the cid a",
"record is advertised and addressed by. It is reimplemented in Go, 2x Rust and 3x TS;",
"they cannot share code (independent submodules, no cargo path-dep). These vectors are",
"the only thing holding them in sync: every implementation's test suite loads this file.",
"",
"This file is VENDORED into each consuming submodule (submodule CI cannot read above its",
"own tree). scripts/check-cid-vectors.sh diffs every copy against this one.",
"",
"Ranking is by UTILITY for external retrieval, not cryptographic strength.",
"Ties break on the lexicographically smaller cid, so the election is deterministic",
"across peers and does not churn.",
"",
"See METADATA_KEYS.md section 2/14.13 and docs/cid-formats.md."
],
"ladder": {
"40": "dag-pb (IPFS block)",
"30": "sha2-256 / sha3-256 / sha3-384 - real content digests",
"20": "BitTorrent infohashes - swarm interop",
"10": "midhash256 - the record address",
"5": "opaque locators (nzb-release, url) - never outrank a real digest",
"0": "weak/unknown digests, unparseable input"
},
"tieBreak": "lexicographically smaller cid wins",
"vectors": [
{
"name": "ipfs-dagpb",
"cid": "bafybeiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 40,
"note": "dag-pb codec 0x70 - chunked IPFS file, broadest network reach"
},
{
"name": "sha2-256-raw-codec",
"cid": "bafkreiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 30,
"note": "fullhash/gateway convention: raw codec 0x55, mh names the algo"
},
{
"name": "sha2-256-codec-eq-mh",
"cid": "baejbeibku6ua2l6r5olnggx4pto2sii2s5jeeawgh4cbkhlbmp52gudbua",
"rank": 30,
"note": "REAL meta-hash fixture; meta-hash sets codec == mh"
},
{
"name": "sha3-256",
"cid": "baelbmidner7x3dd7t2kldh6wiq4w6lsye7qrwo6uoqot2t52zkhqk56fnq",
"rank": 30,
"note": "REAL meta-hash fixture"
},
{
"name": "sha3-384",
"cid": "baekrkmeglipkjf33exqexdvijl4twipruay4wsbvt44akce43lq5rgccxgr7v6dwksmj3sxedakevc53g6ga",
"rank": 30,
"note": "REAL meta-hash fixture"
},
{
"name": "btih-v2-codec-eq-mh",
"cid": "bag3sdnzbeaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 20,
"note": "meta-hash encoder convention"
},
{
"name": "btih-v2-raw-codec",
"cid": "bafk3oijaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 20,
"note": "fullhash encoder convention - SAME tier, both conventions must match"
},
{
"name": "btih-v1-file",
"cid": "bagasbajacqaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 20,
"note": "gateway per-file BitTorrent v1"
},
{
"name": "btih-v2-file",
"cid": "bagbcbaraeaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 20,
"note": "gateway per-file BitTorrent v2"
},
{
"name": "midhash256-codec-eq-mh",
"cid": "bagacbabaeaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 10,
"note": "the record address; internally meaningful, externally invisible"
},
{
"name": "midhash256-raw-codec",
"cid": "bafkyaibaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 10
},
{
"name": "nzb-release-locator",
"cid": "bagcsaaazcbqxa2jonz5gez3fmvvs42lomzx55ln657fp4aab",
"rank": 5,
"note": "codec 0x1005 + IDENTITY multihash 0x00. The locator code is in the CODEC slot, not the mh slot - rank MUST match on codec. Tier 5 = below midhash: a locator never outranks a real digest, it wins only as a record's sole cid."
},
{
"name": "url-locator",
"cid": "bagdcaaaznb2hi4dthixs6zlymfwxa3dffzxxezzpmexhazdg",
"rank": 5,
"note": "codec 0x1006 + IDENTITY multihash 0x00. Same tier/reasoning as nzb-release."
},
{
"name": "sha1",
"cid": "baeircff3eqvxxaqqot2lt242stkdtk3gq4c2vra",
"rank": 0,
"note": "REAL meta-hash fixture; weak digest"
},
{
"name": "md5",
"cid": "bahkqdvibcatpcfl7ine2n625eb5ml5mzkfaa",
"rank": 0,
"note": "REAL meta-hash fixture; weak digest"
},
{
"name": "crc32",
"cid": "bagzafmqcathsbnrk",
"rank": 0,
"note": "REAL meta-hash fixture; weak digest"
},
{
"name": "bt-pieces-root",
"cid": "bafkyf3qceaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 0,
"note": "torrent-internal merkle root, NOT a file-identity cid"
},
{
"name": "cidv0-rejected",
"cid": "QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG",
"rank": 0,
"note": "TRAP: Go rejected this (rank 0) while Rust's MsCid::from_str ACCEPTED it and ranked it 40 (dag-pb). Strict parse = CIDv1 base32-lower only. This vector pins the fix."
},
{
"name": "token-form-rejected",
"cid": "midhash256:bagacbabaeaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 0,
"note": "TRAP: the deprecated <algo>:<cid> token form. Rust used to strip the prefix and rank it 10. It must now be rejected."
},
{
"name": "garbage",
"cid": "not-a-cid",
"rank": 0
},
{
"name": "empty",
"cid": "",
"rank": 0
}
]
}
+43 -7
View File
@@ -20,6 +20,14 @@ import (
// Multicodec / multihash codes used for ranking. Source of truth for the
// digest codes is meta-hash/src/lib/hash-compute/MultiHashData.ts and
// METADATA_KEYS.md §2.
//
// The tier every code maps to is pinned by the golden fixture
// `cid-rank-vectors.json` (vendored next to this package). This is one of
// seven implementations of the same ladder across Go, Rust and TypeScript;
// the fixture is the only thing keeping them in agreement. If you add a code
// here, add a vector — an unranked code silently falls to tier 0, which is
// exactly how the nzb-release/url locators came to be ranked 20 in meta-share
// and 0 here and in meta-search.
const (
CodecRaw = 0x55 // raw — IPFS content-addressed block
CodecDagPB = 0x70 // dag-pb — IPFS content-addressed block
@@ -34,6 +42,23 @@ const (
CodeBtihV2 = 0x10B7 // custom — BitTorrent v2 info hash (BEP 52)
CodeBtihV1File = 0x1001 // custom — per-file BitTorrent v1 CID
CodeBtihV2File = 0x1002 // custom — per-file BitTorrent v2 CID
// Opaque locators. These are NOT digests: the payload rides in an identity
// multihash (0x00), so the code identifying them sits in the *content
// codec* slot, never the multihash slot. Rank must match on the codec for
// these — matching on mh would see 0x00 and rank them 0.
CodeNzbRelease = 0x1005 // custom — self-describing Newznab release locator
CodeURL = 0x1006 // custom — identity-multihash CID wrapping an http(s) URL
)
// Rank tiers, named so the ladder reads as a ladder.
const (
RankIPFS = 40 // dag-pb — broadest network reach
RankDigest = 30 // sha2-256 / sha3-* — real content digests
RankBtih = 20 // BitTorrent infohashes — swarm interop
RankMidhash = 10 // the record address — internally meaningful, externally invisible
RankLocator = 5 // opaque locators — never outrank a real digest
RankUnknown = 0 // weak/unknown digests, unparseable input
)
// Decode parses a multibase-base32 CIDv1 string ("b…") into its content
@@ -89,31 +114,42 @@ func Decode(cidStr string) (codec uint64, mhCode uint64, err error) {
// block of the file IS its sha2-256, so they share this tier).
// - BitTorrent infohashes (btih v2, per-file v1/v2) outrank midhash256 for
// swarm interop, not strength.
// - midhash256 is the floor among real options: internally meaningful,
// - midhash256 is the floor among real *digests*: internally meaningful,
// externally invisible.
// - opaque locators (nzb-release, url) sit BELOW midhash. They are not
// content-derived — two URLs to the same bytes yield two different locator
// CIDs — so a locator must never displace a real digest as the identity a
// record is advertised under. Tier 5 means a locator wins the election only
// when it is the record's sole CID, which is exactly the guarantee
// METADATA_KEYS.md §2 states.
// - md5 / crc32 / plain sha1 / pieces-root / unknown rank 0.
func Rank(cidStr string) int {
codec, mh, err := Decode(cidStr)
if err != nil {
return 0
return RankUnknown
}
if codec == CodecDagPB {
return 40
return RankIPFS
}
// BitTorrent: distinct CID codecs (per-file v1/v2) or the btih-v2
// multihash code, in either encoder convention.
if codec == CodeBtihV1File || codec == CodeBtihV2File || codec == CodeBtihV2 ||
mh == CodeBtihV1File || mh == CodeBtihV2File || mh == CodeBtihV2 {
return 20
return RankBtih
}
// Opaque locators. Codec-only: their multihash is identity (0x00), so
// there is nothing to match on the mh side.
if codec == CodeNzbRelease || codec == CodeURL {
return RankLocator
}
switch mh {
case CodeSha256, CodeSha3_256, CodeSha3_384:
return 30
return RankDigest
case CodeMidhash256:
return 10
return RankMidhash
default:
// md5, crc32, plain sha1, pieces-root, unknown.
return 0
return RankUnknown
}
}
+46
View File
@@ -3,10 +3,56 @@ package cid
import (
"encoding/base32"
"encoding/binary"
"encoding/json"
"os"
"strings"
"testing"
)
// goldenVectors is the CANONICAL cross-language fixture: the same file is
// vendored into meta-search, meta-share and meta-hash, and each of their test
// suites asserts against it. It is the only mechanism keeping the seven
// implementations of this ladder in agreement — they live in independent
// submodules and cannot share code. `scripts/check-cid-vectors.sh` at the
// meta-root fails if any vendored copy drifts from the canonical one.
const goldenVectors = "cid-rank-vectors.json"
type vector struct {
Name string `json:"name"`
CID string `json:"cid"`
Rank int `json:"rank"`
Note string `json:"note"`
}
func loadVectors(t *testing.T) []vector {
t.Helper()
raw, err := os.ReadFile(goldenVectors)
if err != nil {
t.Fatalf("read %s: %v (run ./scripts/check-cid-vectors.sh --fix from the meta-root)", goldenVectors, err)
}
var doc struct {
Vectors []vector `json:"vectors"`
}
if err := json.Unmarshal(raw, &doc); err != nil {
t.Fatalf("parse %s: %v", goldenVectors, err)
}
if len(doc.Vectors) == 0 {
t.Fatalf("%s has no vectors", goldenVectors)
}
return doc.Vectors
}
// TestRank_GoldenVectors is the load-bearing test. Every other test in this
// file is a convenience; this one is the contract with Rust and TypeScript.
func TestRank_GoldenVectors(t *testing.T) {
for _, v := range loadVectors(t) {
if got := Rank(v.CID); got != v.Rank {
t.Errorf("Rank(%s) = %d, want %d\n cid: %s\n note: %s",
v.Name, got, v.Rank, v.CID, v.Note)
}
}
}
// buildCID assembles a multibase-base32 CIDv1 from a content codec, a
// multihash code, and a digest length (digest bytes are zero — only the
// header is under test). Mirrors the encoder in watcher/midhash.go and the
+2 -2
View File
@@ -24,7 +24,7 @@ type Config struct {
// HTTP API configuration
HTTPPort int // HTTP API port (default: 9000)
HTTPHost string // HTTP API host (default: "127.0.0.1")
HTTPHost string // HTTP API host (default: "0.0.0.0" — meta-core always runs in a container and must answer siblings)
// Timing configuration
HealthCheckIntervalMS int // Health check interval in ms (default: 5000)
@@ -54,7 +54,7 @@ func Load() *Config {
PublicURL: getEnv("META_CORE_PUBLIC_URL", ""),
RedisPort: getEnvInt("REDIS_PORT", 6379),
HTTPPort: getEnvInt("META_CORE_HTTP_PORT", 9000),
HTTPHost: getEnv("META_CORE_HTTP_HOST", "127.0.0.1"),
HTTPHost: getEnv("META_CORE_HTTP_HOST", "0.0.0.0"),
HealthCheckIntervalMS: getEnvInt("HEALTH_CHECK_INTERVAL_MS", 5000),
HeartbeatIntervalMS: getEnvInt("HEARTBEAT_INTERVAL_MS", 30000),
StaleThresholdMS: getEnvInt("STALE_THRESHOLD_MS", 60000),
+7 -5
View File
@@ -5,11 +5,13 @@
// reject archives whose schemaVersion exceeds what they know.
//
// v2 (current): root identifiers are UUIDv7 (Crockford Base32). Entries
// under metadata/<root>.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).
// under metadata/<root>.json carry the flat string properties; the cid
// reverse-index and the duplicates SET are NOT serialized — they are rebuilt
// transparently on import because the cids/<cid> key-set members ARE flat
// properties, and SetMetadataFlat / MergeMetadataFlat auto-register a
// reverse-index alias for every one they see (see
// internal/storage/cid_resolution.go: cidFromKeysetField). The legacy flat
// cid_*/midhash256 named fields are not recognised.
//
// v1 (legacy): roots were content-hash-typed tokens like "midhash256:abc".
// v1 archives can no longer be imported — they would resurrect the
+138
View File
@@ -0,0 +1,138 @@
package storage
import (
"testing"
"github.com/alicebob/miniredis/v2"
)
// newTestClient spins an in-memory Redis and returns a connected Client with
// the given namespace prefix. The server is torn down with the test.
func newTestClient(t *testing.T, prefix string) (*Client, *miniredis.Miniredis) {
t.Helper()
s := miniredis.RunT(t)
c := NewClient(prefix)
if err := c.Connect("redis://" + s.Addr()); err != nil {
t.Fatalf("connect to miniredis: %v", err)
}
t.Cleanup(func() { _ = c.Close() })
return c, s
}
// TestClearAllMetadata_PurgesCIDReverseIndex is the regression test for the
// leak: ClearAllMetadata used to delete every file:<root>/* key and the
// file:__index__ set, but left the cid:<cid> → root reverse-index entries
// behind. Those stale aliases are not cosmetic — ResolveRoot consults them, so
// a surviving entry silently redirects a later write into a deleted root.
func TestClearAllMetadata_PurgesCIDReverseIndex(t *testing.T) {
for _, prefix := range []string{"", "mm:"} {
name := "no-prefix"
if prefix != "" {
name = "prefixed"
}
t.Run(name, func(t *testing.T) {
c, s := newTestClient(t, prefix)
// Two records, each with several CID aliases (a midhash plus
// siblings) — exactly what the watcher produces.
aliases := map[string][]string{}
for i, f := range []string{"watch/a.mkv", "watch/b.mkv"} {
uuid, err := c.Mint(f, int64(100+i), int64(1000+i))
if err != nil {
t.Fatalf("mint: %v", err)
}
cids := []string{
"bafkmid" + uuid,
"bafkfull" + uuid,
}
for _, cidStr := range cids {
if err := c.AddAlias(uuid, cidStr); err != nil {
t.Fatalf("add alias %s: %v", cidStr, err)
}
}
aliases[uuid] = cids
}
// An entry already orphaned by a previous (buggy) clear: a
// reverse-index key whose root no longer exists. The sweep is
// expected to collect it too.
if err := s.Set(c.buildCIDIndexKey("bafkorphan"), "01OLDDEADROOT"); err != nil {
t.Fatalf("seed orphan: %v", err)
}
// Sanity: the aliases resolve before the clear.
for uuid, cids := range aliases {
for _, cidStr := range cids {
if got := c.ResolveRoot(cidStr); got != uuid {
t.Fatalf("pre-clear ResolveRoot(%s) = %q, want %q", cidStr, got, uuid)
}
}
}
deleted, err := c.ClearAllMetadata()
if err != nil {
t.Fatalf("ClearAllMetadata: %v", err)
}
// deletedCount counts files, not keys.
if deleted != 2 {
t.Errorf("deletedCount = %d, want 2 (files, not keys)", deleted)
}
// No reverse-index key may survive — not for the cleared records,
// not the pre-existing orphan.
if leaked := s.Keys(); len(leaked) != 0 {
t.Errorf("keys survived the clear: %v", leaked)
}
// And the aliases must no longer resolve: ResolveRoot returns the
// hash unchanged when there is no index entry.
for _, cids := range aliases {
for _, cidStr := range cids {
if got := c.ResolveRoot(cidStr); got != cidStr {
t.Errorf("post-clear ResolveRoot(%s) = %q, want the hash unchanged (stale alias)", cidStr, got)
}
}
}
if got, err := c.GetByCID("bafkorphan"); err != nil || got != "" {
t.Errorf("post-clear GetByCID(bafkorphan) = %q (err %v), want \"\"", got, err)
}
})
}
}
// TestClearAllMetadata_LeavesForeignKeysAlone guards the sweep's blast radius:
// it must key off buildCIDIndexKey (namespace-aware), not a hardcoded "cid:",
// so a clear on a prefixed client cannot reach another namespace's index.
func TestClearAllMetadata_LeavesForeignKeysAlone(t *testing.T) {
c, s := newTestClient(t, "mm:")
uuid, err := c.Mint("watch/a.mkv", 100, 1000)
if err != nil {
t.Fatalf("mint: %v", err)
}
if err := c.AddAlias(uuid, "bafkmine"); err != nil {
t.Fatalf("add alias: %v", err)
}
// Another namespace's reverse-index entry, plus an unrelated key.
if err := s.Set("other:cid:bafktheirs", "01THEIRROOT"); err != nil {
t.Fatalf("seed foreign cid: %v", err)
}
if err := s.Set("unrelated", "keep-me"); err != nil {
t.Fatalf("seed unrelated: %v", err)
}
if _, err := c.ClearAllMetadata(); err != nil {
t.Fatalf("ClearAllMetadata: %v", err)
}
for _, k := range []string{"other:cid:bafktheirs", "unrelated"} {
if !s.Exists(k) {
t.Errorf("clear deleted foreign key %q — the cid sweep is not namespace-aware", k)
}
}
if s.Exists(c.buildCIDIndexKey("bafkmine")) {
t.Errorf("own reverse-index entry survived the clear")
}
}
+45 -7
View File
@@ -192,8 +192,9 @@ 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.
// Auto-register reverse-index aliases for any cids/<cid> key-set members in
// the payload. That is the ONLY shape recognised — the legacy flat cid_* /
// midhash256 named fields are not (see cid_resolution.go: cidFromKeysetField).
c.maybeAddAliasesFromMetadataLocked(ctx, hashID, metadata)
return nil
@@ -461,8 +462,9 @@ 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.
// Auto-register reverse-index aliases for any cids/<cid> key-set members in
// the payload. That is the ONLY shape recognised — the legacy flat cid_* /
// midhash256 named fields are not (see cid_resolution.go: cidFromKeysetField).
c.maybeAddAliasesFromMetadataLocked(ctx, hashID, metadata)
return len(metadata), nil
@@ -819,8 +821,16 @@ func (c *Client) GetMemoryInfo() (string, error) {
return "N/A", nil
}
// ClearAllMetadata deletes all file metadata and index
// Returns the number of files deleted
// ClearAllMetadata deletes all file metadata, the file index, and the CID
// reverse index. Returns the number of files deleted.
//
// The CID reverse index (cid:<cid> → root, see cid_resolution.go) MUST be
// purged too: ResolveRoot consults it, so a surviving cid:<cid> entry pointing
// at a root we just deleted silently redirects the next /meta/<cid> write back
// into that dead root, resurrecting a zombie record under an old identity.
// DeleteRoot already does this per-record; here we sweep the whole prefix
// rather than deriving the CID list per root, which is both simpler and
// self-healing — it also collects entries orphaned by earlier clears.
func (c *Client) ClearAllMetadata() (int64, error) {
c.mu.Lock()
defer c.mu.Unlock()
@@ -868,12 +878,40 @@ func (c *Client) ClearAllMetadata() (int64, error) {
deletedCount++
}
// Delete every CID reverse-index entry (cid:<cid> → root). Swept by
// prefix rather than per-root so entries orphaned by an earlier clear —
// which did not purge this index at all — are collected too. Built via
// buildCIDIndexKey so the configured namespace prefix is honoured.
cidIndexPattern := c.buildCIDIndexKey("*")
var cidIndexDeleted int64
var cidCursor uint64
for {
keys, nextCursor, err := c.client.Scan(ctx, cidCursor, cidIndexPattern, 1000).Result()
if err != nil {
log.Printf("[Storage] Warning: failed to scan %s: %v", cidIndexPattern, err)
break
}
if len(keys) > 0 {
if err := c.client.Del(ctx, keys...).Err(); err != nil {
log.Printf("[Storage] Warning: failed to delete cid index keys: %v", err)
} else {
cidIndexDeleted += int64(len(keys))
}
}
cidCursor = nextCursor
if cidCursor == 0 {
break
}
}
// Delete the index set
if err := c.client.Del(ctx, indexKey).Err(); err != nil {
log.Printf("[Storage] Warning: failed to delete index: %v", err)
}
log.Printf("[Storage] Cleared %d file metadata entries", deletedCount)
log.Printf("[Storage] Cleared %d file metadata entries (%d cid index entries)", deletedCount, cidIndexDeleted)
return deletedCount, nil
}
+284
View File
@@ -0,0 +1,284 @@
package storage
// User Data Layer (UDL) storage.
//
// Holds per-user, identity-signed records keyed by (uid, cid, key) — the
// subjective/per-user state that METADATA_KEYS.md rule #4 bars from the shared
// /file/{cid} hash (likes, ratings, watch progress, local paths). Each record
// is an opaque, client-signed (and, for private-tier keys, client-encrypted)
// CBOR blob; meta-core stores it verbatim and never needs the account key.
//
// Convergence is version-gated: a write is accepted only if its `version`
// strictly exceeds the stored one (a per-cell high-water-mark, replacing the
// op-log merge that used to live in meta-watch's store.rs).
//
// Redis model (all under the literal "udl:" prefix, mirroring the "file:" one):
//
// udl:rec:<uid>/<cid>/<key> HASH { record, version, ts, value? }
// udl:idx:user:<uid>:key:<key> SET of <cid> -> active_for_user_key (My List, Continue Watching)
// udl:idx:user:<uid>:cid:<cid> SET of <key> -> list_for_user_cid
// udl:idx:cid:<cid>:key:<key> SET of <uid> -> aggregate
// udl:idx:cid:<cid>:users SET of <uid> -> uids_for_cid
//
// The `value` hash field is the plaintext JSON scalar for PUBLIC keys only
// (like/rating), used for aggregation; it is absent for private-tier keys
// (whose real value rides encrypted inside `record`) and for tombstones.
import (
"context"
"fmt"
"strconv"
"time"
"github.com/redis/go-redis/v9"
)
// UDLCell is a decoded (uid, cid, key) record cell.
type UDLCell struct {
Record string // base64 CBOR, opaque + signed
Version int64
Ts int64
Value string // plaintext JSON scalar, public keys only
HasValue bool
}
// UDLUserKeyEntry is one row of active_for_user_key (all cids for a uid+key).
type UDLUserKeyEntry struct {
Cid string `json:"cid"`
Record string `json:"record"`
Version int64 `json:"version"`
Ts int64 `json:"ts"`
}
// UDLUserCidEntry is one row of list_for_user_cid (all keys for a uid+cid).
type UDLUserCidEntry struct {
Key string `json:"key"`
Record string `json:"record"`
Version int64 `json:"version"`
Ts int64 `json:"ts"`
}
func udlRecKey(uid, cid, key string) string {
return fmt.Sprintf("udl:rec:%s/%s/%s", uid, cid, key)
}
func udlIdxUserKey(uid, key string) string { return fmt.Sprintf("udl:idx:user:%s:key:%s", uid, key) }
func udlIdxUserCid(uid, cid string) string { return fmt.Sprintf("udl:idx:user:%s:cid:%s", uid, cid) }
func udlIdxCidKey(cid, key string) string { return fmt.Sprintf("udl:idx:cid:%s:key:%s", cid, key) }
func udlIdxCidUsers(cid string) string { return fmt.Sprintf("udl:idx:cid:%s:users", cid) }
// udlUpsertScript performs the version-gated upsert + index maintenance
// atomically. Accepts (returns 1) only when the incoming version strictly
// exceeds the stored one; otherwise returns 0 (stale).
//
// KEYS: 1=cell 2=idx:user:key 3=idx:user:cid 4=idx:cid:key 5=idx:cid:users
// ARGV: 1=version 2=ts 3=record 4=value 5=hasValue("1"/"0")
// 6=cid 7=key 8=uid
var udlUpsertScript = redis.NewScript(`
local stored = redis.call('HGET', KEYS[1], 'version')
if stored and tonumber(stored) >= tonumber(ARGV[1]) then
return 0
end
redis.call('HSET', KEYS[1], 'version', ARGV[1], 'ts', ARGV[2], 'record', ARGV[3])
if ARGV[5] == '1' then
redis.call('HSET', KEYS[1], 'value', ARGV[4])
else
redis.call('HDEL', KEYS[1], 'value')
end
redis.call('SADD', KEYS[2], ARGV[6])
redis.call('SADD', KEYS[3], ARGV[7])
redis.call('SADD', KEYS[4], ARGV[8])
redis.call('SADD', KEYS[5], ARGV[8])
return 1
`)
// UDLUpsertIfNewer stores a record cell iff version strictly exceeds the stored
// one. publicValue/hasValue carry the plaintext JSON scalar for public keys
// (used by aggregation); pass hasValue=false for private-tier keys and
// tombstones. Returns accepted=false when the write is stale.
func (c *Client) UDLUpsertIfNewer(uid, cid, key string, version, ts int64, record, publicValue string, hasValue bool) (bool, error) {
c.mu.Lock()
defer c.mu.Unlock()
if c.client == nil {
return false, fmt.Errorf("not connected")
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
keys := []string{
c.buildKey(udlRecKey(uid, cid, key)),
c.buildKey(udlIdxUserKey(uid, key)),
c.buildKey(udlIdxUserCid(uid, cid)),
c.buildKey(udlIdxCidKey(cid, key)),
c.buildKey(udlIdxCidUsers(cid)),
}
hv := "0"
if hasValue {
hv = "1"
}
res, err := udlUpsertScript.Run(ctx, c.client, keys, version, ts, record, publicValue, hv, cid, key, uid).Result()
if err != nil {
return false, fmt.Errorf("udl upsert failed: %w", err)
}
n, _ := res.(int64)
return n == 1, nil
}
// UDLGetCell returns the record cell for (uid, cid, key). exists is false when
// no record has ever been written for the cell.
func (c *Client) UDLGetCell(uid, cid, key string) (UDLCell, bool, error) {
c.mu.RLock()
defer c.mu.RUnlock()
if c.client == nil {
return UDLCell{}, false, fmt.Errorf("not connected")
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
return c.udlReadCell(ctx, udlRecKey(uid, cid, key))
}
// udlReadCell HMGETs a cell hash. Caller holds the lock. exists is false when
// the version field is absent (i.e. the hash does not exist).
func (c *Client) udlReadCell(ctx context.Context, recKey string) (UDLCell, bool, error) {
vals, err := c.client.HMGet(ctx, c.buildKey(recKey), "record", "version", "ts", "value").Result()
if err != nil {
return UDLCell{}, false, fmt.Errorf("hmget failed: %w", err)
}
// vals: [record, version, ts, value] — nil for absent fields.
if len(vals) < 4 || vals[1] == nil {
return UDLCell{}, false, nil
}
cell := UDLCell{}
if s, ok := vals[0].(string); ok {
cell.Record = s
}
if s, ok := vals[1].(string); ok {
cell.Version, _ = strconv.ParseInt(s, 10, 64)
}
if s, ok := vals[2].(string); ok {
cell.Ts, _ = strconv.ParseInt(s, 10, 64)
}
if s, ok := vals[3].(string); ok {
cell.Value = s
cell.HasValue = true
}
return cell, true, nil
}
// UDLListUserKey returns every cell for (uid, key) across all cids — the
// projection behind active_for_user_key (My List, Continue Watching). Includes
// tombstones; the caller decodes each record and filters.
func (c *Client) UDLListUserKey(uid, key string) ([]UDLUserKeyEntry, error) {
c.mu.RLock()
defer c.mu.RUnlock()
if c.client == nil {
return nil, fmt.Errorf("not connected")
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
cids, err := c.client.SMembers(ctx, c.buildKey(udlIdxUserKey(uid, key))).Result()
if err != nil {
return nil, fmt.Errorf("smembers failed: %w", err)
}
out := make([]UDLUserKeyEntry, 0, len(cids))
for _, cid := range cids {
cell, ok, err := c.udlReadCell(ctx, udlRecKey(uid, cid, key))
if err != nil {
return nil, err
}
if !ok {
continue
}
out = append(out, UDLUserKeyEntry{Cid: cid, Record: cell.Record, Version: cell.Version, Ts: cell.Ts})
}
return out, nil
}
// UDLListUserCid returns every cell for (uid, cid) across all keys — the
// projection behind list_for_user_cid.
func (c *Client) UDLListUserCid(uid, cid string) ([]UDLUserCidEntry, error) {
c.mu.RLock()
defer c.mu.RUnlock()
if c.client == nil {
return nil, fmt.Errorf("not connected")
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
keys, err := c.client.SMembers(ctx, c.buildKey(udlIdxUserCid(uid, cid))).Result()
if err != nil {
return nil, fmt.Errorf("smembers failed: %w", err)
}
out := make([]UDLUserCidEntry, 0, len(keys))
for _, key := range keys {
cell, ok, err := c.udlReadCell(ctx, udlRecKey(uid, cid, key))
if err != nil {
return nil, err
}
if !ok {
continue
}
out = append(out, UDLUserCidEntry{Key: key, Record: cell.Record, Version: cell.Version, Ts: cell.Ts})
}
return out, nil
}
// UDLCidUsers returns every uid that has any record for cid — uids_for_cid.
func (c *Client) UDLCidUsers(cid string) ([]string, error) {
c.mu.RLock()
defer c.mu.RUnlock()
if c.client == nil {
return nil, fmt.Errorf("not connected")
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
uids, err := c.client.SMembers(ctx, c.buildKey(udlIdxCidUsers(cid))).Result()
if err != nil {
return nil, fmt.Errorf("smembers failed: %w", err)
}
return uids, nil
}
// UDLPublicValues returns the live plaintext public JSON scalar values for
// (cid, key) across all uids — the input to aggregation. Cells without a value
// field (private-tier or tombstoned) are skipped.
func (c *Client) UDLPublicValues(cid, key string) ([]string, error) {
c.mu.RLock()
defer c.mu.RUnlock()
if c.client == nil {
return nil, fmt.Errorf("not connected")
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
uids, err := c.client.SMembers(ctx, c.buildKey(udlIdxCidKey(cid, key))).Result()
if err != nil {
return nil, fmt.Errorf("smembers failed: %w", err)
}
out := make([]string, 0, len(uids))
for _, uid := range uids {
v, err := c.client.HGet(ctx, c.buildKey(udlRecKey(uid, cid, key)), "value").Result()
if err == redis.Nil {
continue // private-tier or tombstoned
}
if err != nil {
return nil, fmt.Errorf("hget failed: %w", err)
}
out = append(out, v)
}
return out, nil
}