meta-core/internal/storage/udl.go

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:user:<uid>:cids         SET of <cid>   -> whole-profile export (recovery)
//	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"
	"sort"
	"strconv"
	"strings"
	"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) }

// udlIdxUserCids is the set of every cid a uid has ANY record for — the one
// index that makes "export this whole profile" a set read instead of a
// keyspace scan.
//
// The other four indexes each answer a question with one coordinate already
// known (a key, a cid). Cross-instance recovery knows neither: it has a uid and
// nothing else. Without this set the only ways to enumerate a profile are to
// iterate the hardcoded key list — which silently misses any key added later —
// or to SCAN `udl:rec:<uid>/*`, which is the full-keyspace-read pathology that
// already made `/api/udl/users/stats` admin-only.
func udlIdxUserCids(uid string) string { return fmt.Sprintf("udl:idx:user:%s:cids", uid) }

// 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
//	      6=idx:user:cids
//	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 four indexes are left
// alone: they answer "who has an opinion about this cid", where a tombstone is
// still an answer.
//
// ⚠ `idx:user:cids` (KEYS[6]) in particular must NEVER be pruned on a
// tombstone, even though it looks symmetric with KEYS[2]. It is the export
// index: a cid dropped from it is a cid whose tombstone never travels, and a
// tombstone that does not travel is exactly how an un-liked title comes back
// from the dead on the other box — the peer still holds the older `like` at a
// lower version, and with no tombstone to beat it, nothing ever supersedes it.
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])
redis.call('SADD', KEYS[6], ARGV[6])
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)),
		c.buildKey(udlIdxUserCids(uid)),
	}
	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
}

// UDLUserAllEntry is one row of a whole-profile export.
type UDLUserAllEntry struct {
	Cid     string `json:"cid"`
	Key     string `json:"key"`
	Record  string `json:"record"`
	Version int64  `json:"version"`
	Ts      int64  `json:"ts"`
}

// UDLAllForUser returns every cell a uid owns, for cross-instance recovery.
//
// Ordered by cid (sorted, so paging is stable across calls even though the
// underlying index is a Redis SET), starting strictly after `afterCid`. A cid's
// keys are never split across a page: the page ends on a cid boundary once
// `limit` rows have been emitted, and `nextCid` is the last cid included — pass
// it back as `afterCid` to continue. `nextCid` is empty when the export is
// complete.
//
// Tombstones are included. They are the CRDT state that stops a deleted value
// resurrecting on the box that receives this export, so omitting them would
// make recovery actively wrong rather than merely incomplete.
func (c *Client) UDLAllForUser(uid, afterCid string, limit int) ([]UDLUserAllEntry, string, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return nil, "", fmt.Errorf("not connected")
	}
	if limit <= 0 {
		limit = 1000
	}

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	cids, err := c.client.SMembers(ctx, c.buildKey(udlIdxUserCids(uid))).Result()
	if err != nil {
		return nil, "", fmt.Errorf("smembers failed: %w", err)
	}
	sort.Strings(cids)

	out := make([]UDLUserAllEntry, 0, limit)
	nextCid := ""
	for _, cid := range cids {
		if afterCid != "" && cid <= afterCid {
			continue
		}
		if len(out) >= limit {
			nextCid = out[len(out)-1].Cid
			break
		}
		keys, err := c.client.SMembers(ctx, c.buildKey(udlIdxUserCid(uid, cid))).Result()
		if err != nil {
			return nil, "", fmt.Errorf("smembers failed: %w", err)
		}
		sort.Strings(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, UDLUserAllEntry{
				Cid: cid, Key: key, Record: cell.Record, Version: cell.Version, Ts: cell.Ts,
			})
		}
	}
	return out, nextCid, nil
}

// UDLCountUserCids is the number of distinct cids a uid holds records for.
//
// O(1) (a Redis SCARD), which is what makes it usable as a per-push quota
// check on the gateway mirror — a quota that cost a keyspace walk would be a
// denial-of-service in its own right. Counts cids rather than cells because
// that is what the index holds; a profile is bounded by the titles it has
// touched, not by how many keys it has per title.
func (c *Client) UDLCountUserCids(uid string) (int64, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()

	if c.client == nil {
		return 0, fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	n, err := c.client.SCard(ctx, c.buildKey(udlIdxUserCids(uid))).Result()
	if err != nil {
		return 0, fmt.Errorf("scard failed: %w", err)
	}
	return n, nil
}

// UDLBackfillUserCids rebuilds the `idx:user:<uid>:cids` sets for data written
// before that index existed, and reports how many cids it added.
//
// One-shot: the upsert script maintains the set from now on. It is a SCAN, but
// a targeted one (`udl:idx:user:*:cid:*`, which is one key per (uid,cid), not
// per record) run once at boot rather than per read — the distinction that
// makes the difference between a migration and the pathology this index exists
// to avoid.
func (c *Client) UDLBackfillUserCids() (int, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.client == nil {
		return 0, fmt.Errorf("not connected")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()

	pattern := c.buildKey("udl:idx:user:*:cid:*")
	prefix := c.buildKey("udl:idx:user:")
	added := 0
	var cursor uint64
	for {
		keys, next, err := c.client.Scan(ctx, cursor, pattern, 500).Result()
		if err != nil {
			return added, fmt.Errorf("scan failed: %w", err)
		}
		for _, k := range keys {
			// <prefix><uid>:cid:<cid> — split on the FIRST separator, not the
			// last. A uid is multibase base58btc, whose alphabet contains no
			// colon at all, so the first ":cid:" always ends the uid. A cid
			// that happens to contain ":cid:" in its own name would make the
			// last occurrence split in the wrong place.
			rest := strings.TrimPrefix(k, prefix)
			i := strings.Index(rest, ":cid:")
			if i < 0 {
				continue
			}
			uid, cid := rest[:i], rest[i+len(":cid:"):]
			if uid == "" || cid == "" {
				continue
			}
			n, err := c.client.SAdd(ctx, c.buildKey(udlIdxUserCids(uid)), cid).Result()
			if err != nil {
				return added, fmt.Errorf("sadd failed: %w", err)
			}
			added += int(n)
		}
		cursor = next
		if cursor == 0 {
			break
		}
	}
	return added, 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
}