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 os
import threading
import uuid
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from typing import ClassVar

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

    _ENV: ClassVar[tuple[str, ...]] = (
        "NISSE_MATRIX_HOMESERVER",
        "NISSE_MATRIX_USER_ID",
        "NISSE_MATRIX_TOKEN",
    )

    @classmethod
    def from_env(
        cls,
        env: Mapping[str, str] = os.environ,
        room_var: str = "NISSE_MATRIX_ROOM_ID",
    ) -> "MatrixConfig":
        needed = (*cls._ENV, room_var)
        missing = [var for var in needed if not env.get(var)]
        if missing:
            raise TransportError(f"matrix config incomplete; set: {', '.join(missing)}")
        return cls(
            homeserver=env["NISSE_MATRIX_HOMESERVER"].rstrip("/"),
            user_id=env["NISSE_MATRIX_USER_ID"],
            token=env["NISSE_MATRIX_TOKEN"],
            room_id=env[room_var],
        )


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