signal-bot/signal_bot/app.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The ASGI entrypoint, and the one endpoint that says whether the bot works.

Wiring lives in bot.py and the startup sequence in startup.py. What is left
here is the part uvicorn imports: an app that begins serving immediately, so
that "the bot is still connecting" is a state you can ask about rather than a
process that is not there.
"""

from __future__ import annotations

import asyncio
import contextlib
import logging
from typing import TYPE_CHECKING

import httpx
from fastapi import FastAPI, Response, status
from pydantic import BaseModel

from signal_bot.config_env import load_config
from signal_bot.selftest import SelfTestError
from signal_bot.selftest import run as run_selftest
from signal_bot.startup_retry import Starter
from signal_bot.store import Store
from signal_bot.versioning import POPE, __version__

if TYPE_CHECKING:
    from collections.abc import AsyncIterator

logger = logging.getLogger(__name__)

HTTP_TIMEOUT_SECONDS = 30.0
# iNaturalist asks automated clients to identify themselves, and Wikimedia's
# policy outright refuses the default "python-httpx/x.y" this used to send.
USER_AGENT = f"signal-bot/{__version__} (+https://github.com/kuhyx/signal-bot)"


@contextlib.asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Serve first, connect to Signal afterwards.

    Two earlier versions of this got it wrong in opposite directions. The
    first awaited asyncio.gather() on two infinite tasks, so uvicorn never
    finished starting and never served at all. The second connected to
    signal-cli inline, so a container that was not ready yet failed startup
    outright and burned the unit's restart limit in 25 seconds.

    Config errors still fail here and exit non-zero: load_config() raising is
    an answer, not an outage, and no amount of waiting fixes a missing
    PHONE_NUMBER. Everything that needs the network is handed to Starter.
    """
    config = load_config()
    client = httpx.AsyncClient(timeout=HTTP_TIMEOUT_SECONDS, headers={"User-Agent": USER_AGENT})
    # Opened once, here, rather than inside the retried connection attempt:
    # a Store per attempt would leak one SQLite handle per failed try, for
    # as long as the container stayed down.
    store = Store(config.database_path)
    starter = Starter(config, client, store)
    app.state.starter = starter
    app.state.client = client
    connecting = asyncio.create_task(starter.run())
    logger.info("serving on %s:%d while connecting to Signal", config.bind_host, config.bind_port)
    try:
        yield
    finally:
        connecting.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await connecting
        if starter.runtime is not None:
            await starter.runtime.stop()
        await client.aclose()
        store.close()


class SelfTestRequest(BaseModel):
    """One command to type at the bot, and where."""

    group: str
    text: str


app = FastAPI(lifespan=lifespan)


@app.get("/healthz")
async def healthz(response: Response) -> dict[str, str]:
    """Liveness probe, and the only place the bot admits it is not working.

    Answering 200 while nothing can reach Signal is what let a three-hour
    outage look like a quiet evening, so this must go red before anybody
    thinks to look at the logs. It has three answers, not two: a bot that has
    not connected yet is "connecting", which is what a booting machine looks
    like, and is not the same thing as one that has gone deaf.
    """
    starter: Starter = app.state.starter
    build = {"version": __version__, "pope": POPE.name}
    runtime = starter.runtime
    if runtime is None:
        response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
        return {"status": "connecting", "reason": starter.reason, **build}
    liveness = runtime.bot.liveness
    if liveness.healthy:
        return {"status": "ok", **build}
    response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
    return {"status": "deaf", "reason": liveness.reason, **build}


@app.post("/_selftest")
async def selftest(request: SelfTestRequest, response: Response) -> dict[str, str]:
    """Type a command at the bot on its own behalf, for verification.

    Bound to loopback with the rest of the app, and refuses any group not
    named in SELFTEST_GROUPS. See signal_bot/selftest.py for why this exists.
    """
    starter: Starter = app.state.starter
    runtime = starter.runtime
    if runtime is None:
        response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
        return {"status": "connecting", "reason": starter.reason}
    try:
        group_id = await run_selftest(runtime.bot, app.state.client, request.group, request.text)
    except SelfTestError as refusal:
        response.status_code = status.HTTP_403_FORBIDDEN
        return {"status": "refused", "reason": str(refusal)}
    return {"status": "delivered", "group": group_id, "text": request.text}