Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
55d08e77ed | ||
|
|
1f084b817c | ||
|
|
80050cd2a8 | ||
|
|
75dda1e7d6 | ||
|
|
76169920d6 | ||
|
|
df6a2d78f8 | ||
|
|
0ca03ac95a | ||
|
|
1798821c53 | ||
|
|
d268bb83e6 | ||
|
|
075d6f9bcb |
@@ -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
|
||||
@@ -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. |
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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. |
|
||||
|
||||
@@ -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'
|
||||
|
||||
@@ -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,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
|
||||
)
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -50,6 +50,26 @@ type identityImportResponse struct {
|
||||
UID string `json:"uid"`
|
||||
Curve string `json:"curve"`
|
||||
CreatedAt int64 `json:"createdAt"`
|
||||
// Created distinguishes "this key was new here" (201) from "this key was
|
||||
// already in the keystore" (200). Import is the sign-in primitive for
|
||||
// clients: presenting the secret key is the proof of ownership, so a
|
||||
// re-import must succeed rather than conflict. Clients use the flag only
|
||||
// to word the confirmation ("new profile" vs "welcome back").
|
||||
Created bool `json:"created"`
|
||||
}
|
||||
|
||||
// identityRevealRequest asks for an existing account's private key back. The
|
||||
// key was shown once at generate; without a way to see it again, a client with
|
||||
// a real sign-out has no recovery path for a user who never saved it.
|
||||
type identityRevealRequest struct {
|
||||
Confirm bool `json:"confirm"`
|
||||
UID string `json:"uid"`
|
||||
}
|
||||
|
||||
type identityRevealResponse struct {
|
||||
UID string `json:"uid"`
|
||||
Curve string `json:"curve"`
|
||||
PrivateKeyHex string `json:"privateKeyHex"`
|
||||
}
|
||||
|
||||
type identityDeleteRequest struct {
|
||||
@@ -172,9 +192,17 @@ func (s *Server) handleIdentityImport(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
dir := s.config.IdentityAccountsDir()
|
||||
if identity.ExistsByUID(dir, id.UID) {
|
||||
writeErrorSlug(w, http.StatusConflict, "identity_exists",
|
||||
"an account with this key already exists", false)
|
||||
// Idempotent: importing a key that is already in the keystore is how a
|
||||
// second device signs in to an existing account. Answer 200 with the
|
||||
// stored account (preserving its original createdAt) instead of 409 —
|
||||
// the caller proved ownership by presenting the key.
|
||||
if existing, err := identity.LoadByUID(dir, id.UID); err == nil && existing != nil {
|
||||
writeJSON(w, http.StatusOK, identityImportResponse{
|
||||
UID: existing.UID,
|
||||
Curve: existing.Curve,
|
||||
CreatedAt: existing.CreatedAt,
|
||||
Created: false,
|
||||
})
|
||||
return
|
||||
}
|
||||
if err := identity.SaveAccount(dir, id); err != nil {
|
||||
@@ -185,6 +213,35 @@ func (s *Server) handleIdentityImport(w http.ResponseWriter, r *http.Request) {
|
||||
UID: id.UID,
|
||||
Curve: id.Curve,
|
||||
CreatedAt: id.CreatedAt,
|
||||
Created: true,
|
||||
})
|
||||
}
|
||||
|
||||
// handleIdentityReveal returns an existing account's private key. Deliberately
|
||||
// POST + explicit confirm (never a bare GET) so it cannot be triggered by a
|
||||
// link or an image tag, and so it reads as the deliberate act it is. The
|
||||
// endpoint is auth-gated at the perimeter like the rest of /api/identity/*.
|
||||
func (s *Server) handleIdentityReveal(w http.ResponseWriter, r *http.Request) {
|
||||
var body identityRevealRequest
|
||||
_ = json.NewDecoder(r.Body).Decode(&body)
|
||||
if !body.Confirm && r.URL.Query().Get("confirm") != "true" {
|
||||
writeErrorSlug(w, http.StatusBadRequest, "confirm_required",
|
||||
"revealing a private key requires {\"confirm\": true} or ?confirm=true", false)
|
||||
return
|
||||
}
|
||||
uid := strings.TrimSpace(body.UID)
|
||||
if uid == "" {
|
||||
uid = strings.TrimSpace(r.URL.Query().Get("uid"))
|
||||
}
|
||||
id, status, slug, msg := s.resolveAccount(uid)
|
||||
if status != 0 {
|
||||
writeErrorSlug(w, status, slug, msg, false)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, identityRevealResponse{
|
||||
UID: id.UID,
|
||||
Curve: id.Curve,
|
||||
PrivateKeyHex: id.PrivateKeyHex,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
+32
-26
@@ -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())
|
||||
@@ -395,7 +415,18 @@ func (s *Server) handleSearchMetadata(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
// Get all hash IDs
|
||||
// Free-text query: served from the in-memory search index (search_index.go)
|
||||
// so we match in memory instead of doing a SCAN+MGET Redis round-trip per
|
||||
// record (that was 40s+ over ~72k records). Full metadata is fetched only
|
||||
// for the <=limit matches.
|
||||
if req.Query != "" && req.Property == "" {
|
||||
s.serveIndexedSearch(w, req)
|
||||
return
|
||||
}
|
||||
|
||||
// Property filter (rare) or empty query (returns the first `limit`): the
|
||||
// legacy scan. Empty query only fetches `limit` records; property search is
|
||||
// low-volume and not on the hot search path.
|
||||
allHashIDs, err := s.storage.GetAllHashIDs()
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, err.Error())
|
||||
@@ -405,9 +436,6 @@ func (s *Server) handleSearchMetadata(w http.ResponseWriter, r *http.Request) {
|
||||
var results []MetadataSearchResult
|
||||
total := len(allHashIDs)
|
||||
|
||||
// Search fields for query
|
||||
searchFields := []string{"title", "fileName", "originaltitle", "showtitle", "filePath"}
|
||||
|
||||
for _, hashID := range allHashIDs {
|
||||
if len(results) >= req.Limit {
|
||||
break
|
||||
@@ -428,28 +456,6 @@ func (s *Server) handleSearchMetadata(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
// General query search
|
||||
if req.Query != "" && req.Property == "" {
|
||||
queryLower := strings.ToLower(req.Query)
|
||||
|
||||
// Check if query matches hashId
|
||||
if strings.Contains(strings.ToLower(hashID), queryLower) {
|
||||
matches = true
|
||||
}
|
||||
|
||||
// Check search fields
|
||||
if !matches {
|
||||
for _, field := range searchFields {
|
||||
if value, ok := metadata[field]; ok {
|
||||
if strings.Contains(strings.ToLower(value), queryLower) {
|
||||
matches = true
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// If no query or property filter, return all
|
||||
if req.Query == "" && req.Property == "" {
|
||||
matches = true
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"log"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/metazla/meta-core/internal/storage"
|
||||
)
|
||||
|
||||
// searchIndexRefreshInterval is how often the in-memory search index is rebuilt
|
||||
// from Redis in the background. Records ingest continuously, so a newly stored
|
||||
// record becomes searchable within one interval; the rebuild is a single
|
||||
// batched bulk read (storage.GetSearchIndexTuples), off the request path.
|
||||
const searchIndexRefreshInterval = 60 * time.Second
|
||||
|
||||
// searchIndex is an in-memory snapshot of every record's search haystack
|
||||
// (hashID + lowercased searchable fields), rebuilt periodically from Redis. It
|
||||
// exists so POST /api/metadata/search matches in memory (string ops over a
|
||||
// slice) instead of doing a SCAN+MGET Redis round-trip per record — the latter
|
||||
// took 40s+ over ~72k records and made a live parallel search path unusable.
|
||||
type searchIndex struct {
|
||||
mu sync.RWMutex
|
||||
tuples []storage.SearchTuple
|
||||
ready atomic.Bool
|
||||
}
|
||||
|
||||
// snapshot returns the current tuples slice and whether the index has built at
|
||||
// least once. The refresh swaps in a fresh slice under the write lock (never
|
||||
// mutates in place), so the returned slice is safe to read lock-free.
|
||||
func (idx *searchIndex) snapshot() ([]storage.SearchTuple, bool) {
|
||||
idx.mu.RLock()
|
||||
defer idx.mu.RUnlock()
|
||||
return idx.tuples, idx.ready.Load()
|
||||
}
|
||||
|
||||
// refresh rebuilds the index from Redis once (a batched bulk read).
|
||||
func (idx *searchIndex) refresh(stor *storage.Client) {
|
||||
if stor == nil || !stor.IsConnected() {
|
||||
return
|
||||
}
|
||||
tuples, err := stor.GetSearchIndexTuples()
|
||||
if err != nil {
|
||||
log.Printf("[API] search index refresh failed: %v", err)
|
||||
return
|
||||
}
|
||||
idx.mu.Lock()
|
||||
idx.tuples = tuples
|
||||
idx.mu.Unlock()
|
||||
idx.ready.Store(true)
|
||||
}
|
||||
|
||||
// runRefreshLoop builds the index immediately, then rebuilds on a ticker for the
|
||||
// process lifetime.
|
||||
func (idx *searchIndex) runRefreshLoop(stor *storage.Client) {
|
||||
idx.refresh(stor)
|
||||
ticker := time.NewTicker(searchIndexRefreshInterval)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
idx.refresh(stor)
|
||||
}
|
||||
}
|
||||
|
||||
// serveIndexedSearch answers a free-text (non-property) query from the in-memory
|
||||
// index: token-AND match against each record's haystack (or a whole-query hashID
|
||||
// substring, preserving the legacy behaviour), then fetch full metadata for only
|
||||
// the <=limit matches. No Redis round-trip per scanned record.
|
||||
func (s *Server) serveIndexedSearch(w http.ResponseWriter, req MetadataSearchRequest) {
|
||||
tuples, ready := s.search.snapshot()
|
||||
if !ready {
|
||||
// First call before the background build finished: warm inline with the
|
||||
// efficient batched read (seconds), never the per-record SCAN fallback.
|
||||
s.search.refresh(s.storage)
|
||||
tuples, _ = s.search.snapshot()
|
||||
}
|
||||
tokens := strings.Fields(strings.ToLower(req.Query))
|
||||
queryLower := strings.ToLower(req.Query)
|
||||
results := make([]MetadataSearchResult, 0, req.Limit)
|
||||
for _, t := range tuples {
|
||||
if len(results) >= req.Limit {
|
||||
break
|
||||
}
|
||||
if !matchAllTokens(t.Haystack, tokens) &&
|
||||
!strings.Contains(strings.ToLower(t.HashID), queryLower) {
|
||||
continue
|
||||
}
|
||||
// Full metadata is cached in the index — return it straight from memory,
|
||||
// no Redis round-trip per result.
|
||||
results = append(results, MetadataSearchResult{HashID: t.HashID, Metadata: t.Fields})
|
||||
}
|
||||
writeJSON(w, http.StatusOK, MetadataSearchResponse{
|
||||
Results: results,
|
||||
Count: len(results),
|
||||
Total: len(tuples),
|
||||
})
|
||||
}
|
||||
|
||||
// matchAllTokens reports whether haystack contains every token (AND semantics).
|
||||
// An empty token list matches everything.
|
||||
func matchAllTokens(haystack string, tokens []string) bool {
|
||||
for _, t := range tokens {
|
||||
if !strings.Contains(haystack, t) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
+59
-10
@@ -41,6 +41,7 @@ type Server struct {
|
||||
metaPublisher *events.MetaPublisher
|
||||
schemaIndexer *schema.Indexer
|
||||
webdavHandler *webdav.Handler
|
||||
search *searchIndex
|
||||
router *mux.Router
|
||||
server *http.Server
|
||||
}
|
||||
@@ -59,6 +60,7 @@ func NewServer(
|
||||
discovery: disc,
|
||||
cleaner: cleaner,
|
||||
storage: stor,
|
||||
search: &searchIndex{},
|
||||
router: mux.NewRouter(),
|
||||
}
|
||||
|
||||
@@ -170,6 +172,7 @@ func (s *Server) setupRoutes() {
|
||||
s.router.HandleFunc("/api/identity/accounts", s.handleIdentityAccounts).Methods("GET")
|
||||
s.router.HandleFunc("/api/identity/generate", s.handleIdentityGenerate).Methods("POST")
|
||||
s.router.HandleFunc("/api/identity/import", s.handleIdentityImport).Methods("POST")
|
||||
s.router.HandleFunc("/api/identity/reveal", s.handleIdentityReveal).Methods("POST")
|
||||
s.router.HandleFunc("/api/identity", s.handleIdentityDelete).Methods("DELETE")
|
||||
s.router.HandleFunc("/api/identity/sign", s.handleIdentitySign).Methods("POST")
|
||||
s.router.HandleFunc("/api/identity/aead-key", s.handleIdentityAEADKey).Methods("GET")
|
||||
@@ -199,6 +202,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")
|
||||
@@ -250,29 +262,22 @@ func (s *Server) setupRoutes() {
|
||||
// Mount management routes (if manager initialized)
|
||||
if s.mountsHandlers != nil {
|
||||
s.mountsHandlers.RegisterRoutes(s.router)
|
||||
log.Println("[API] Mount management routes registered at /api/mounts/*")
|
||||
}
|
||||
|
||||
log.Println("[API] Metadata Editor routes registered at /api/metadata/*")
|
||||
log.Println("[API] KV Browser routes registered at /api/kv/*")
|
||||
|
||||
// File watcher routes (if watcher initialized)
|
||||
if s.watcherHandlers != nil {
|
||||
s.watcherHandlers.RegisterRoutes(s.router)
|
||||
log.Println("[API] File watcher routes registered at /api/events/*, /api/scan/*")
|
||||
}
|
||||
|
||||
// Watchers management routes (if watchers initialized)
|
||||
if s.watchersHandlers != nil {
|
||||
s.watchersHandlers.RegisterRoutes(s.router)
|
||||
log.Println("[API] Watchers management routes registered at /api/watchers/*")
|
||||
}
|
||||
|
||||
// WebDAV handler (caching is handled by nginx proxy_cache)
|
||||
if s.webdavHandler != nil {
|
||||
s.router.PathPrefix("/webdav/").Handler(s.webdavHandler)
|
||||
s.router.PathPrefix("/webdav").Handler(s.webdavHandler)
|
||||
log.Println("[API] WebDAV routes registered at /webdav/*")
|
||||
}
|
||||
|
||||
// Add middleware
|
||||
@@ -321,6 +326,13 @@ func (s *Server) Start() error {
|
||||
s.mountStatsPoller.Start()
|
||||
}
|
||||
|
||||
// Build + periodically refresh the in-memory search index so
|
||||
// POST /api/metadata/search matches in memory instead of a Redis
|
||||
// round-trip per record (see search_index.go).
|
||||
if s.storage != nil {
|
||||
go s.search.runRefreshLoop(s.storage)
|
||||
}
|
||||
|
||||
// Start meta publisher (if storage is connected)
|
||||
// Publishes keyspace notifications to meta:events stream
|
||||
if s.storage != nil && s.storage.IsConnected() {
|
||||
@@ -421,12 +433,49 @@ func (s *Server) RepublishMetadata() (int, error) {
|
||||
return s.metaPublisher.RepublishAllMetadata(republishFunc)
|
||||
}
|
||||
|
||||
// loggingMiddleware logs all requests
|
||||
// statusRecorder captures the response status so loggingMiddleware can stay
|
||||
// silent on success and only report failures. It forwards Flush/Unwrap so SSE
|
||||
// and other streaming handlers keep working through the wrapper.
|
||||
type statusRecorder struct {
|
||||
http.ResponseWriter
|
||||
status int
|
||||
written bool
|
||||
}
|
||||
|
||||
func (r *statusRecorder) WriteHeader(code int) {
|
||||
if !r.written {
|
||||
r.status = code
|
||||
r.written = true
|
||||
}
|
||||
r.ResponseWriter.WriteHeader(code)
|
||||
}
|
||||
|
||||
func (r *statusRecorder) Write(b []byte) (int, error) {
|
||||
if !r.written {
|
||||
r.status = http.StatusOK
|
||||
r.written = true
|
||||
}
|
||||
return r.ResponseWriter.Write(b)
|
||||
}
|
||||
|
||||
func (r *statusRecorder) Flush() {
|
||||
if f, ok := r.ResponseWriter.(http.Flusher); ok {
|
||||
f.Flush()
|
||||
}
|
||||
}
|
||||
|
||||
func (r *statusRecorder) Unwrap() http.ResponseWriter { return r.ResponseWriter }
|
||||
|
||||
// loggingMiddleware logs a request only when it fails (status >= 400), keeping
|
||||
// the nominal case quiet. Successful requests are not logged.
|
||||
func loggingMiddleware(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
start := time.Now()
|
||||
next.ServeHTTP(w, r)
|
||||
log.Printf("[API] %s %s %s", r.Method, r.URL.Path, time.Since(start))
|
||||
rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
|
||||
next.ServeHTTP(rec, r)
|
||||
if rec.status >= 400 {
|
||||
log.Printf("[API] %s %s -> %d (%s)", r.Method, r.URL.Path, rec.status, time.Since(start))
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,247 @@
|
||||
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"`
|
||||
// Tombstone marks this write as a deletion of (uid, cid, key). The record
|
||||
// itself is opaque here — meta-core never decodes the CBOR, and a private-
|
||||
// tier value and a tombstone are equally "no plaintext value" — so the
|
||||
// writer has to say. Without it the cid is never removed from
|
||||
// udl:idx:user:<uid>:key:<key>, which then grows for the life of the
|
||||
// account and is returned in full on every My List / Continue Watching
|
||||
// read. Omitted by older writers, which keeps the previous behaviour.
|
||||
Tombstone bool `json:"tombstone,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, req.Tombstone)
|
||||
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
|
||||
}
|
||||
@@ -0,0 +1,166 @@
|
||||
{
|
||||
"$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",
|
||||
"25": "nzb-posting - content-derived (Message-ID set digest), but not itself fetchable",
|
||||
"20": "BitTorrent infohashes - swarm interop",
|
||||
"10": "midhash256 - the record address",
|
||||
"5": "opaque locators (nzb-release, url, card) - 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": "nzb-posting",
|
||||
"cid": "bagbsaeraaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
|
||||
"rank": 25,
|
||||
"note": "codec 0x1003 + REAL sha2-256 multihash (0x12) over the normalised, sorted Message-ID set - NOT a locator, so it sits ABOVE tier 5, but it addresses no fetchable block (bitswap can never satisfy a want for it - the digest is over an id set, not over any block's bytes), so it stays BELOW a real content digest (30): once meta-share materialises the media and re-seeds it, that genuine sha2-256 cid must win the canonical election, not this one. Rank MUST match on codec BEFORE falling through to the mh switch, or this scores 30 (same trap as nzb-release/url, in reverse: those need a codec check because their mh is identity 0x00; this needs one because its mh is REAL sha2-256 and would otherwise tie with tier 30)."
|
||||
},
|
||||
{
|
||||
"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": "card-locator",
|
||||
"cid": "bagdsaaanar2g2zdcor3duojvgq3ts",
|
||||
"rank": 5,
|
||||
"note": "codec 0x1007 + IDENTITY multihash 0x00, digest = varint(len(source)) || source || id — here (\"tmdb\",\"tv:95479\"). Same codec-slot rule as nzb-release/url: match on the CODEC, never the mh. The limit case of the family — the only CID in MetaMesh with NO bytes behind it anywhere, ever: it addresses a *work* (a series, a film), not a file, so the byte path 404s by design and bitswap never offers it (seeding is gated on sha2-256). Tier 5 is load-bearing for the planned cross-source merge: one card record may carry BOTH a TMDB and a MyAnimeList locator, and if a real card body is ever added (see card-tier-search.md §4.3) its sha2-256 must win the canonical election over both locators without any addressing change. Ranking it 0 instead would still lose to a digest, but would put the implementations back out of agreement — which is the exact failure this fixture exists to prevent."
|
||||
},
|
||||
{
|
||||
"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
|
||||
}
|
||||
]
|
||||
}
|
||||
+75
-7
@@ -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,44 @@ 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
|
||||
// CodeCard is the limit case of the family: the only CID with NO bytes
|
||||
// behind it anywhere, ever. It addresses a *work* (a series, a film) via
|
||||
// varint(len(source))‖source‖id — ("tmdb","tv:95479") — so `/api/file/{cid}`
|
||||
// correctly 404s and bitswap never offers it. Ranking it here (rather than
|
||||
// letting it fall to RankUnknown) is what makes the planned "one card, two
|
||||
// CIDs" cross-source merge safe: a TMDB and a MyAnimeList locator can sit on
|
||||
// one record, and a real digest added later still wins the election.
|
||||
CodeCard = 0x1007 // custom — identity-multihash CID wrapping a card (work) id
|
||||
|
||||
// CodeNzbPosting is the INVERSE trap of the two above: its multihash is a
|
||||
// REAL sha2-256 digest (over the normalised, sorted article Message-ID
|
||||
// set of a Usenet posting we scanned ourselves — never a locator, no
|
||||
// indexer host embedded), so it would silently fall into RankDigest via
|
||||
// the mh switch unless matched on the codec FIRST. It is content-derived
|
||||
// (so it outranks the opaque locators) but addresses no fetchable block —
|
||||
// the digest is over an id set, not over any block's bytes, so it can
|
||||
// never be bitswap-fetched under itself — so it must stay BELOW a real
|
||||
// digest (once the media is materialised and re-seeded, that genuine
|
||||
// sha2-256 cid has to win the canonical election, not this one).
|
||||
CodeNzbPosting = 0x1003 // custom — self-scanned Usenet posting identity
|
||||
)
|
||||
|
||||
// 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
|
||||
RankNzbPosting = 25 // content-derived, but the digest addresses no fetchable block
|
||||
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
|
||||
@@ -87,33 +133,55 @@ func Decode(cidStr string) (codec uint64, mhCode uint64, err error) {
|
||||
// - dag-pb (chunked IPFS file) wins: best retrieval over the IPFS network.
|
||||
// - sha2-256 / sha3 digests follow: standard content hashes (a raw IPFS
|
||||
// block of the file IS its sha2-256, so they share this tier).
|
||||
// - nzb-posting is content-derived (a digest over a Usenet posting's article
|
||||
// Message-ID set) so it outranks the opaque locators below, but the
|
||||
// digest addresses no fetchable block, so it stays below a real content
|
||||
// digest — once the media is materialised and re-seeded under a genuine
|
||||
// sha2-256 cid, that cid must win the election, not this one.
|
||||
// - 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, card) 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 || codec == CodeCard {
|
||||
return RankLocator
|
||||
}
|
||||
// nzb-posting: codec-only, checked BEFORE the mh switch below — its
|
||||
// multihash is a real sha2-256 and would otherwise match CodeSha256 there
|
||||
// and rank 30, tying with genuine content digests.
|
||||
if codec == CodeNzbPosting {
|
||||
return RankNzbPosting
|
||||
}
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
@@ -104,12 +150,31 @@ func TestRank_Tiers(t *testing.T) {
|
||||
func TestRank_StrictOrdering(t *testing.T) {
|
||||
ipfs := Rank(buildCID(CodecDagPB, CodeSha256, 32))
|
||||
sha := Rank(cidSha256)
|
||||
posting := Rank(buildCID(CodeNzbPosting, CodeSha256, 32))
|
||||
btih := Rank(buildCID(CodeBtihV2, CodeBtihV2, 32))
|
||||
mid := Rank(buildCID(CodeMidhash256, CodeMidhash256, 32))
|
||||
unknown := Rank(cidMd5)
|
||||
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)
|
||||
if !(ipfs > sha && sha > posting && posting > btih && btih > mid && mid > unknown) {
|
||||
t.Fatalf("rank ordering broken: ipfs=%d sha=%d posting=%d btih=%d mid=%d unknown=%d",
|
||||
ipfs, sha, posting, btih, mid, unknown)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRank_NzbPostingMatchesCodecNotMultihash pins the inverse trap:
|
||||
// nzb-posting's multihash is a REAL sha2-256 (unlike nzb-release/url, whose mh
|
||||
// is identity 0x00), so Rank must match on the codec BEFORE falling into the
|
||||
// mh switch — otherwise this ties with a genuine content digest at 30 instead
|
||||
// of sitting at 25.
|
||||
func TestRank_NzbPostingMatchesCodecNotMultihash(t *testing.T) {
|
||||
posting := buildCID(CodeNzbPosting, CodeSha256, 32)
|
||||
if got := Rank(posting); got != RankNzbPosting {
|
||||
t.Errorf("Rank(nzb-posting) = %d, want %d (RankNzbPosting) — codec check must precede the mh switch", got, RankNzbPosting)
|
||||
}
|
||||
// A real sha2-256 with the SAME digest bytes must still rank 30 — the
|
||||
// codec, not the digest, is what distinguishes the two.
|
||||
realDigest := buildCID(CodecRaw, CodeSha256, 32)
|
||||
if got := Rank(realDigest); got != RankDigest {
|
||||
t.Errorf("Rank(sha2-256) regressed to %d, want %d", got, RankDigest)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -44,6 +44,8 @@ type PollingStatus struct {
|
||||
CurrentScanStartedAt int64 `json:"currentScanStartedAt,omitempty"` // ms; start of in-flight scan
|
||||
LastScanDurationMs int64 `json:"lastScanDurationMs,omitempty"` // duration of the previous scan
|
||||
NextScanAt int64 `json:"nextScanAt,omitempty"` // ms; planned start of next adaptive scan
|
||||
LastFileCount int `json:"-"` // last observed file count; used to suppress no-change scan logs
|
||||
scanned bool // true once at least one scan has completed for this mount
|
||||
}
|
||||
|
||||
// Poller manages per-mount polling goroutines
|
||||
@@ -309,16 +311,22 @@ func (p *Poller) executeScan(mountID, mountPath string) (time.Duration, error) {
|
||||
end := time.Now()
|
||||
duration := end.Sub(start)
|
||||
|
||||
// Only log a completed scan when the file count changed (or on the very
|
||||
// first scan) so a steady mount doesn't emit one line per poll tick.
|
||||
changed := true
|
||||
p.mu.Lock()
|
||||
if status, exists := p.status[mountID]; exists {
|
||||
status.Scanning = false
|
||||
status.CurrentScanStartedAt = 0
|
||||
status.LastScan = end.UnixMilli()
|
||||
status.LastScanDurationMs = duration.Milliseconds()
|
||||
changed = !status.scanned || status.LastFileCount != fileCount
|
||||
status.LastFileCount = fileCount
|
||||
status.scanned = true
|
||||
}
|
||||
p.mu.Unlock()
|
||||
|
||||
if err == nil {
|
||||
if err == nil && changed {
|
||||
log.Printf("[Poller] Poll scan complete for mount %s: %d files in %s", mountID, fileCount, duration.Round(time.Millisecond))
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
+154
-7
@@ -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
|
||||
@@ -683,6 +685,115 @@ func (c *Client) GetTuplesForAllFiles() ([]FileTuple, error) {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// SearchTuple is one record's search state: its hashID, a lowercased haystack of
|
||||
// its searchable text fields (for token/substring match), and the full flat
|
||||
// metadata map (so a search hit is returned straight from memory — no per-result
|
||||
// Redis fetch).
|
||||
type SearchTuple struct {
|
||||
HashID string
|
||||
Haystack string
|
||||
Fields map[string]string
|
||||
}
|
||||
|
||||
// searchIndexFields are the metadata fields folded into a record's search
|
||||
// haystack — must stay in sync with the fields the search handler documents.
|
||||
var searchIndexFields = []string{"title", "fileName", "originaltitle", "showtitle", "filePath"}
|
||||
|
||||
// GetSearchIndexTuples reads EVERY record's full metadata in a single keyspace
|
||||
// SCAN pass over the flat `file:{hashID}/{field}` keys plus batched MGETs,
|
||||
// grouped by record — one bulk pass instead of a SCAN+MGET per record (which is
|
||||
// O(records) Redis round-trips, ~40s over 72k field keys). It is the bulk source
|
||||
// the API folds into its in-memory search index, so per-request search does zero
|
||||
// Redis round-trips (match AND result metadata both come from memory).
|
||||
func (c *Client) GetSearchIndexTuples() ([]SearchTuple, error) {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
|
||||
if c.client == nil {
|
||||
return nil, fmt.Errorf("not connected")
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
|
||||
defer cancel()
|
||||
|
||||
filePrefix := c.buildKey("file:") // key layout: {filePrefix}{hashID}/{field}
|
||||
indexKey := c.buildIndexKey() // {filePrefix}__index__ (a SET, skip it)
|
||||
pattern := filePrefix + "*"
|
||||
|
||||
records := make(map[string]map[string]string)
|
||||
keyBatch := make([]string, 0, 1000)
|
||||
|
||||
flush := func() error {
|
||||
if len(keyBatch) == 0 {
|
||||
return nil
|
||||
}
|
||||
values, err := c.client.MGet(ctx, keyBatch...).Result()
|
||||
if err != nil {
|
||||
return fmt.Errorf("mget: %w", err)
|
||||
}
|
||||
for i, k := range keyBatch {
|
||||
v, ok := values[i].(string)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
// {filePrefix}{hashID}/{field}; hashID has no '/', field may.
|
||||
rest := strings.TrimPrefix(k, filePrefix)
|
||||
slash := strings.IndexByte(rest, '/')
|
||||
if slash < 0 {
|
||||
continue
|
||||
}
|
||||
hashID, field := rest[:slash], rest[slash+1:]
|
||||
m := records[hashID]
|
||||
if m == nil {
|
||||
m = make(map[string]string)
|
||||
records[hashID] = m
|
||||
}
|
||||
m[field] = v
|
||||
}
|
||||
keyBatch = keyBatch[:0]
|
||||
return nil
|
||||
}
|
||||
|
||||
var cursor uint64
|
||||
for {
|
||||
keys, next, err := c.client.Scan(ctx, cursor, pattern, 1000).Result()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("scan: %w", err)
|
||||
}
|
||||
for _, k := range keys {
|
||||
if k == indexKey {
|
||||
continue
|
||||
}
|
||||
keyBatch = append(keyBatch, k)
|
||||
if len(keyBatch) >= 1000 {
|
||||
if err := flush(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
cursor = next
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
if err := flush(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
out := make([]SearchTuple, 0, len(records))
|
||||
for hashID, fields := range records {
|
||||
var sb strings.Builder
|
||||
for _, f := range searchIndexFields {
|
||||
if v, ok := fields[f]; ok && v != "" {
|
||||
sb.WriteString(strings.ToLower(v))
|
||||
sb.WriteByte('\n')
|
||||
}
|
||||
}
|
||||
out = append(out, SearchTuple{HashID: hashID, Haystack: sb.String(), Fields: fields})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// GetMemoryInfo returns Redis memory usage information
|
||||
func (c *Client) GetMemoryInfo() (string, error) {
|
||||
c.mu.RLock()
|
||||
@@ -710,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()
|
||||
@@ -759,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
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,302 @@
|
||||
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 9=tombstone("1"/"0")
|
||||
//
|
||||
// A tombstone still writes the cell — that is the CRDT state, and dropping it
|
||||
// would let an older write resurrect the value — but it SREMs the cid from the
|
||||
// user+key index. That index is what `active_for_user_key` returns, so leaving
|
||||
// deleted cids in it means every My List / Continue Watching read carries every
|
||||
// title ever un-liked or finished, forever. The other three indexes are left
|
||||
// alone: they answer "who has an opinion about this cid", where a tombstone is
|
||||
// still an answer.
|
||||
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
|
||||
if ARGV[9] == '1' then
|
||||
redis.call('SREM', KEYS[2], ARGV[6])
|
||||
else
|
||||
redis.call('SADD', KEYS[2], ARGV[6])
|
||||
end
|
||||
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. Set tombstone=true when the record deletes the cell, which also
|
||||
// prunes the cid from the user+key index. Returns accepted=false when the write
|
||||
// is stale.
|
||||
func (c *Client) UDLUpsertIfNewer(uid, cid, key string, version, ts int64, record, publicValue string, hasValue, tombstone 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"
|
||||
}
|
||||
tomb := "0"
|
||||
if tombstone {
|
||||
tomb = "1"
|
||||
}
|
||||
res, err := udlUpsertScript.Run(ctx, c.client, keys, version, ts, record, publicValue, hv, cid, key, uid, tomb).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
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
// A tombstone must leave the cell in place but drop the cid from the user+key
|
||||
// index.
|
||||
//
|
||||
// The cell has to stay: it *is* the CRDT state, and deleting it would let an
|
||||
// older write resurrect the value the tombstone retracted. The index entry has
|
||||
// to go: `active_for_user_key` returns every member of that set, so a set that
|
||||
// only ever grows means each My List / Continue Watching read carries every
|
||||
// title the user has ever un-liked or finished — with its full record blob —
|
||||
// forever. The reader filters them out again, so the cost is invisible until
|
||||
// the account is old.
|
||||
func TestUDLUpsert_TombstonePrunesUserKeyIndexButKeepsCell(t *testing.T) {
|
||||
c, _ := newTestClient(t, "")
|
||||
|
||||
const (
|
||||
uid = "zAlice"
|
||||
cid = "bagcsaaalocator"
|
||||
key = "seek"
|
||||
)
|
||||
|
||||
// A live position: indexed, readable.
|
||||
if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 1, 100, "rec-v1", "", false, false); err != nil || !ok {
|
||||
t.Fatalf("live upsert: ok=%v err=%v", ok, err)
|
||||
}
|
||||
entries, err := c.UDLListUserKey(uid, key)
|
||||
if err != nil {
|
||||
t.Fatalf("active_for_user_key: %v", err)
|
||||
}
|
||||
if len(entries) != 1 || entries[0].Cid != cid {
|
||||
t.Fatalf("live position should be indexed, got %+v", entries)
|
||||
}
|
||||
|
||||
// Finishing it tombstones the seek.
|
||||
if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 2, 200, "rec-v2-tombstone", "", false, true); err != nil || !ok {
|
||||
t.Fatalf("tombstone upsert: ok=%v err=%v", ok, err)
|
||||
}
|
||||
|
||||
entries, err = c.UDLListUserKey(uid, key)
|
||||
if err != nil {
|
||||
t.Fatalf("active_for_user_key after tombstone: %v", err)
|
||||
}
|
||||
if len(entries) != 0 {
|
||||
t.Fatalf("tombstoned cid must leave the user+key index, got %+v", entries)
|
||||
}
|
||||
|
||||
// ...but the cell itself survives, at the tombstone's version.
|
||||
cell, exists, err := c.UDLGetCell(uid, cid, key)
|
||||
if err != nil {
|
||||
t.Fatalf("get cell: %v", err)
|
||||
}
|
||||
if !exists {
|
||||
t.Fatal("tombstone must not delete the cell — an older write could then resurrect the value")
|
||||
}
|
||||
if cell.Version != 2 || cell.Record != "rec-v2-tombstone" {
|
||||
t.Fatalf("cell should hold the tombstone record at v2, got %+v", cell)
|
||||
}
|
||||
|
||||
// And the high-water mark still rejects the superseded write.
|
||||
if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 1, 300, "rec-v1-replay", "", false, false); err != nil || ok {
|
||||
t.Fatalf("stale replay must be rejected: ok=%v err=%v", ok, err)
|
||||
}
|
||||
|
||||
// Re-watching re-establishes the cell and puts it back in the index.
|
||||
if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 3, 400, "rec-v3", "", false, false); err != nil || !ok {
|
||||
t.Fatalf("re-establish upsert: ok=%v err=%v", ok, err)
|
||||
}
|
||||
entries, err = c.UDLListUserKey(uid, key)
|
||||
if err != nil {
|
||||
t.Fatalf("active_for_user_key after re-establish: %v", err)
|
||||
}
|
||||
if len(entries) != 1 || entries[0].Cid != cid {
|
||||
t.Fatalf("re-established cid must return to the index, got %+v", entries)
|
||||
}
|
||||
}
|
||||
@@ -422,10 +422,9 @@ func (w *Watcher) ScanMountPath(mountPath string) (int, error) {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// Scan the directory
|
||||
log.Printf("[Watcher] Scanning mount path: %s", mountPath)
|
||||
// Scan the directory. The mount poller reports completed scans (only when
|
||||
// the file count changes), so nothing is logged per scan here.
|
||||
count := w.ScanDirectory(mountPath)
|
||||
log.Printf("[Watcher] Mount scan complete: %s (%d files)", mountPath, count)
|
||||
|
||||
return count, nil
|
||||
}
|
||||
|
||||
@@ -50,8 +50,6 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
path = "/"
|
||||
}
|
||||
|
||||
log.Printf("[WebDAV] %s %s", r.Method, path)
|
||||
|
||||
switch r.Method {
|
||||
case "GET", "HEAD":
|
||||
h.handleRead(w, r, path)
|
||||
|
||||
Reference in New Issue
Block a user