signal-bot/signal_bot/transport_rest.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Transport adapter for bbernhard/signal-cli-rest-api.

Receive is a WebSocket, which the container only exposes when it runs in one
of the json-rpc modes; in normal/native mode the same path silently degrades
to a polling GET, so the mode is verified at startup rather than assumed.
"""

from __future__ import annotations

import asyncio
import logging
from collections.abc import AsyncIterator, Awaitable, Callable
from typing import TYPE_CHECKING, Any, Final

from websockets.asyncio.client import connect as ws_connect
from websockets.exceptions import WebSocketException

if TYPE_CHECKING:
    import httpx

    from signal_bot.config import Config
    from signal_bot.quoting import Quote

logger = logging.getLogger(__name__)

INITIAL_BACKOFF: Final = 1.0
MAX_BACKOFF: Final = 30.0
# Exactly this mode. See verify_mode for why "json-rpc-native" is not enough.
REQUIRED_MODE: Final = "json-rpc"
BACKOFF_FACTOR: Final = 2.0
_OK_STATUSES: Final = frozenset({200, 201, 204})

Connector = Callable[[str], Any]
Sleeper = Callable[[float], Awaitable[None]]


class TransportError(RuntimeError):
    """Raised when the Signal REST API rejects a request."""


class RestTransport:
    """Talks to the signal-cli-rest-api container over REST plus a WebSocket."""

    def __init__(
        self,
        config: Config,
        client: httpx.AsyncClient,
        *,
        connector: Connector | None = None,
        sleeper: Sleeper | None = None,
        max_reconnects: int | None = None,
    ) -> None:
        """Wire the adapter to its config and its injectable I/O.

        Args:
            config: Resolved runtime configuration.
            client: HTTP client used for every REST call.
            connector: WebSocket connect callable. Injected by the tests.
            sleeper: Async sleep used for reconnect backoff.
            max_reconnects: Stop after this many reconnects. None means forever.
        """
        self._config = config
        self._client = client
        self._connect = connector if connector is not None else ws_connect
        self._sleep = sleeper if sleeper is not None else asyncio.sleep
        self._max_reconnects = max_reconnects

    async def verify_mode(self) -> str:
        """Return the container's mode, raising unless it exposes a WebSocket.

        Raises:
            TransportError: The container is not running a json-rpc mode.
        """
        response = await self._client.get(self._config.about_url)
        response.raise_for_status()
        body = response.json()
        mode = str(body.get("mode", "")) if isinstance(body, dict) else ""
        if mode != REQUIRED_MODE:
            msg = (
                f"signal-cli-rest-api is running in mode {mode!r}, not "
                f"{REQUIRED_MODE!r}. 'normal' exposes no receive WebSocket at "
                "all, and 'json-rpc-native' is a GraalVM build with no AWT, so "
                'every attachment send fails with "Can\'t load library: awt" '
                "while text replies keep working. Recreate the container with "
                f"MODE={REQUIRED_MODE}."
            )
            raise TransportError(msg)
        return mode

    async def receive(self) -> AsyncIterator[str]:
        """Yield raw envelope frames, reconnecting with backoff when dropped."""
        backoff = INITIAL_BACKOFF
        reconnects = 0
        while True:
            try:
                async with self._connect(self._config.receive_url) as socket:
                    backoff = INITIAL_BACKOFF
                    async for frame in socket:
                        yield frame if isinstance(frame, str) else frame.decode()
            except (WebSocketException, OSError) as exc:
                logger.warning("signal websocket dropped: %s", exc)
            if self._max_reconnects is not None and reconnects >= self._max_reconnects:
                return
            reconnects += 1
            await self._sleep(backoff)
            backoff = min(backoff * BACKOFF_FACTOR, MAX_BACKOFF)

    async def _post_send(self, payload: dict[str, Any]) -> None:
        """POST one send payload and raise unless the API accepted it."""
        response = await self._client.post(self._config.send_url, json=payload)
        if response.status_code not in _OK_STATUSES:
            msg = f"send failed with HTTP {response.status_code}: {response.text}"
            raise TransportError(msg)

    async def send_text(self, text: str, recipient: str, quote: Quote | None = None) -> None:
        """Send a plain-text message, optionally quoting one message."""
        payload: dict[str, Any] = {
            "message": text,
            "number": self._config.phone_number,
            "recipients": [recipient],
        }
        if quote is not None:
            payload.update(quote.payload())
        await self._post_send(payload)

    async def send_image(
        self, image_base64: str, recipient: str, quote: Quote | None = None
    ) -> None:
        """Send a base64-encoded image attachment, optionally quoting one message."""
        payload: dict[str, Any] = {
            "message": "",
            "base64_attachments": [image_base64],
            "number": self._config.phone_number,
            "recipients": [recipient],
        }
        if quote is not None:
            payload.update(quote.payload())
        await self._post_send(payload)

    async def list_attachments(self) -> list[str]:
        """Return the ids of attachments stored server-side."""
        response = await self._client.get(self._config.attachments_url)
        response.raise_for_status()
        body = response.json()
        if not isinstance(body, list):
            return []
        return [str(item) for item in body]

    async def delete_attachment(self, attachment_id: str) -> None:
        """Delete one stored attachment by id."""
        response = await self._client.delete(f"{self._config.attachments_url}/{attachment_id}")
        if response.status_code not in _OK_STATUSES:
            msg = f"attachment delete failed with HTTP {response.status_code}"
            raise TransportError(msg)