path: root/packages/meshbay-node/src/meshbay_node/transport/webrtc/chat.py

"""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)