1 files changed, 142 insertions, 0 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index af41061..298b3cd 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -29,6 +29,7 @@ import hashlib
 import hmac
 import logging
 import os
+import re
 import struct
 import tempfile
 import time
@@ -146,6 +147,32 @@ log = logging.getLogger(__name__)
 CHUNK_SIZE = 1024 * 1024
 MAX_MSG = 64 * 1024 * 1024
 
+# Per-account blobs (docs/playlists.md §4.3). These are an unbounded write
+# primitive pointed at somebody else's disk, so every one of them is checked —
+# and every check **refuses**, never truncates. A truncating cap silently loses
+# tracks, which is the one failure the whole playlist design exists to prevent.
+#
+# The numbers are sized against the measured shape: ~300 bytes per track before
+# compression, deflate worth about three on a payload this repetitive. A 1 MB
+# body is therefore roughly ten thousand tracks in one playlist, and the
+# manifest holds names and revisions only.
+USER_BLOB_MANIFEST_MAX = 64 * 1024
+USER_BLOB_BODY_MAX = 1024 * 1024
+USER_BLOB_ACCOUNT_MAX = 8 * 1024 * 1024
+
+# "playlists" is the manifest; "playlist:<id>" is one playlist's tracks. A
+# pattern rather than a set, because the ids are client-generated — but a
+# pattern, not anything at all, or the table becomes a key/value store for
+# whatever a client feels like writing.
+#
+# The character class is deliberately wider than a UUID: the reserved id is the
+# word "favorites" (docs/playlists.md §5.1), so a hex-only pattern refuses the
+# one playlist every account has. It stays narrow enough to carry no structure
+# of its own — no "/", no ".", no second ":" — so a kind can never be read as a
+# path or as anything but one name in one namespace.
+_USER_BLOB_KIND_RE = re.compile(
+    r"^(playlists|playlist:[A-Za-z0-9_-]{1,64})$")
+
 # Chat link-preview results, kept in memory only (draft-v6 §2.7: the node
 # produces enrichment on demand and keeps nothing durable — the asking device
 # caches). Bounded and time-limited so a busy group cannot grow it without end
@@ -682,6 +709,14 @@ class WebRTCPeerSession:
                 self._spawn(self._do_keypair_bundle_store(msg))
             elif mtype == MNP.KEYPAIR_BUNDLE_DELETE:
                 self._spawn(self._do_keypair_bundle_delete())
+            elif mtype == MNP.USER_BLOB_STORE:
+                self._spawn(self._do_user_blob_store(msg))
+            elif mtype == MNP.USER_BLOB_FETCH:
+                self._spawn(self._do_user_blob_fetch(msg))
+            elif mtype == MNP.USER_BLOB_LIST:
+                self._spawn(self._do_user_blob_list())
+            elif mtype == MNP.USER_BLOB_DELETE:
+                self._spawn(self._do_user_blob_delete(msg))
             elif mtype == MNP.STREAM_REQUEST:
                 sem = self._ctx.get("_transcode_sem")
                 log.info("stream: req file=%s credits=%s slots_free=%s prev=%s",
@@ -1136,6 +1171,113 @@ class WebRTCPeerSession:
             "detail": "keypair_bundle_stored",
         })
 
