improve meta core user management

internal/api/server.go

@@ -201,6 +201,15 @@ func (s *Server) setupRoutes() {
 	s.router.HandleFunc("/api/kv/value", s.handleKVValueDelete).Methods("DELETE")
 	s.router.HandleFunc("/api/kv/key/{key:.*}", s.handleKVGetKey).Methods("GET")
 
+	// User Data Layer (UDL) routes — per-user signed records (My List,
+	// Continue Watching, ratings, profile). See internal/api/udl.go.
+	s.router.HandleFunc("/api/udl/record", s.handleUDLRecordGet).Methods("GET")
+	s.router.HandleFunc("/api/udl/record", s.handleUDLRecordPut).Methods("PUT")
+	s.router.HandleFunc("/api/udl/user/{uid}/key/{key}", s.handleUDLUserKey).Methods("GET")
+	s.router.HandleFunc("/api/udl/user/{uid}/cid/{cid}", s.handleUDLUserCid).Methods("GET")
+	s.router.HandleFunc("/api/udl/cid/{cid}/users", s.handleUDLCidUsers).Methods("GET")
+	s.router.HandleFunc("/api/udl/cid/{cid}/aggregate", s.handleUDLAggregate).Methods("GET")
+
 	// Metadata operations - base endpoints
 	s.router.HandleFunc("/meta/{hash}", s.handleGetMeta).Methods("GET")
 	s.router.HandleFunc("/meta/{hash}", s.handlePutMeta).Methods("PUT")

internal/api/udl.go

@@ -0,0 +1,239 @@
+package api
+
+// User Data Layer (UDL) HTTP surface.
+//
+// Backs meta-watch's per-user store (My List, Continue Watching, ratings,
+// profile) now that the authoritative store lives in meta-core instead of
+// meta-watch's local op-log. Records arrive already signed (and, for
+// private-tier keys, already encrypted) as opaque base64 CBOR blobs; meta-core
+// version-gates and indexes them but never inspects the crypto. See
+// internal/storage/udl.go for the Redis model.
+
+import (
+	"encoding/json"
+	"math"
+	"net/http"
+	"strings"
+
+	"github.com/gorilla/mux"
+)
+
+// UDLPutRequest is the body of PUT /api/udl/record.
+type UDLPutRequest struct {
+	Uid     string           `json:"uid"`
+	Cid     string           `json:"cid"`
+	Key     string           `json:"key"`
+	Version int64            `json:"version"`
+	Ts      int64            `json:"ts"`
+	Record  string           `json:"record"` // base64 CBOR, opaque + signed
+	Value   *json.RawMessage `json:"value,omitempty"`
+}
+
+// UDLPutResponse is the response of PUT /api/udl/record.
+type UDLPutResponse struct {
+	Accepted bool `json:"accepted"`
+}
+
+// UDLRecordResponse is the response of GET /api/udl/record.
+type UDLRecordResponse struct {
+	Record  string `json:"record"`
+	Version int64  `json:"version"`
+	Ts      int64  `json:"ts"`
+	Exists  bool   `json:"exists"`
+}
+
+// UDLAggregate mirrors meta-watch's Aggregate wire shape exactly (snake_case
+// field names; `avg` is null for boolean keys like `like`), so the meta-watch
+// client deserializes it into the same struct it serializes to the UI.
+type UDLAggregate struct {
+	Key        string   `json:"key"`
+	Count      int      `json:"count"`
+	Avg        *float64 `json:"avg"`
+	Trues      int      `json:"trues"`
+	SampleSize int      `json:"sample_size"`
+}
+
+// handleUDLRecordGet handles GET /api/udl/record?uid&cid&key
+func (s *Server) handleUDLRecordGet(w http.ResponseWriter, r *http.Request) {
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+	q := r.URL.Query()
+	uid, cid, key := q.Get("uid"), q.Get("cid"), q.Get("key")
+	if uid == "" || cid == "" || key == "" {
+		writeError(w, http.StatusBadRequest, "uid, cid and key are required")
+		return
+	}
+
+	cell, exists, err := s.storage.UDLGetCell(uid, cid, key)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, UDLRecordResponse{
+		Record:  cell.Record,
+		Version: cell.Version,
+		Ts:      cell.Ts,
+		Exists:  exists,
+	})
+}
+
+// handleUDLRecordPut handles PUT /api/udl/record. Version-gated: a stale write
+// (version <= stored) returns 409 stale_version so the client can re-read and
+// retry with a higher version.
+func (s *Server) handleUDLRecordPut(w http.ResponseWriter, r *http.Request) {
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+	var req UDLPutRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid JSON body")
+		return
+	}
+	if req.Uid == "" || req.Cid == "" || req.Key == "" || req.Record == "" {
+		writeError(w, http.StatusBadRequest, "uid, cid, key and record are required")
+		return
+	}
+
+	publicValue := ""
+	hasValue := false
+	if req.Value != nil {
+		publicValue = string(*req.Value)
+		hasValue = true
+	}
+
+	accepted, err := s.storage.UDLUpsertIfNewer(req.Uid, req.Cid, req.Key, req.Version, req.Ts, req.Record, publicValue, hasValue)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	if !accepted {
+		writeErrorSlug(w, http.StatusConflict, "stale_version", "a record with an equal or newer version already exists", true)
+		return
+	}
+	writeJSON(w, http.StatusOK, UDLPutResponse{Accepted: true})
+}
+
+// handleUDLUserKey handles GET /api/udl/user/{uid}/key/{key}
+// (active_for_user_key — My List, Continue Watching).
+func (s *Server) handleUDLUserKey(w http.ResponseWriter, r *http.Request) {
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+	vars := mux.Vars(r)
+	uid, key := vars["uid"], vars["key"]
+	if uid == "" || key == "" {
+		writeError(w, http.StatusBadRequest, "uid and key are required")
+		return
+	}
+	entries, err := s.storage.UDLListUserKey(uid, key)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]interface{}{"entries": entries})
+}
+
+// handleUDLUserCid handles GET /api/udl/user/{uid}/cid/{cid}
+// (list_for_user_cid).
+func (s *Server) handleUDLUserCid(w http.ResponseWriter, r *http.Request) {
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+	vars := mux.Vars(r)
+	uid, cid := vars["uid"], vars["cid"]
+	if uid == "" || cid == "" {
+		writeError(w, http.StatusBadRequest, "uid and cid are required")
+		return
+	}
+	entries, err := s.storage.UDLListUserCid(uid, cid)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]interface{}{"entries": entries})
+}
+
+// handleUDLCidUsers handles GET /api/udl/cid/{cid}/users (uids_for_cid).
+func (s *Server) handleUDLCidUsers(w http.ResponseWriter, r *http.Request) {
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+	cid := mux.Vars(r)["cid"]
+	if cid == "" {
+		writeError(w, http.StatusBadRequest, "cid is required")
+		return
+	}
+	users, err := s.storage.UDLCidUsers(cid)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]interface{}{"users": users})
+}
+
+// handleUDLAggregate handles GET /api/udl/cid/{cid}/aggregate?keys=like,rating.
+// Computes per-key population stats from the stored public value fields
+// (absorbs meta-watch's aggregate.rs). Reflects only records this core holds.
+func (s *Server) handleUDLAggregate(w http.ResponseWriter, r *http.Request) {
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+	cid := mux.Vars(r)["cid"]
+	if cid == "" {
+		writeError(w, http.StatusBadRequest, "cid is required")
+		return
+	}
+	keysParam := r.URL.Query().Get("keys")
+	if keysParam == "" {
+		keysParam = "like,rating"
+	}
+	aggregates := make([]UDLAggregate, 0)
+	for _, key := range strings.Split(keysParam, ",") {
+		key = strings.TrimSpace(key)
+		if key == "" {
+			continue
+		}
+		values, err := s.storage.UDLPublicValues(cid, key)
+		if err != nil {
+			writeError(w, http.StatusInternalServerError, err.Error())
+			return
+		}
+		aggregates = append(aggregates, aggregatePublicValues(key, values))
+	}
+	writeJSON(w, http.StatusOK, map[string]interface{}{"aggregates": aggregates})
+}
+
+// aggregatePublicValues folds raw JSON scalar values into an UDLAggregate:
+// counts booleans as trues and averages numeric values.
+func aggregatePublicValues(key string, values []string) UDLAggregate {
+	agg := UDLAggregate{Key: key, SampleSize: len(values), Count: len(values)}
+	var sum float64
+	var numeric int
+	for _, raw := range values {
+		var v interface{}
+		if err := json.Unmarshal([]byte(raw), &v); err != nil {
+			continue
+		}
+		switch t := v.(type) {
+		case bool:
+			if t {
+				agg.Trues++
+			}
+		case float64:
+			sum += t
+			numeric++
+		}
+	}
+	if numeric > 0 {
+		avg := math.Round((sum/float64(numeric))*100) / 100
+		agg.Avg = &avg
+	}
+	return agg
+}

