refactor(node): move group chat out of webrtc_server

ChatMixin in transport/webrtc/chat.py: sealed messages, history, epoch
keys, link previews with their cache and rate bounds, and the operator's
chat ops.

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

4 files changed, 670 insertions, 654 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/chat.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/chat.py
new file mode 100644
index 0000000..26ec27c
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/chat.py
@@ -0,0 +1,660 @@
+"""Group chat: sealed messages, history, the epoch keys, link previews, and the
+operator's chat settings."""
+
+import base64
+import logging
+import time
+
+import blake3
+from meshbay_common import MNP_VERSION
+from meshbay_common.adminop import OP_CHAT_DIRECTORY, OP_CHAT_EPOCH, OP_CHAT_LINK_PREVIEW
+from meshbay_common.chatbox import NONCE_LEN as CHAT_NONCE_LEN
+from meshbay_common.chatbox import SIG_LEN as CHAT_SIG_LEN
+from meshbay_common.groupbox import PURPOSE_CHAT_KEYS, seal
+from meshbay_common.protocol import MNP
+
+from meshbay_node import linkpreview, ops
+from meshbay_node.chat import FORMAT_SEALED_V1, ReplayedMessage
+
+log = logging.getLogger("meshbay_node.transport.webrtc_server")
+
+
+# Chat link-preview results, kept in memory only (docs/MESHBAY_DESIGN.md §6.5:
+# 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
+# and a page that changed its card is picked up within the hour.
+_LINK_PREVIEW_TTL = 3600
+_LINK_PREVIEW_MAX = 256
+_link_preview_cache: dict[str, tuple[float, dict]] = {}
+
+
+def _link_preview_cache_get(url: str) -> dict | None:
+    hit = _link_preview_cache.get(url)
+    if hit is None:
+        return None
+    ts, value = hit
+    if time.time() - ts > _LINK_PREVIEW_TTL:
+        _link_preview_cache.pop(url, None)
+        return None
+    return value
+
+
+def _link_preview_cache_put(url: str, value: dict) -> None:
+    if not url:
+        return
+    if len(_link_preview_cache) >= _LINK_PREVIEW_MAX:
+        oldest = min(_link_preview_cache, key=lambda k: _link_preview_cache[k][0])
+        _link_preview_cache.pop(oldest, None)
+    _link_preview_cache[url] = (time.time(), value)
+
+
+# A member pasting a link is normal; a member — or a hub minting tokens for many
+# accounts — firing hundreds is amplification/DoS and a way to make the node
+# reach arbitrary hosts on demand (finding M3). Only a real outbound fetch is
+# counted (a cache hit costs nothing), and the ceilings are generous enough that
+# ordinary chat never meets them.
+_LINK_PREVIEW_RATE_WINDOW = 60.0
+_LINK_PREVIEW_RATE_PER_CONN = 15
+_LINK_PREVIEW_RATE_NODE = 60
+
+
+# Chat limits. A message is a member-supplied write onto the operator's disk
+# (`chat.db`, where retention is a manual CLI command — §6.6), relayed from there
+# to every other connected member and turned into a notification for every member
+# of the group. Nothing bounded any of it: the only ceiling was the frame size,
+# 64 MB once the handshake is done, so one member in a loop could fill the
+# operator's disk and saturate everyone else's connection. Uploads — the other
+# member-supplied write — have carried four protections and a size cap since
+# C5a; this is the same question asked of the path nobody had asked it of.
+#
+# 64 KB of ciphertext is about sixty thousand characters. The sealed payload is
+# the text, a thread id, a display name and a timestamp: an attachment is a file
+# on a root and travels as a reference (§4.5), so nothing legitimate comes close.
+MAX_CHAT_CIPHERTEXT = 64 * 1024
+# Per account per group, not per connection: a second tab does not make a person
+# type faster, and keying on the session would hand a script one budget per
+# socket. Sixty a minute is far above a human and far below a flood.
+_CHAT_RATE_WINDOW = 60.0
+_CHAT_RATE_PER_ACCOUNT = 60
+# When the map of senders grows past this, the stale entries are dropped. A node
+# with more live chatters than this in one window is not the case being bounded.
+_CHAT_RATE_MAX_TRACKED = 1000
+
+
+class ChatMixin:
+    async def _new_chat_epoch(self, group_id: str, reason: str) -> None:
+        """
+        Open a chat epoch because the set of devices that may read future
+        messages just shrank.
+
+        Called on every removal — a member, a device, an unpin — and on group
+        key rotation, because the operator rotates precisely when someone has
+        left. It is the exact counterpart of "still rotate the GEK, the
+        ex-member holds the current one": revocation stops the node handing
+        over the *next* key, and nothing else takes the current one away.
+
+        Best effort by design: a failure here must never turn a successful
+        revocation into a refused one — the revocation is the control, and this
+        is the follow-through. It is logged loudly instead, because an operator
+        who removed someone needs to know if the chat key did not move.
+        """
+        if not group_id:
+            return
+        try:
+            result = await self._run_op(ops.open_chat_epoch, group_id)
+        except Exception as e:
+            log.error("chat: could not open a new epoch for group %s after "
+                      "%s (%s) — the removed party still holds the current "
+                      "chat key", group_id[:8], reason, e)
+            self._audit("chat_epoch_failed", reason)
+            return
+        self._audit("chat_epoch", f"{reason}:{result['epoch']}")
+        # Everyone still connected picks the new key up without reconnecting.
+        for session in list(
+                (self._ctx.get("groups") or {}).get(group_id, {})
+                .get("_peers", {}).values()):
+            try:
+                session._send({"type": MNP.CHAT_EPOCH_ACK, "v": MNP_VERSION,
+                               "epoch": result["epoch"]})
+            except Exception:
+                pass
+
+    def _do_chat_directory(self, msg: dict) -> None:
+        """
+        Where chat attachments are written.
+
+        Unlike every other app directory this one is a destination, so it has
+        to be on a read-write root — checked by `ops.set_chat_directory` after
+        the signature, which is where the refusal actually lives.
+        """
+        path = msg.get("path")
+        if not isinstance(path, str):
+            self._send({"type": "error", "detail": "Missing or invalid 'path'"})
+            return
+        path = path.strip("/")
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_CHAT_DIRECTORY, path)
+
+    async def _admin_exec_chat_directory(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        path = pending["subject"]
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"chat_directory:{path}")
+            return
+        try:
+            await self._run_op(
+                ops.set_chat_directory, self._group_id or "", path)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("chat_directory", path)
+        self._broadcast_to_group(
+            {"type": MNP.CHAT_DIRECTORY_ACK, "v": MNP_VERSION, "path": path})
+
+    def _do_chat_link_preview(self, msg: dict) -> None:
+        """
+        Whether the node fetches a page's title and image when a member posts
+        a link — outbound traffic on the operator's connection, from a message
+        they did not write, so it is signed like everything else that decides
+        what leaves this machine.
+        """
+        enabled = msg.get("enabled")
+        if not isinstance(enabled, bool):
+            self._send({"type": "error", "detail": "Missing or invalid 'enabled'"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(
+            OP_CHAT_LINK_PREVIEW, "on" if enabled else "off")
+
+    async def _admin_exec_chat_link_preview(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        enabled = pending["subject"] == "on"
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed",
+                        f"chat_link_preview:{pending['subject']}")
+            return
+        try:
+            await self._run_op(
+                ops.set_chat_link_preview, self._group_id or "", enabled)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("chat_link_preview", pending["subject"])
+        self._broadcast_to_group({"type": MNP.CHAT_LINK_PREVIEW_ACK,
+                                  "v": MNP_VERSION, "enabled": enabled})
+
+    def _do_chat_epoch(self, msg: dict) -> None:
+        """
+        Open a new chat epoch by hand. Operator only, and signed.
+
+        There is no switch to turn chat encryption on: MNP 2.0 has no plaintext
+        chat to fall back to. What an operator may want to do deliberately is
+        move the key on — the same instruction as `gek_rotate`, and signed for
+        the same reason. The removals that matter (member revoke, member unpin,
+        device revoke, `gek_rotate`) already open one by themselves.
+        """
+        group_id = str(msg.get("group_id", "")).strip() or self._group_id
+        if not group_id:
+            self._send({"type": "error", "detail": "No group on this connection"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_CHAT_EPOCH, group_id, group_id=group_id)
+
+    async def _admin_exec_chat_epoch(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"chat_epoch:{pending['subject'][:8]}")
+            return
+        try:
+            result = await self._run_op(ops.open_chat_epoch, pending["subject"])
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("chat_epoch", f"manual:{result['epoch']}")
+        self._broadcast_to_group({"type": MNP.CHAT_EPOCH_ACK,
+                                  "v": MNP_VERSION, "epoch": result["epoch"]})
+
+    async def _do_chat_keys_req(self, msg: dict) -> None:
+        """
+        Hand this member every chat epoch key the group has, sealed.
+
+        Sealed under a group-derived subkey rather than sent in clear: the same
+        reasoning as the index and the handshake ack, and one step stronger
+        here, because the payload *is* key material. A peer that has completed
+        the handshake holds the group key and can open it; anything short of
+        that gets a ciphertext.
+
+        **Every** live epoch, not just the current one, which is what keeps the
+        history readable to a member who joined after it was written and to a
+        device linked this morning. Whether a new member should receive the back
+        catalogue at all is a policy question with a per-group answer; the shape
+        is here so that answer can be given without a wire change.
+        """
+        gctx = self._group_ctx()
+        gek = gctx.get("gek")
+        if not gek:
+            self._send({"type": "error", "detail": "Group encryption not initialized"})
+            return
+        try:
+            keys = await self._run_op(ops.chat_epoch_keys, self._group_id or "")
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+
+        payload = {"epochs": [{"epoch": k["epoch"], "key": k["key"]}
+                              for k in keys],
+                   "current": keys[-1]["epoch"] if keys else 0}
+        sealed = seal(gek, PURPOSE_CHAT_KEYS, MNP.CHAT_KEYS_RESP,
+                      self._group_id or "", payload)
+        self._send({"type": MNP.CHAT_KEYS_RESP, "v": MNP_VERSION,
+                    "group_id": self._group_id or "", **sealed})
+
+    def _do_chat_message(self, msg: dict) -> None:
+        """
+        Store one message and hand it to everyone else in this group.
+
+        The node is a relay and an archive here, not a reader: once a group has
+        chat encryption on, `payload` is a ciphertext it cannot open, and every
+        decision below is made from fields that stay in clear — which is why
+        those fields are the ones that must be *authenticated* rather than
+        merely present.
+
+        `sender_id` comes from the authenticated session and never from the wire
+        (NS6). What the wire may now assert is the sending *device*, and that is
+        checked against this connection rather than believed: a member who could
+        name any device could sign as anyone once receivers verify signatures.
+        """
+        # Per-group store — see _peer_registry() and finding H1. Reading
+        # chat_store off the shared transport context sent every group's
+        # messages to the first group's database, and served them back to
+        # anyone on the node.
+        gctx = self._group_ctx()
+        chat_store = gctx.get("chat_store")
+        sender_name = msg.get("sender_name", "")
+
+        # Two shapes, and keeping them apart is what makes this deployable.
+        #
+        # A plaintext message is exactly what it has always been: a string in
+        # `payload`. A sealed one carries its ciphertext in `ct`, beside the
+        # `nonce`/`device`/`sig` that authenticate it. Putting the ciphertext in
+        # `payload` instead would have been tidier and wrong: `payload` reaches
+        # older clients — the UI ships inside the desktop package now, so it can
+        # be months behind the node — and they would render bytes where they
+        # expect text. A field they have never heard of is ignored instead.
+        fmt = int(msg.get("format", 0) or 0)
+        epoch = int(msg.get("epoch", 0) or 0)
+        device = msg.get("device")
+        nonce = msg.get("nonce")
+        sig = msg.get("sig")
+
+        if fmt == FORMAT_SEALED_V1:
+            payload = ""
+            raw = bytes(msg.get("ct") or b"")
+        else:
+            payload = msg.get("payload", "")
+            raw = (payload.encode() if isinstance(payload, str)
+                   else bytes(payload or b""))
+
+        refusal = self._check_chat_envelope(gctx, fmt, raw, device, nonce, sig)
+        if refusal:
+            self._send({"type": "error", "detail": refusal})
+            self._audit("chat_refused", refusal)
+            return
+
+        # Separate from the envelope check above, and deliberately: that one asks
+        # whether the message is well formed and authentic, this one asks what it
+        # costs everyone else. Before either is written to disk or relayed.
+        detail, code = self._chat_bounds_refusal(raw)
+        if detail:
+            self._send({"type": "error", "detail": detail, "code": code})
+            self._audit("chat_refused", code)
+            return
+
+        if sender_name:
+            self._user_names()[self._user_id] = sender_name
+        if chat_store:
+            self._spawn(self._store_chat_message(
+                chat_store,
+                iteration=msg.get("iteration", 0), payload=raw,
+                thread_id=msg.get("thread_id"), sender_name=sender_name,
+                format=fmt, epoch=epoch, device=device, nonce=nonce, sig=sig,
+            ))
+
+        peers = self._peer_registry()
+        broadcast = {
+            "type": MNP.CHAT_MESSAGE,
+            "v": MNP_VERSION,
+            "sender_id": self._user_id,
+            "sender_name": sender_name,
+            "payload": payload,
+            "thread_id": msg.get("thread_id"),
+            "timestamp": time.time(),
+            "format": fmt,
+            "epoch": epoch,
+            "device": device,
+            "nonce": nonce,
+            "sig": sig,
+        }
+        if fmt == FORMAT_SEALED_V1:
+            broadcast["ct"] = raw
+        # Excludes this connection, not this account. The sender's other
+        # devices are ordinary recipients: they did not compose the message and
+        # have no local echo of it, so skipping them by user_id left a person's
+        # second device silently missing everything they said from the first.
+        for session in list(peers.values()):
+            if session is not self:
+                try:
+                    session._send(broadcast)
+                except Exception:
+                    pass
+
+        hub_ws = self._ctx.get("hub_ws")
+        if hub_ws and self._group_id:
+            try:
+                import json as _json
+                self._spawn(hub_ws.send(_json.dumps({
+                    "type": "chat_notify",
+                    "group_id": self._group_id,
+                    # No sender_name. The body is unreadable to the hub the
+                    # moment a group turns encryption on, and shipping the
+                    # author's display name beside it would leave the hub a
+                    # per-message record of who spoke where — the metadata the
+                    # feature is otherwise about not producing. The hub renders
+                    # "New message in <group>".
+                    #
+                    # `sender_user_id` stays: the hub needs it to not notify
+                    # the author of their own message, and it already knows the
+                    # group's membership.
+                    "sender_user_id": self._user_id,
+                })))
+            except Exception:
+                pass
+
+        self._send({"type": "ack", "v": MNP_VERSION})
+        self._audit("chat_message")
+
+    def _check_chat_envelope(self, gctx: dict, fmt: int, ct: bytes, device,
+                             nonce, sig) -> str:
+        """
+        Why this message is refused, or "" to accept it.
+
+        Two rules, and the first is the one that matters:
+
+        **A device may only send as itself.** `device` is what receivers verify
+        a signature against, so a member free to name another member's key could
+        be that member to everyone — worse than the node-asserted attribution it
+        replaces (NS6), not better. The connection has proved which device it is
+        (`device_hello`), and this must match it.
+
+        **Plaintext is refused, always.** Not "accepts and marks", and not
+        "unless a switch says otherwise": a member who can post in clear into a
+        group whose members believe their chat is encrypted is a downgrade, and
+        C6 is the standing lesson that the bypass left open is the one that gets
+        used. There is no switch to leave open — MNP 2.0 refuses a 1.x peer at
+        the handshake, so nothing that reaches here is unable to seal.
+
+        `FORMAT_PLAIN` still exists, because rows written before 2.0 are still
+        in `chat.db` and still served. It is a *storage* state, never something
+        this accepts from the wire.
+        """
+        if fmt != FORMAT_SEALED_V1:
+            return "Chat messages must be encrypted"
+
+        if not (isinstance(device, (bytes, bytearray))
+                and isinstance(nonce, (bytes, bytearray))
+                and isinstance(sig, (bytes, bytearray))):
+            return "Sealed chat message is missing its envelope"
+        if len(nonce) != CHAT_NONCE_LEN or len(sig) != CHAT_SIG_LEN:
+            return "Sealed chat message has a malformed envelope"
+        if not ct:
+            return "Sealed chat message has no ciphertext"
+
+        claimed = base64.b64encode(bytes(device)).decode()
+        if not self._device_confirmed:
+            return ("Identify this device before sending chat (device_hello)")
+        if claimed != self._pinned_pk:
+            return "That is not the device on this connection"
+
+        return ""
+
+    def _chat_bounds_refusal(self, ct: bytes) -> tuple[str, str]:
+        """What this message would cost the others, or ("", "") to accept it.
+
+        Two bounds, and each answers a different half of "who pays". A message
+        is written to `chat.db` on the operator's disk and kept — retention is a
+        manual command (§6.6) — then relayed to every other connected member and
+        turned into a notification for every member of the group. So **size**
+        bounds what one message costs, and **rate** bounds how often one member
+        may impose it.
+
+        There is deliberately no node-wide ceiling to go with the per-account
+        one. The link-preview limiter has both because a preview spends the
+        *node's* egress and its third-party quota, which is one shared thing; a
+        chat message spends the sender's own group. A node-wide chat ceiling
+        would let a busy group silence a quiet one, which is the same class of
+        defect this bound exists to close, one level up.
+        """
+        if len(ct) > MAX_CHAT_CIPHERTEXT:
+            return ("This message is too large to send in chat — "
+                    "send a large file as an attachment instead.",
+                    "chat_too_large")
+        if not self._chat_rate_ok():
+            return ("Too many messages just now — wait a moment.",
+                    "chat_rate_limited")
+        return ("", "")
+
+    def _chat_rate_ok(self) -> bool:
+        """True when this sender is within their window; records it when so.
+
+        Keyed by (group, account) on the transport context rather than on the
+        session: the sender is authenticated, so this is the one identifier a
+        second tab — or fifty of them — cannot multiply. The window is trimmed
+        on every call, and the map of senders is swept when it grows, so neither
+        can be the memory leak the bound was added to prevent.
+        """
+        now = time.monotonic()
+        hits: dict = self._ctx.setdefault("chat_hits", {})
+        if len(hits) > _CHAT_RATE_MAX_TRACKED:
+            for key, times in list(hits.items()):
+                if not times or now - times[-1] >= _CHAT_RATE_WINDOW:
+                    hits.pop(key, None)
+        key = (self._group_id or "", self._user_id or "")
+        mine = [t for t in hits.get(key, ()) if now - t < _CHAT_RATE_WINDOW]
+        if len(mine) >= _CHAT_RATE_PER_ACCOUNT:
+            hits[key] = mine
+            return False
+        mine.append(now)
+        hits[key] = mine
+        return True
+
+    async def _store_chat_message(self, chat_store, **kwargs) -> None:
+        """
+        Persist one message, treating a replay as already-done.
+
+        A replayed message is a *validly signed* copy of a real one, so nothing
+        about the signature refuses it; the unique `(device, nonce)` does. It is
+        logged and dropped rather than raised at the sender: the message it
+        duplicates is already stored, so there is nothing for anyone to retry.
+        """
+        try:
+            await chat_store.save_message(sender_id=self._user_id, **kwargs)
+        except ReplayedMessage:
+            log.warning("Replayed chat message from %s dropped",
+                        (self._user_id or "?")[:8])
+            self._audit("chat_replay_dropped")
+
+    def _do_chat_history(self, msg: dict) -> None:
+        chat_store = self._group_ctx().get("chat_store")
+        if not chat_store:
+            self._send({
+                "type": MNP.CHAT_HISTORY_RESPONSE,
+                "v": MNP_VERSION,
+                "messages": [],
+                "has_more": False,
+            })
+            return
+
+        # `before` pages backwards from the newest, which is the direction a chat
+        # is actually read. `since` remains for callers that want everything
+        # after a point in time; the browser no longer uses it.
+        before = msg.get("before")
+        limit = max(1, min(int(msg.get("limit", 100)), 200))
+        self._spawn(self._send_chat_history(chat_store, before, limit))
+
+    async def _send_chat_history(self, chat_store, before, limit: int) -> None:
+        if before:
+            msgs = await chat_store.get_before(int(before), limit=limit)
+        else:
+            msgs = await chat_store.get_recent(limit=limit)
+        # Whether the "load older" control has anything left to fetch. Asked
+        # about the oldest row returned, so an empty page correctly says no.
+        has_more = await chat_store.has_before(msgs[0].id) if msgs else False
+        names = self._user_names()
+        self._send({
+            "type": MNP.CHAT_HISTORY_RESPONSE,
+            "v": MNP_VERSION,
+            "has_more": has_more,
+            # `payload` goes out as **bytes**, never decoded here. It used to be
+            # `.decode("utf-8", errors="replace")`, which substitutes U+FFFD for
+            # every byte that is not valid UTF-8 — fine while chat was text, and
+            # silent destruction of a ciphertext. Live messages would have kept
+            # working (they are relayed, not re-read), so the symptom would have
+            # been "history won't decrypt", which is the hardest possible place
+            # to look. msgpack carries `bin` on both sides; the client decides
+            # how to read it from `format`.
+            "messages": [self._history_row(m, names) for m in msgs],
+        })
+
+    @staticmethod
+    def _history_row(m, names: dict) -> dict:
+        """One stored message on the wire.
+
+        A plaintext row goes out under `payload` as a string, exactly as it
+        always has — an older client reads this response and must keep working.
+        A sealed row's ciphertext goes out under `ct` as bytes and `payload`
+        stays empty: decoding a ciphertext as UTF-8 (which is what this did,
+        with `errors="replace"`) substitutes U+FFFD for most of it, and the
+        symptom would have been history that will not decrypt while live
+        messages worked — the hardest possible place to look.
+        """
+        row = {
+            "id": m.id,
+            "sender_id": m.sender_id,
+            "sender_name": m.sender_name or names.get(m.sender_id, ""),
+            "timestamp": m.timestamp,
+            "thread_id": m.thread_id,
+            "format": m.format,
+            "epoch": m.epoch,
+            "device": m.device,
+            "nonce": m.nonce,
+            "sig": m.sig,
+        }
+        if m.format == FORMAT_SEALED_V1:
+            row["payload"] = ""
+            row["ct"] = m.payload
+        else:
+            row["payload"] = (m.payload.decode("utf-8", errors="replace")
+                              if isinstance(m.payload, bytes) else m.payload)
+        return row
+
+    def _link_preview_rate_ok(self) -> bool:
+        """
+        True when this preview fetch is within both the per-connection and the
+        node-wide window; records it when so, and both counts are trimmed to the
+        window on every call so the lists cannot grow without bound.
+        """
+        now = time.monotonic()
+        w = _LINK_PREVIEW_RATE_WINDOW
+        mine = [t for t in getattr(self, "_link_preview_hits", []) if now - t < w]
+        node = [t for t in self._ctx.get("link_preview_hits", []) if now - t < w]
+        if (len(mine) >= _LINK_PREVIEW_RATE_PER_CONN
+                or len(node) >= _LINK_PREVIEW_RATE_NODE):
+            self._link_preview_hits = mine
+            self._ctx["link_preview_hits"] = node
+            return False
+        mine.append(now)
+        node.append(now)
+        self._link_preview_hits = mine
+        self._ctx["link_preview_hits"] = node
+        return True
+
+    async def _do_link_preview_request(self, msg: dict) -> None:
+        """
+        Unfurl a URL a member pasted into chat (docs/MESHBAY_DESIGN.md §6.5's
+        enrichment rule:
+        the client asks, the node produces on demand, the asking device
+        caches — nothing durable here).
+
+        `linkpreview.safe_url` is the SSRF gate: the URL a *member* chose
+        decides an outbound request from the operator's machine, so http(s)
+        only and the resolved address must be globally routable. Failure of
+        any kind — blocked, unreachable, not HTML, nothing worth showing —
+        comes back as `ok: false`, the way a TMDB miss does; the client then
+        just shows the bare link.
+        """
+        url = msg.get("url")
+        key = url if isinstance(url, str) else ""
+
+        # Checked before the cache, not after: the operator turning previews
+        # off has to stop serving the ones already fetched too, or the setting
+        # takes effect only for links nobody has posted yet. Refused as an
+        # ordinary miss — the client shows the bare link, which is exactly what
+        # "no preview" looks like for a page that has none.
+        if not self._group_ctx().get("chat_link_preview", True):
+            self._send({"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION,
+                        "url": key, "ok": False})
+            return
+
+        cached = _link_preview_cache_get(key)
+        if cached is not None:
+            self._send({**cached, "type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION})
+            return
+
+        if not self._link_preview_rate_ok():
+            # Same shape as any other miss — the client shows the bare link. A
+            # rate-limited result is not cached, so it is retried once the
+            # window clears rather than pinned as "no preview".
+            log.debug("link_preview_req: rate-limited (peer=%s)", self._peer_id)
+            self._send({"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION,
+                        "url": key, "ok": False})
+            return
+
+        resp: dict = {"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION,
+                      "url": key, "ok": False}
+        try:
+            meta = await linkpreview.fetch_preview(url)
+            if meta is not None:
+                resp.update(ok=True, title=meta["title"],
+                            description=meta["description"],
+                            site_name=meta["site_name"])
+                image_url = meta.get("image_url")
+                media_cache = self._ctx.get("media_cache")
+                if image_url and media_cache is not None:
+                    synthetic_id = f"linkpreview:{image_url}"
+                    thumb_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id)
+                    if thumb_hash is None:
+                        jpeg = await linkpreview.fetch_image(image_url)
+                        if jpeg:
+                            thumb_hash = blake3.blake3(jpeg).hexdigest()
+                            await media_cache.put_thumb(thumb_hash, synthetic_id, jpeg)
+                    if thumb_hash:
+                        resp["image_thumb_hash"] = thumb_hash
+        except Exception as e:
+            log.debug("link_preview_req %s: %s", key[:80], e)
+
+        _link_preview_cache_put(key, {k: v for k, v in resp.items()
+                                      if k not in ("type", "v")})
+        self._send(resp)
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 77f2721..889c411 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -32,7 +32,6 @@ import uuid
 from pathlib import Path
 from typing import Any
 
-import blake3
 from aiortc import RTCDataChannel, RTCPeerConnection, RTCSessionDescription
 from cryptography.hazmat.primitives.asymmetric.ed25519 import (
     Ed25519PrivateKey,
@@ -71,12 +70,6 @@ from meshbay_common.adminop import (
     OP_TRANSFER_LIMITS,
     admin_transcript,
 )
-from meshbay_common.chatbox import (
-    NONCE_LEN as CHAT_NONCE_LEN,
-)
-from meshbay_common.chatbox import (
-    SIG_LEN as CHAT_SIG_LEN,
-)
 from meshbay_common.crypto import pk_to_b64, wrap_gek_aes
 from meshbay_common.device import (
     DEVICE_TTL,
@@ -86,7 +79,6 @@ from meshbay_common.device import (
 )
 from meshbay_common.groupbox import (
     PURPOSE_ACK,
-    PURPOSE_CHAT_KEYS,
     PURPOSE_ROSTER,
     seal,
 )
@@ -118,10 +110,9 @@ from meshbay_common.protocol import (
     file_upload_payload,
 )
 
-from meshbay_node import linkpreview, ops
+from meshbay_node import ops
 from meshbay_node import transfers as transfers_mod
 from meshbay_node import uploads as uploads_mod
-from meshbay_node.chat import FORMAT_SEALED_V1, ReplayedMessage
 from meshbay_node.indexer import GroupIndex
 from meshbay_node.indexer.indexer import DirectoryIndexer
 
@@ -151,6 +142,7 @@ from meshbay_node.transport.webrtc.channel import (
     _get_remote_ip,
     _pack,
 )
+from meshbay_node.transport.webrtc.chat import ChatMixin
 from meshbay_node.transport.webrtc.disk import _locate
 from meshbay_node.transport.webrtc.limits import CHUNK_SIZE, MAX_MSG
 from meshbay_node.transport.wire import index_sync_message
@@ -160,66 +152,6 @@ log = logging.getLogger(__name__)
 # An invitation link's handle, as `roster.create_link_invite` mints it.
 _INVITE_ID_RE = re.compile(r"[0-9a-f]{32}")
 
-# Chat link-preview results, kept in memory only (docs/MESHBAY_DESIGN.md §6.5:
-# 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
-# and a page that changed its card is picked up within the hour.
-_LINK_PREVIEW_TTL = 3600
-_LINK_PREVIEW_MAX = 256
-_link_preview_cache: dict[str, tuple[float, dict]] = {}
-
-
-def _link_preview_cache_get(url: str) -> dict | None:
-    hit = _link_preview_cache.get(url)
-    if hit is None:
-        return None
-    ts, value = hit
-    if time.time() - ts > _LINK_PREVIEW_TTL:
-        _link_preview_cache.pop(url, None)
-        return None
-    return value
-
-
-def _link_preview_cache_put(url: str, value: dict) -> None:
-    if not url:
-        return
-    if len(_link_preview_cache) >= _LINK_PREVIEW_MAX:
-        oldest = min(_link_preview_cache, key=lambda k: _link_preview_cache[k][0])
-        _link_preview_cache.pop(oldest, None)
-    _link_preview_cache[url] = (time.time(), value)
-
-
-# A member pasting a link is normal; a member — or a hub minting tokens for many
-# accounts — firing hundreds is amplification/DoS and a way to make the node
-# reach arbitrary hosts on demand (finding M3). Only a real outbound fetch is
-# counted (a cache hit costs nothing), and the ceilings are generous enough that
-# ordinary chat never meets them.
-_LINK_PREVIEW_RATE_WINDOW = 60.0
-_LINK_PREVIEW_RATE_PER_CONN = 15
-_LINK_PREVIEW_RATE_NODE = 60
-
-# Chat limits. A message is a member-supplied write onto the operator's disk
-# (`chat.db`, where retention is a manual CLI command — §6.6), relayed from there
-# to every other connected member and turned into a notification for every member
-# of the group. Nothing bounded any of it: the only ceiling was the frame size,
-# 64 MB once the handshake is done, so one member in a loop could fill the
-# operator's disk and saturate everyone else's connection. Uploads — the other
-# member-supplied write — have carried four protections and a size cap since
-# C5a; this is the same question asked of the path nobody had asked it of.
-#
-# 64 KB of ciphertext is about sixty thousand characters. The sealed payload is
-# the text, a thread id, a display name and a timestamp: an attachment is a file
-# on a root and travels as a reference (§4.5), so nothing legitimate comes close.
-MAX_CHAT_CIPHERTEXT = 64 * 1024
-# Per account per group, not per connection: a second tab does not make a person
-# type faster, and keying on the session would hand a script one budget per
-# socket. Sixty a minute is far above a human and far below a flood.
-_CHAT_RATE_WINDOW = 60.0
-_CHAT_RATE_PER_ACCOUNT = 60
-# When the map of senders grows past this, the stale entries are dropped. A node
-# with more live chatters than this in one window is not the case being bounded.
-_CHAT_RATE_MAX_TRACKED = 1000
 
 # Upload limits (finding C5a). Uploads used to land directly in the shared root under
 # a name the client chose, overwriting whatever was already there — which both violated
@@ -297,7 +229,7 @@ _WEBRTC_TRACE_INTERVAL_S = 30.0
 
 
 class WebRTCPeerSession(
-    BlobsMixin,
+    BlobsMixin, ChatMixin,
     StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin,
 ):
     """One WebRTC peer connection, handling MNP over a DataChannel."""
@@ -1553,43 +1485,6 @@ class WebRTCPeerSession(
             ],
         })
 
-    async def _new_chat_epoch(self, group_id: str, reason: str) -> None:
-        """
-        Open a chat epoch because the set of devices that may read future
-        messages just shrank.
-
-        Called on every removal — a member, a device, an unpin — and on group
-        key rotation, because the operator rotates precisely when someone has
-        left. It is the exact counterpart of "still rotate the GEK, the
-        ex-member holds the current one": revocation stops the node handing
-        over the *next* key, and nothing else takes the current one away.
-
-        Best effort by design: a failure here must never turn a successful
-        revocation into a refused one — the revocation is the control, and this
-        is the follow-through. It is logged loudly instead, because an operator
-        who removed someone needs to know if the chat key did not move.
-        """
-        if not group_id:
-            return
-        try:
-            result = await self._run_op(ops.open_chat_epoch, group_id)
-        except Exception as e:
-            log.error("chat: could not open a new epoch for group %s after "
-                      "%s (%s) — the removed party still holds the current "
-                      "chat key", group_id[:8], reason, e)
-            self._audit("chat_epoch_failed", reason)
-            return
-        self._audit("chat_epoch", f"{reason}:{result['epoch']}")
-        # Everyone still connected picks the new key up without reconnecting.
-        for session in list(
-                (self._ctx.get("groups") or {}).get(group_id, {})
-                .get("_peers", {}).values()):
-            try:
-                session._send({"type": MNP.CHAT_EPOCH_ACK, "v": MNP_VERSION,
-                               "epoch": result["epoch"]})
-            except Exception:
-                pass
-
     async def _do_device_revoke(self, msg: dict) -> None:
         """
         Retire one of this account's devices — a lost laptop.
@@ -2140,78 +2035,6 @@ class WebRTCPeerSession(
 
     # ── Chat ─────────────────────────────────────────────────────────────
 
-    def _do_chat_directory(self, msg: dict) -> None:
-        """
-        Where chat attachments are written.
-
-        Unlike every other app directory this one is a destination, so it has
-        to be on a read-write root — checked by `ops.set_chat_directory` after
-        the signature, which is where the refusal actually lives.
-        """
-        path = msg.get("path")
-        if not isinstance(path, str):
-            self._send({"type": "error", "detail": "Missing or invalid 'path'"})
-            return
-        path = path.strip("/")
-        if not self._has_admin_authority():
-            self._send({"type": "error", "detail": "No authorized key for this"})
-            return
-        self._issue_admin_challenge(OP_CHAT_DIRECTORY, path)
-
-    async def _admin_exec_chat_directory(
-        self, pending: dict, transcript: bytes, sig: bytes,
-    ) -> None:
-        path = pending["subject"]
-        if not await self._verify_admin_sig(transcript, sig):
-            self._send({"type": "error", "detail": "Signature verification failed"})
-            self._audit("admin_auth_failed", f"chat_directory:{path}")
-            return
-        try:
-            await self._run_op(
-                ops.set_chat_directory, self._group_id or "", path)
-        except ops.OpError as e:
-            self._send({"type": "error", "detail": e.message})
-            return
-        self._audit("chat_directory", path)
-        self._broadcast_to_group(
-            {"type": MNP.CHAT_DIRECTORY_ACK, "v": MNP_VERSION, "path": path})
-
-    def _do_chat_link_preview(self, msg: dict) -> None:
-        """
-        Whether the node fetches a page's title and image when a member posts
-        a link — outbound traffic on the operator's connection, from a message
-        they did not write, so it is signed like everything else that decides
-        what leaves this machine.
-        """
-        enabled = msg.get("enabled")
-        if not isinstance(enabled, bool):
-            self._send({"type": "error", "detail": "Missing or invalid 'enabled'"})
-            return
-        if not self._has_admin_authority():
-            self._send({"type": "error", "detail": "No authorized key for this"})
-            return
-        self._issue_admin_challenge(
-            OP_CHAT_LINK_PREVIEW, "on" if enabled else "off")
-
-    async def _admin_exec_chat_link_preview(
-        self, pending: dict, transcript: bytes, sig: bytes,
-    ) -> None:
-        enabled = pending["subject"] == "on"
-        if not await self._verify_admin_sig(transcript, sig):
-            self._send({"type": "error", "detail": "Signature verification failed"})
-            self._audit("admin_auth_failed",
-                        f"chat_link_preview:{pending['subject']}")
-            return
-        try:
-            await self._run_op(
-                ops.set_chat_link_preview, self._group_id or "", enabled)
-        except ops.OpError as e:
-            self._send({"type": "error", "detail": e.message})
-            return
-        self._audit("chat_link_preview", pending["subject"])
-        self._broadcast_to_group({"type": MNP.CHAT_LINK_PREVIEW_ACK,
-                                  "v": MNP_VERSION, "enabled": enabled})
-
     def _do_search_listed(self, msg: dict) -> None:
         """
         Whether this group's files appear in members' cross-group Search.
@@ -2244,41 +2067,6 @@ class WebRTCPeerSession(
         self._broadcast_to_group({"type": MNP.SEARCH_LISTED_ACK,
                                   "v": MNP_VERSION, "listed": listed})
 
-    def _do_chat_epoch(self, msg: dict) -> None:
-        """
-        Open a new chat epoch by hand. Operator only, and signed.
-
-        There is no switch to turn chat encryption on: MNP 2.0 has no plaintext
-        chat to fall back to. What an operator may want to do deliberately is
-        move the key on — the same instruction as `gek_rotate`, and signed for
-        the same reason. The removals that matter (member revoke, member unpin,
-        device revoke, `gek_rotate`) already open one by themselves.
-        """
-        group_id = str(msg.get("group_id", "")).strip() or self._group_id
-        if not group_id:
-            self._send({"type": "error", "detail": "No group on this connection"})
-            return
-        if not self._has_admin_authority():
-            self._send({"type": "error", "detail": "No authorized key for this"})
-            return
-        self._issue_admin_challenge(OP_CHAT_EPOCH, group_id, group_id=group_id)
-
-    async def _admin_exec_chat_epoch(
-        self, pending: dict, transcript: bytes, sig: bytes,
-    ) -> None:
-        if not await self._verify_admin_sig(transcript, sig):
-            self._send({"type": "error", "detail": "Signature verification failed"})
-            self._audit("admin_auth_failed", f"chat_epoch:{pending['subject'][:8]}")
-            return
-        try:
-            result = await self._run_op(ops.open_chat_epoch, pending["subject"])
-        except ops.OpError as e:
-            self._send({"type": "error", "detail": e.message})
-            return
-        self._audit("chat_epoch", f"manual:{result['epoch']}")
-        self._broadcast_to_group({"type": MNP.CHAT_EPOCH_ACK,
-                                  "v": MNP_VERSION, "epoch": result["epoch"]})
-
     async def _do_group_roster_req(self, msg: dict) -> None:
         """
         Who is in this group, and which device keys they hold.
@@ -2319,41 +2107,6 @@ class WebRTCPeerSession(
         self._send({"type": MNP.GROUP_ROSTER_RESP, "v": MNP_VERSION,
                     "group_id": self._group_id or "", **sealed})
 
-    async def _do_chat_keys_req(self, msg: dict) -> None:
-        """
-        Hand this member every chat epoch key the group has, sealed.
-
-        Sealed under a group-derived subkey rather than sent in clear: the same
-        reasoning as the index and the handshake ack, and one step stronger
-        here, because the payload *is* key material. A peer that has completed
-        the handshake holds the group key and can open it; anything short of
-        that gets a ciphertext.
-
-        **Every** live epoch, not just the current one, which is what keeps the
-        history readable to a member who joined after it was written and to a
-        device linked this morning. Whether a new member should receive the back
-        catalogue at all is a policy question with a per-group answer; the shape
-        is here so that answer can be given without a wire change.
-        """
-        gctx = self._group_ctx()
-        gek = gctx.get("gek")
-        if not gek:
-            self._send({"type": "error", "detail": "Group encryption not initialized"})
-            return
-        try:
-            keys = await self._run_op(ops.chat_epoch_keys, self._group_id or "")
-        except ops.OpError as e:
-            self._send({"type": "error", "detail": e.message})
-            return
-
-        payload = {"epochs": [{"epoch": k["epoch"], "key": k["key"]}
-                              for k in keys],
-                   "current": keys[-1]["epoch"] if keys else 0}
-        sealed = seal(gek, PURPOSE_CHAT_KEYS, MNP.CHAT_KEYS_RESP,
-                      self._group_id or "", payload)
-        self._send({"type": MNP.CHAT_KEYS_RESP, "v": MNP_VERSION,
-                    "group_id": self._group_id or "", **sealed})
-
     def _broadcast_to_group(self, notice: dict) -> None:
         """
         Tell everyone connected to this group about a setting that changed.
@@ -3515,240 +3268,6 @@ class WebRTCPeerSession(
         if not leased and (chunk_index + 1) * CHUNK_SIZE >= entry.size:
             self._leaseless.finish(str(file_id))
 
-    def _do_chat_message(self, msg: dict) -> None:
-        """
-        Store one message and hand it to everyone else in this group.
-
-        The node is a relay and an archive here, not a reader: once a group has
-        chat encryption on, `payload` is a ciphertext it cannot open, and every
-        decision below is made from fields that stay in clear — which is why
-        those fields are the ones that must be *authenticated* rather than
-        merely present.
-
-        `sender_id` comes from the authenticated session and never from the wire
-        (NS6). What the wire may now assert is the sending *device*, and that is
-        checked against this connection rather than believed: a member who could
-        name any device could sign as anyone once receivers verify signatures.
-        """
-        # Per-group store — see _peer_registry() and finding H1. Reading
-        # chat_store off the shared transport context sent every group's
-        # messages to the first group's database, and served them back to
-        # anyone on the node.
-        gctx = self._group_ctx()
-        chat_store = gctx.get("chat_store")
-        sender_name = msg.get("sender_name", "")
-
-        # Two shapes, and keeping them apart is what makes this deployable.
-        #
-        # A plaintext message is exactly what it has always been: a string in
-        # `payload`. A sealed one carries its ciphertext in `ct`, beside the
-        # `nonce`/`device`/`sig` that authenticate it. Putting the ciphertext in
-        # `payload` instead would have been tidier and wrong: `payload` reaches
-        # older clients — the UI ships inside the desktop package now, so it can
-        # be months behind the node — and they would render bytes where they
-        # expect text. A field they have never heard of is ignored instead.
-        fmt = int(msg.get("format", 0) or 0)
-        epoch = int(msg.get("epoch", 0) or 0)
-        device = msg.get("device")
-        nonce = msg.get("nonce")
-        sig = msg.get("sig")
-
-        if fmt == FORMAT_SEALED_V1:
-            payload = ""
-            raw = bytes(msg.get("ct") or b"")
-        else:
-            payload = msg.get("payload", "")
-            raw = (payload.encode() if isinstance(payload, str)
-                   else bytes(payload or b""))
-
-        refusal = self._check_chat_envelope(gctx, fmt, raw, device, nonce, sig)
-        if refusal:
-            self._send({"type": "error", "detail": refusal})
-            self._audit("chat_refused", refusal)
-            return
-
-        # Separate from the envelope check above, and deliberately: that one asks
-        # whether the message is well formed and authentic, this one asks what it
-        # costs everyone else. Before either is written to disk or relayed.
-        detail, code = self._chat_bounds_refusal(raw)
-        if detail:
-            self._send({"type": "error", "detail": detail, "code": code})
-            self._audit("chat_refused", code)
-            return
-
-        if sender_name:
-            self._user_names()[self._user_id] = sender_name
-        if chat_store:
-            self._spawn(self._store_chat_message(
-                chat_store,
-                iteration=msg.get("iteration", 0), payload=raw,
-                thread_id=msg.get("thread_id"), sender_name=sender_name,
-                format=fmt, epoch=epoch, device=device, nonce=nonce, sig=sig,
-            ))
-
-        peers = self._peer_registry()
-        broadcast = {
-            "type": MNP.CHAT_MESSAGE,
-            "v": MNP_VERSION,
-            "sender_id": self._user_id,
-            "sender_name": sender_name,
-            "payload": payload,
-            "thread_id": msg.get("thread_id"),
-            "timestamp": time.time(),
-            "format": fmt,
-            "epoch": epoch,
-            "device": device,
-            "nonce": nonce,
-            "sig": sig,
-        }
-        if fmt == FORMAT_SEALED_V1:
-            broadcast["ct"] = raw
-        # Excludes this connection, not this account. The sender's other
-        # devices are ordinary recipients: they did not compose the message and
-        # have no local echo of it, so skipping them by user_id left a person's
-        # second device silently missing everything they said from the first.
-        for session in list(peers.values()):
-            if session is not self:
-                try:
-                    session._send(broadcast)
-                except Exception:
-                    pass
-
-        hub_ws = self._ctx.get("hub_ws")
-        if hub_ws and self._group_id:
-            try:
-                import json as _json
-                self._spawn(hub_ws.send(_json.dumps({
-                    "type": "chat_notify",
-                    "group_id": self._group_id,
-                    # No sender_name. The body is unreadable to the hub the
-                    # moment a group turns encryption on, and shipping the
-                    # author's display name beside it would leave the hub a
-                    # per-message record of who spoke where — the metadata the
-                    # feature is otherwise about not producing. The hub renders
-                    # "New message in <group>".
-                    #
-                    # `sender_user_id` stays: the hub needs it to not notify
-                    # the author of their own message, and it already knows the
-                    # group's membership.
-                    "sender_user_id": self._user_id,
-                })))
-            except Exception:
-                pass
-
-        self._send({"type": "ack", "v": MNP_VERSION})
-        self._audit("chat_message")
-
-    def _check_chat_envelope(self, gctx: dict, fmt: int, ct: bytes, device,
-                             nonce, sig) -> str:
-        """
-        Why this message is refused, or "" to accept it.
-
-        Two rules, and the first is the one that matters:
-
-        **A device may only send as itself.** `device` is what receivers verify
-        a signature against, so a member free to name another member's key could
-        be that member to everyone — worse than the node-asserted attribution it
-        replaces (NS6), not better. The connection has proved which device it is
-        (`device_hello`), and this must match it.
-
-        **Plaintext is refused, always.** Not "accepts and marks", and not
-        "unless a switch says otherwise": a member who can post in clear into a
-        group whose members believe their chat is encrypted is a downgrade, and
-        C6 is the standing lesson that the bypass left open is the one that gets
-        used. There is no switch to leave open — MNP 2.0 refuses a 1.x peer at
-        the handshake, so nothing that reaches here is unable to seal.
-
-        `FORMAT_PLAIN` still exists, because rows written before 2.0 are still
-        in `chat.db` and still served. It is a *storage* state, never something
-        this accepts from the wire.
-        """
-        if fmt != FORMAT_SEALED_V1:
-            return "Chat messages must be encrypted"
-
-        if not (isinstance(device, (bytes, bytearray))
-                and isinstance(nonce, (bytes, bytearray))
-                and isinstance(sig, (bytes, bytearray))):
-            return "Sealed chat message is missing its envelope"
-        if len(nonce) != CHAT_NONCE_LEN or len(sig) != CHAT_SIG_LEN:
-            return "Sealed chat message has a malformed envelope"
-        if not ct:
-            return "Sealed chat message has no ciphertext"
-
-        claimed = base64.b64encode(bytes(device)).decode()
-        if not self._device_confirmed:
-            return ("Identify this device before sending chat (device_hello)")
-        if claimed != self._pinned_pk:
-            return "That is not the device on this connection"
-
-        return ""
-
-    def _chat_bounds_refusal(self, ct: bytes) -> tuple[str, str]:
-        """What this message would cost the others, or ("", "") to accept it.
-
-        Two bounds, and each answers a different half of "who pays". A message
-        is written to `chat.db` on the operator's disk and kept — retention is a
-        manual command (§6.6) — then relayed to every other connected member and
-        turned into a notification for every member of the group. So **size**
-        bounds what one message costs, and **rate** bounds how often one member
-        may impose it.
-
-        There is deliberately no node-wide ceiling to go with the per-account
-        one. The link-preview limiter has both because a preview spends the
-        *node's* egress and its third-party quota, which is one shared thing; a
-        chat message spends the sender's own group. A node-wide chat ceiling
-        would let a busy group silence a quiet one, which is the same class of
-        defect this bound exists to close, one level up.
-        """
-        if len(ct) > MAX_CHAT_CIPHERTEXT:
-            return ("This message is too large to send in chat — "
-                    "send a large file as an attachment instead.",
-                    "chat_too_large")
-        if not self._chat_rate_ok():
-            return ("Too many messages just now — wait a moment.",
-                    "chat_rate_limited")
-        return ("", "")
-
-    def _chat_rate_ok(self) -> bool:
-        """True when this sender is within their window; records it when so.
-
-        Keyed by (group, account) on the transport context rather than on the
-        session: the sender is authenticated, so this is the one identifier a
-        second tab — or fifty of them — cannot multiply. The window is trimmed
-        on every call, and the map of senders is swept when it grows, so neither
-        can be the memory leak the bound was added to prevent.
-        """
-        now = time.monotonic()
-        hits: dict = self._ctx.setdefault("chat_hits", {})
-        if len(hits) > _CHAT_RATE_MAX_TRACKED:
-            for key, times in list(hits.items()):
-                if not times or now - times[-1] >= _CHAT_RATE_WINDOW:
-                    hits.pop(key, None)
-        key = (self._group_id or "", self._user_id or "")
-        mine = [t for t in hits.get(key, ()) if now - t < _CHAT_RATE_WINDOW]
-        if len(mine) >= _CHAT_RATE_PER_ACCOUNT:
-            hits[key] = mine
-            return False
-        mine.append(now)
-        hits[key] = mine
-        return True
-
-    async def _store_chat_message(self, chat_store, **kwargs) -> None:
-        """
-        Persist one message, treating a replay as already-done.
-
-        A replayed message is a *validly signed* copy of a real one, so nothing
-        about the signature refuses it; the unique `(device, nonce)` does. It is
-        logged and dropped rather than raised at the sender: the message it
-        duplicates is already stored, so there is nothing for anyone to retry.
-        """
-        try:
-            await chat_store.save_message(sender_id=self._user_id, **kwargs)
-        except ReplayedMessage:
-            log.warning("Replayed chat message from %s dropped",
-                        (self._user_id or "?")[:8])
-            self._audit("chat_replay_dropped")
-
     def _do_ping(self, msg: dict) -> None:
         """Answer a liveness probe on an open channel, echoing the caller's token.
 
@@ -3758,169 +3277,6 @@ class WebRTCPeerSession(
         """
         self._send({"type": MNP.PONG, "v": MNP_VERSION, "token": msg.get("token")})
 
-    def _do_chat_history(self, msg: dict) -> None:
-        chat_store = self._group_ctx().get("chat_store")
-        if not chat_store:
-            self._send({
-                "type": MNP.CHAT_HISTORY_RESPONSE,
-                "v": MNP_VERSION,
-                "messages": [],
-                "has_more": False,
-            })
-            return
-
-        # `before` pages backwards from the newest, which is the direction a chat
-        # is actually read. `since` remains for callers that want everything
-        # after a point in time; the browser no longer uses it.
-        before = msg.get("before")
-        limit = max(1, min(int(msg.get("limit", 100)), 200))
-        self._spawn(self._send_chat_history(chat_store, before, limit))
-
-    async def _send_chat_history(self, chat_store, before, limit: int) -> None:
-        if before:
-            msgs = await chat_store.get_before(int(before), limit=limit)
-        else:
-            msgs = await chat_store.get_recent(limit=limit)
-        # Whether the "load older" control has anything left to fetch. Asked
-        # about the oldest row returned, so an empty page correctly says no.
-        has_more = await chat_store.has_before(msgs[0].id) if msgs else False
-        names = self._user_names()
-        self._send({
-            "type": MNP.CHAT_HISTORY_RESPONSE,
-            "v": MNP_VERSION,
-            "has_more": has_more,
-            # `payload` goes out as **bytes**, never decoded here. It used to be
-            # `.decode("utf-8", errors="replace")`, which substitutes U+FFFD for
-            # every byte that is not valid UTF-8 — fine while chat was text, and
-            # silent destruction of a ciphertext. Live messages would have kept
-            # working (they are relayed, not re-read), so the symptom would have
-            # been "history won't decrypt", which is the hardest possible place
-            # to look. msgpack carries `bin` on both sides; the client decides
-            # how to read it from `format`.
-            "messages": [self._history_row(m, names) for m in msgs],
-        })
-
-    @staticmethod
-    def _history_row(m, names: dict) -> dict:
-        """One stored message on the wire.
-
-        A plaintext row goes out under `payload` as a string, exactly as it
-        always has — an older client reads this response and must keep working.
-        A sealed row's ciphertext goes out under `ct` as bytes and `payload`
-        stays empty: decoding a ciphertext as UTF-8 (which is what this did,
-        with `errors="replace"`) substitutes U+FFFD for most of it, and the
-        symptom would have been history that will not decrypt while live
-        messages worked — the hardest possible place to look.
-        """
-        row = {
-            "id": m.id,
-            "sender_id": m.sender_id,
-            "sender_name": m.sender_name or names.get(m.sender_id, ""),
-            "timestamp": m.timestamp,
-            "thread_id": m.thread_id,
-            "format": m.format,
-            "epoch": m.epoch,
-            "device": m.device,
-            "nonce": m.nonce,
-            "sig": m.sig,
-        }
-        if m.format == FORMAT_SEALED_V1:
-            row["payload"] = ""
-            row["ct"] = m.payload
-        else:
-            row["payload"] = (m.payload.decode("utf-8", errors="replace")
-                              if isinstance(m.payload, bytes) else m.payload)
-        return row
-
-    def _link_preview_rate_ok(self) -> bool:
-        """
-        True when this preview fetch is within both the per-connection and the
-        node-wide window; records it when so, and both counts are trimmed to the
-        window on every call so the lists cannot grow without bound.
-        """
-        now = time.monotonic()
-        w = _LINK_PREVIEW_RATE_WINDOW
-        mine = [t for t in getattr(self, "_link_preview_hits", []) if now - t < w]
-        node = [t for t in self._ctx.get("link_preview_hits", []) if now - t < w]
-        if (len(mine) >= _LINK_PREVIEW_RATE_PER_CONN
-                or len(node) >= _LINK_PREVIEW_RATE_NODE):
-            self._link_preview_hits = mine
-            self._ctx["link_preview_hits"] = node
-            return False
-        mine.append(now)
-        node.append(now)
-        self._link_preview_hits = mine
-        self._ctx["link_preview_hits"] = node
-        return True
-
-    async def _do_link_preview_request(self, msg: dict) -> None:
-        """
-        Unfurl a URL a member pasted into chat (docs/MESHBAY_DESIGN.md §6.5's
-        enrichment rule:
-        the client asks, the node produces on demand, the asking device
-        caches — nothing durable here).
-
-        `linkpreview.safe_url` is the SSRF gate: the URL a *member* chose
-        decides an outbound request from the operator's machine, so http(s)
-        only and the resolved address must be globally routable. Failure of
-        any kind — blocked, unreachable, not HTML, nothing worth showing —
-        comes back as `ok: false`, the way a TMDB miss does; the client then
-        just shows the bare link.
-        """
-        url = msg.get("url")
-        key = url if isinstance(url, str) else ""
-
-        # Checked before the cache, not after: the operator turning previews
-        # off has to stop serving the ones already fetched too, or the setting
-        # takes effect only for links nobody has posted yet. Refused as an
-        # ordinary miss — the client shows the bare link, which is exactly what
-        # "no preview" looks like for a page that has none.
-        if not self._group_ctx().get("chat_link_preview", True):
-            self._send({"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION,
-                        "url": key, "ok": False})
-            return
-
-        cached = _link_preview_cache_get(key)
-        if cached is not None:
-            self._send({**cached, "type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION})
-            return
-
-        if not self._link_preview_rate_ok():
-            # Same shape as any other miss — the client shows the bare link. A
-            # rate-limited result is not cached, so it is retried once the
-            # window clears rather than pinned as "no preview".
-            log.debug("link_preview_req: rate-limited (peer=%s)", self._peer_id)
-            self._send({"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION,
-                        "url": key, "ok": False})
-            return
-
-        resp: dict = {"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION,
-                      "url": key, "ok": False}
-        try:
-            meta = await linkpreview.fetch_preview(url)
-            if meta is not None:
-                resp.update(ok=True, title=meta["title"],
-                            description=meta["description"],
-                            site_name=meta["site_name"])
-                image_url = meta.get("image_url")
-                media_cache = self._ctx.get("media_cache")
-                if image_url and media_cache is not None:
-                    synthetic_id = f"linkpreview:{image_url}"
-                    thumb_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id)
-                    if thumb_hash is None:
-                        jpeg = await linkpreview.fetch_image(image_url)
-                        if jpeg:
-                            thumb_hash = blake3.blake3(jpeg).hexdigest()
-                            await media_cache.put_thumb(thumb_hash, synthetic_id, jpeg)
-                    if thumb_hash:
-                        resp["image_thumb_hash"] = thumb_hash
-        except Exception as e:
-            log.debug("link_preview_req %s: %s", key[:80], e)
-
-        _link_preview_cache_put(key, {k: v for k, v in resp.items()
-                                      if k not in ("type", "v")})
-        self._send(resp)
-
     def _partial_uploads(self, ctx: dict) -> uploads_mod.PartialUploads:
         """This group's uploads in progress, created on first use.
 
diff --git a/packages/meshbay-node/tests/test_chat_is_bounded.py b/packages/meshbay-node/tests/test_chat_is_bounded.py
index 5dc1128..3332601 100644
--- a/packages/meshbay-node/tests/test_chat_is_bounded.py
+++ b/packages/meshbay-node/tests/test_chat_is_bounded.py
@@ -36,7 +36,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from meshbay_common.chatbox import NONCE_LEN, SIG_LEN
 from meshbay_common.crypto import pk_to_b64
 from meshbay_node.chat import FORMAT_SEALED_V1, ChatStore
-from meshbay_node.transport import webrtc_server as ws
+from meshbay_node.transport.webrtc import chat as ws
 from meshbay_node.transport.webrtc_server import WebRTCPeerSession
 
 # Read with a default rather than imported. Against the source these were
diff --git a/packages/meshbay-node/tests/test_link_preview_request.py b/packages/meshbay-node/tests/test_link_preview_request.py
index fe7dec6..0e1bd06 100644
--- a/packages/meshbay-node/tests/test_link_preview_request.py
+++ b/packages/meshbay-node/tests/test_link_preview_request.py
@@ -13,7 +13,7 @@ import pytest
 from meshbay_common.protocol import MNP
 from meshbay_node import linkpreview
 from meshbay_node.media_cache import MediaCache
-from meshbay_node.transport import webrtc_server
+from meshbay_node.transport.webrtc import chat
 from meshbay_node.transport.webrtc_server import WebRTCPeerSession
 
 pytestmark = pytest.mark.asyncio
@@ -29,9 +29,9 @@ async def media_cache(tmp_path):
 
 @pytest.fixture(autouse=True)
 def _clear_cache():
-    webrtc_server._link_preview_cache.clear()
+    chat._link_preview_cache.clear()
     yield
-    webrtc_server._link_preview_cache.clear()
+    chat._link_preview_cache.clear()
 
 
 def _session(media_cache, ctx=None):
@@ -83,7 +83,7 @@ async def test_unfurlable_failure_is_ok_false(media_cache, monkeypatch):
 async def test_rate_limit_per_connection(media_cache, monkeypatch):
     """A member firing many previews is bounded; over the ceiling the reply is
     a plain `ok: false` (bare link) and no outbound fetch is made."""
-    monkeypatch.setattr(webrtc_server, "_LINK_PREVIEW_RATE_PER_CONN", 3)
+    monkeypatch.setattr(chat, "_LINK_PREVIEW_RATE_PER_CONN", 3)
     calls = {"n": 0}
 
     async def counting_preview(url, **k):
@@ -105,8 +105,8 @@ async def test_rate_limit_per_connection(media_cache, monkeypatch):
 
 async def test_rate_limit_is_node_wide(media_cache, monkeypatch):
     """Two connections share the node-wide ceiling."""
-    monkeypatch.setattr(webrtc_server, "_LINK_PREVIEW_RATE_PER_CONN", 100)
-    monkeypatch.setattr(webrtc_server, "_LINK_PREVIEW_RATE_NODE", 2)
+    monkeypatch.setattr(chat, "_LINK_PREVIEW_RATE_PER_CONN", 100)
+    monkeypatch.setattr(chat, "_LINK_PREVIEW_RATE_NODE", 2)
     calls = {"n": 0}
 
     async def counting_preview(url, **k):