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>

1 files changed, 660 insertions, 0 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)