mnp 3.1: per-account blobs the node cannot read

One row per playlist plus a manifest, so starring a track rewrites that
playlist rather than the whole collection. blob_enc is a BLOB, not
base64 TEXT: these run to hundreds of kilobytes.

user_id comes from the session and never from the message; kind is
validated against a pattern; every cap refuses with a stated reason
rather than truncating.

Additive, so MNP_MIN_SUPPORTED does not move — a 3.0 node answers
"unknown message type" and the client writes to the next one it reaches.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

1 files changed, 115 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/bundle_store.py b/packages/meshbay-node/src/meshbay_node/bundle_store.py
index 86b8de4..c65b94d 100644
--- a/packages/meshbay-node/src/meshbay_node/bundle_store.py
+++ b/packages/meshbay-node/src/meshbay_node/bundle_store.py
@@ -63,6 +63,38 @@ CREATE TABLE IF NOT EXISTS keypair_bundles (
 );
 """
 
+# Per-account blobs the node holds and cannot read — playlists today
+# (docs/playlists.md §3.3). The same shape as a keypair bundle, with a
+# different payload, so this adds no trust boundary the node was not already
+# on the wrong side of for this same account.
+#
+# Two deliberate differences from keypair_bundles, each a mistake avoided
+# rather than a preference:
+#
+#   `blob_enc` is a BLOB, not base64 TEXT. A keypair bundle is a few hundred
+#   bytes and nobody will ever notice the 33% tax; a playlist is hundreds of
+#   kilobytes, where a third is not a rounding error.
+#
+#   `kind` is a namespace, not an enum: "playlists" is the manifest and
+#   "playlist:<uuid>" is one playlist's tracks. That is what lets one playlist
+#   be rewritten without re-uploading the whole collection, and it costs no
+#   second table. webrtc_server.py validates the shape.
+#
+# blob_enc_recovery is declared now and left NULL — the same column, on the
+# same kind of table, that keypair_bundles needed a PRAGMA migration for one
+# release late. CREATE TABLE IF NOT EXISTS never adds a column.
+_SCHEMA_USER_BLOBS = """\
+CREATE TABLE IF NOT EXISTS user_blobs (
+    user_id           TEXT NOT NULL,
+    kind              TEXT NOT NULL,
+    rev               INTEGER NOT NULL,
+    blob_enc          BLOB NOT NULL,
+    blob_enc_recovery BLOB,
+    stored_at         TEXT NOT NULL DEFAULT (datetime('now')),
+    PRIMARY KEY (user_id, kind)
+);
+"""
+
 
 class BundleStore:
     def __init__(self, db_path: Path):
@@ -75,6 +107,7 @@ class BundleStore:
         await self._db.execute(_SCHEMA_GEK)
         await self._db.execute(_SCHEMA_KEYPAIR)
         await self._db.execute(_SCHEMA_CHAT_EPOCHS)
+        await self._db.execute(_SCHEMA_USER_BLOBS)
         await self._migrate_keypair_recovery()
         await self._db.commit()
 
@@ -180,6 +213,88 @@ class BundleStore:
         await self._db.commit()
         return cur.rowcount > 0
 
+    # ── Per-account blobs (playlists) ────────────────────────────────────
+    #
+    # Opaque throughout: the node stores bytes it cannot read, returns them, and
+    # never looks inside. Every method takes `user_id` from the caller, which
+    # takes it from the authenticated session and never from a message body —
+    # a user_id off the wire would let any member read or overwrite any other
+    # member's blob.
+
+    async def store_user_blob(
+        self,
+        user_id: str,
+        kind: str,
+        rev: int,
+        blob_enc: bytes,
+    ) -> None:
+        """
+        Write one blob, replacing whatever was there.
+
+        Older revisions are not kept. The client is the authority on what the
+        merged state is (docs/playlists.md §6.4) and holds its own copy, so a
+        node keeping history would buy nothing and would mean the node deciding
+        which revision is current — which is exactly what it must not do.
+        """
+        assert self._db
+        await self._db.execute(
+            "INSERT INTO user_blobs (user_id, kind, rev, blob_enc, stored_at) "
+            "VALUES (?, ?, ?, ?, datetime('now')) "
+            "ON CONFLICT(user_id, kind) DO UPDATE SET "
+            "  rev = excluded.rev, "
+            "  blob_enc = excluded.blob_enc, "
+            "  stored_at = excluded.stored_at",
+            (user_id, kind, rev, blob_enc),
+        )
+        await self._db.commit()
+
+    async def fetch_user_blob(self, user_id: str, kind: str) -> dict | None:
+        assert self._db
+        async with self._db.execute(
+            "SELECT rev, blob_enc FROM user_blobs WHERE user_id = ? AND kind = ?",
+            (user_id, kind),
+        ) as cursor:
+            row = await cursor.fetchone()
+            if not row:
+                return None
+            return {"rev": row[0], "blob_enc": row[1]}
+
+    async def list_user_blobs(self, user_id: str) -> list[dict]:
+        """
+        Which blobs exist and at what revision — never a payload.
+
+        A client that has lost its local state (a cache clear, a new device)
+        cannot otherwise discover which playlists exist: their kinds carry
+        client-generated UUIDs, and guessing is not a plan.
+        """
+        assert self._db
+        async with self._db.execute(
+            "SELECT kind, rev FROM user_blobs WHERE user_id = ? ORDER BY kind",
+            (user_id,),
+        ) as cursor:
+            return [{"kind": r[0], "rev": r[1]} for r in await cursor.fetchall()]
+
+    async def delete_user_blob(self, user_id: str, kind: str) -> bool:
+        assert self._db
+        cur = await self._db.execute(
+            "DELETE FROM user_blobs WHERE user_id = ? AND kind = ?",
+            (user_id, kind))
+        await self._db.commit()
+        return cur.rowcount > 0
+
+    async def user_blob_total_bytes(self, user_id: str) -> int:
+        """
+        What this account is already using, so a store can refuse to take it
+        over the per-account cap. An unbounded write primitive pointed at
+        somebody else's disk needs a number that is actually checked.
+        """
+        assert self._db
+        async with self._db.execute(
+            "SELECT COALESCE(SUM(LENGTH(blob_enc)), 0) FROM user_blobs "
+            "WHERE user_id = ?", (user_id,)) as cursor:
+            row = await cursor.fetchone()
+            return int(row[0]) if row else 0
+
     # ── Chat epoch keys ──────────────────────────────────────────────────
 
     async def store_chat_epoch(