nisse/nisse-dev/src/nisse_dev/matrix.py

"""MatrixTransport (spec A1): Matrix as strictly a UI channel.

Implements the Transport interface directly over the Matrix client-server HTTP
API (token auth, /sync long-poll, /send) with the httpx client already in the
dependency tree — no Matrix SDK. The surface used is four endpoints; an SDK
earns its keep when E2E arrives, and bot rooms are created unencrypted until
then, so encrypted rooms are not supported here.

Inbound handling follows B2's polling posture: on homeserver errors the sync
loop backs off (capped) and retries indefinitely — an outage delays reactions
but never fails anything.
"""

import threading
import uuid
from collections.abc import Callable
from dataclasses import dataclass

import httpx

from nisse_dev.transport import InboundMessage


class TransportError(RuntimeError):
    pass


@dataclass(frozen=True)
class MatrixConfig:
    homeserver: str
    user_id: str
    token: str
    room_id: str


class MatrixTransport:
    def __init__(
        self,
        config: MatrixConfig,
        *,
        poll_timeout_ms: int = 30000,
        http_transport: httpx.BaseTransport | None = None,
    ) -> None:
        self._config = config
        self._poll_timeout_ms = poll_timeout_ms
        self._http = httpx.Client(
            base_url=f"{config.homeserver}/_matrix/client/v3",
            headers={"Authorization": f"Bearer {config.token}"},
            timeout=(poll_timeout_ms / 1000) + 30,
            transport=http_transport,
        )
        self._callback: Callable[[InboundMessage], None] | None = None
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None

    # -- outbound ---------------------------------------------------------

    def send_message(self, text: str) -> None:
        self._send(text)

    def send_approval_request(self, text: str) -> None:
        self._send(text)

    def ask_question(self, text: str) -> None:
        self._send(text)

    def _send(self, text: str) -> None:
        room = self._config.room_id
        txn = uuid.uuid4().hex
        try:
            response = self._http.put(
                f"/rooms/{room}/send/m.room.message/{txn}",
                json={"msgtype": "m.text", "body": text},
            )
        except httpx.HTTPError as exc:
            raise TransportError(f"matrix send failed: {exc}") from exc
        if response.status_code != 200:
            raise TransportError(
                f"matrix send returned {response.status_code}: {response.text[:300]}"
            )

    # -- inbound ----------------------------------------------------------

    def on_message(self, callback: Callable[[InboundMessage], None]) -> None:
        self._callback = callback

    def start(self) -> None:
        """Begin delivering new room messages to the registered callback.
        History from before start() is never delivered (Matrix room history
        must not become load-bearing context, A1)."""
        if self._callback is None:
            raise TransportError("no message handler registered; call on_message first")
        self._stop.clear()
        self._thread = threading.Thread(target=self._sync_loop, daemon=True)
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()
        if self._thread is not None:
            self._thread.join(timeout=(self._poll_timeout_ms / 1000) + 5)
            self._thread = None

    def _sync_loop(self) -> None:
        since: str | None = None
        backoff = 1.0
        while not self._stop.is_set():
            params: dict[str, str] = {}
            if since is None:
                params["timeout"] = "0"  # first sync: learn position, skip backlog
            else:
                params["timeout"] = str(self._poll_timeout_ms)
                params["since"] = since
            try:
                response = self._http.get("/sync", params=params)
                response.raise_for_status()
            except httpx.HTTPError:
                self._stop.wait(backoff)
                backoff = min(backoff * 2, 60.0)
                continue
            backoff = 1.0
            data = response.json()
            if since is not None:
                self._deliver(data)
            since = data.get("next_batch", since)

    def _deliver(self, sync_data: dict) -> None:
        room = sync_data.get("rooms", {}).get("join", {}).get(self._config.room_id, {})
        for event in room.get("timeline", {}).get("events", []):
            if event.get("type") != "m.room.message":
                continue
            sender = event.get("sender", "")
            if sender == self._config.user_id:
                continue
            body = event.get("content", {}).get("body")
            if not body:
                continue
            assert self._callback is not None
            self._callback(InboundMessage(sender=sender, text=body))

    def close(self) -> None:
        self.stop()
        self._http.close()