internal/storage/udl.go

@@ -0,0 +1,284 @@
+package storage
+
+// User Data Layer (UDL) storage.
+//
+// Holds per-user, identity-signed records keyed by (uid, cid, key) — the
+// subjective/per-user state that METADATA_KEYS.md rule #4 bars from the shared
+// /file/{cid} hash (likes, ratings, watch progress, local paths). Each record
+// is an opaque, client-signed (and, for private-tier keys, client-encrypted)
+// CBOR blob; meta-core stores it verbatim and never needs the account key.
+//
+// Convergence is version-gated: a write is accepted only if its `version`
+// strictly exceeds the stored one (a per-cell high-water-mark, replacing the
+// op-log merge that used to live in meta-watch's store.rs).
+//
+// Redis model (all under the literal "udl:" prefix, mirroring the "file:" one):
+//
+//	udl:rec:<uid>/<cid>/<key>       HASH { record, version, ts, value? }
+//	udl:idx:user:<uid>:key:<key>    SET of <cid>   -> active_for_user_key (My List, Continue Watching)
+//	udl:idx:user:<uid>:cid:<cid>    SET of <key>   -> list_for_user_cid
+//	udl:idx:cid:<cid>:key:<key>     SET of <uid>   -> aggregate
+//	udl:idx:cid:<cid>:users         SET of <uid>   -> uids_for_cid
+//
+// The `value` hash field is the plaintext JSON scalar for PUBLIC keys only
+// (like/rating), used for aggregation; it is absent for private-tier keys
+// (whose real value rides encrypted inside `record`) and for tombstones.
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// UDLCell is a decoded (uid, cid, key) record cell.
+type UDLCell struct {
+	Record   string // base64 CBOR, opaque + signed
+	Version  int64
+	Ts       int64
+	Value    string // plaintext JSON scalar, public keys only
+	HasValue bool
+}
+
+// UDLUserKeyEntry is one row of active_for_user_key (all cids for a uid+key).
+type UDLUserKeyEntry struct {
+	Cid     string `json:"cid"`
+	Record  string `json:"record"`
+	Version int64  `json:"version"`
+	Ts      int64  `json:"ts"`
+}
+
+// UDLUserCidEntry is one row of list_for_user_cid (all keys for a uid+cid).
+type UDLUserCidEntry struct {
+	Key     string `json:"key"`
+	Record  string `json:"record"`
+	Version int64  `json:"version"`
+	Ts      int64  `json:"ts"`
+}
+
+func udlRecKey(uid, cid, key string) string {
+	return fmt.Sprintf("udl:rec:%s/%s/%s", uid, cid, key)
+}
+func udlIdxUserKey(uid, key string) string { return fmt.Sprintf("udl:idx:user:%s:key:%s", uid, key) }
+func udlIdxUserCid(uid, cid string) string { return fmt.Sprintf("udl:idx:user:%s:cid:%s", uid, cid) }
+func udlIdxCidKey(cid, key string) string  { return fmt.Sprintf("udl:idx:cid:%s:key:%s", cid, key) }
+func udlIdxCidUsers(cid string) string      { return fmt.Sprintf("udl:idx:cid:%s:users", cid) }
+
+// udlUpsertScript performs the version-gated upsert + index maintenance
+// atomically. Accepts (returns 1) only when the incoming version strictly
+// exceeds the stored one; otherwise returns 0 (stale).
+//
+//	KEYS: 1=cell 2=idx:user:key 3=idx:user:cid 4=idx:cid:key 5=idx:cid:users
+//	ARGV: 1=version 2=ts 3=record 4=value 5=hasValue("1"/"0")
+//	      6=cid 7=key 8=uid
+var udlUpsertScript = redis.NewScript(`
+local stored = redis.call('HGET', KEYS[1], 'version')
+if stored and tonumber(stored) >= tonumber(ARGV[1]) then
+  return 0
+end
+redis.call('HSET', KEYS[1], 'version', ARGV[1], 'ts', ARGV[2], 'record', ARGV[3])
+if ARGV[5] == '1' then
+  redis.call('HSET', KEYS[1], 'value', ARGV[4])
+else
+  redis.call('HDEL', KEYS[1], 'value')
+end
+redis.call('SADD', KEYS[2], ARGV[6])
+redis.call('SADD', KEYS[3], ARGV[7])
+redis.call('SADD', KEYS[4], ARGV[8])
+redis.call('SADD', KEYS[5], ARGV[8])
+return 1
+`)
+
+// UDLUpsertIfNewer stores a record cell iff version strictly exceeds the stored
+// one. publicValue/hasValue carry the plaintext JSON scalar for public keys
+// (used by aggregation); pass hasValue=false for private-tier keys and
+// tombstones. Returns accepted=false when the write is stale.
+func (c *Client) UDLUpsertIfNewer(uid, cid, key string, version, ts int64, record, publicValue string, hasValue bool) (bool, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.client == nil {
+		return false, fmt.Errorf("not connected")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	keys := []string{
+		c.buildKey(udlRecKey(uid, cid, key)),
+		c.buildKey(udlIdxUserKey(uid, key)),
+		c.buildKey(udlIdxUserCid(uid, cid)),
+		c.buildKey(udlIdxCidKey(cid, key)),
+		c.buildKey(udlIdxCidUsers(cid)),
+	}
+	hv := "0"
+	if hasValue {
+		hv = "1"
+	}
+	res, err := udlUpsertScript.Run(ctx, c.client, keys, version, ts, record, publicValue, hv, cid, key, uid).Result()
+	if err != nil {
+		return false, fmt.Errorf("udl upsert failed: %w", err)
+	}
+	n, _ := res.(int64)
+	return n == 1, nil
+}
+
+// UDLGetCell returns the record cell for (uid, cid, key). exists is false when
+// no record has ever been written for the cell.
+func (c *Client) UDLGetCell(uid, cid, key string) (UDLCell, bool, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	if c.client == nil {
+		return UDLCell{}, false, fmt.Errorf("not connected")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	return c.udlReadCell(ctx, udlRecKey(uid, cid, key))
+}
+
+// udlReadCell HMGETs a cell hash. Caller holds the lock. exists is false when
+// the version field is absent (i.e. the hash does not exist).
+func (c *Client) udlReadCell(ctx context.Context, recKey string) (UDLCell, bool, error) {
+	vals, err := c.client.HMGet(ctx, c.buildKey(recKey), "record", "version", "ts", "value").Result()
+	if err != nil {
+		return UDLCell{}, false, fmt.Errorf("hmget failed: %w", err)
+	}
+	// vals: [record, version, ts, value] — nil for absent fields.
+	if len(vals) < 4 || vals[1] == nil {
+		return UDLCell{}, false, nil
+	}
+	cell := UDLCell{}
+	if s, ok := vals[0].(string); ok {
+		cell.Record = s
+	}
+	if s, ok := vals[1].(string); ok {
+		cell.Version, _ = strconv.ParseInt(s, 10, 64)
+	}
+	if s, ok := vals[2].(string); ok {
+		cell.Ts, _ = strconv.ParseInt(s, 10, 64)
+	}
+	if s, ok := vals[3].(string); ok {
+		cell.Value = s
+		cell.HasValue = true
+	}
+	return cell, true, nil
+}
+
+// UDLListUserKey returns every cell for (uid, key) across all cids — the
+// projection behind active_for_user_key (My List, Continue Watching). Includes
+// tombstones; the caller decodes each record and filters.
+func (c *Client) UDLListUserKey(uid, key string) ([]UDLUserKeyEntry, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	if c.client == nil {
+		return nil, fmt.Errorf("not connected")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	cids, err := c.client.SMembers(ctx, c.buildKey(udlIdxUserKey(uid, key))).Result()
+	if err != nil {
+		return nil, fmt.Errorf("smembers failed: %w", err)
+	}
+	out := make([]UDLUserKeyEntry, 0, len(cids))
+	for _, cid := range cids {
+		cell, ok, err := c.udlReadCell(ctx, udlRecKey(uid, cid, key))
+		if err != nil {
+			return nil, err
+		}
+		if !ok {
+			continue
+		}
+		out = append(out, UDLUserKeyEntry{Cid: cid, Record: cell.Record, Version: cell.Version, Ts: cell.Ts})
+	}
+	return out, nil
+}
+
+// UDLListUserCid returns every cell for (uid, cid) across all keys — the
+// projection behind list_for_user_cid.
+func (c *Client) UDLListUserCid(uid, cid string) ([]UDLUserCidEntry, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	if c.client == nil {
+		return nil, fmt.Errorf("not connected")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	keys, err := c.client.SMembers(ctx, c.buildKey(udlIdxUserCid(uid, cid))).Result()
+	if err != nil {
+		return nil, fmt.Errorf("smembers failed: %w", err)
+	}
+	out := make([]UDLUserCidEntry, 0, len(keys))
+	for _, key := range keys {
+		cell, ok, err := c.udlReadCell(ctx, udlRecKey(uid, cid, key))
+		if err != nil {
+			return nil, err
+		}
+		if !ok {
+			continue
+		}
+		out = append(out, UDLUserCidEntry{Key: key, Record: cell.Record, Version: cell.Version, Ts: cell.Ts})
+	}
+	return out, nil
+}
+
+// UDLCidUsers returns every uid that has any record for cid — uids_for_cid.
+func (c *Client) UDLCidUsers(cid string) ([]string, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	if c.client == nil {
+		return nil, fmt.Errorf("not connected")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	uids, err := c.client.SMembers(ctx, c.buildKey(udlIdxCidUsers(cid))).Result()
+	if err != nil {
+		return nil, fmt.Errorf("smembers failed: %w", err)
+	}
+	return uids, nil
+}
+
+// UDLPublicValues returns the live plaintext public JSON scalar values for
+// (cid, key) across all uids — the input to aggregation. Cells without a value
+// field (private-tier or tombstoned) are skipped.
+func (c *Client) UDLPublicValues(cid, key string) ([]string, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	if c.client == nil {
+		return nil, fmt.Errorf("not connected")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	uids, err := c.client.SMembers(ctx, c.buildKey(udlIdxCidKey(cid, key))).Result()
+	if err != nil {
+		return nil, fmt.Errorf("smembers failed: %w", err)
+	}
+	out := make([]string, 0, len(uids))
+	for _, uid := range uids {
+		v, err := c.client.HGet(ctx, c.buildKey(udlRecKey(uid, cid, key)), "value").Result()
+		if err == redis.Nil {
+			continue // private-tier or tombstoned
+		}
+		if err != nil {
+			return nil, fmt.Errorf("hget failed: %w", err)
+		}
+		out = append(out, v)
+	}
+	return out, nil
+}