add identity management

internal/api/handlers_identity.go

@@ -4,6 +4,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"net/http"
+	"strconv"
 	"strings"
 
 	"github.com/metazla/meta-core/internal/identity"
@@ -88,6 +89,31 @@ type identitySignResponse struct {
 	SigB64 string `json:"sigB64"`
 }
 
+// identitySignBatchRequest signs many payloads under one account in one call.
+//
+// Exists because a User Data Layer write signs exactly one record, and the bulk
+// verbs above it ("mark this whole series seen") produce one record per episode:
+// a 300-episode show is 300 sequential sign round trips through
+// meta-watch -> meta-search -> here, which dominates the operation. The private
+// key still never leaves this process; only the transport is batched.
+type identitySignBatchRequest struct {
+	// PayloadsB64 are the canonical byte strings to sign, in order. The
+	// response's signatures line up index-for-index.
+	PayloadsB64 []string `json:"payloadsB64"`
+	// UID selects which account signs — same optional single-identity
+	// back-compat as identitySignRequest.
+	UID string `json:"uid,omitempty"`
+}
+
+type identitySignBatchResponse struct {
+	SigsB64 []string `json:"sigsB64"`
+}
+
+// maxSignBatch bounds one batch. A signature is cheap but not free, and the
+// request holds the account lock for its duration; a caller with more than this
+// should page. Sized well above the longest real series (Supernatural, 327).
+const maxSignBatch = 1000
+
 type identityAEADKeyResponse struct {
 	AEADKeyB64 string `json:"aeadKeyB64"`
 }
@@ -317,6 +343,53 @@ func (s *Server) handleIdentitySign(w http.ResponseWriter, r *http.Request) {
 	})
 }
 
