remove in metadata cid_* to put in the cids/ set
This commit is contained in:
+158
-208
@@ -12,37 +12,38 @@ import (
|
||||
)
|
||||
|
||||
// This file implements the CID-resolution layer that lives on top of the
|
||||
// flat metadata storage in client.go. The shape is described in detail in
|
||||
// docs/uuid-rooted-metadata.md; the short version:
|
||||
// flat metadata storage in client.go. The short version:
|
||||
//
|
||||
// - Roots are opaque UUIDv7s — file:<uuid>/<property> for every property
|
||||
// - Roots are opaque UUIDv7s — file:<uuid>/<property> for every property.
|
||||
// The UUID is an internal implementation detail; the mental model is
|
||||
// "N CIDs → one record of keys→metadata".
|
||||
// - Every known CID for a file is registered as a reverse-index entry:
|
||||
// cid:<algorithm>:<value> → <uuid> (Redis STRING)
|
||||
// and mirrored in a per-root set for fast enumeration:
|
||||
// file:<uuid>/cids (Redis SET)
|
||||
// - A canonical_cid field tracks the best (rank-wise) CID to advertise
|
||||
// externally; updated by reconcileCanonicalCIDLocked whenever the set
|
||||
// changes
|
||||
// cid:<bareCid> → <uuid> (Redis STRING)
|
||||
// and as a member of the per-root bare-CID key-set:
|
||||
// file:<uuid>/cids/<bareCid> = "true" (flat key, sentinel value)
|
||||
// - A CID is self-describing: its algorithm is recoverable from the CIDv1
|
||||
// multicodec (see internal/cid), so there is no per-algorithm field name
|
||||
// and no <algo>: token prefix. The canonical externally-visible CID is
|
||||
// *derived by rank* from the key-set on read (GetMetadataDocument), never
|
||||
// stored.
|
||||
//
|
||||
// Auto-alias hooks in SetProperty/SetMetadataFlat/MergeMetadataFlat catch
|
||||
// any write of a cid_* or midhash256 field and register the alias, so
|
||||
// plugins that just write their result fields don't need to know about
|
||||
// the reverse index at all. The hook helpers are *Locked — they assume
|
||||
// c.mu is already held by the caller, avoiding nested RLock (which can
|
||||
// deadlock against a waiting writer).
|
||||
// any write of a cids/<cid> field and register the reverse-index entry, so
|
||||
// writers that just emit the key-set members don't need to know about the
|
||||
// reverse index at all. The hook helpers are *Locked — they assume c.mu is
|
||||
// already held by the caller, avoiding nested RLock (which can deadlock
|
||||
// against a waiting writer).
|
||||
|
||||
// CIDIndexPrefix is the Redis key prefix for the reverse index. Combined
|
||||
// with a CID token, it forms the full key: "cid:midhash256:bafk…".
|
||||
// with a bare CID, it forms the full key: "cid:bafk…".
|
||||
const CIDIndexPrefix = "cid:"
|
||||
|
||||
// CIDsField is the per-root SET key that mirrors all CID aliases for a file.
|
||||
// Stored at file:<uuid>/cids — read by reconcileCanonicalCIDLocked and by
|
||||
// DeleteRoot to find every reverse-index entry that needs to be cleaned up.
|
||||
const CIDsField = "cids"
|
||||
|
||||
// CanonicalCIDField is the per-root STRING key that stores the externally
|
||||
// preferred CID for this file. Picked by cid.Better over the cids set.
|
||||
const CanonicalCIDField = "canonical_cid"
|
||||
// CIDsKeyPrefix is the field-name prefix for the bare-CID key-set. Sibling
|
||||
// CIDs live as flat keys file:<uuid>/cids/<cid> = "true"; the set IS the
|
||||
// keyset, the value is the sentinel. Read by GetMetadataDocument (hoisted
|
||||
// into the typed CIDs array) and by DeleteRoot/DeleteMetadata (to find the
|
||||
// reverse-index entries that need cleanup).
|
||||
const CIDsKeyPrefix = "cids/"
|
||||
|
||||
// DuplicatesField is the per-root SET key holding additional file paths
|
||||
// whose content hashes to the same CIDs as this root. The watcher writes
|
||||
@@ -51,9 +52,9 @@ const CanonicalCIDField = "canonical_cid"
|
||||
const DuplicatesField = "duplicates"
|
||||
|
||||
// buildCIDIndexKey returns the absolute Redis key for a reverse-index entry.
|
||||
// token is a CID in prefixed form ("midhash256:bafk…", "sha256:…", etc.).
|
||||
func (c *Client) buildCIDIndexKey(token string) string {
|
||||
return c.buildKey(CIDIndexPrefix + token)
|
||||
// cidStr is a bare multibase CIDv1 ("bafk…").
|
||||
func (c *Client) buildCIDIndexKey(cidStr string) string {
|
||||
return c.buildKey(CIDIndexPrefix + cidStr)
|
||||
}
|
||||
|
||||
// Mint creates a new root with filePath/sizeByte/mtimeNano and registers it
|
||||
@@ -88,11 +89,10 @@ func (c *Client) Mint(filePath string, size, mtimeNano int64) (string, error) {
|
||||
return uuid, nil
|
||||
}
|
||||
|
||||
// AddAlias registers a CID token as resolving to uuid. Idempotent: writing
|
||||
// the same alias twice is a no-op. After the alias is registered, the
|
||||
// canonical_cid field is reconciled — picks the highest-ranked token in
|
||||
// the cids set and writes it to canonical_cid if it changed.
|
||||
func (c *Client) AddAlias(uuid, token string) error {
|
||||
// AddAlias registers cidStr as a sibling CID of uuid: it writes the reverse
|
||||
// index cid:<cidStr> → uuid and the key-set member cids/<cidStr> = "true".
|
||||
// Idempotent: registering the same CID twice is a no-op.
|
||||
func (c *Client) AddAlias(uuid, cidStr string) error {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
|
||||
@@ -103,119 +103,102 @@ func (c *Client) AddAlias(uuid, token string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
return c.addAliasLocked(ctx, uuid, token)
|
||||
return c.addAliasLocked(ctx, uuid, cidStr)
|
||||
}
|
||||
|
||||
// addAliasLocked is the lock-free body of AddAlias. The caller must already
|
||||
// hold c.mu (read or write). Splits out so the alias-maintenance hooks
|
||||
// hold c.mu (read or write). Split out so the alias-maintenance hooks
|
||||
// invoked from SetProperty / SetMetadataFlat / MergeMetadataFlat can reuse
|
||||
// the work without re-entering the lock — nested RLock can deadlock
|
||||
// against a waiting writer.
|
||||
func (c *Client) addAliasLocked(ctx context.Context, uuid, token string) error {
|
||||
if uuid == "" || token == "" {
|
||||
return fmt.Errorf("uuid and token are required")
|
||||
}
|
||||
if cid.AlgorithmOf(token) == "" {
|
||||
return fmt.Errorf("token %q is not in <algorithm>:<value> form", token)
|
||||
// the work without re-entering the lock — nested RLock can deadlock against
|
||||
// a waiting writer.
|
||||
func (c *Client) addAliasLocked(ctx context.Context, uuid, cidStr string) error {
|
||||
if uuid == "" || cidStr == "" {
|
||||
return fmt.Errorf("uuid and cid are required")
|
||||
}
|
||||
|
||||
keysetKey := c.buildKeyPrefix(uuid) + CIDsKeyPrefix + cidStr
|
||||
|
||||
// Self-pointing alias guard. If the caller is trying to register
|
||||
// `cid:<algo>:<value>` → `<value>` (i.e. the root IS the cid value),
|
||||
// that's the dual-root failure mode in disguise — it means somebody
|
||||
// wrote /meta/<value> without first resolving via ResolveRoot. Refuse
|
||||
// the alias write; leave the existing alias (if any) alone. Still
|
||||
// add to this root's cids set so the field-level write succeeds.
|
||||
if uuid == cid.ValueOf(token) {
|
||||
if _, err := c.client.SAdd(ctx, c.buildKeyPrefix(uuid)+CIDsField, token).Result(); err != nil {
|
||||
return fmt.Errorf("sadd cids set: %w", err)
|
||||
}
|
||||
return c.reconcileCanonicalCIDLocked(ctx, uuid)
|
||||
// cid:<cid> → <cid> (i.e. the root IS the cid value), that's the
|
||||
// dual-root failure mode in disguise — it means somebody wrote
|
||||
// /meta/<cid> without first resolving via ResolveRoot. Refuse the alias
|
||||
// write; leave any existing alias alone. Still record the key-set
|
||||
// membership so the field-level write succeeds.
|
||||
if uuid == cidStr {
|
||||
return c.client.Set(ctx, keysetKey, "true", 0).Err()
|
||||
}
|
||||
|
||||
// Guard against the dual-root pattern: if `cid:<token>` already points
|
||||
// at a DIFFERENT uuid, don't silently overwrite. The historical bug:
|
||||
// meta-sort writing cid_midhash256=<v> to root "<v>" caused the hook
|
||||
// to register cid:midhash256:<v> → <v>, severing the watcher's prior
|
||||
// cid:midhash256:<v> → <real-uuid> alias. With write-path resolution
|
||||
// now in place callers shouldn't hit this anymore, but the guard is
|
||||
// cheap and prevents regression if a new caller bypasses ResolveRoot.
|
||||
indexKey := c.buildCIDIndexKey(token)
|
||||
// Guard against the dual-root pattern: if cid:<cid> already points at a
|
||||
// DIFFERENT uuid, don't silently overwrite. Leave the existing alias and
|
||||
// just record the key-set membership so it's still discoverable via the
|
||||
// root → cids enumeration.
|
||||
indexKey := c.buildCIDIndexKey(cidStr)
|
||||
existing, err := c.client.Get(ctx, indexKey).Result()
|
||||
if err != nil && err != redis.Nil {
|
||||
return fmt.Errorf("read existing alias: %w", err)
|
||||
}
|
||||
if existing != "" && existing != uuid {
|
||||
// Existing alias points elsewhere. Leave it alone and just add the
|
||||
// token to this root's cids set so it's still discoverable via the
|
||||
// root → cids enumeration.
|
||||
if _, err := c.client.SAdd(ctx, c.buildKeyPrefix(uuid)+CIDsField, token).Result(); err != nil {
|
||||
return fmt.Errorf("sadd cids set: %w", err)
|
||||
}
|
||||
return c.reconcileCanonicalCIDLocked(ctx, uuid)
|
||||
return c.client.Set(ctx, keysetKey, "true", 0).Err()
|
||||
}
|
||||
|
||||
prefix := c.buildKeyPrefix(uuid)
|
||||
pipe := c.client.Pipeline()
|
||||
pipe.Set(ctx, indexKey, uuid, 0)
|
||||
pipe.SAdd(ctx, prefix+CIDsField, token)
|
||||
pipe.Set(ctx, keysetKey, "true", 0)
|
||||
if _, err := pipe.Exec(ctx); err != nil {
|
||||
return fmt.Errorf("add alias pipeline: %w", err)
|
||||
}
|
||||
return c.reconcileCanonicalCIDLocked(ctx, uuid)
|
||||
return nil
|
||||
}
|
||||
|
||||
// ResolveRoot maps a hash supplied by a caller (a meta-sort write, an
|
||||
// editor lookup, an SSE consumer) to the actual storage root key.
|
||||
// ResolveRoot maps a hash supplied by a caller (a meta-sort write, an editor
|
||||
// lookup, an SSE consumer) to the actual storage root key.
|
||||
//
|
||||
// Callers historically send a bare midhash256 value (e.g.
|
||||
// "bagacba…") because that's what the watcher publishes on file:events.
|
||||
// The UUID-rooted-metadata design (see docs/uuid-rooted-metadata.md)
|
||||
// keeps roots opaque and aliases CIDs into the reverse index:
|
||||
// Callers send a bare CID (e.g. the midhash "bafk…" the watcher publishes on
|
||||
// file:events, or any sibling CID). Roots are kept opaque and every CID is
|
||||
// aliased into the reverse index:
|
||||
//
|
||||
// cid:midhash256:bagacba… → 01KSBHM*
|
||||
// cid:bafk… → 01KSBHM*
|
||||
//
|
||||
// Without resolution, every /meta/<midhash> write creates a parallel
|
||||
// midhash-rooted entry that the watcher's UUID never sees. The editor
|
||||
// then shows two entries per file: the (rich) midhash root and the
|
||||
// (empty) UUID root containing only `duplicates`.
|
||||
// Without resolution, every /meta/<cid> write creates a parallel cid-rooted
|
||||
// entry that the watcher's UUID never sees.
|
||||
//
|
||||
// Behaviour:
|
||||
// - empty hash → empty (caller's problem)
|
||||
// - hash matches cid:midhash256:<hash> in the reverse index → return UUID
|
||||
// - hash matches a reverse-index entry → return UUID
|
||||
// - otherwise return hash unchanged (legacy / explicit-UUID writes)
|
||||
//
|
||||
// Errors are converted to "no resolution" — if Redis is down or the
|
||||
// lookup fails we let the caller proceed with the bare hash and surface
|
||||
// the Redis failure on the actual operation.
|
||||
// Errors are converted to "no resolution" — if Redis is down or the lookup
|
||||
// fails we let the caller proceed with the bare hash and surface the Redis
|
||||
// failure on the actual operation.
|
||||
func (c *Client) ResolveRoot(hash string) string {
|
||||
if hash == "" {
|
||||
return hash
|
||||
}
|
||||
uuid, err := c.GetByCID("midhash256:" + hash)
|
||||
uuid, err := c.GetByCID(hash)
|
||||
if err != nil || uuid == "" {
|
||||
return hash
|
||||
}
|
||||
return uuid
|
||||
}
|
||||
|
||||
// GetByCID resolves a CID token to the UUID of its root. Returns ("", nil)
|
||||
// when the token is unknown; the empty string is the "not found" signal
|
||||
// rather than a typed error so callers can branch without errors.As.
|
||||
func (c *Client) GetByCID(token string) (string, error) {
|
||||
// GetByCID resolves a bare CID to the UUID of its root. Returns ("", nil)
|
||||
// when the CID is unknown; the empty string is the "not found" signal rather
|
||||
// than a typed error so callers can branch without errors.As.
|
||||
func (c *Client) GetByCID(cidStr string) (string, error) {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
|
||||
if c.client == nil {
|
||||
return "", fmt.Errorf("not connected")
|
||||
}
|
||||
if token == "" {
|
||||
if cidStr == "" {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
uuid, err := c.client.Get(ctx, c.buildCIDIndexKey(token)).Result()
|
||||
uuid, err := c.client.Get(ctx, c.buildCIDIndexKey(cidStr)).Result()
|
||||
if err == redis.Nil {
|
||||
return "", nil
|
||||
}
|
||||
@@ -250,9 +233,32 @@ func (c *Client) AddDuplicatePath(uuid, path string) (bool, error) {
|
||||
return added > 0, nil
|
||||
}
|
||||
|
||||
// DeleteRoot removes everything keyed off uuid: every file:<uuid>/* property,
|
||||
// every reverse-index entry listed in the cids set, and the index membership.
|
||||
// Idempotent — a uuid with no remaining keys is a successful no-op.
|
||||
// cidsForRootLocked scans the bare-CID key-set for a root and returns its
|
||||
// members (bare CIDs, with the cids/ prefix stripped). Caller must hold c.mu.
|
||||
func (c *Client) cidsForRootLocked(ctx context.Context, uuid string) ([]string, error) {
|
||||
prefix := c.buildKeyPrefix(uuid) + CIDsKeyPrefix
|
||||
var out []string
|
||||
var cursor uint64
|
||||
for {
|
||||
keys, next, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("scan cids: %w", err)
|
||||
}
|
||||
for _, k := range keys {
|
||||
out = append(out, strings.TrimPrefix(k, prefix))
|
||||
}
|
||||
cursor = next
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// DeleteRoot removes everything keyed off uuid: every file:<uuid>/* property
|
||||
// (including the cids/<cid> key-set), every reverse-index entry for those
|
||||
// CIDs, and the index membership. Idempotent — a uuid with no remaining keys
|
||||
// is a successful no-op.
|
||||
func (c *Client) DeleteRoot(uuid string) error {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
@@ -269,27 +275,25 @@ func (c *Client) DeleteRoot(uuid string) error {
|
||||
|
||||
prefix := c.buildKeyPrefix(uuid)
|
||||
|
||||
// 1. Read the cids set first — once we delete file:<uuid>/* we lose the
|
||||
// list of reverse-index entries to clean up. Tolerate a missing set
|
||||
// (empty list = nothing to unmap).
|
||||
tokens, err := c.client.SMembers(ctx, prefix+CIDsField).Result()
|
||||
if err != nil && err != redis.Nil {
|
||||
return fmt.Errorf("smembers cids: %w", err)
|
||||
// 1. Read the cids key-set first — once we delete file:<uuid>/* we lose
|
||||
// the list of reverse-index entries to clean up.
|
||||
cids, err := c.cidsForRootLocked(ctx, uuid)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 2. Delete every reverse-index entry. Build the keys upfront, then DEL
|
||||
// in one shot so we don't pay per-RTT.
|
||||
if len(tokens) > 0 {
|
||||
keys := make([]string, 0, len(tokens))
|
||||
for _, t := range tokens {
|
||||
keys = append(keys, c.buildCIDIndexKey(t))
|
||||
// 2. Delete every reverse-index entry in one shot.
|
||||
if len(cids) > 0 {
|
||||
keys := make([]string, 0, len(cids))
|
||||
for _, cidStr := range cids {
|
||||
keys = append(keys, c.buildCIDIndexKey(cidStr))
|
||||
}
|
||||
if err := c.client.Del(ctx, keys...).Err(); err != nil {
|
||||
return fmt.Errorf("del cid index: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 3. Scan and delete every file:<uuid>/* key.
|
||||
// 3. Scan and delete every file:<uuid>/* key (cids/<cid> keys included).
|
||||
var cursor uint64
|
||||
for {
|
||||
keys, next, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
|
||||
@@ -314,115 +318,59 @@ func (c *Client) DeleteRoot(uuid string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// reconcileCanonicalCIDLocked reads the cids set, picks the highest-ranked
|
||||
// token, and updates the canonical_cid field if it changed. Called from
|
||||
// addAliasLocked after a new alias is registered. Caller must hold c.mu.
|
||||
func (c *Client) reconcileCanonicalCIDLocked(ctx context.Context, uuid string) error {
|
||||
prefix := c.buildKeyPrefix(uuid)
|
||||
|
||||
tokens, err := c.client.SMembers(ctx, prefix+CIDsField).Result()
|
||||
if err != nil {
|
||||
return fmt.Errorf("smembers cids: %w", err)
|
||||
}
|
||||
if len(tokens) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
best := tokens[0]
|
||||
for _, t := range tokens[1:] {
|
||||
best = cid.Better(best, t)
|
||||
}
|
||||
|
||||
canonicalKey := prefix + CanonicalCIDField
|
||||
current, err := c.client.Get(ctx, canonicalKey).Result()
|
||||
if err != nil && err != redis.Nil {
|
||||
return fmt.Errorf("get canonical_cid: %w", err)
|
||||
}
|
||||
if current == best {
|
||||
return nil
|
||||
}
|
||||
if err := c.client.Set(ctx, canonicalKey, best, 0).Err(); err != nil {
|
||||
return fmt.Errorf("set canonical_cid: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// maybeAddAliasFromFieldLocked inspects a (field, value) pair written into
|
||||
// a root. If it looks like a CID field — name is "midhash256" or starts
|
||||
// with "cid_" — it registers the implied CID token as an alias. Plugins
|
||||
// write CID fields without knowing about the reverse index; this hook
|
||||
// keeps the index in sync transparently.
|
||||
// maybeAddAliasFromFieldLocked inspects a (field, value) pair written into a
|
||||
// root. If it's a key-set member — field name starts with "cids/" — it
|
||||
// registers the implied CID in the reverse index. Writers emit cids/<cid>
|
||||
// members without knowing about the reverse index; this hook keeps it in
|
||||
// sync transparently.
|
||||
//
|
||||
// Caller must hold c.mu. Errors are intentionally swallowed at the call
|
||||
// site (logged, not returned): an out-of-sync reverse index is recoverable
|
||||
// via a sweep, while a failed property write is not.
|
||||
// Caller must hold c.mu. Errors are intentionally swallowed at the call site
|
||||
// (logged, not returned): an out-of-sync reverse index is recoverable via a
|
||||
// sweep, while a failed property write is not.
|
||||
func (c *Client) maybeAddAliasFromFieldLocked(ctx context.Context, uuid, field, value string) error {
|
||||
token := cidTokenFromField(field, value)
|
||||
if token == "" {
|
||||
cidStr := cidFromKeysetField(field)
|
||||
if cidStr == "" {
|
||||
return nil
|
||||
}
|
||||
return c.addAliasLocked(ctx, uuid, token)
|
||||
return c.addAliasLocked(ctx, uuid, cidStr)
|
||||
}
|
||||
|
||||
// maybeAddAliasesFromMetadataLocked bulk-applies maybeAddAliasFromFieldLocked
|
||||
// over every entry in a metadata map. Used by SetMetadataFlat /
|
||||
// MergeMetadataFlat after the underlying pipeline write succeeds. Caller
|
||||
// must hold c.mu.
|
||||
// MergeMetadataFlat after the underlying pipeline write succeeds. Caller must
|
||||
// hold c.mu.
|
||||
func (c *Client) maybeAddAliasesFromMetadataLocked(ctx context.Context, uuid string, metadata map[string]string) {
|
||||
for k, v := range metadata {
|
||||
_ = c.maybeAddAliasFromFieldLocked(ctx, uuid, k, v)
|
||||
for k := range metadata {
|
||||
_ = c.maybeAddAliasFromFieldLocked(ctx, uuid, k, metadata[k])
|
||||
}
|
||||
}
|
||||
|
||||
// cidTokenFromField turns a (field-name, raw-value) pair into a normalized
|
||||
// CID token "<algorithm>:<value>". Returns "" for non-CID fields.
|
||||
//
|
||||
// Acceptable inputs:
|
||||
// - field="midhash256", value="bafk…" → "midhash256:bafk…"
|
||||
// - field="cid_sha256", value="bafk…" → "sha256:bafk…"
|
||||
// - field="cid_ipfs", value="ipfs:bafy…" → "ipfs:bafy…" (trusted)
|
||||
//
|
||||
// Plugins that already use the prefixed form get their value passed through;
|
||||
// plugins that store the bare CID get prefixed from the field name. The
|
||||
// detection heuristic is just "value contains a colon" — base32lower CIDs
|
||||
// never contain colons, so this is unambiguous.
|
||||
func cidTokenFromField(field, value string) string {
|
||||
if value == "" {
|
||||
// cidFromKeysetField returns the bare CID encoded in a cids/<cid> field name,
|
||||
// or "" if the field isn't a key-set member.
|
||||
func cidFromKeysetField(field string) string {
|
||||
if !strings.HasPrefix(field, CIDsKeyPrefix) {
|
||||
return ""
|
||||
}
|
||||
var algo string
|
||||
switch {
|
||||
case field == "midhash256":
|
||||
algo = "midhash256"
|
||||
case strings.HasPrefix(field, "cid_"):
|
||||
algo = strings.TrimPrefix(field, "cid_")
|
||||
default:
|
||||
return ""
|
||||
}
|
||||
if algo == "" {
|
||||
return ""
|
||||
}
|
||||
if strings.IndexByte(value, ':') > 0 {
|
||||
return value
|
||||
}
|
||||
return cid.Token(algo, value)
|
||||
return strings.TrimPrefix(field, CIDsKeyPrefix)
|
||||
}
|
||||
|
||||
// MetadataDocument is the complete state of one file as the public API
|
||||
// returns it: every flat string field, plus the cids and duplicates sets
|
||||
// hoisted into typed arrays. Distinct from GetMetadataFlat (which only
|
||||
// returns string keys) — clients querying /api/meta/<cid> want the SETs
|
||||
// expanded too, since "what other CIDs identify this file" and "what other
|
||||
// paths point at this content" are core to the response.
|
||||
// returns it: every flat string field (the cids/<cid> members included),
|
||||
// plus the CIDs hoisted into a typed array, the canonical CID derived by
|
||||
// rank, and the duplicates set. Distinct from GetMetadataFlat (which only
|
||||
// returns the raw string keys).
|
||||
type MetadataDocument struct {
|
||||
Flat map[string]string `json:"metadata"`
|
||||
CIDs []string `json:"cids"`
|
||||
Canonical string `json:"canonical_cid,omitempty"`
|
||||
Duplicates []string `json:"duplicates"`
|
||||
}
|
||||
|
||||
// GetMetadataDocument returns the full document for a root: every string
|
||||
// property under file:<uuid>/* plus the cids and duplicates SETs. Returns
|
||||
// nil if the root has no keys at all (deleted or never existed).
|
||||
// property under file:<uuid>/* (cids/<cid> members included in Flat), the
|
||||
// CIDs hoisted into a typed array, the canonical CID computed by rank, and
|
||||
// the duplicates SET. Returns nil if the root has no keys at all (deleted or
|
||||
// never existed).
|
||||
func (c *Client) GetMetadataDocument(uuid string) (*MetadataDocument, error) {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
@@ -443,10 +391,24 @@ func (c *Client) GetMetadataDocument(uuid string) (*MetadataDocument, error) {
|
||||
}
|
||||
prefix := c.buildKeyPrefix(uuid)
|
||||
|
||||
cids, err := c.client.SMembers(ctx, prefix+CIDsField).Result()
|
||||
if err != nil && err != redis.Nil {
|
||||
return nil, fmt.Errorf("smembers cids: %w", err)
|
||||
// Hoist the cids/<cid> key-set members into a typed array and derive the
|
||||
// canonical CID by rank. The members remain in Flat too — that's how
|
||||
// flat-map consumers (meta-share ingest) see them.
|
||||
cids := make([]string, 0)
|
||||
canonical := ""
|
||||
for k := range flat {
|
||||
if !strings.HasPrefix(k, CIDsKeyPrefix) {
|
||||
continue
|
||||
}
|
||||
cidStr := strings.TrimPrefix(k, CIDsKeyPrefix)
|
||||
cids = append(cids, cidStr)
|
||||
if canonical == "" {
|
||||
canonical = cidStr
|
||||
} else {
|
||||
canonical = cid.Better(canonical, cidStr)
|
||||
}
|
||||
}
|
||||
|
||||
dups, err := c.client.SMembers(ctx, prefix+DuplicatesField).Result()
|
||||
if err != nil && err != redis.Nil {
|
||||
return nil, fmt.Errorf("smembers duplicates: %w", err)
|
||||
@@ -458,19 +420,7 @@ func (c *Client) GetMetadataDocument(uuid string) (*MetadataDocument, error) {
|
||||
return &MetadataDocument{
|
||||
Flat: flat,
|
||||
CIDs: cids,
|
||||
Canonical: canonical,
|
||||
Duplicates: dups,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// lookupSidecarPathByCID — REMOVED.
|
||||
//
|
||||
// This function used to walk every metadata entry looking for poster/backdrop
|
||||
// matches and return their `posterPath`/`backdropPath` companion fields. It
|
||||
// was a workaround for plugin output (TMDB images, etc.) not being indexed
|
||||
// by the watcher.
|
||||
//
|
||||
// Now that the plugin output directory is a first-class watcher root
|
||||
// (see config.DefaultWatcherPaths), every plugin artefact has its own
|
||||
// midhash256 alias and resolves through the standard reverse index. Storing
|
||||
// paths in metadata broke the CID-as-identity contract; the watcher
|
||||
// indexing fixes the root cause.
|
||||
|
||||
+38
-47
@@ -291,16 +291,16 @@ func (c *Client) DeleteMetadata(hashID string) (int64, error) {
|
||||
|
||||
prefix := c.buildKeyPrefix(hashID)
|
||||
|
||||
// Read the cids set first — once file:<uuid>/* is gone we've lost the
|
||||
// Read the cids key-set first — once file:<uuid>/* is gone we've lost the
|
||||
// list of reverse-index keys that need cleanup.
|
||||
tokens, err := c.client.SMembers(ctx, prefix+CIDsField).Result()
|
||||
if err != nil && err != redis.Nil {
|
||||
return 0, fmt.Errorf("smembers cids: %w", err)
|
||||
cids, err := c.cidsForRootLocked(ctx, hashID)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if len(tokens) > 0 {
|
||||
idxKeys := make([]string, 0, len(tokens))
|
||||
for _, t := range tokens {
|
||||
idxKeys = append(idxKeys, c.buildCIDIndexKey(t))
|
||||
if len(cids) > 0 {
|
||||
idxKeys := make([]string, 0, len(cids))
|
||||
for _, cidStr := range cids {
|
||||
idxKeys = append(idxKeys, c.buildCIDIndexKey(cidStr))
|
||||
}
|
||||
if err := c.client.Del(ctx, idxKeys...).Err(); err != nil {
|
||||
return 0, fmt.Errorf("del cid index: %w", err)
|
||||
@@ -347,32 +347,24 @@ func (c *Client) CountFiles() (int, error) {
|
||||
return len(hashIDs), nil
|
||||
}
|
||||
|
||||
// LookupPathByCID resolves a CID to its file path on disk via the reverse
|
||||
// index. O(1) lookup followed by a single GET of file:<uuid>/filePath.
|
||||
// LookupPathByCID resolves a bare CID to its file path on disk via the
|
||||
// reverse index. O(1) lookup followed by a single GET of file:<uuid>/filePath.
|
||||
//
|
||||
// Accepts both prefixed tokens ("midhash256:bafk…", "sha256:…",
|
||||
// "ipfs:bafy…") and bare midhash256 CIDs — the bare form is what the
|
||||
// editor passes to /api/file/{cid}/info because that's how plugins
|
||||
// write the value (e.g. `backdrop=<bare-cid>`). For bare tokens we
|
||||
// try the midhash256 prefix as a fallback.
|
||||
// Any registered CID resolves — the midhash, a sibling sha2-256, an IPFS
|
||||
// CID, a btih, etc. (they all alias to the same root). The bare CID is what
|
||||
// the editor passes to /api/file/{cid}/info and what plugins write (e.g.
|
||||
// `backdrop=<bare-cid>`).
|
||||
//
|
||||
// Plugin output (TMDB posters/backdrops, extracted subtitles, etc.) is
|
||||
// resolved through this same path: the plugin output directory is a
|
||||
// watcher root (see config.DefaultWatcherPaths), so every plugin
|
||||
// artefact gets a midhash256 alias and resolves through the reverse
|
||||
// index like any other file. There is no path-companion sidecar field.
|
||||
func (c *Client) LookupPathByCID(token string) (string, error) {
|
||||
uuid, err := c.GetByCID(token)
|
||||
// watcher root (see config.DefaultWatcherPaths), so every plugin artefact
|
||||
// gets a midhash alias and resolves through the reverse index like any
|
||||
// other file. There is no path-companion sidecar field.
|
||||
func (c *Client) LookupPathByCID(cidStr string) (string, error) {
|
||||
uuid, err := c.GetByCID(cidStr)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if uuid == "" && !strings.Contains(token, ":") {
|
||||
// Bare CID — most likely a midhash256 written by a plugin.
|
||||
uuid, err = c.GetByCID("midhash256:" + token)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
}
|
||||
if uuid == "" {
|
||||
return "", nil
|
||||
}
|
||||
@@ -599,23 +591,24 @@ func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) {
|
||||
// file. Used by the watcher hydrator at startup to avoid re-mid-hashing
|
||||
// files whose tuple matches what's already in Redis.
|
||||
//
|
||||
// HashID is the root identifier (UUIDv7 in the current model); MidHash256
|
||||
// is the actual midhash CID for the file, present iff the watcher already
|
||||
// computed and stored it. Hydration uses HashID for the state-registry
|
||||
// key and MidHash256 for the in-memory "I already know this hash" flag.
|
||||
// HashID is the root identifier (UUIDv7 in the current model). Hydration
|
||||
// uses HashID + (FilePath, Size, MtimeNano) to seed the in-memory state so
|
||||
// the next scan can skip re-mid-hashing files whose size+mtime are unchanged.
|
||||
type FileTuple struct {
|
||||
HashID string
|
||||
FilePath string
|
||||
Size int64
|
||||
MtimeNano int64
|
||||
MidHash256 string
|
||||
HashID string
|
||||
FilePath string
|
||||
Size int64
|
||||
MtimeNano int64
|
||||
}
|
||||
|
||||
// GetTuplesForAllFiles returns one FileTuple per indexed root, fetched in
|
||||
// batched MGETs (no SCAN per root). Skips files missing any of filePath,
|
||||
// sizeByte, or mtimeNano. midhash256 is optional — files that haven't
|
||||
// been hashed yet still appear in the result with an empty MidHash256.
|
||||
// Sub-second on local Redis for ~10K files.
|
||||
// sizeByte, or mtimeNano. Sub-second on local Redis for ~10K files.
|
||||
//
|
||||
// The midhash is intentionally NOT fetched here: it now lives only as a
|
||||
// cids/<cid> key-set member (no named field), and the skip-rehash decision
|
||||
// is purely size+mtime — so reading it would cost a SCAN per file for no
|
||||
// benefit.
|
||||
func (c *Client) GetTuplesForAllFiles() ([]FileTuple, error) {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
@@ -638,8 +631,8 @@ func (c *Client) GetTuplesForAllFiles() ([]FileTuple, error) {
|
||||
// Build the key list: 4 keys per root, in a fixed order so we can decode
|
||||
// the MGET response by position. fields[i] for hashIDs[j] sits at
|
||||
// keys[j*4+i].
|
||||
const fieldsPerFile = 4
|
||||
fields := [fieldsPerFile]string{"filePath", "sizeByte", "mtimeNano", "midhash256"}
|
||||
const fieldsPerFile = 3
|
||||
fields := [fieldsPerFile]string{"filePath", "sizeByte", "mtimeNano"}
|
||||
keys := make([]string, 0, len(hashIDs)*fieldsPerFile)
|
||||
for _, h := range hashIDs {
|
||||
prefix := c.buildKeyPrefix(h)
|
||||
@@ -669,7 +662,6 @@ func (c *Client) GetTuplesForAllFiles() ([]FileTuple, error) {
|
||||
filePath, _ := values[base].(string)
|
||||
sizeStr, _ := values[base+1].(string)
|
||||
mtimeStr, _ := values[base+2].(string)
|
||||
midhash, _ := values[base+3].(string)
|
||||
if filePath == "" || sizeStr == "" || mtimeStr == "" {
|
||||
continue
|
||||
}
|
||||
@@ -682,11 +674,10 @@ func (c *Client) GetTuplesForAllFiles() ([]FileTuple, error) {
|
||||
continue
|
||||
}
|
||||
out = append(out, FileTuple{
|
||||
HashID: h,
|
||||
FilePath: filePath,
|
||||
Size: size,
|
||||
MtimeNano: mtime,
|
||||
MidHash256: midhash,
|
||||
HashID: h,
|
||||
FilePath: filePath,
|
||||
Size: size,
|
||||
MtimeNano: mtime,
|
||||
})
|
||||
}
|
||||
return out, nil
|
||||
|
||||
@@ -3,160 +3,39 @@ package storage
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// MigrateDualRoots scans file:__index__ for hashIds that look like raw
|
||||
// midhash256 CID values (i.e. NOT UUIDs/ULIDs) and have a matching alias
|
||||
// `cid:midhash256:<hash>` pointing to a different UUID. For each such
|
||||
// "stranded" root we:
|
||||
// MigrateDualRoots was a one-shot repair for the "dual-root" pattern in the
|
||||
// legacy storage model: /meta/{hash} write handlers that didn't resolve
|
||||
// CID → UUID first created a parallel midhash-rooted entry alongside the
|
||||
// watcher's UUID root.
|
||||
//
|
||||
// 1. Copy every file:<midhash>/<property> into file:<uuid>/<property>
|
||||
// using MergeMetadataFlat (preserves anything the UUID root already
|
||||
// holds, e.g. duplicates and watcher-owned fields).
|
||||
// 2. Delete the entire file:<midhash>/* subtree.
|
||||
// 3. Drop the midhash hashId from file:__index__ (kept on the UUID).
|
||||
// That pattern is now structurally impossible:
|
||||
// - writes resolve via ResolveRoot before landing (see handlers), and
|
||||
// - CIDs are stored as a bare-CID key-set (cids/<cid>) keyed off opaque
|
||||
// UUID roots, with the reverse index cid:<cid> → uuid as the single
|
||||
// resolution path.
|
||||
//
|
||||
// This is the one-shot fix for the dual-root pattern described in
|
||||
// docs/api-mediated-access.md. The pattern was caused by the (now-fixed)
|
||||
// /meta/{hash} write handlers not resolving CID → UUID before writing.
|
||||
//
|
||||
// Returns the number of stranded roots fixed.
|
||||
// The legacy on-disk shape this migration targeted (midhash-rooted entries,
|
||||
// <algo>:<value> tokens, named midhash256 fields) no longer exists, and the
|
||||
// cut-over wipes Redis, so there is nothing to migrate. The function is kept
|
||||
// as a no-op so the admin endpoint and its wrappers continue to compile and
|
||||
// respond.
|
||||
func (c *Client) MigrateDualRoots(ctx context.Context) (int, error) {
|
||||
c.mu.RLock()
|
||||
if c.client == nil {
|
||||
c.mu.RUnlock()
|
||||
return 0, fmt.Errorf("not connected")
|
||||
}
|
||||
hashIDs, err := c.getAllHashIDsInternal(ctx)
|
||||
c.mu.RUnlock()
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("get hash ids: %w", err)
|
||||
}
|
||||
|
||||
// Build the orphaned-UUID index: for every UUID-rooted entry, record
|
||||
// midhash256 → UUID. This lets us find the "real" target when an alias
|
||||
// has been overwritten to self-point (the failure mode the in-flight
|
||||
// fix prevents going forward, but a one-shot still needs to repair
|
||||
// existing dual-rooted data).
|
||||
midhashToUUID := make(map[string]string)
|
||||
for _, hashID := range hashIDs {
|
||||
if !looksLikeULID(hashID) {
|
||||
continue
|
||||
}
|
||||
mh, err := c.GetProperty(hashID, "midhash256")
|
||||
if err != nil || mh == "" {
|
||||
continue
|
||||
}
|
||||
midhashToUUID[mh] = hashID
|
||||
}
|
||||
|
||||
fixed := 0
|
||||
for _, hashID := range hashIDs {
|
||||
// Skip likely-UUID hashIds. ULIDs are 26 chars, uppercase Crockford
|
||||
// Base32. We treat anything that's not all-uppercase-alphanumeric
|
||||
// of length ~26 as a CID-shaped hashId worth checking.
|
||||
if looksLikeULID(hashID) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Is there an alias pointing to a different uuid?
|
||||
token := "midhash256:" + hashID
|
||||
uuid, err := c.GetByCID(token)
|
||||
if err != nil {
|
||||
log.Printf("[MigrateDualRoots] GetByCID(%s): %v (skipping)", token, err)
|
||||
continue
|
||||
}
|
||||
// Three cases:
|
||||
// - uuid empty → no alias, this midhash root is standalone (skip)
|
||||
// - uuid == hashID → self-pointing alias; check the orphan index
|
||||
// for a UUID that claims this midhash via its midhash256 field
|
||||
// - uuid != hashID and != "" → proper alias, migrate to that UUID
|
||||
if uuid == "" || uuid == hashID {
|
||||
if orphan, ok := midhashToUUID[hashID]; ok && orphan != "" {
|
||||
uuid = orphan
|
||||
} else {
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
// Pull every key under file:<hashID>/* and merge into file:<uuid>/.
|
||||
flat, err := c.GetMetadataFlat(hashID)
|
||||
if err != nil {
|
||||
log.Printf("[MigrateDualRoots] GetMetadataFlat(%s): %v (skipping)", hashID, err)
|
||||
continue
|
||||
}
|
||||
if len(flat) > 0 {
|
||||
if _, err := c.MergeMetadataFlat(uuid, flat); err != nil {
|
||||
log.Printf("[MigrateDualRoots] MergeMetadataFlat(%s ← %s): %v", uuid, hashID, err)
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
// Delete the stranded root entirely.
|
||||
if _, err := c.DeleteMetadata(hashID); err != nil {
|
||||
log.Printf("[MigrateDualRoots] DeleteMetadata(%s): %v", hashID, err)
|
||||
continue
|
||||
}
|
||||
|
||||
// Re-point the alias to the surviving UUID. DeleteMetadata wipes
|
||||
// the cid:* entries for the deleted root, so the alias for this
|
||||
// midhash may have been removed. Restore it pointing at the
|
||||
// canonical UUID so future ResolveRoot calls land correctly.
|
||||
if err := c.AddAlias(uuid, "midhash256:"+hashID); err != nil {
|
||||
log.Printf("[MigrateDualRoots] AddAlias(%s → midhash256:%s): %v", uuid, hashID, err)
|
||||
// Not fatal — data is migrated; the alias mismatch can be
|
||||
// re-fixed by another sweep.
|
||||
}
|
||||
|
||||
fixed++
|
||||
}
|
||||
|
||||
if fixed > 0 {
|
||||
log.Printf("[MigrateDualRoots] Reunited %d stranded midhash roots into their UUID counterparts", fixed)
|
||||
}
|
||||
return fixed, nil
|
||||
}
|
||||
|
||||
// looksLikeULID is a cheap heuristic for "this hashId is opaque internal
|
||||
// UUID, not a CID we should resolve." ULIDs are 26-char Crockford Base32;
|
||||
// CIDs in this codebase are longer and lowercase. Tolerates dashes for
|
||||
// future-proofing.
|
||||
func looksLikeULID(s string) bool {
|
||||
if len(s) != 26 {
|
||||
return false
|
||||
}
|
||||
for _, r := range s {
|
||||
if r >= 'A' && r <= 'Z' {
|
||||
continue
|
||||
}
|
||||
if r >= '0' && r <= '9' {
|
||||
continue
|
||||
}
|
||||
return false
|
||||
}
|
||||
return true
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// MigrateDualRootsWithTimeout is a convenience wrapper used by the API
|
||||
// handler and the startup hook. Default to a generous 5-minute deadline —
|
||||
// the per-entry work is small but the index can be large.
|
||||
// handler and the startup hook.
|
||||
func (c *Client) MigrateDualRootsWithTimeout() (int, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
|
||||
defer cancel()
|
||||
return c.MigrateDualRoots(ctx)
|
||||
return c.MigrateDualRoots(context.Background())
|
||||
}
|
||||
|
||||
// MigrateDualRootsReport returns a human-readable summary of what
|
||||
// MigrateDualRoots would do or did. Useful for the admin endpoint.
|
||||
// MigrateDualRootsReport returns a human-readable summary. With the dual-root
|
||||
// pattern retired (see MigrateDualRoots), this always reports nothing to do.
|
||||
func (c *Client) MigrateDualRootsReport(ctx context.Context) (string, int, error) {
|
||||
fixed, err := c.MigrateDualRoots(ctx)
|
||||
if err != nil {
|
||||
return "", fixed, err
|
||||
}
|
||||
parts := []string{}
|
||||
parts = append(parts, fmt.Sprintf("merged %d stranded midhash roots into UUID roots", fixed))
|
||||
return strings.Join(parts, "; "), fixed, nil
|
||||
return "dual-root migration is obsolete under the bare-CID key-set model; nothing to do", 0, nil
|
||||
}
|
||||
|
||||
@@ -52,26 +52,23 @@ func TestNewUUID_TimeSortable(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestCIDTokenFromField(t *testing.T) {
|
||||
func TestCIDFromKeysetField(t *testing.T) {
|
||||
cases := []struct {
|
||||
field, value string
|
||||
want string
|
||||
field string
|
||||
want string
|
||||
}{
|
||||
{"midhash256", "bafk", "midhash256:bafk"},
|
||||
{"cid_sha256", "bafk", "sha256:bafk"},
|
||||
{"cid_ipfs", "bafy", "ipfs:bafy"},
|
||||
{"cid_sha256", "sha256:bafk", "sha256:bafk"}, // already prefixed
|
||||
{"cid_ipfs", "ipfs:bafy", "ipfs:bafy"}, // already prefixed
|
||||
{"title", "Inception", ""}, // non-CID field
|
||||
{"tmdb/title", "Inception", ""}, // non-CID nested
|
||||
{"cid_", "bafk", ""}, // empty algo
|
||||
{"midhash256", "", ""}, // empty value
|
||||
{"cid_sha256", "", ""}, // empty value
|
||||
{"cids/bafkmidhash", "bafkmidhash"},
|
||||
{"cids/baejbeisha256", "baejbeisha256"},
|
||||
{"cids/bafyipfs", "bafyipfs"},
|
||||
{"title", ""}, // non-CID field
|
||||
{"tmdb/title", ""}, // non-CID nested
|
||||
{"cid_sha2-256", ""}, // legacy flat field — no longer recognized
|
||||
{"midhash256", ""}, // legacy named field — no longer recognized
|
||||
{"cids", ""}, // prefix without a member
|
||||
}
|
||||
for _, tc := range cases {
|
||||
if got := cidTokenFromField(tc.field, tc.value); got != tc.want {
|
||||
t.Errorf("cidTokenFromField(%q, %q) = %q, want %q",
|
||||
tc.field, tc.value, got, tc.want)
|
||||
if got := cidFromKeysetField(tc.field); got != tc.want {
|
||||
t.Errorf("cidFromKeysetField(%q) = %q, want %q", tc.field, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user