diff --git a/internal/api/server.go b/internal/api/server.go index fa7f638..d3f455a 100644 --- a/internal/api/server.go +++ b/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") diff --git a/internal/api/udl.go b/internal/api/udl.go new file mode 100644 index 0000000..d8dd0b1 --- /dev/null +++ b/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 +} diff --git a/internal/storage/udl.go b/internal/storage/udl.go new file mode 100644 index 0000000..f402fa7 --- /dev/null +++ b/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:// HASH { record, version, ts, value? } +// udl:idx:user::key: SET of -> active_for_user_key (My List, Continue Watching) +// udl:idx:user::cid: SET of -> list_for_user_cid +// udl:idx:cid::key: SET of -> aggregate +// udl:idx:cid::users SET of -> 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 +}