+// handleIdentitySignBatch signs every payload in the request under one account.
+//
+// All-or-nothing: a single bad base64 or signing failure fails the whole
+// request rather than returning a short array. The caller pairs signatures with
+// records by index, so a partial response would silently misalign them — and a
+// misaligned signature verifies against the wrong record, which is worse than
+// an error.
+func (s *Server) handleIdentitySignBatch(w http.ResponseWriter, r *http.Request) {
+	var body identitySignBatchRequest
+	if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid JSON body")
+		return
+	}
+	uid := body.UID
+	if strings.TrimSpace(uid) == "" {
+		uid = r.URL.Query().Get("uid")
+	}
+	if len(body.PayloadsB64) == 0 {
+		writeError(w, http.StatusBadRequest, "payloadsB64 is required")
+		return
+	}
+	if len(body.PayloadsB64) > maxSignBatch {
+		writeError(w, http.StatusBadRequest, "too many payloads in one batch")
+		return
+	}
+	id, status, slug, msg := s.resolveAccount(uid)
+	if status != 0 {
+		writeErrorSlug(w, status, slug, msg, false)
+		return
+	}
+	sigs := make([]string, 0, len(body.PayloadsB64))
+	for i, p := range body.PayloadsB64 {
+		raw, err := base64.StdEncoding.DecodeString(p)
+		if err != nil {
+			writeError(w, http.StatusBadRequest, "payloadsB64["+strconv.Itoa(i)+"] is not valid base64")
+			return
+		}
+		sig, err := identity.Sign(id, raw)
+		if err != nil {
+			writeError(w, http.StatusInternalServerError, err.Error())
+			return
+		}
+		sigs = append(sigs, base64.StdEncoding.EncodeToString(sig))
+	}
+	writeJSON(w, http.StatusOK, identitySignBatchResponse{SigsB64: sigs})
+}
+
 func (s *Server) handleIdentityAEADKey(w http.ResponseWriter, r *http.Request) {
 	id, status, slug, msg := s.resolveAccount(r.URL.Query().Get("uid"))
 	if status != 0 {

internal/api/server.go

@@ -175,6 +175,7 @@ func (s *Server) setupRoutes() {
 	s.router.HandleFunc("/api/identity/reveal", s.handleIdentityReveal).Methods("POST")
 	s.router.HandleFunc("/api/identity", s.handleIdentityDelete).Methods("DELETE")
 	s.router.HandleFunc("/api/identity/sign", s.handleIdentitySign).Methods("POST")
+	s.router.HandleFunc("/api/identity/sign-batch", s.handleIdentitySignBatch).Methods("POST")
 	s.router.HandleFunc("/api/identity/aead-key", s.handleIdentityAEADKey).Methods("GET")
 
 	// Snapshot (export / import / wipe)
@@ -206,6 +207,7 @@ func (s *Server) setupRoutes() {
 	// 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/records", s.handleUDLRecordsPut).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")

internal/api/udl.go

@@ -42,6 +42,32 @@ type UDLPutResponse struct {
 	Accepted bool `json:"accepted"`
 }
 
+// UDLPutBatchRequest is the body of PUT /api/udl/records — many cells, one
+// request. Each entry is exactly a UDLPutRequest, so a client can build the
+// same signed record for either endpoint.
+type UDLPutBatchRequest struct {
+	Records []UDLPutRequest `json:"records"`
+}
+
+// UDLPutBatchResult is one cell's outcome. Accepted=false with an empty Error
+// means the write was stale (the single-record endpoint's 409), which in a
+// batch is an ordinary outcome rather than a failure.
+type UDLPutBatchResult struct {
+	Key      string `json:"key"`
+	Accepted bool   `json:"accepted"`
+	Error    string `json:"error,omitempty"`
+}
+
+// UDLPutBatchResponse mirrors the request order index-for-index.
+type UDLPutBatchResponse struct {
+	Results []UDLPutBatchResult `json:"results"`
+}
+
+// maxUDLBatch bounds one batch. Each record is a Redis round trip while the
+// request is held open; sized to match the sign batch on the other side of the
+// same operation (see maxSignBatch).
+const maxUDLBatch = 1000
+
 // UDLRecordResponse is the response of GET /api/udl/record.
 type UDLRecordResponse struct {
 	Record  string `json:"record"`
@@ -144,6 +170,64 @@ func (s *Server) handleUDLRecordPut(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, UDLPutResponse{Accepted: true})
 }
 
+// handleUDLRecordsPut handles PUT /api/udl/records — the batch form of
+// handleUDLRecordPut.
+//
+// Exists for the bulk verbs above this layer ("mark this whole series seen"),
+// where one user action produces one cell per episode. Each cell still goes
+// through the same version-gated Lua upsert, so the CRDT semantics are
+// identical — this batches the transport only.
+//
+// **Not transactional, and deliberately not 409.** A per-record `accepted` flag
+// replaces the single-record conflict status: in a batch, one stale cell is a
+// normal outcome (another device already marked that episode) and must not fail
+// the 299 cells beside it. Callers read the array, not the status code.
+func (s *Server) handleUDLRecordsPut(w http.ResponseWriter, r *http.Request) {
+	if !s.storage.IsConnected() {
+		writeError(w, http.StatusServiceUnavailable, "storage not connected")
+		return
+	}
+	var req UDLPutBatchRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid JSON body")
+		return
+	}
+	if len(req.Records) == 0 {
+		writeError(w, http.StatusBadRequest, "records is required")
+		return
+	}
+	if len(req.Records) > maxUDLBatch {
+		writeError(w, http.StatusBadRequest, "too many records in one batch")
+		return
+	}
+	results := make([]UDLPutBatchResult, 0, len(req.Records))
+	for _, rec := range req.Records {
+		if rec.Uid == "" || rec.Cid == "" || rec.Key == "" || rec.Record == "" {
+			results = append(results, UDLPutBatchResult{
+				Key:   rec.Key,
+				Error: "uid, cid, key and record are required",
+			})
+			continue
+		}
+		publicValue := ""
+		hasValue := false
+		if rec.Value != nil {
+			publicValue = string(*rec.Value)
+			hasValue = true
+		}
+		accepted, err := s.storage.UDLUpsertIfNewer(rec.Uid, rec.Cid, rec.Key, rec.Version, rec.Ts, rec.Record, publicValue, hasValue, rec.Tombstone)
+		if err != nil {
+			// A storage error is per-record here rather than fatal for the whole
+			// batch: the cells already written stay written, and reporting which
+			// ones failed lets the caller retry exactly those.
+			results = append(results, UDLPutBatchResult{Key: rec.Key, Error: err.Error()})
+			continue
+		}
+		results = append(results, UDLPutBatchResult{Key: rec.Key, Accepted: accepted})
+	}
+	writeJSON(w, http.StatusOK, UDLPutBatchResponse{Results: results})
+}
+
 // 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) {

internal/api/udl_test.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 
 	"github.com/alicebob/miniredis/v2"
@@ -128,3 +129,98 @@ func TestUDLRecordGet_MissingCellHasNoValue(t *testing.T) {
 		t.Fatalf("missing cell must omit `value`, got %s", body["value"])
 	}
 }
+
+// putRecords drives handleUDLRecordsPut with a raw JSON body and decodes the
+// per-record results.
+func putRecords(t *testing.T, s *Server, body string) (int, UDLPutBatchResponse) {
+	t.Helper()
+
+	req := httptest.NewRequest(http.MethodPut, "/api/udl/records", strings.NewReader(body))
+	rr := httptest.NewRecorder()
+	s.handleUDLRecordsPut(rr, req)
+
+	var out UDLPutBatchResponse
+	if rr.Code == http.StatusOK {
+		if err := json.Unmarshal(rr.Body.Bytes(), &out); err != nil {
+			t.Fatalf("decode response %q: %v", rr.Body.String(), err)
+		}
+	}
+	return rr.Code, out
+}
+
+// **The load-bearing property of the batch endpoint: one stale cell does not
+// fail the batch.**
+//
+// The single-record PUT answers a stale version with 409, which is right when
+// the caller wrote one thing. In a batch it would be wrong: bulk verbs ("mark
+// this whole series seen") routinely include episodes another device already
+// marked, and 409-ing there would abandon every cell after it. So staleness is
+// reported per record as accepted:false, with 200 on the request itself.
+func TestUDLRecordsPut_StaleCellDoesNotFailTheBatch(t *testing.T) {
+	s := newUDLTestServer(t)
+
+	const uid = "zCarol"
+	const cid = "bagdsaaaseries"
+	// Seed s01e01 at version 4 — the batch below tries to write it at 2.
+	if ok, err := s.storage.UDLUpsertIfNewer(uid, cid, "seen:s01e01", 4, 100, "b2xk", "", false, false); err != nil || !ok {
+		t.Fatalf("seed: ok=%v err=%v", ok, err)
+	}
+
+	code, out := putRecords(t, s, `{"records":[
+		{"uid":"`+uid+`","cid":"`+cid+`","key":"seen:s01e01","version":2,"ts":200,"record":"c3RhbGU"},
+		{"uid":"`+uid+`","cid":"`+cid+`","key":"seen:s01e02","version":1,"ts":200,"record":"ZnJlc2g"}
+	]}`)
+
+	if code != http.StatusOK {
+		t.Fatalf("status = %d, want 200 — a stale cell is an outcome, not an error", code)
+	}
+	if len(out.Results) != 2 {
+		t.Fatalf("results = %d, want 2 (one per record, in order)", len(out.Results))
+	}
+	if out.Results[0].Accepted || out.Results[0].Error != "" {
+		t.Fatalf("stale cell = %+v, want accepted:false with no error", out.Results[0])
+	}
+	if !out.Results[1].Accepted {
+		t.Fatalf("fresh cell = %+v, want accepted:true", out.Results[1])
+	}
+
+	// The fresh cell really landed, and the stale one did not overwrite.
+	if _, body := getRecord(t, s, uid, cid, "seen:s01e02"); string(body["exists"]) != "true" {
+		t.Fatalf("s01e02 was not written")
+	}
+	if _, body := getRecord(t, s, uid, cid, "seen:s01e01"); string(body["version"]) != "4" {
+		t.Fatalf("s01e01 version = %s, want 4 (the stale write must not apply)", body["version"])
+	}
+}
+
+// A malformed record is reported in its own slot rather than rejecting the
+// request, for the same reason: the caller retries that cell, not the batch.
+func TestUDLRecordsPut_ReportsBadRecordPerSlot(t *testing.T) {
+	s := newUDLTestServer(t)
+
+	code, out := putRecords(t, s, `{"records":[
+		{"uid":"zDan","cid":"bagdsaaa","key":"","version":1,"ts":1,"record":"eA"},
+		{"uid":"zDan","cid":"bagdsaaa","key":"seen:s01e01","version":1,"ts":1,"record":"eA"}
+	]}`)
+
+	if code != http.StatusOK {
+		t.Fatalf("status = %d, want 200", code)
+	}
+	if out.Results[0].Error == "" {
+		t.Fatalf("record with no key must report an error, got %+v", out.Results[0])
+	}
+	if !out.Results[1].Accepted {
+		t.Fatalf("the valid record beside it must still land, got %+v", out.Results[1])
+	}
+}
+
+// An empty batch is a client bug, not a no-op: it almost always means the
+// caller built its record list wrong, and answering 200 with an empty array
+// would hide that.
+func TestUDLRecordsPut_RejectsEmptyBatch(t *testing.T) {
+	s := newUDLTestServer(t)
+
+	if code, _ := putRecords(t, s, `{"records":[]}`); code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want 400", code)
+	}
+}