add identity management

dashboard/src/pages/Identity.tsx

@@ -1,21 +1,43 @@
-import { useEffect, useState } from 'react';
+import { useEffect, useRef, useState } from 'react';
 
-// Status returned by GET /api/identity. Private key is never present here —
+// Admin console for meta-core's account keystore.
-// it is shown exactly once at generation time (see GeneratedKeyPanel).
+//
-interface IdentityStatus {
+// meta-core stores one secp256k1 keypair per file under
-  hasIdentity: boolean;
+// {META_CORE_PATH}/identity/accounts/<uid>.json and has NO notion of a current
-  uid?: string;
+// user — which account is "you" is decided by whichever client signs in
-  curve?: string;
+// (meta-watch). So this page lists accounts; it never speaks of an active one.
-  createdAt?: number;
+//
-}
+// The private key is returned by exactly two calls: POST /api/identity/generate
+// (once, at creation) and POST /api/identity/reveal (on demand, for an account
+// that already exists). Both land in the same exclusive SecretKeyPanel.
 
-interface GeneratedKey {
+interface Account {
   uid: string;
   curve: string;
   createdAt: number;
-  privateKeyHex: string;
 }
 
+// A secret on screen. Carries no createdAt: /api/identity/reveal answers with
+// {uid, curve, privateKeyHex} only.
+interface SecretKeyView {
+  uid: string;
+  privateKeyHex: string;
+  kind: 'generated' | 'revealed';
+}
+
+// Which row has an open confirmation, and for what. One discriminated value
+// rather than a pair of ids, so at most one confirm panel can ever be open.
+interface Pending {
+  uid: string;
+  kind: 'remove' | 'reveal';
+}
+
+// meta-watch's sentinel cell for profile records: display name and avatar are
+// about the *user*, not a content file, so they live at (uid, "self", ...).
+// meta-core stores them opaquely and knows nothing of the convention.
+const PROFILE_CID = 'self';
+const PROFILE_NAME_KEY = 'profile:name';
+
 const cardStyle: React.CSSProperties = {
   background: '#16213e',
   borderRadius: '8px',
@@ -59,6 +81,21 @@ const monoStyle: React.CSSProperties = {
   borderRadius: '4px',
 };
 
+const dangerPanelStyle: React.CSSProperties = {
+  background: '#1f0a0a',
+  border: '1px solid #7f1d1d',
+  padding: '0.75rem 1rem',
+  borderRadius: '4px',
+  marginTop: '0.75rem',
+};
+
+const errorTextStyle: React.CSSProperties = {
+  color: '#f87171',
+  background: '#1f0a0a',
+  padding: '0.5rem 0.75rem',
+  borderRadius: '4px',
+};
+
 function formatCreatedAt(unixSec?: number): string {
   if (!unixSec) return '';
   try {
@@ -79,25 +116,76 @@ async function errorMessage(res: Response, fallback: string): Promise<string> {
   return `${fallback} (HTTP ${res.status})`;
 }
 
-function GeneratedKeyPanel({ data, onDismiss }: { data: GeneratedKey; onDismiss: () => void }) {
+// Copy text, falling back to selecting the element that displays it.
-  const [copied, setCopied] = useState(false);
+//
-  const copy = async () => {
+// The fallback is load-bearing, not politeness: the dashboard is served over
-    try {
+// plain HTTP, where navigator.clipboard is undefined — without it the copy
-      await navigator.clipboard.writeText(data.privateKeyHex);
+// buttons do nothing at all on a real deployment.
-      setCopied(true);
+async function copyToClipboard(text: string, fallbackElementId: string): Promise<void> {
-      setTimeout(() => setCopied(false), 2000);
+  try {
-    } catch {
+    await navigator.clipboard.writeText(text);
-      // Clipboard API blocked (non-HTTPS context, etc.) — fall back to selection.
+  } catch {
-      const sel = window.getSelection();
+    const sel = window.getSelection();
+    const el = document.getElementById(fallbackElementId);
+    if (sel && el) {
       const range = document.createRange();
-      const el = document.getElementById('mm-privkey-box');
+      range.selectNodeContents(el);
-      if (sel && el) {
+      sel.removeAllRanges();
-        range.selectNodeContents(el);
+      sel.addRange(range);
-        sel.removeAllRanges();
-        sel.addRange(range);
-      }
     }
+  }
+}
+
+async function fetchAccounts(): Promise<Account[]> {
+  const res = await fetch('/api/identity/accounts');
+  if (!res.ok) throw new Error(await errorMessage(res, 'Failed to load accounts'));
+  const data = (await res.json()) as { accounts?: Account[] };
+  // Backend order is CreatedAt ascending then uid (identity.List) — stable, so
+  // render it as given rather than re-sorting here.
+  return data.accounts ?? [];
+}
+
+// Resolve an account's display name from the User Data Layer, or null.
+//
+// Never throws and never reports an error upward: a name is decoration on top
+// of the uid, and the account list must render identically whether Redis is up
+// (identity itself is file-backed, the UDL is not). The name is also *unsigned*
+// — meta-core stores the plaintext beside the signed record and verifies
+// nothing — so the uid stays on screen as the real identity, and the length cap
+// bounds what a hostile writer can do to the layout.
+async function fetchDisplayName(uid: string): Promise<string | null> {
+  const url =
+    `/api/udl/record?uid=${encodeURIComponent(uid)}` +
+    `&cid=${encodeURIComponent(PROFILE_CID)}` +
+    `&key=${encodeURIComponent(PROFILE_NAME_KEY)}`;
+  try {
+    const res = await fetch(url);
+    if (!res.ok) return null; // 503 when storage is down, 500, …
+    const data = (await res.json()) as { exists?: boolean; value?: unknown };
+    // Key off `value`, not `exists`: clearing a name writes a tombstone, which
+    // leaves the cell existing with no plaintext. Private-tier cells look the
+    // same, which is exactly right — there is nothing to show for either.
+    if (typeof data.value !== 'string') return null;
+    const name = data.value.trim();
+    return name ? name.slice(0, 64) : null;
+  } catch {
+    return null;
+  }
+}
+
+// The one place a private key is ever displayed. Rendered exclusively (the
+// account list is hidden while it is open) so the "save it now" contract of a
+// freshly generated key is not competing with anything, and so the hardcoded
+// element id below stays unique.
+function SecretKeyPanel({ data, onDismiss }: { data: SecretKeyView; onDismiss: () => void }) {
+  const [copied, setCopied] = useState(false);
+
+  const copy = async () => {
+    await copyToClipboard(data.privateKeyHex, 'mm-privkey-box');
+    setCopied(true);
+    setTimeout(() => setCopied(false), 2000);
   };
+
   return (
     <div style={{ ...cardStyle, border: '2px solid #dc2626' }}>
       <div
@@ -109,9 +197,19 @@ function GeneratedKeyPanel({ data, onDismiss }: { data: GeneratedKey; onDismiss:
           marginBottom: '1rem',
         }}
       >
-        <strong>This private key is shown ONCE.</strong> Save it now in a password manager.
+        {data.kind === 'generated' ? (
-        It will never be displayed again — there is no recovery path. Losing it means losing access
+          <>
-        to every record signed under this identity.
+            <strong>This private key is shown ONCE.</strong> Save it now in a password manager.
+            You can ask for it again later with <em>Reveal private key</em>, but only for as long
+            as this account exists on this node — delete it and the key is gone for good.
+          </>
+        ) : (
+          <>
+            <strong>This is the live private key for an existing account.</strong> Anyone holding
+            it can sign as this uid forever, on any machine. There is no revocation and no
+            rotation. Do not paste it into chat, tickets or screenshots.
+          </>
+        )}
       </div>
       <p style={{ color: '#9ca3af', margin: '0 0 0.5rem' }}>uid (public):</p>
       <div style={monoStyle}>{data.uid}</div>
@@ -124,36 +222,40 @@ function GeneratedKeyPanel({ data, onDismiss }: { data: GeneratedKey; onDismiss:
           {copied ? 'Copied' : 'Copy private key'}
         </button>
         <button style={dangerButtonStyle} onClick={onDismiss}>
-          I have saved it
+          {data.kind === 'generated' ? 'I have saved it' : 'Done — hide it'}
         </button>
       </div>
     </div>
   );
 }
 
-function NoIdentityPanel({
+// Create / import. Always available, not just on an empty keystore: this is an
+// admin console, so adding an account to a populated node has to be possible.
+function NewAccountPanel({
   onGenerated,
   onImported,
 }: {
-  onGenerated: (g: GeneratedKey) => void;
+  onGenerated: (s: SecretKeyView) => void;
   onImported: () => void;
 }) {
   const [busy, setBusy] = useState(false);
   const [importing, setImporting] = useState(false);
   const [importHex, setImportHex] = useState('');
   const [error, setError] = useState<string | null>(null);
+  const [notice, setNotice] = useState<string | null>(null);
 
   const generate = async () => {
     setBusy(true);
     setError(null);
+    setNotice(null);
     try {
       const res = await fetch('/api/identity/generate', { method: 'POST' });
       if (!res.ok) {
         setError(await errorMessage(res, 'Generate failed'));
         return;
       }
-      const data = (await res.json()) as GeneratedKey;
+      const data = (await res.json()) as { uid: string; privateKeyHex: string };
-      onGenerated(data);
+      onGenerated({ uid: data.uid, privateKeyHex: data.privateKeyHex, kind: 'generated' });
     } catch (e) {
       setError(`Generate failed: ${e instanceof Error ? e.message : String(e)}`);
     } finally {
@@ -165,6 +267,7 @@ function NoIdentityPanel({
     e.preventDefault();
     setBusy(true);
     setError(null);
+    setNotice(null);
     try {
       const res = await fetch('/api/identity/import', {
         method: 'POST',
@@ -175,6 +278,15 @@ function NoIdentityPanel({
         setError(await errorMessage(res, 'Import failed'));
         return;
       }
+      // Import is idempotent — presenting the key IS the proof of ownership, so
+      // a key this node already holds is a success, not a conflict. `created`
+      // only decides the wording.
+      const data = (await res.json()) as { created?: boolean };
+      setNotice(
+        data.created
+          ? 'Imported — a new account was added to this keystore.'
+          : 'That key was already in this keystore; nothing changed.'
+      );
       setImportHex('');
       setImporting(false);
       onImported();
@@ -187,15 +299,16 @@ function NoIdentityPanel({
 
   return (
     <div style={cardStyle}>
-      <h3 style={{ marginTop: 0 }}>No identity yet</h3>
+      <h3 style={{ marginTop: 0 }}>Add an account</h3>
-      <p style={{ color: '#9ca3af' }}>
+      <p style={{ color: '#9ca3af', marginTop: 0 }}>
-        The User Data Layer needs a signing key for your ratings, watch progress, and "liked" flags
+        Generating mints a fresh secp256k1 keypair here. Importing adds a key created elsewhere —
-        to be attributed to you across devices. Nothing happens automatically — generate one below,
+        it is also how an existing account is moved onto this node, so importing a key the
-        or import a key you already have.
+        keystore already holds succeeds and changes nothing.
       </p>
-      {error && (
+      {error && <p style={errorTextStyle}>{error}</p>}
-        <p style={{ color: '#f87171', background: '#1f0a0a', padding: '0.5rem 0.75rem', borderRadius: '4px' }}>
+      {notice && (
-          {error}
+        <p style={{ color: '#9ca3af', background: '#0a0a18', padding: '0.5rem 0.75rem', borderRadius: '4px' }}>
+          {notice}
         </p>
       )}
       <div>
@@ -235,71 +348,125 @@ function NoIdentityPanel({
   );
 }
 
-function HasIdentityPanel({ status, onReplace }: { status: IdentityStatus; onReplace: () => void }) {
+function AccountCard({
-  const [confirming, setConfirming] = useState(false);
+  account,
-  const [busy, setBusy] = useState(false);
+  name,
-  const [error, setError] = useState<string | null>(null);
+  busy,
+  disabled,
+  pending,
+  error,
+  onAsk,
+  onCancel,
+  onReveal,
+  onRemove,
+}: {
+  account: Account;
+  name?: string;
+  busy: boolean;
+  disabled: boolean;
+  pending: Pending | null;
+  error: string | null;
+  onAsk: (p: Pending) => void;
+  onCancel: () => void;
+  onReveal: (uid: string) => void;
+  onRemove: (uid: string) => void;
+}) {
+  const [copied, setCopied] = useState(false);
+  const uidBoxId = `mm-uid-${account.uid}`;
+  const label = name ?? account.uid;
 
-  const remove = async () => {
+  const copyUid = async () => {
-    setBusy(true);
+    await copyToClipboard(account.uid, uidBoxId);
-    setError(null);
+    setCopied(true);
-    try {
+    setTimeout(() => setCopied(false), 2000);
-      const res = await fetch('/api/identity?confirm=true', { method: 'DELETE' });
-      if (!res.ok && res.status !== 204) {
-        setError(await errorMessage(res, 'Delete failed'));
-        return;
-      }
-      onReplace();
-    } catch (e) {
-      setError(`Delete failed: ${e instanceof Error ? e.message : String(e)}`);
-    } finally {
-      setBusy(false);
-      setConfirming(false);
-    }
   };
 
   return (
     <div style={cardStyle}>
-      <h3 style={{ marginTop: 0 }}>Active identity</h3>
+      <div style={{ display: 'flex', justifyContent: 'space-between', alignItems: 'flex-start', gap: '1rem' }}>
-      <p style={{ color: '#9ca3af', margin: '0 0 0.5rem' }}>uid:</p>
+        <div style={{ minWidth: 0, flex: 1 }}>
-      <div style={monoStyle}>{status.uid}</div>
+          <h3 style={{ margin: '0 0 0.5rem', color: name ? '#fff' : '#888' }}>
-      <p style={{ color: '#888', marginTop: '1rem' }}>
+            {name ?? 'Unnamed account'}
-        Curve: {status.curve} · Created: {formatCreatedAt(status.createdAt)}
+          </h3>
-      </p>
+          <div id={uidBoxId} style={monoStyle}>
-      <p style={{ color: '#9ca3af', fontSize: '0.85rem' }}>
+            {account.uid}
-        The private key lives at <code>/meta-core/identity/identity.json</code> on this volume and
+          </div>
-        is never returned through the API after generation. Other services on this stack (e.g.
+          <p style={{ color: '#888', margin: '0.75rem 0 0', fontSize: '0.85rem' }}>
-        meta-watch) sign records by calling <code>POST /api/identity/sign</code>.
+            Curve: {account.curve} · Created: {formatCreatedAt(account.createdAt)}
-      </p>
-      {error && (
-        <p style={{ color: '#f87171', background: '#1f0a0a', padding: '0.5rem 0.75rem', borderRadius: '4px' }}>
-          {error}
-        </p>
-      )}
-      {!confirming ? (
-        <button style={dangerButtonStyle} onClick={() => setConfirming(true)} disabled={busy}>
-          Replace identity
-        </button>
-      ) : (
-        <div
-          style={{
-            background: '#1f0a0a',
-            border: '1px solid #7f1d1d',
-            padding: '0.75rem 1rem',
-            borderRadius: '4px',
-            marginTop: '0.5rem',
-          }}
-        >
-          <p style={{ color: '#fecaca', marginTop: 0 }}>
-            Replacing the identity deletes the current keypair from disk. Records signed by the
-            current uid will keep verifying for anyone who has them cached, but this node will no
-            longer be able to write under that uid. <strong>Make sure you have backed up the key
-            first.</strong>
           </p>
-          <button style={dangerButtonStyle} onClick={remove} disabled={busy}>
+        </div>
-            {busy ? 'Deleting…' : 'Yes, delete identity'}
+        <div style={{ display: 'flex', flexDirection: 'column', gap: '0.5rem', flexShrink: 0 }}>
+          <button style={{ ...buttonStyle, marginRight: 0 }} onClick={copyUid} disabled={disabled}>
+            {copied ? 'Copied' : 'Copy uid'}
           </button>
-          <button style={buttonStyle} onClick={() => setConfirming(false)} disabled={busy}>
+          <button
+            style={{ ...dangerButtonStyle, marginRight: 0 }}
+            onClick={() => onAsk({ uid: account.uid, kind: 'reveal' })}
+            disabled={disabled}
+          >
+            Reveal private key
+          </button>
+          <button
+            style={{ ...dangerButtonStyle, marginRight: 0 }}
+            onClick={() => onAsk({ uid: account.uid, kind: 'remove' })}
+            disabled={disabled}
+          >
+            Remove
+          </button>
+        </div>
+      </div>
+
+      {error && <p style={{ ...errorTextStyle, marginBottom: 0 }}>{error}</p>}
+
+      {pending?.uid === account.uid && pending.kind === 'reveal' && (
+        <div style={dangerPanelStyle}>
+          <p style={{ color: '#fecaca', marginTop: 0 }}>
+            Show the private key for <strong>{label}</strong>?
+          </p>
+          <p style={{ color: '#fecaca', marginTop: 0 }}>
+            The key <em>is</em> the account. Anyone who copies it can sign records as this uid on
+            any machine, forever — there is no way to revoke or rotate it afterwards. meta-core
+            will hand it over without asking who you are; see the note at the top of this page.
+          </p>
+          <button style={dangerButtonStyle} onClick={() => onReveal(account.uid)} disabled={busy}>
+            {busy ? 'Revealing…' : 'Yes, show the private key'}
+          </button>
+          <button style={buttonStyle} onClick={onCancel} disabled={busy}>
+            Cancel
+          </button>
+        </div>
+      )}
+
+      {pending?.uid === account.uid && pending.kind === 'remove' && (
+        <div style={dangerPanelStyle}>
+          <p style={{ color: '#fecaca', marginTop: 0 }}>
+            Remove <strong>{label}</strong> from this node?
+          </p>
+          <ul style={{ color: '#fecaca', paddingLeft: '1.2rem', margin: '0 0 0.75rem' }}>
+            <li>
+              <strong>The key is gone for good.</strong> The uid <em>is</em> the public key, and
+              the private half exists only in this node's keystore. If nobody saved it, nothing —
+              not you, not meta-core — can ever sign as this uid again. Reveal and copy it first
+              if there is any doubt.
+            </li>
+            <li>
+              <strong>Nothing is retracted.</strong> Records already signed under this uid keep
+              verifying on every peer that replicated them. Deleting a key is not a recall.
+            </li>
+            <li>
+              <strong>Its data stays, but goes read-only.</strong> The account's User Data Layer
+              cells — likes, ratings, watch progress — survive in Redis, and no newer version of
+              any of them can ever be written.
+            </li>
+            <li>
+              <strong>This signs nobody out elsewhere.</strong> Any device that holds this key
+              still holds it, and can import it back here at any time.
+            </li>
+          </ul>
+          <button style={dangerButtonStyle} onClick={() => onRemove(account.uid)} disabled={busy}>
+            {busy ? 'Removing…' : 'Yes, remove this account'}
+          </button>
+          <button style={buttonStyle} onClick={onCancel} disabled={busy}>
             Cancel
           </button>
         </div>
@@ -309,66 +476,183 @@ function HasIdentityPanel({ status, onReplace }: { status: IdentityStatus; onRep
 }
 
 export default function Identity() {
-  const [status, setStatus] = useState<IdentityStatus | null>(null);
+  const [accounts, setAccounts] = useState<Account[] | null>(null);
-  const [generated, setGenerated] = useState<GeneratedKey | null>(null);
+  const [names, setNames] = useState<Record<string, string>>({});
   const [loading, setLoading] = useState(true);
   const [fetchError, setFetchError] = useState<string | null>(null);
+  const [secret, setSecret] = useState<SecretKeyView | null>(null);
+  const [pending, setPending] = useState<Pending | null>(null);
+  const [busyUid, setBusyUid] = useState<string | null>(null);
+  const [rowError, setRowError] = useState<{ uid: string; message: string } | null>(null);
+
+  // Bumped per refresh so a slow response from a superseded load cannot
+  // overwrite a newer one. The page does not poll (unlike the rest of the
+  // dashboard) — a key list has no business changing under the cursor.
+  const reqGen = useRef(0);
 
   const refresh = async () => {
+    const gen = ++reqGen.current;
     setLoading(true);
     setFetchError(null);
+    let list: Account[];
     try {
-      const res = await fetch('/api/identity');
+      list = await fetchAccounts();
-      if (!res.ok) {
-        setFetchError(await errorMessage(res, 'Failed to load identity status'));
-        return;
-      }
-      setStatus((await res.json()) as IdentityStatus);
     } catch (e) {
-      setFetchError(`Failed to load identity status: ${e instanceof Error ? e.message : String(e)}`);
+      if (gen !== reqGen.current) return;
-    } finally {
+      setFetchError(e instanceof Error ? e.message : String(e));
       setLoading(false);
+      return;
     }
+    if (gen !== reqGen.current) return;
+    // Render the list before resolving names, so a down Redis delays nothing.
+    setAccounts(list);
+    setLoading(false);
+
+    const settled = await Promise.allSettled(list.map((a) => fetchDisplayName(a.uid)));
+    if (gen !== reqGen.current) return;
+    const next: Record<string, string> = {};
+    settled.forEach((r, i) => {
+      if (r.status === 'fulfilled' && r.value) next[list[i].uid] = r.value;
+    });
+    // Replace rather than merge, so a removed account's name goes with it.
+    setNames(next);
   };
 
   useEffect(() => {
     refresh();
   }, []);
 
+  const reveal = async (uid: string) => {
+    setBusyUid(uid);
+    setRowError(null);
+    try {
+      // POST with a body rather than ?uid=, so the target stays out of the
+      // proxy's access log. The handler accepts either.
+      const res = await fetch('/api/identity/reveal', {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json' },
+        body: JSON.stringify({ confirm: true, uid }),
+      });
+      if (!res.ok) {
+        setRowError({ uid, message: await errorMessage(res, 'Reveal failed') });
+        return;
+      }
+      const data = (await res.json()) as { uid: string; privateKeyHex: string };
+      setPending(null);
+      setSecret({ uid: data.uid, privateKeyHex: data.privateKeyHex, kind: 'revealed' });
+    } catch (e) {
+      setRowError({ uid, message: `Reveal failed: ${e instanceof Error ? e.message : String(e)}` });
+    } finally {
+      setBusyUid(null);
+    }
+  };
+
+  const remove = async (uid: string) => {
+    setBusyUid(uid);
+    setRowError(null);
+    try {
+      // The uid is mandatory: with more than one account the backend refuses an
+      // unqualified delete (400 uid_required) rather than guess.
+      const res = await fetch(`/api/identity?uid=${encodeURIComponent(uid)}&confirm=true`, {
+        method: 'DELETE',
+      });
+      if (!res.ok) {
+        setRowError({ uid, message: await errorMessage(res, 'Remove failed') });
+        return;
+      }
+      setPending(null);
+      await refresh();
+    } catch (e) {
+      setRowError({ uid, message: `Remove failed: ${e instanceof Error ? e.message : String(e)}` });
+    } finally {
+      setBusyUid(null);
+    }
+  };
+
+  const dismissSecret = () => {
+    setSecret(null);
+    refresh();
+  };
+
+  const showList = !loading && !fetchError && !secret;
+
   return (
     <div style={{ padding: '2rem', maxWidth: '900px', margin: '0 auto' }}>
-      <h1>Identity</h1>
+      <div style={{ display: 'flex', justifyContent: 'space-between', alignItems: 'center' }}>
+        <h1 style={{ marginBottom: 0 }}>Identity</h1>
+        <button style={{ ...buttonStyle, marginRight: 0 }} onClick={refresh} disabled={loading}>
+          {loading ? 'Loading…' : 'Refresh'}
+        </button>
+      </div>
       <p style={{ color: '#9ca3af' }}>
-        Signing key for the User Data Layer (per-user ratings, progress, "liked" flags). One
+        Every signing key in this node's keystore. meta-core has no "current user" — it stores
-        identity per node. The same key on another device makes that device count as "you".
+        accounts, and which one is <em>you</em> is decided by whichever client signs in (e.g.
+        meta-watch). Keys live one JSON file per account under{' '}
+        <code>{'{META_CORE_PATH}'}/identity/accounts/&lt;uid&gt;.json</code>, and are used to sign
+        User Data Layer records (ratings, watch progress, "liked" flags).
       </p>
 
+      <div style={{ ...dangerPanelStyle, marginTop: 0, marginBottom: '1rem' }}>
+        <p style={{ color: '#fecaca', margin: 0 }}>
+          <strong>Admin page — no authorisation happens here.</strong> <code>/api/identity/*</code>{' '}
+          asks no questions: anything that can reach this API can list, create, reveal or delete{' '}
+          <em>any</em> account's private key. The only access control is the perimeter in front of
+          this port. Client-facing screens (meta-watch) authorise reveal and delete against the
+          caller's own session and expose no delete at all; this page has neither check. Treat
+          reaching it as equivalent to holding every key on this node.
+        </p>
+      </div>
+
       {loading && <div style={cardStyle}>Loading…</div>}
+
       {fetchError && (
         <div style={{ ...cardStyle, border: '1px solid #7f1d1d' }}>
-          <p style={{ color: '#f87171', margin: 0 }}>{fetchError}</p>
+          <p style={{ color: '#f87171', margin: '0 0 1rem' }}>{fetchError}</p>
           <button style={buttonStyle} onClick={refresh}>
             Retry
           </button>
         </div>
       )}
 
-      {!loading && !fetchError && generated && (
+      {!loading && !fetchError && secret && <SecretKeyPanel data={secret} onDismiss={dismissSecret} />}
-        <GeneratedKeyPanel
-          data={generated}
-          onDismiss={() => {
-            setGenerated(null);
-            refresh();
-          }}
-        />
-      )}
 
-      {!loading && !fetchError && !generated && status && !status.hasIdentity && (
+      {showList && (
-        <NoIdentityPanel onGenerated={(g) => setGenerated(g)} onImported={refresh} />
+        <>
-      )}
+          <NewAccountPanel onGenerated={(s) => setSecret(s)} onImported={refresh} />
 
-      {!loading && !fetchError && !generated && status && status.hasIdentity && (
+          <h2 style={{ fontSize: '1.1rem', color: '#9ca3af', margin: '1.5rem 0 1rem' }}>
-        <HasIdentityPanel status={status} onReplace={refresh} />
+            Accounts ({accounts?.length ?? 0})
+          </h2>
+
+          {accounts && accounts.length === 0 && (
+            <div style={cardStyle}>
+              <p style={{ color: '#9ca3af', margin: 0 }}>
+                No accounts in this keystore. Accounts are normally created by a client when
+                someone signs up; use <em>Add an account</em> above if you are bootstrapping this
+                node.
+              </p>
+            </div>
+          )}
+
+          {accounts?.map((a) => (
+            <AccountCard
+              key={a.uid}
+              account={a}
+              name={names[a.uid]}
+              busy={busyUid === a.uid}
+              disabled={busyUid !== null}
+              pending={pending}
+              error={rowError?.uid === a.uid ? rowError.message : null}
+              onAsk={(p) => {
+                setRowError(null);
+                setPending(p);
+              }}
+              onCancel={() => setPending(null)}
+              onReveal={reveal}
+              onRemove={remove}
+            />
+          ))}
+        </>
       )}
     </div>
   );

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")
@@ -351,6 +353,21 @@ func (s *Server) Start() error {
 
 			// Republish events for all existing metadata
 			go func() {
+				// Build any missing per-record field indexes first, in ONE
+				// keyspace pass. The republish below reads every record, and
+				// an un-indexed record read costs a full-keyspace SCAN — on a
+				// real store that walk warmed at ~1.3 records/s (a ~2.6h tail,
+				// paid again on every restart). Warming first turns the whole
+				// republish into index-served reads. Idempotent, so a
+				// fully-indexed store just pays the single scan.
+				// See storage/field_index.go.
+				started := time.Now()
+				if n, err := s.storage.WarmFieldIndexes(); err != nil {
+					log.Printf("[API] Warning: field index warm failed: %v (reads fall back to SCAN)", err)
+				} else if n > 0 {
+					log.Printf("[API] Warmed %d field indexes in %s", n, time.Since(started).Round(time.Millisecond))
+				}
+
 				count, err := s.RepublishMetadata()
 				if err != nil {
 					log.Printf("[API] Warning: failed to republish metadata on startup: %v", err)

internal/api/udl.go

@@ -27,6 +27,14 @@ type UDLPutRequest struct {
 	Ts      int64            `json:"ts"`
 	Record  string           `json:"record"` // base64 CBOR, opaque + signed
 	Value   *json.RawMessage `json:"value,omitempty"`
+	// Tombstone marks this write as a deletion of (uid, cid, key). The record
+	// itself is opaque here — meta-core never decodes the CBOR, and a private-
+	// tier value and a tombstone are equally "no plaintext value" — so the
+	// writer has to say. Without it the cid is never removed from
+	// udl:idx:user:<uid>:key:<key>, which then grows for the life of the
+	// account and is returned in full on every My List / Continue Watching
+	// read. Omitted by older writers, which keeps the previous behaviour.
+	Tombstone bool `json:"tombstone,omitempty"`
 }
 
 // UDLPutResponse is the response of PUT /api/udl/record.
@@ -34,12 +42,50 @@ 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"`
 	Version int64  `json:"version"`
 	Ts      int64  `json:"ts"`
 	Exists  bool   `json:"exists"`
+	// Value is the plaintext JSON scalar stored beside the opaque record, and
+	// is present only for PUBLIC-tier keys (like, rating, profile:name,
+	// profile:avatar). json.RawMessage rather than string because the field is
+	// generic over keys — a bool for `like`, a number for `rating`, a string
+	// for a profile name (see aggregatePublicValues) — and because it is the
+	// exact type UDLPutRequest.Value uses, so a value written through PUT reads
+	// back byte-identical.
+	//
+	// Absent, never null, when there is no plaintext: private-tier cells,
+	// tombstones and never-written cells must not read as an empty public
+	// value. Readers key off presence, not `exists`.
+	Value *json.RawMessage `json:"value,omitempty"`
 }
 
 // UDLAggregate mirrors meta-watch's Aggregate wire shape exactly (snake_case
@@ -71,12 +117,20 @@ func (s *Server) handleUDLRecordGet(w http.ResponseWriter, r *http.Request) {
 		writeError(w, http.StatusInternalServerError, err.Error())
 		return
 	}
-	writeJSON(w, http.StatusOK, UDLRecordResponse{
+	resp := UDLRecordResponse{
 		Record:  cell.Record,
 		Version: cell.Version,
 		Ts:      cell.Ts,
 		Exists:  exists,
-	})
+	}
+	// json.Valid guards the encoder: a malformed `value` hash field (hand-edited
+	// Redis, a future writer bug) would otherwise fail mid-encode and emit a
+	// truncated body instead of a readable error.
+	if exists && cell.HasValue && json.Valid([]byte(cell.Value)) {
+		raw := json.RawMessage(cell.Value)
+		resp.Value = &raw
+	}
+	writeJSON(w, http.StatusOK, resp)
 }
 
 // handleUDLRecordPut handles PUT /api/udl/record. Version-gated: a stale write
@@ -104,7 +158,7 @@ func (s *Server) handleUDLRecordPut(w http.ResponseWriter, r *http.Request) {
 		hasValue = true
 	}
 
-	accepted, err := s.storage.UDLUpsertIfNewer(req.Uid, req.Cid, req.Key, req.Version, req.Ts, req.Record, publicValue, hasValue)
+	accepted, err := s.storage.UDLUpsertIfNewer(req.Uid, req.Cid, req.Key, req.Version, req.Ts, req.Record, publicValue, hasValue, req.Tombstone)
 	if err != nil {
 		writeError(w, http.StatusInternalServerError, err.Error())
 		return
@@ -116,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

@@ -0,0 +1,226 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/metazla/meta-core/internal/storage"
+)
+
+// newUDLTestServer returns a Server wired to an in-memory Redis.
+//
+// The struct is built directly rather than through NewServer: NewServer boots
+// the file watcher, mounts manager, watchers poller and WebDAV handler, none of
+// which the UDL handlers touch — they use s.storage and nothing else.
+func newUDLTestServer(t *testing.T) *Server {
+	t.Helper()
+
+	mr := miniredis.RunT(t)
+	c := storage.NewClient("")
+	if err := c.Connect("redis://" + mr.Addr()); err != nil {
+		t.Fatalf("connect to miniredis: %v", err)
+	}
+	t.Cleanup(func() { _ = c.Close() })
+	return &Server{storage: c}
+}
+
+// getRecord drives handleUDLRecordGet and decodes the body into raw JSON
+// fields, so a test can assert on the *encoding* of `value` (is it a quoted
+// string? is the key there at all?) rather than on what a typed struct would
+// have coerced it into.
+func getRecord(t *testing.T, s *Server, uid, cid, key string) (int, map[string]json.RawMessage) {
+	t.Helper()
+
+	req := httptest.NewRequest(http.MethodGet, "/api/udl/record?uid="+uid+"&cid="+cid+"&key="+key, nil)
+	rr := httptest.NewRecorder()
+	s.handleUDLRecordGet(rr, req)
+
+	var body map[string]json.RawMessage
+	if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil {
+		t.Fatalf("decode response %q: %v", rr.Body.String(), err)
+	}
+	return rr.Code, body
+}
+
+// The plaintext value of a PUBLIC-tier cell must come back over the wire, and
+// come back *as it was written* — a JSON string stays a quoted string, a bool
+// stays a bool.
+//
+// Both halves matter. Without the value at all, the meta-core dashboard can
+// only ever show 44-character base58 uids: display names live in the User Data
+// Layer as the public cell (uid, "self", "profile:name"), and the plaintext
+// sits in Redis where the storage layer has always read it — the handler just
+// used to drop it. And the type has to stay json.RawMessage: this endpoint is
+// generic over keys, so a `*string` would mangle `like` (bool) and `rating`
+// (number) while looking correct for names.
+func TestUDLRecordGet_ReturnsPublicValue(t *testing.T) {
+	s := newUDLTestServer(t)
+
+	const uid = "zAlice"
+	if ok, err := s.storage.UDLUpsertIfNewer(uid, "self", "profile:name", 1, 100, "cmVj", `"Ada Lovelace"`, true, false); err != nil || !ok {
+		t.Fatalf("seed name: ok=%v err=%v", ok, err)
+	}
+	if ok, err := s.storage.UDLUpsertIfNewer(uid, "bagcsaaa", "like", 1, 100, "cmVj", `true`, true, false); err != nil || !ok {
+		t.Fatalf("seed like: ok=%v err=%v", ok, err)
+	}
+
+	code, body := getRecord(t, s, uid, "self", "profile:name")
+	if code != http.StatusOK {
+		t.Fatalf("status = %d, want 200", code)
+	}
+	if got := string(body["value"]); got != `"Ada Lovelace"` {
+		t.Fatalf("value = %s, want a quoted JSON string", got)
+	}
+	var name string
+	if err := json.Unmarshal(body["value"], &name); err != nil || name != "Ada Lovelace" {
+		t.Fatalf("value should unmarshal to the plain name, got %q err=%v", name, err)
+	}
+
+	// A boolean key survives unquoted — the assertion a *string field fails.
+	if _, body = getRecord(t, s, uid, "bagcsaaa", "like"); string(body["value"]) != "true" {
+		t.Fatalf("like value = %s, want the bare JSON literal true", body["value"])
+	}
+}
+
+// A cell with no plaintext must omit `value` entirely rather than send null.
+//
+// That is the shape of every private-tier key (seek, watched, card, localpath —
+// their real value rides encrypted inside the opaque record) and of every
+// tombstone. A reader keys off presence: "there is no plaintext here" and "the
+// plaintext is null" are different facts, and a name cleared by its owner
+// leaves the cell existing with no value, which must not read as a name.
+func TestUDLRecordGet_OmitsValueForPrivateCell(t *testing.T) {
+	s := newUDLTestServer(t)
+
+	const uid = "zBob"
+	if ok, err := s.storage.UDLUpsertIfNewer(uid, "bagcsaaa", "seek", 1, 100, "ZW5jcnlwdGVk", "", false, false); err != nil || !ok {
+		t.Fatalf("seed private cell: ok=%v err=%v", ok, err)
+	}
+
+	code, body := getRecord(t, s, uid, "bagcsaaa", "seek")
+	if code != http.StatusOK {
+		t.Fatalf("status = %d, want 200", code)
+	}
+	if _, ok := body["value"]; ok {
+		t.Fatalf("private cell must omit `value`, got %s", body["value"])
+	}
+	if string(body["exists"]) != "true" {
+		t.Fatalf("exists = %s, want true", body["exists"])
+	}
+}
+
+// A cell that was never written answers 200 with exists:false and no value —
+// not a 404, and not an empty-string value that would render as a blank name.
+func TestUDLRecordGet_MissingCellHasNoValue(t *testing.T) {
+	s := newUDLTestServer(t)
+
+	code, body := getRecord(t, s, "zNobody", "self", "profile:name")
+	if code != http.StatusOK {
+		t.Fatalf("status = %d, want 200", code)
+	}
+	if string(body["exists"]) != "false" {
+		t.Fatalf("exists = %s, want false", body["exists"])
+	}
+	if _, ok := body["value"]; ok {
+		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)
+	}
+}

internal/events/meta_publisher.go

@@ -9,6 +9,8 @@ import (
 	"time"
 
 	"github.com/redis/go-redis/v9"
+
+	"github.com/metazla/meta-core/internal/storage"
 )
 
 const (
@@ -126,6 +128,14 @@ func (p *MetaPublisher) publishEvent(channel, operation string) {
 	if strings.Contains(key, "__index__") || !strings.HasPrefix(key, "file:") {
 		return
 	}
+	// Skip the per-record field index. It is bookkeeping (see
+	// storage.FieldsField), not a metadata field — publishing it would make
+	// every downstream consumer that parses `file:<id>/<field>` off this
+	// stream (meta-search, meta-dup, meta-fuse, meta-sort) see a phantom
+	// field change on a key they can never read.
+	if strings.HasSuffix(key, "/"+storage.FieldsField) {
+		return
+	}
 
 	// Build event fields
 	fields := map[string]interface{}{

internal/storage/cid_resolution.go

@@ -83,6 +83,7 @@ func (c *Client) Mint(filePath string, size, mtimeNano int64) (string, error) {
 	pipe.Set(ctx, prefix+"sizeByte", strconv.FormatInt(size, 10), 0)
 	pipe.Set(ctx, prefix+"mtimeNano", strconv.FormatInt(mtimeNano, 10), 0)
 	pipe.SAdd(ctx, c.buildIndexKey(), uuid)
+	c.addFieldsToPipe(ctx, pipe, uuid, []string{"filePath", "sizeByte", "mtimeNano"})
 	if _, err := pipe.Exec(ctx); err != nil {
 		return "", fmt.Errorf("mint pipeline: %w", err)
 	}
@@ -116,7 +117,13 @@ func (c *Client) addAliasLocked(ctx context.Context, uuid, cidStr string) error
 		return fmt.Errorf("uuid and cid are required")
 	}
 
-	keysetKey := c.buildKeyPrefix(uuid) + CIDsKeyPrefix + cidStr
+	keysetField := CIDsKeyPrefix + cidStr
+	keysetKey := c.buildKeyPrefix(uuid) + keysetField
+
+	// Every branch below writes the key-set member, so index the field name
+	// once here — including the two guard branches that skip the alias itself.
+	// Missing it would leave `cids/<cid>` invisible to a record read.
+	defer c.addFieldLocked(ctx, uuid, keysetField)
 
 	// Self-pointing alias guard. If the caller is trying to register
 	// cid:<cid> → <cid> (i.e. the root IS the cid value), that's the
@@ -233,26 +240,13 @@ func (c *Client) AddDuplicatePath(uuid, path string) (bool, error) {
 	return added > 0, nil
 }
 
-// cidsForRootLocked scans the bare-CID key-set for a root and returns its
+// cidsForRootLocked returns a root's bare-CID key-set members (with the cids/
-// members (bare CIDs, with the cids/ prefix stripped). Caller must hold c.mu.
+// prefix stripped). Caller must hold c.mu.
+//
+// Reads the field index and filters by prefix; only an un-indexed root falls
+// back to a SCAN, and then over the narrower cids/ glob. See field_index.go.
 func (c *Client) cidsForRootLocked(ctx context.Context, uuid string) ([]string, error) {
-	prefix := c.buildKeyPrefix(uuid) + CIDsKeyPrefix
+	return c.fieldsWithPrefixLocked(ctx, 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

internal/storage/client.go

@@ -109,8 +109,12 @@ func (c *Client) buildIndexKey() string {
 	return c.buildKey("file:__index__")
 }
 
-// GetMetadataFlat retrieves all metadata for a file as a flat map
+// GetMetadataFlat retrieves all metadata for a file as a flat map.
-// Uses flat keys: SCAN file:{hashId}/* then MGET
+//
+// Reads the record's field-name index and MGETs those keys — two round-trips,
+// independent of keyspace size. Records written before the index existed fall
+// back to the old SCAN and are indexed on the way out. See field_index.go for
+// why the index exists and why it is safe to derive.
 func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
@@ -122,41 +126,7 @@ func (c *Client) GetMetadataFlat(hashID string) (map[string]string, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
-	prefix := c.buildKeyPrefix(hashID)
+	return c.readMetadataLocked(ctx, hashID)
-	result := make(map[string]string)
-
-	// Scan for all keys with this prefix
-	var cursor uint64
-	for {
-		keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
-		if err != nil {
-			return nil, fmt.Errorf("scan failed: %w", err)
-		}
-
-		if len(keys) > 0 {
-			values, err := c.client.MGet(ctx, keys...).Result()
-			if err != nil {
-				return nil, fmt.Errorf("mget failed: %w", err)
-			}
-			for i, key := range keys {
-				field := strings.TrimPrefix(key, prefix)
-				if values[i] != nil {
-					result[field] = values[i].(string)
-				}
-			}
-		}
-
-		cursor = nextCursor
-		if cursor == 0 {
-			break
-		}
-	}
-
-	if len(result) == 0 {
-		return nil, nil
-	}
-
-	return result, nil
 }
 
 // SetMetadataFlat stores metadata for a file using flat keys
@@ -179,9 +149,21 @@ func (c *Client) SetMetadataFlat(hashID string, metadata map[string]string) erro
 	prefix := c.buildKeyPrefix(hashID)
 	pipe := c.client.Pipeline()
 
+	fields := make([]string, 0, len(metadata))
 	for field, value := range metadata {
+		// A payload carrying the reserved name (a hand-made snapshot, a buggy
+		// client) would SET a string over the index SET and break every
+		// subsequent read of this record. Drop it — it is derived state, so
+		// there is never a reason to accept it from outside.
+		if isReservedFieldName(field) {
+			continue
+		}
 		pipe.Set(ctx, prefix+field, value, 0)
+		fields = append(fields, field)
 	}
+	// Same pipeline as the SETs, so indexing the field names costs no extra
+	// round-trip. See field_index.go.
+	c.addFieldsToPipe(ctx, pipe, hashID, fields)
 
 	if _, err := pipe.Exec(ctx); err != nil {
 		return fmt.Errorf("pipeline exec failed: %w", err)
@@ -256,6 +238,12 @@ func (c *Client) SetProperty(hashID, property, value string) error {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
+	// See SetMetadataFlat: the reserved name addresses the derived index SET,
+	// not a metadata field.
+	if isReservedFieldName(property) {
+		return fmt.Errorf("field name %q is reserved", property)
+	}
+
 	key := c.buildKeyPrefix(hashID) + property
 
 	// Use SET to set the key
@@ -263,6 +251,8 @@ func (c *Client) SetProperty(hashID, property, value string) error {
 		return fmt.Errorf("set failed: %w", err)
 	}
 
+	c.addFieldLocked(ctx, hashID, property)
+
 	// Add to index set
 	if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
 		return fmt.Errorf("sadd index failed: %w", err)
@@ -387,32 +377,14 @@ func normalizeFilesRelativePath(p string) string {
 // getMetadataFlatInternal mirrors GetMetadataFlat without re-acquiring the lock.
 // Callers must already hold c.mu.
 func (c *Client) getMetadataFlatInternal(ctx context.Context, hashID string) (map[string]string, error) {
-	prefix := c.buildKeyPrefix(hashID)
+	result, err := c.readMetadataLocked(ctx, hashID)
-	result := make(map[string]string)
+	if err != nil {
-
+		return nil, err
-	var cursor uint64
+	}
-	for {
+	// Unlike GetMetadataFlat, this variant hands back an empty map rather than
-		keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
+	// nil for a record with no fields — GetMetadataDocument writes into it.
-		if err != nil {
+	if result == nil {
-			return nil, fmt.Errorf("scan failed: %w", err)
+		result = make(map[string]string)
-		}
-		if len(keys) > 0 {
-			values, err := c.client.MGet(ctx, keys...).Result()
-			if err != nil {
-				return nil, fmt.Errorf("mget failed: %w", err)
-			}
-			for i, key := range keys {
-				if values[i] == nil {
-					continue
-				}
-				field := strings.TrimPrefix(key, prefix)
-				result[field] = values[i].(string)
-			}
-		}
-		cursor = nextCursor
-		if cursor == 0 {
-			break
-		}
 	}
 	return result, nil
 }
@@ -449,9 +421,17 @@ func (c *Client) MergeMetadataFlat(hashID string, metadata map[string]string) (i
 	prefix := c.buildKeyPrefix(hashID)
 	pipe := c.client.Pipeline()
 
+	fields := make([]string, 0, len(metadata))
 	for field, value := range metadata {
+		// See SetMetadataFlat: never let an inbound payload overwrite the
+		// derived index SET with a string.
+		if isReservedFieldName(field) {
+			continue
+		}
 		pipe.Set(ctx, prefix+field, value, 0)
+		fields = append(fields, field)
 	}
+	c.addFieldsToPipe(ctx, pipe, hashID, fields)
 
 	if _, err := pipe.Exec(ctx); err != nil {
 		return 0, fmt.Errorf("pipeline exec failed: %w", err)
@@ -484,7 +464,11 @@ func (c *Client) DeleteProperty(hashID, property string) error {
 	defer cancel()
 
 	key := c.buildKeyPrefix(hashID) + property
-	return c.client.Del(ctx, key).Err()
+	if err := c.client.Del(ctx, key).Err(); err != nil {
+		return err
+	}
+	c.removeFieldLocked(ctx, hashID, property)
+	return nil
 }
 
 // AddToSet adds a value to a set-type field (stored as pipe-delimited string in flat key)
@@ -530,6 +514,9 @@ func (c *Client) AddToSet(hashID, property, value string) (bool, error) {
 		return false, err
 	}
 
+	// The SET above may have created the key, so the field name is new.
+	c.addFieldLocked(ctx, hashID, property)
+
 	// Add to index set
 	if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
 		return true, fmt.Errorf("sadd index failed: %w", err)
@@ -582,7 +569,12 @@ func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) {
 
 	// Save back (or delete key if empty)
 	if len(newValues) == 0 {
-		return true, c.client.Del(ctx, key).Err()
+		if err := c.client.Del(ctx, key).Err(); err != nil {
+			return true, err
+		}
+		// The key is gone, so the field name must leave the index too.
+		c.removeFieldLocked(ctx, hashID, property)
+		return true, nil
 	}
 
 	newValue := strings.Join(newValues, "|")
@@ -764,6 +756,12 @@ func (c *Client) GetSearchIndexTuples() ([]SearchTuple, error) {
 			if k == indexKey {
 				continue
 			}
+			// Per-record field indexes live under the same prefix and are SETs
+			// — MGET would error on them, and they are bookkeeping, not
+			// searchable metadata.
+			if strings.HasSuffix(k, "/"+FieldsField) {
+				continue
+			}
 			keyBatch = append(keyBatch, k)
 			if len(keyBatch) >= 1000 {
 				if err := flush(); err != nil {
@@ -1039,6 +1037,12 @@ func (c *Client) FindByValue(contains string, fields []string, limit int) ([]Fin
 		if field == "" {
 			continue
 		}
+		// The per-record field index is a SET, so an MGET over those keys
+		// would fail the whole query with WRONGTYPE. It holds field names,
+		// not values, so there is nothing here to search anyway.
+		if isReservedFieldName(field) {
+			continue
+		}
 		if len(matches) >= limit {
 			truncated = true
 			break
@@ -1207,7 +1211,17 @@ func (c *Client) SetRawValue(key, value string) error {
 	if t != "none" && t != "string" {
 		return fmt.Errorf("key has incompatible type %q (only string is editable)", t)
 	}
-	return c.client.Set(ctx, full, value, 0).Err()
+	if err := c.client.Set(ctx, full, value, 0).Err(); err != nil {
+		return err
+	}
+	// The raw editor can mint a record property out of band, so it has to keep
+	// the field index honest like every other write path. Non-record keys
+	// (`cid:*`, `udl:*`, `file:__index__`, the index set itself) split false
+	// and are left alone.
+	if hashID, field, ok := c.splitRecordKey(full); ok {
+		c.addFieldLocked(ctx, hashID, field)
+	}
+	return nil
 }
 
 // DeleteRawKey deletes the exact key. Returns true if the key existed.
@@ -1222,10 +1236,16 @@ func (c *Client) DeleteRawKey(key string) (bool, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
-	n, err := c.client.Del(ctx, c.buildKey(key)).Result()
+	full := c.buildKey(key)
+	n, err := c.client.Del(ctx, full).Result()
 	if err != nil {
 		return false, err
 	}
+	if n > 0 {
+		if hashID, field, ok := c.splitRecordKey(full); ok {
+			c.removeFieldLocked(ctx, hashID, field)
+		}
+	}
 	return n > 0, nil
 }
 
@@ -1257,14 +1277,20 @@ func (c *Client) IndexRemoveIfEmpty(hashID string) (bool, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
-	prefix := c.buildKeyPrefix(hashID)
+	nonEmpty, err := c.recordHasFieldsLocked(ctx, hashID)
-	keys, _, err := c.client.Scan(ctx, 0, prefix+"*", 1).Result()
 	if err != nil {
-		return false, fmt.Errorf("scan failed: %w", err)
+		return false, err
 	}
-	if len(keys) > 0 {
+	if nonEmpty {
 		return false, nil
 	}
+
+	// Nothing left but (possibly) an empty field index — drop it too, so a
+	// de-indexed root leaves no orphan key behind.
+	if err := c.client.Del(ctx, c.buildFieldsKey(hashID)).Err(); err != nil {
+		return false, fmt.Errorf("del field index: %w", err)
+	}
+
 	removed, err := c.client.SRem(ctx, c.buildIndexKey(), hashID).Result()
 	if err != nil {
 		return false, fmt.Errorf("srem failed: %w", err)

internal/storage/field_index.go

@@ -0,0 +1,486 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// The per-record field-name index.
+//
+// A record is stored as one flat Redis STRING per leaf
+// (`file:<uuid>/<field>`), and that shape is deliberate — it is what lets
+// keyspace notifications fire per *field*, so meta-fuse / meta-editor /
+// meta-search can react to one field changing instead of diffing a whole hash
+// (see docs/project-architecture/persistent-storage.md, and
+// dev/test/suites/flat-keys.bats which asserts records are never hashes).
+//
+// The cost of that shape was that reading one record meant asking Redis "which
+// keys start with `file:<uuid>/`?", and Redis has no prefix index — `SCAN
+// MATCH` walks the ENTIRE keyspace. On a real store (445k keys, ~36 fields per
+// record) a single record read cost ~670 SCAN round-trips and ~0.5s of Redis
+// CPU; SCAN accounted for ~97.5% of all Redis CPU time on the box. That sat on
+// the byte path for every poster meta-share served, which is how it surfaced.
+//
+// So we keep the flat keys and index their *names*:
+//
+//	file:<uuid>/__fields__   SET{ "filePath", "sizeByte", "cids/bafk…", … }
+//
+// A read becomes SMEMBERS + a chunked MGET — two round-trips instead of ~670.
+// This mirrors `file:__index__` (the set of every root) one level down, and the
+// bulk reads already worked this way: `GetTuplesForAllFiles` composes its keys
+// from the index rather than scanning per root.
+//
+// **The set is derived, never authoritative.** The flat keys remain the source
+// of truth. A record whose set is missing (anything written before this landed)
+// falls back to the old SCAN and populates the set on the way out, so existing
+// deployments heal themselves on first touch with no migration and no
+// SchemaVersion bump — which matters, because a bump is a flag day that makes
+// operators wipe Redis (see schema_version.go).
+//
+// The consistency argument that makes a derived index safe here is meta-core's
+// single-writer rule: leader election publishes an empty `redisUrl` on purpose
+// (internal/leader/election.go), so nothing outside this process can write a
+// `file:` key behind the index's back.
+
+// FieldsField is the reserved field name holding a record's field-name index.
+//
+// It is a Redis SET living at `file:<uuid>/__fields__`, i.e. inside the same
+// prefix as the record's own fields. That keeps every existing sweep correct
+// for free — `DeleteMetadata`, `DeleteRoot` and `ClearAllMetadata` all
+// SCAN+DEL `file:<uuid>/*` and therefore drop the index along with the record.
+// The flip side is that every reader which enumerates that prefix has to skip
+// it, exactly as they already skip `file:__index__`.
+const FieldsField = "__fields__"
+
+// mgetBatch caps how many keys go into one MGET. Redis limits arguments per
+// command; 1000 is the batch size `GetTuplesForAllFiles` already uses.
+const mgetBatch = 1000
+
+// buildFieldsKey constructs the key for a record's field-name index set.
+func (c *Client) buildFieldsKey(hashID string) string {
+	return c.buildKeyPrefix(hashID) + FieldsField
+}
+
+// isReservedFieldName reports whether a field name is bookkeeping rather than
+// metadata, and so must never be returned to an API caller, exported into a
+// snapshot, or folded into the search haystack.
+func isReservedFieldName(field string) bool {
+	return field == FieldsField
+}
+
+// readMetadataLocked returns a record's full flat metadata. Caller must hold
+// c.mu (either mode).
+//
+// Fast path: SMEMBERS the field index, then MGET those keys. Fallback: the
+// legacy full-keyspace SCAN, whose result is used to populate the index so the
+// next read takes the fast path.
+//
+// Returns (nil, nil) for a record with no fields — callers treat that as 404.
+func (c *Client) readMetadataLocked(ctx context.Context, hashID string) (map[string]string, error) {
+	fields, err := c.client.SMembers(ctx, c.buildFieldsKey(hashID)).Result()
+	if err != nil && err != redis.Nil {
+		return nil, fmt.Errorf("smembers fields: %w", err)
+	}
+
+	if len(fields) > 0 {
+		return c.mgetFieldsLocked(ctx, hashID, fields)
+	}
+
+	// No index yet: a record written before the index existed, or one that
+	// genuinely has no fields. Fall back to the scan, then backfill.
+	result, scanned, err := c.scanMetadataLocked(ctx, hashID)
+	if err != nil {
+		return nil, err
+	}
+	if len(scanned) > 0 {
+		// Best-effort: a failed backfill costs the next read a scan, it does
+		// not make the read wrong. Never fail the caller's read for it.
+		if err := c.client.SAdd(ctx, c.buildFieldsKey(hashID), toAny(scanned)...).Err(); err != nil {
+			logFieldIndexBackfillFailure(hashID, err)
+		}
+	}
+	if len(result) == 0 {
+		return nil, nil
+	}
+	return result, nil
+}
+
+// mgetFieldsLocked reads the named fields of a record in chunked MGETs.
+//
+// A field named in the index whose key is gone reads back as nil and is simply
+// skipped: a *stale* index entry is harmless. (A *missing* one would silently
+// drop data, which is why every write path maintains the set.)
+func (c *Client) mgetFieldsLocked(ctx context.Context, hashID string, fields []string) (map[string]string, error) {
+	prefix := c.buildKeyPrefix(hashID)
+
+	// Keep the reserved name out of the key list AND out of the result — it is
+	// a SET, so MGET would error on it in real Redis rather than return nil.
+	wanted := make([]string, 0, len(fields))
+	for _, f := range fields {
+		if !isReservedFieldName(f) {
+			wanted = append(wanted, f)
+		}
+	}
+	if len(wanted) == 0 {
+		return nil, nil
+	}
+
+	result := make(map[string]string, len(wanted))
+	for start := 0; start < len(wanted); start += mgetBatch {
+		end := start + mgetBatch
+		if end > len(wanted) {
+			end = len(wanted)
+		}
+		chunk := wanted[start:end]
+		keys := make([]string, len(chunk))
+		for i, f := range chunk {
+			keys[i] = prefix + f
+		}
+		values, err := c.client.MGet(ctx, keys...).Result()
+		if err != nil {
+			return nil, fmt.Errorf("mget failed: %w", err)
+		}
+		for i, v := range values {
+			if s, ok := v.(string); ok {
+				result[chunk[i]] = s
+			}
+		}
+	}
+
+	if len(result) == 0 {
+		return nil, nil
+	}
+	return result, nil
+}
+
+// scanMetadataLocked is the pre-index read: SCAN the record's key prefix, then
+// MGET. Retained as the fallback for un-indexed records and as the source for
+// the backfill. Returns the metadata plus the field names it saw (which
+// includes fields whose value came back nil, so the index stays complete).
+func (c *Client) scanMetadataLocked(ctx context.Context, hashID string) (map[string]string, []string, error) {
+	prefix := c.buildKeyPrefix(hashID)
+	result := make(map[string]string)
+	var fields []string
+
+	var cursor uint64
+	for {
+		keys, nextCursor, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
+		if err != nil {
+			return nil, nil, fmt.Errorf("scan failed: %w", err)
+		}
+
+		if len(keys) > 0 {
+			wanted := make([]string, 0, len(keys))
+			for _, key := range keys {
+				field := strings.TrimPrefix(key, prefix)
+				if isReservedFieldName(field) {
+					continue
+				}
+				fields = append(fields, field)
+				wanted = append(wanted, key)
+			}
+			if len(wanted) > 0 {
+				values, err := c.client.MGet(ctx, wanted...).Result()
+				if err != nil {
+					return nil, nil, fmt.Errorf("mget failed: %w", err)
+				}
+				for i, key := range wanted {
+					if s, ok := values[i].(string); ok {
+						result[strings.TrimPrefix(key, prefix)] = s
+					}
+				}
+			}
+		}
+
+		cursor = nextCursor
+		if cursor == 0 {
+			break
+		}
+	}
+
+	return result, fields, nil
+}
+
+// addFieldsToPipe records field names in the index as part of an existing
+// pipeline, so a multi-field write stays one round-trip.
+func (c *Client) addFieldsToPipe(ctx context.Context, pipe redis.Pipeliner, hashID string, fields []string) {
+	names := filterReserved(fields)
+	if len(names) == 0 {
+		return
+	}
+	pipe.SAdd(ctx, c.buildFieldsKey(hashID), toAny(names)...)
+}
+
+// addFieldLocked records a single field name in the index. Caller must hold
+// c.mu. Best-effort by the same argument as the backfill: the flat key is the
+// source of truth, and a lost index entry is repaired by the next scan
+// fallback — so it must not fail the write it accompanies.
+func (c *Client) addFieldLocked(ctx context.Context, hashID, field string) {
+	if isReservedFieldName(field) {
+		return
+	}
+	if err := c.client.SAdd(ctx, c.buildFieldsKey(hashID), field).Err(); err != nil {
+		logFieldIndexWriteFailure(hashID, field, err)
+	}
+}
+
+// removeFieldLocked drops a field name from the index. Caller must hold c.mu.
+//
+// A stale entry left behind by a failure is harmless (the MGET reads nil and
+// skips it), so this is best-effort too.
+func (c *Client) removeFieldLocked(ctx context.Context, hashID, field string) {
+	if isReservedFieldName(field) {
+		return
+	}
+	if err := c.client.SRem(ctx, c.buildFieldsKey(hashID), field).Err(); err != nil {
+		logFieldIndexWriteFailure(hashID, field, err)
+	}
+}
+
+// fieldsWithPrefixLocked returns the record's indexed field names under a
+// field-name prefix (e.g. "cids/"), falling back to a SCAN when the record has
+// no index yet. Caller must hold c.mu.
+func (c *Client) fieldsWithPrefixLocked(ctx context.Context, hashID, fieldPrefix string) ([]string, error) {
+	fields, err := c.client.SMembers(ctx, c.buildFieldsKey(hashID)).Result()
+	if err != nil && err != redis.Nil {
+		return nil, fmt.Errorf("smembers fields: %w", err)
+	}
+	if len(fields) == 0 {
+		// Un-indexed record: scan the narrower `<prefix><fieldPrefix>*` glob
+		// rather than the whole record, and do not backfill from it — a
+		// partial view must never be mistaken for a complete index.
+		return c.scanFieldsWithPrefixLocked(ctx, hashID, fieldPrefix)
+	}
+
+	var out []string
+	for _, f := range fields {
+		if isReservedFieldName(f) {
+			continue
+		}
+		if strings.HasPrefix(f, fieldPrefix) {
+			out = append(out, strings.TrimPrefix(f, fieldPrefix))
+		}
+	}
+	// SMEMBERS order is arbitrary and SCAN order was too, so the CID array in
+	// GetMetadataDocument used to shuffle between calls. Sorting costs nothing
+	// at these sizes and makes the response byte-stable, which matters for
+	// anything diffing or caching it.
+	sort.Strings(out)
+	return out, nil
+}
+
+func (c *Client) scanFieldsWithPrefixLocked(ctx context.Context, hashID, fieldPrefix string) ([]string, error) {
+	prefix := c.buildKeyPrefix(hashID) + fieldPrefix
+	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 fields: %w", err)
+		}
+		for _, k := range keys {
+			out = append(out, strings.TrimPrefix(k, prefix))
+		}
+		cursor = next
+		if cursor == 0 {
+			break
+		}
+	}
+	sort.Strings(out)
+	return out, nil
+}
+
+// WarmFieldIndexes builds every missing field index in a SINGLE keyspace pass.
+//
+// Without it the indexes still appear — each record's first read falls back to
+// a scan and backfills — but that transition is O(records x keyspace) and it
+// runs again from scratch after any wipe. Measured on the real store (12,345
+// records / 445k keys) the lazy path warmed at ~1.3 records/s: a ~2.6 HOUR tail
+// during which meta-core stays pinned exactly as it was before the fix. Worse,
+// `RepublishMetadata` walks every record at startup, so that tail is paid on
+// every restart.
+//
+// One bulk `file:*` scan grouped by record collapses it to a few seconds. This
+// is the same shape `GetSearchIndexTuples` already uses for the bulk read, and
+// for the same reason.
+//
+// Idempotent, and safe to run against a partially-indexed store: SADD of a
+// member that is already there is a no-op. Returns the number of records it
+// wrote an index for.
+func (c *Client) WarmFieldIndexes() (int, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	if c.client == nil {
+		return 0, fmt.Errorf("not connected")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
+	defer cancel()
+
+	filePrefix := c.buildKey("file:") // {filePrefix}{hashID}/{field}
+	indexKey := c.buildIndexKey()
+
+	byRecord := make(map[string][]string)
+	var cursor uint64
+	for {
+		keys, next, err := c.client.Scan(ctx, cursor, filePrefix+"*", 1000).Result()
+		if err != nil {
+			return 0, fmt.Errorf("scan: %w", err)
+		}
+		for _, k := range keys {
+			if k == indexKey {
+				continue
+			}
+			// hashID has no '/'; field may contain them.
+			rest := strings.TrimPrefix(k, filePrefix)
+			slash := strings.IndexByte(rest, '/')
+			if slash < 0 {
+				continue
+			}
+			hashID, field := rest[:slash], rest[slash+1:]
+			if isReservedFieldName(field) {
+				// Already indexed — remember the record so we don't treat it
+				// as missing, but there is nothing to add for this key.
+				if _, seen := byRecord[hashID]; !seen {
+					byRecord[hashID] = nil
+				}
+				continue
+			}
+			byRecord[hashID] = append(byRecord[hashID], field)
+		}
+		cursor = next
+		if cursor == 0 {
+			break
+		}
+	}
+
+	written := 0
+	pipe := c.client.Pipeline()
+	pending := 0
+	for hashID, fields := range byRecord {
+		if len(fields) == 0 {
+			continue
+		}
+		pipe.SAdd(ctx, c.buildFieldsKey(hashID), toAny(fields)...)
+		written++
+		pending++
+		// Bound the pipeline so a large store doesn't buffer every command.
+		if pending >= 500 {
+			if _, err := pipe.Exec(ctx); err != nil {
+				return written, fmt.Errorf("sadd pipeline: %w", err)
+			}
+			pipe = c.client.Pipeline()
+			pending = 0
+		}
+	}
+	if pending > 0 {
+		if _, err := pipe.Exec(ctx); err != nil {
+			return written, fmt.Errorf("sadd pipeline: %w", err)
+		}
+	}
+	return written, nil
+}
+
+// recordHasFieldsLocked reports whether a record still has at least one
+// property key. Caller must hold c.mu.
+//
+// `__fields__` must not count — it lives under the record's own prefix, so a
+// naive "does any `file:<id>/*` key exist?" probe would answer yes for a record
+// whose every property has been deleted, and the root could never be
+// de-indexed.
+func (c *Client) recordHasFieldsLocked(ctx context.Context, hashID string) (bool, error) {
+	n, err := c.client.SCard(ctx, c.buildFieldsKey(hashID)).Result()
+	if err != nil && err != redis.Nil {
+		return false, fmt.Errorf("scard fields: %w", err)
+	}
+	if n > 0 {
+		return true, nil
+	}
+
+	// Either genuinely empty, or an un-indexed record from before the index
+	// existed — the scan is the only way to tell those apart.
+	prefix := c.buildKeyPrefix(hashID)
+	var cursor uint64
+	for {
+		keys, next, err := c.client.Scan(ctx, cursor, prefix+"*", 1000).Result()
+		if err != nil {
+			return false, fmt.Errorf("scan failed: %w", err)
+		}
+		for _, k := range keys {
+			if !isReservedFieldName(strings.TrimPrefix(k, prefix)) {
+				return true, nil
+			}
+		}
+		cursor = next
+		if cursor == 0 {
+			return false, nil
+		}
+	}
+}
+
+// splitRecordKey decomposes a full Redis key into (hashID, field) when it is a
+// record property key. Reports ok=false for anything else — the index set
+// itself, `file:__index__`, and any key outside the `file:` namespace.
+//
+// This is what lets the raw KV escape hatch (SetRawValue / DeleteRawKey) keep
+// the index honest without its callers knowing the layout.
+func (c *Client) splitRecordKey(key string) (hashID, field string, ok bool) {
+	rest := key
+	if c.prefix != "" {
+		var found bool
+		rest, found = strings.CutPrefix(rest, c.prefix)
+		if !found {
+			return "", "", false
+		}
+	}
+	rest, found := strings.CutPrefix(rest, "file:")
+	if !found {
+		return "", "", false
+	}
+	hashID, field, found = strings.Cut(rest, "/")
+	if !found || hashID == "" || field == "" {
+		// `file:__index__` has no slash and lands here.
+		return "", "", false
+	}
+	if isReservedFieldName(field) {
+		return "", "", false
+	}
+	return hashID, field, true
+}
+
+// The index is derived and self-healing, so a write failure degrades
+// performance (the next read falls back to a scan and re-populates) rather
+// than correctness. Log it rather than failing the caller's write — but log it
+// loudly, because a *persistent* failure means every read is scanning again.
+func logFieldIndexWriteFailure(hashID, field string, err error) {
+	log.Printf("[Storage] field index update failed for %s/%s: %v (reads fall back to SCAN)", hashID, field, err)
+}
+
+func logFieldIndexBackfillFailure(hashID string, err error) {
+	log.Printf("[Storage] field index backfill failed for %s: %v (reads keep falling back to SCAN)", hashID, err)
+}
+
+func filterReserved(fields []string) []string {
+	out := make([]string, 0, len(fields))
+	for _, f := range fields {
+		if !isReservedFieldName(f) {
+			out = append(out, f)
+		}
+	}
+	return out
+}
+
+func toAny(ss []string) []interface{} {
+	out := make([]interface{}, len(ss))
+	for i, s := range ss {
+		out[i] = s
+	}
+	return out
+}

internal/storage/field_index_test.go

@@ -0,0 +1,464 @@
+package storage
+
+import (
+	"context"
+	"reflect"
+	"sort"
+	"testing"
+)
+
+// Tests for the per-record field-name index (field_index.go).
+//
+// The index is what makes a record read O(fields) instead of O(keyspace). It is
+// derived state, so the properties worth pinning are: (1) a read served from
+// the index returns exactly what a scan would have, (2) a record with no index
+// still reads correctly and gets one, and (3) every write path keeps it in
+// sync — because a *missing* entry silently drops a field from every
+// subsequent read, which is the one failure mode that would be invisible.
+//
+// Every case runs with and without a namespace prefix, matching clear_test.go.
+
+func forEachPrefix(t *testing.T, fn func(t *testing.T, c *Client)) {
+	t.Helper()
+	for _, prefix := range []string{"", "mm:"} {
+		name := "no-prefix"
+		if prefix != "" {
+			name = "prefix-" + prefix
+		}
+		t.Run(name, func(t *testing.T) {
+			c, _ := newTestClient(t, prefix)
+			fn(t, c)
+		})
+	}
+}
+
+func fieldNames(t *testing.T, c *Client, hashID string) []string {
+	t.Helper()
+	got, err := c.client.SMembers(context.Background(), c.buildFieldsKey(hashID)).Result()
+	if err != nil {
+		t.Fatalf("smembers fields: %v", err)
+	}
+	sort.Strings(got)
+	return got
+}
+
+func mustGetMetadata(t *testing.T, c *Client, hashID string) map[string]string {
+	t.Helper()
+	got, err := c.GetMetadataFlat(hashID)
+	if err != nil {
+		t.Fatalf("GetMetadataFlat: %v", err)
+	}
+	return got
+}
+
+// SetMetadataFlat must index every field it writes, and the read must come back
+// through the index rather than a scan.
+func TestFieldIndex_SetMetadataFlatIndexesFields(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		want := map[string]string{
+			"filePath":                      "/files/watch/a.mkv",
+			"sizeByte":                      "123",
+			"cids/bafkreiaaa":               "true",
+			"titles/jpn":                    "ナルト",
+			"categories/TV/UHD":             "true",
+			"info/files/0/path":             "a.mkv",
+			"provenance":                    `{"source":"gateway"}`,
+			"plot/eng":                      "a plot with / slashes",
+			"stream/3":                      "eng",
+			"genres/action":                 "true",
+			"originaltitle":                 "Naruto",
+			"deeply/nested/field/name/here": "x",
+		}
+		if err := c.SetMetadataFlat("root1", want); err != nil {
+			t.Fatalf("SetMetadataFlat: %v", err)
+		}
+
+		if got := mustGetMetadata(t, c, "root1"); !reflect.DeepEqual(got, want) {
+			t.Errorf("GetMetadataFlat mismatch\n got %v\nwant %v", got, want)
+		}
+
+		// The index must name every field — including the ones containing '/',
+		// which is where a naive "split on the first slash" would go wrong.
+		wantNames := make([]string, 0, len(want))
+		for k := range want {
+			wantNames = append(wantNames, k)
+		}
+		sort.Strings(wantNames)
+		if got := fieldNames(t, c, "root1"); !reflect.DeepEqual(got, wantNames) {
+			t.Errorf("field index mismatch\n got %v\nwant %v", got, wantNames)
+		}
+	})
+}
+
+// A record written before the index existed has flat keys but no set. The read
+// must still be correct, and must leave an index behind so the next read takes
+// the fast path. This is the migration path for the ~12k records already on
+// disk: no flag day, no SchemaVersion bump.
+func TestFieldIndex_LazyBackfillForUnindexedRecord(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		ctx := context.Background()
+		prefix := c.buildKeyPrefix("legacy")
+		want := map[string]string{
+			"filePath":        "/files/watch/old.mkv",
+			"sizeByte":        "42",
+			"cids/bafkreiold": "true",
+		}
+		// Write the flat keys directly, bypassing every index-maintaining path
+		// — exactly the shape a pre-upgrade store is in.
+		for f, v := range want {
+			if err := c.client.Set(ctx, prefix+f, v, 0).Err(); err != nil {
+				t.Fatalf("seed %s: %v", f, err)
+			}
+		}
+		if names := fieldNames(t, c, "legacy"); len(names) != 0 {
+			t.Fatalf("precondition: expected no index, got %v", names)
+		}
+
+		if got := mustGetMetadata(t, c, "legacy"); !reflect.DeepEqual(got, want) {
+			t.Errorf("scan-fallback read mismatch\n got %v\nwant %v", got, want)
+		}
+
+		wantNames := []string{"cids/bafkreiold", "filePath", "sizeByte"}
+		if got := fieldNames(t, c, "legacy"); !reflect.DeepEqual(got, wantNames) {
+			t.Errorf("backfill mismatch\n got %v\nwant %v", got, wantNames)
+		}
+
+		// Second read, now served from the index, must agree with the first.
+		if got := mustGetMetadata(t, c, "legacy"); !reflect.DeepEqual(got, want) {
+			t.Errorf("indexed read disagrees with scan read\n got %v\nwant %v", got, want)
+		}
+	})
+}
+
+// Each single-field write path must add to the index, and each delete path must
+// remove from it. A field written but not indexed vanishes from every later
+// read — silently — so this is the property that matters most.
+func TestFieldIndex_SingleFieldWritePathsStayInSync(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		if err := c.SetProperty("root2", "title", "Naruto"); err != nil {
+			t.Fatalf("SetProperty: %v", err)
+		}
+		if _, err := c.AddToSet("root2", "tags", "anime"); err != nil {
+			t.Fatalf("AddToSet: %v", err)
+		}
+		if _, err := c.MergeMetadataFlat("root2", map[string]string{"year": "2002"}); err != nil {
+			t.Fatalf("MergeMetadataFlat: %v", err)
+		}
+
+		want := map[string]string{"title": "Naruto", "tags": "anime", "year": "2002"}
+		if got := mustGetMetadata(t, c, "root2"); !reflect.DeepEqual(got, want) {
+			t.Errorf("after writes\n got %v\nwant %v", got, want)
+		}
+
+		// DeleteProperty drops the key and the index entry.
+		if err := c.DeleteProperty("root2", "title"); err != nil {
+			t.Fatalf("DeleteProperty: %v", err)
+		}
+		// RemoveFromSet deletes the key once its last member is gone.
+		if _, err := c.RemoveFromSet("root2", "tags", "anime"); err != nil {
+			t.Fatalf("RemoveFromSet: %v", err)
+		}
+
+		want = map[string]string{"year": "2002"}
+		if got := mustGetMetadata(t, c, "root2"); !reflect.DeepEqual(got, want) {
+			t.Errorf("after deletes\n got %v\nwant %v", got, want)
+		}
+		if got := fieldNames(t, c, "root2"); !reflect.DeepEqual(got, []string{"year"}) {
+			t.Errorf("index after deletes = %v, want [year]", got)
+		}
+	})
+}
+
+// The raw KV editor can mint or drop a record property out of band. It is the
+// one write path whose caller does not know the record layout, so the storage
+// layer has to keep the index honest for it.
+func TestFieldIndex_RawKeyEditorStaysInSync(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		if err := c.SetRawValue("file:root3/hand-written", "v"); err != nil {
+			t.Fatalf("SetRawValue: %v", err)
+		}
+		if got := fieldNames(t, c, "root3"); !reflect.DeepEqual(got, []string{"hand-written"}) {
+			t.Errorf("index after SetRawValue = %v", got)
+		}
+		if got := mustGetMetadata(t, c, "root3"); !reflect.DeepEqual(got, map[string]string{"hand-written": "v"}) {
+			t.Errorf("read after SetRawValue = %v", got)
+		}
+
+		if _, err := c.DeleteRawKey("file:root3/hand-written"); err != nil {
+			t.Fatalf("DeleteRawKey: %v", err)
+		}
+		if got := fieldNames(t, c, "root3"); len(got) != 0 {
+			t.Errorf("index after DeleteRawKey = %v, want empty", got)
+		}
+
+		// Keys outside the record namespace must be left alone entirely.
+		if err := c.SetRawValue("cid:bafkreiaaa", "root3"); err != nil {
+			t.Fatalf("SetRawValue non-record: %v", err)
+		}
+		if _, _, ok := c.splitRecordKey(c.buildKey("cid:bafkreiaaa")); ok {
+			t.Error("cid: key was treated as a record property")
+		}
+		if _, _, ok := c.splitRecordKey(c.buildIndexKey()); ok {
+			t.Error("file:__index__ was treated as a record property")
+		}
+	})
+}
+
+// Mint and AddAlias write fields through their own pipelines, so they need the
+// index too. AddAlias in particular is how `cids/<cid>` members appear — miss
+// it and a record's CID key-set becomes invisible to every read.
+func TestFieldIndex_MintAndAliasIndexTheirFields(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		uuid, err := c.Mint("/files/watch/x.mkv", 10, 20)
+		if err != nil {
+			t.Fatalf("Mint: %v", err)
+		}
+		if got := fieldNames(t, c, uuid); !reflect.DeepEqual(got, []string{"filePath", "mtimeNano", "sizeByte"}) {
+			t.Errorf("index after Mint = %v", got)
+		}
+
+		if err := c.AddAlias(uuid, "bafkreiaaa"); err != nil {
+			t.Fatalf("AddAlias: %v", err)
+		}
+		got := mustGetMetadata(t, c, uuid)
+		if got["cids/bafkreiaaa"] != "true" {
+			t.Errorf("alias key-set member missing from read: %v", got)
+		}
+
+		// And the CID enumeration, which now reads the index instead of
+		// scanning, must see it.
+		cids, err := c.cidsForRootLocked(context.Background(), uuid)
+		if err != nil {
+			t.Fatalf("cidsForRootLocked: %v", err)
+		}
+		if !reflect.DeepEqual(cids, []string{"bafkreiaaa"}) {
+			t.Errorf("cidsForRoot = %v, want [bafkreiaaa]", cids)
+		}
+	})
+}
+
+// The index lives under the record's own prefix, so the "is this record empty?"
+// probe would otherwise always find it and the root could never be de-indexed.
+func TestFieldIndex_IndexRemoveIfEmptyIgnoresTheIndexItself(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		if err := c.SetProperty("root4", "title", "x"); err != nil {
+			t.Fatalf("SetProperty: %v", err)
+		}
+
+		removed, err := c.IndexRemoveIfEmpty("root4")
+		if err != nil {
+			t.Fatalf("IndexRemoveIfEmpty: %v", err)
+		}
+		if removed {
+			t.Error("de-indexed a record that still has a field")
+		}
+
+		if err := c.DeleteProperty("root4", "title"); err != nil {
+			t.Fatalf("DeleteProperty: %v", err)
+		}
+		removed, err = c.IndexRemoveIfEmpty("root4")
+		if err != nil {
+			t.Fatalf("IndexRemoveIfEmpty: %v", err)
+		}
+		if !removed {
+			t.Fatal("record with no fields left was not de-indexed — the field index masked emptiness")
+		}
+		// ...and the now-empty index key must not be left orphaned.
+		n, err := c.client.Exists(context.Background(), c.buildFieldsKey("root4")).Result()
+		if err != nil {
+			t.Fatalf("exists: %v", err)
+		}
+		if n != 0 {
+			t.Error("orphaned field index left behind after de-indexing")
+		}
+	})
+}
+
+// The same probe must still work for a record that predates the index.
+func TestFieldIndex_IndexRemoveIfEmptyHandlesUnindexedRecord(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		ctx := context.Background()
+		if err := c.client.Set(ctx, c.buildKeyPrefix("legacy2")+"title", "x", 0).Err(); err != nil {
+			t.Fatalf("seed: %v", err)
+		}
+		if err := c.IndexAdd("legacy2"); err != nil {
+			t.Fatalf("IndexAdd: %v", err)
+		}
+		removed, err := c.IndexRemoveIfEmpty("legacy2")
+		if err != nil {
+			t.Fatalf("IndexRemoveIfEmpty: %v", err)
+		}
+		if removed {
+			t.Error("de-indexed an un-indexed record that still has a field")
+		}
+	})
+}
+
+// The reserved name is bookkeeping. It must never reach an API caller as a
+// metadata field, and an inbound payload must never be able to overwrite the
+// SET with a string — that would break every subsequent read of the record.
+func TestFieldIndex_ReservedNameIsNeverMetadata(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		err := c.SetMetadataFlat("root5", map[string]string{
+			"title":     "x",
+			FieldsField: "malicious",
+		})
+		if err != nil {
+			t.Fatalf("SetMetadataFlat: %v", err)
+		}
+
+		got := mustGetMetadata(t, c, "root5")
+		if _, present := got[FieldsField]; present {
+			t.Errorf("%s surfaced as a metadata field: %v", FieldsField, got)
+		}
+		if !reflect.DeepEqual(got, map[string]string{"title": "x"}) {
+			t.Errorf("read = %v, want just title", got)
+		}
+
+		// The index must still be a SET — i.e. not clobbered by the payload.
+		if got := fieldNames(t, c, "root5"); !reflect.DeepEqual(got, []string{"title"}) {
+			t.Errorf("index = %v, want [title]", got)
+		}
+
+		if err := c.SetProperty("root5", FieldsField, "malicious"); err == nil {
+			t.Error("SetProperty accepted the reserved field name")
+		}
+	})
+}
+
+// Deleting a record must take its index with it, or a recreated root would
+// inherit phantom field names.
+func TestFieldIndex_DeleteMetadataRemovesTheIndex(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		if err := c.SetMetadataFlat("root6", map[string]string{"title": "x", "year": "2002"}); err != nil {
+			t.Fatalf("SetMetadataFlat: %v", err)
+		}
+		if _, err := c.DeleteMetadata("root6"); err != nil {
+			t.Fatalf("DeleteMetadata: %v", err)
+		}
+		if got := fieldNames(t, c, "root6"); len(got) != 0 {
+			t.Errorf("index survived DeleteMetadata: %v", got)
+		}
+		got, err := c.GetMetadataFlat("root6")
+		if err != nil {
+			t.Fatalf("GetMetadataFlat: %v", err)
+		}
+		if got != nil {
+			t.Errorf("deleted record still reads back: %v", got)
+		}
+	})
+}
+
+// A stale index entry (name present, key gone) is the benign direction: the
+// MGET reads nil and the field is skipped rather than erroring.
+func TestFieldIndex_StaleEntryIsHarmless(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		ctx := context.Background()
+		if err := c.SetMetadataFlat("root7", map[string]string{"title": "x"}); err != nil {
+			t.Fatalf("SetMetadataFlat: %v", err)
+		}
+		// Name a field that has no key behind it.
+		if err := c.client.SAdd(ctx, c.buildFieldsKey("root7"), "ghost").Err(); err != nil {
+			t.Fatalf("sadd ghost: %v", err)
+		}
+
+		got := mustGetMetadata(t, c, "root7")
+		if !reflect.DeepEqual(got, map[string]string{"title": "x"}) {
+			t.Errorf("read = %v, want just title", got)
+		}
+	})
+}
+
+// WarmFieldIndexes is what keeps an upgrade from spending hours reading
+// un-indexed records one full-keyspace scan at a time. It must index every
+// record in one pass, be idempotent, and leave already-indexed records alone.
+func TestFieldIndex_WarmBuildsEveryMissingIndex(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		ctx := context.Background()
+
+		// Two legacy records seeded straight into Redis (no index), plus one
+		// written through the normal path (already indexed).
+		legacy := map[string]map[string]string{
+			"old1": {"filePath": "/a.mkv", "cids/bafkreiaaa": "true"},
+			"old2": {"filePath": "/b.mkv", "titles/jpn": "ナルト", "deep/a/b": "x"},
+		}
+		for root, fields := range legacy {
+			for f, v := range fields {
+				if err := c.client.Set(ctx, c.buildKeyPrefix(root)+f, v, 0).Err(); err != nil {
+					t.Fatalf("seed: %v", err)
+				}
+			}
+			if err := c.IndexAdd(root); err != nil {
+				t.Fatalf("IndexAdd: %v", err)
+			}
+		}
+		if err := c.SetMetadataFlat("fresh", map[string]string{"title": "x"}); err != nil {
+			t.Fatalf("SetMetadataFlat: %v", err)
+		}
+
+		n, err := c.WarmFieldIndexes()
+		if err != nil {
+			t.Fatalf("WarmFieldIndexes: %v", err)
+		}
+		if n < 2 {
+			t.Errorf("warmed %d records, want at least the 2 legacy ones", n)
+		}
+
+		if got := fieldNames(t, c, "old1"); !reflect.DeepEqual(got, []string{"cids/bafkreiaaa", "filePath"}) {
+			t.Errorf("old1 index = %v", got)
+		}
+		if got := fieldNames(t, c, "old2"); !reflect.DeepEqual(got, []string{"deep/a/b", "filePath", "titles/jpn"}) {
+			t.Errorf("old2 index = %v", got)
+		}
+		if got := fieldNames(t, c, "fresh"); !reflect.DeepEqual(got, []string{"title"}) {
+			t.Errorf("fresh index disturbed = %v", got)
+		}
+
+		// Reads must now agree with what the records actually hold.
+		for root, want := range legacy {
+			if got := mustGetMetadata(t, c, root); !reflect.DeepEqual(got, want) {
+				t.Errorf("%s read = %v, want %v", root, got, want)
+			}
+		}
+
+		// Idempotent: a second pass changes nothing.
+		if _, err := c.WarmFieldIndexes(); err != nil {
+			t.Fatalf("second WarmFieldIndexes: %v", err)
+		}
+		if got := fieldNames(t, c, "old2"); !reflect.DeepEqual(got, []string{"deep/a/b", "filePath", "titles/jpn"}) {
+			t.Errorf("old2 index after second warm = %v", got)
+		}
+	})
+}
+
+// The bulk search read enumerates the whole file: keyspace and MGETs what it
+// finds. The index keys are SETs living in that same namespace, so it has to
+// skip them or the MGET fails with WRONGTYPE and search dies wholesale.
+func TestFieldIndex_SearchIndexTuplesSkipTheIndexKeys(t *testing.T) {
+	forEachPrefix(t, func(t *testing.T, c *Client) {
+		if err := c.SetMetadataFlat("root8", map[string]string{
+			"title":    "Naruto Shippuden",
+			"fileName": "naruto.mkv",
+		}); err != nil {
+			t.Fatalf("SetMetadataFlat: %v", err)
+		}
+
+		tuples, err := c.GetSearchIndexTuples()
+		if err != nil {
+			t.Fatalf("GetSearchIndexTuples: %v", err)
+		}
+		var found bool
+		for _, tu := range tuples {
+			if tu.HashID != "root8" {
+				continue
+			}
+			found = true
+			if _, present := tu.Fields[FieldsField]; present {
+				t.Errorf("%s leaked into the search tuple: %v", FieldsField, tu.Fields)
+			}
+		}
+		if !found {
+			t.Fatal("record missing from search tuples")
+		}
+	})
+}

internal/storage/udl.go

@@ -72,7 +72,15 @@ func udlIdxCidUsers(cid string) string      { return fmt.Sprintf("udl:idx:cid:%s
 //
 //	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
+//	      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 three indexes are left
+// alone: they answer "who has an opinion about this cid", where a tombstone is
+// still an answer.
 var udlUpsertScript = redis.NewScript(`
 local stored = redis.call('HGET', KEYS[1], 'version')
 if stored and tonumber(stored) >= tonumber(ARGV[1]) then
@@ -84,7 +92,11 @@ if ARGV[5] == '1' then
 else
   redis.call('HDEL', KEYS[1], 'value')
 end
-redis.call('SADD', KEYS[2], ARGV[6])
+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])
@@ -94,8 +106,10 @@ 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.
+// tombstones. Set tombstone=true when the record deletes the cell, which also
-func (c *Client) UDLUpsertIfNewer(uid, cid, key string, version, ts int64, record, publicValue string, hasValue bool) (bool, error) {
+// 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()
 
@@ -117,7 +131,11 @@ func (c *Client) UDLUpsertIfNewer(uid, cid, key string, version, ts int64, recor
 	if hasValue {
 		hv = "1"
 	}
-	res, err := udlUpsertScript.Run(ctx, c.client, keys, version, ts, record, publicValue, hv, cid, key, uid).Result()
+	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)
 	}

internal/storage/udl_test.go

@@ -0,0 +1,79 @@
+package storage
+
+import (
+	"testing"
+)
+
+// A tombstone must leave the cell in place but drop the cid from the user+key
+// index.
+//
+// The cell has to stay: it *is* the CRDT state, and deleting it would let an
+// older write resurrect the value the tombstone retracted. The index entry has
+// to go: `active_for_user_key` returns every member of that set, so a set that
+// only ever grows means each My List / Continue Watching read carries every
+// title the user has ever un-liked or finished — with its full record blob —
+// forever. The reader filters them out again, so the cost is invisible until
+// the account is old.
+func TestUDLUpsert_TombstonePrunesUserKeyIndexButKeepsCell(t *testing.T) {
+	c, _ := newTestClient(t, "")
+
+	const (
+		uid = "zAlice"
+		cid = "bagcsaaalocator"
+		key = "seek"
+	)
+
+	// A live position: indexed, readable.
+	if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 1, 100, "rec-v1", "", false, false); err != nil || !ok {
+		t.Fatalf("live upsert: ok=%v err=%v", ok, err)
+	}
+	entries, err := c.UDLListUserKey(uid, key)
+	if err != nil {
+		t.Fatalf("active_for_user_key: %v", err)
+	}
+	if len(entries) != 1 || entries[0].Cid != cid {
+		t.Fatalf("live position should be indexed, got %+v", entries)
+	}
+
+	// Finishing it tombstones the seek.
+	if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 2, 200, "rec-v2-tombstone", "", false, true); err != nil || !ok {
+		t.Fatalf("tombstone upsert: ok=%v err=%v", ok, err)
+	}
+
+	entries, err = c.UDLListUserKey(uid, key)
+	if err != nil {
+		t.Fatalf("active_for_user_key after tombstone: %v", err)
+	}
+	if len(entries) != 0 {
+		t.Fatalf("tombstoned cid must leave the user+key index, got %+v", entries)
+	}
+
+	// ...but the cell itself survives, at the tombstone's version.
+	cell, exists, err := c.UDLGetCell(uid, cid, key)
+	if err != nil {
+		t.Fatalf("get cell: %v", err)
+	}
+	if !exists {
+		t.Fatal("tombstone must not delete the cell — an older write could then resurrect the value")
+	}
+	if cell.Version != 2 || cell.Record != "rec-v2-tombstone" {
+		t.Fatalf("cell should hold the tombstone record at v2, got %+v", cell)
+	}
+
+	// And the high-water mark still rejects the superseded write.
+	if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 1, 300, "rec-v1-replay", "", false, false); err != nil || ok {
+		t.Fatalf("stale replay must be rejected: ok=%v err=%v", ok, err)
+	}
+
+	// Re-watching re-establishes the cell and puts it back in the index.
+	if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 3, 400, "rec-v3", "", false, false); err != nil || !ok {
+		t.Fatalf("re-establish upsert: ok=%v err=%v", ok, err)
+	}
+	entries, err = c.UDLListUserKey(uid, key)
+	if err != nil {
+		t.Fatalf("active_for_user_key after re-establish: %v", err)
+	}
+	if len(entries) != 1 || entries[0].Cid != cid {
+		t.Fatalf("re-established cid must return to the index, got %+v", entries)
+	}
+}