+    # ── Per-account blobs (playlists, docs/playlists.md §8.2) ────────────────
+    #
+    # The node stores bytes it cannot read and hands them back. `self._user_id`
+    # comes from the authenticated session and never from the message: a
+    # `user_id` in the body would let any member of this group read or
+    # overwrite any other member's blob, which is finding C5 one size down.
+
+    def _user_blob_kind(self, msg: dict) -> str | None:
+        """The validated `kind`, or None having already refused."""
+        kind = msg.get("kind")
+        if not isinstance(kind, str) or not _USER_BLOB_KIND_RE.match(kind):
+            self._send({"type": "error", "detail": "Unknown blob kind"})
+            return None
+        return kind
+
+    def _user_blob_store_ok(self):
+        store = self._ctx.get("bundle_store")
+        if not store:
+            self._send({"type": "error", "detail": "Bundle store not available"})
+            return None
+        if not self._user_id:
+            self._send({"type": "error", "detail": "Not authenticated"})
+            return None
+        return store
+
+    async def _do_user_blob_store(self, msg: dict) -> None:
+        store = self._user_blob_store_ok()
+        if not store:
+            return
+        kind = self._user_blob_kind(msg)
+        if not kind:
+            return
+
+        blob = msg.get("blob_enc")
+        if not isinstance(blob, (bytes, bytearray)) or not blob:
+            self._send({"type": "error", "detail": "Missing blob_enc"})
+            return
+        blob = bytes(blob)
+
+        rev = msg.get("rev")
+        if not isinstance(rev, int) or rev < 0:
+            self._send({"type": "error", "detail": "Missing rev"})
+            return
+
+        limit = (USER_BLOB_MANIFEST_MAX if kind == "playlists"
+                 else USER_BLOB_BODY_MAX)
+        if len(blob) > limit:
+            # A stated reason, not a bare error: the client turns this into a
+            # sentence the reader can act on ("this playlist is too large"),
+            # and a refusal nobody can read is a support case.
+            self._send({"type": "error",
+                        "detail": f"Blob too large ({len(blob)} > {limit})"})
+            return
+
+        # What the account already uses, minus whatever this call replaces.
+        used = await store.user_blob_total_bytes(self._user_id)
+        existing = await store.fetch_user_blob(self._user_id, kind)
+        if existing:
+            used -= len(existing["blob_enc"])
+        if used + len(blob) > USER_BLOB_ACCOUNT_MAX:
+            self._send({"type": "error",
+                        "detail": f"Account blob quota exceeded "
+                                  f"({used + len(blob)} > {USER_BLOB_ACCOUNT_MAX})"})
+            return
+
+        await store.store_user_blob(self._user_id, kind, rev, blob)
+        self._audit("user_blob_store", f"{kind} rev={rev} bytes={len(blob)}")
+        self._send({"type": "ack", "v": MNP_VERSION, "detail": "user_blob_stored"})
+
+    async def _do_user_blob_fetch(self, msg: dict) -> None:
+        store = self._user_blob_store_ok()
+        if not store:
+            return
+        kind = self._user_blob_kind(msg)
+        if not kind:
+            return
+        row = await store.fetch_user_blob(self._user_id, kind)
+        self._audit("user_blob_fetch", kind)
+        self._send({
+            "type": MNP.USER_BLOB_RESP, "v": MNP_VERSION, "kind": kind,
+            # A kind this account has never written is `null`, not an error:
+            # "no playlist here yet" is the ordinary state of a fresh node and
+            # the client must not read it as a failure.
+            "rev": row["rev"] if row else None,
+            "blob_enc": row["blob_enc"] if row else None,
+        })
+
+    async def _do_user_blob_list(self) -> None:
+        store = self._user_blob_store_ok()
+        if not store:
+            return
+        blobs = await store.list_user_blobs(self._user_id)
+        self._send({"type": MNP.USER_BLOB_LIST_RESP, "v": MNP_VERSION,
+                    "blobs": blobs})
+
+    async def _do_user_blob_delete(self, msg: dict) -> None:
+        store = self._user_blob_store_ok()
+        if not store:
+            return
+        kind = self._user_blob_kind(msg)
+        if not kind:
+            return
+        removed = await store.delete_user_blob(self._user_id, kind)
+        self._audit("user_blob_delete", kind)
+        self._send({"type": "ack", "v": MNP_VERSION,
+                    "detail": "user_blob_deleted" if removed else "user_blob_absent"})
+
     # ── Pairing and join (H3, M3) ────────────────────────────────────────────
 
     def _join_refuse(self, reason: str, audit_detail: str = "") -> None: