signal-bot/signal_bot/config.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Runtime configuration, read once from the environment and validated eagerly."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Final
from urllib.parse import urlsplit, urlunsplit

from signal_bot.endpoints import (
    DEFAULT_CAT_API_URL,
    DEFAULT_COMMONS_API_URL,
    DEFAULT_DOG_API_URL,
    DEFAULT_GECKO_API_URL,
)

# Re-exported: every caller has always imported these from config, and the
# split is a filing change, not an interface change.
from signal_bot.groups import (
    ConfigError,
    build_alias_map,
    canonical_group_id,
    group_id_aliases,
)
from signal_bot.trust import parse_trusted_inviters

__all__ = [
    "DEFAULT_CAT_API_URL",
    "DEFAULT_COMMONS_API_URL",
    "DEFAULT_DOG_API_URL",
    "DEFAULT_GECKO_API_URL",
    "Config",
    "ConfigError",
    "canonical_group_id",
    "group_id_aliases",
    "parse_trusted_inviters",
]

DEFAULT_API_BASE_URL: Final = "http://localhost:9922"
DEFAULT_DATABASE_PATH: Final = "signal_bot.sqlite3"
DEFAULT_REPORT_HOUR: Final = 21
DEFAULT_REPORT_MINUTE: Final = 37
DEFAULT_COOLDOWN_SECONDS: Final = 10
DEFAULT_BIND_HOST: Final = "127.0.0.1"
# Not 8000: that port is a magnet for unrelated dev servers and was already
# taken here by a long-running enforcer. 8137 sits below the ephemeral range
# (/proc/sys/net/ipv4/ip_local_port_range, 32768-60999 on Arch), so the kernel
# will never hand it out to an outbound connection and steal it back.
DEFAULT_BIND_PORT: Final = 8137
MAX_HOUR: Final = 23
MAX_MINUTE: Final = 59
MAX_PORT: Final = 65535
DEFAULT_LOG_LEVEL: Final = "INFO"
# How long the bot may hear nothing at all before it stops believing its own
# health. Silence is normal -- a group can be quiet all evening -- so this is
# only the point at which silence becomes worth *asking* about.
DEFAULT_SILENCE_SECONDS: Final = 120

_WS_SCHEMES: Final = {"http": "ws", "https": "wss"}


@dataclass(frozen=True, slots=True)
class Config:
    """Everything the bot needs to know, resolved and validated."""

    phone_number: str
    # The optional allowlist. Empty means every group the account is a
    # full member of, discovered at runtime -- see signal_bot.discovery.
    group_ids: tuple[str, ...] = ()
    api_base_url: str = DEFAULT_API_BASE_URL
    database_path: str = DEFAULT_DATABASE_PATH
    report_hour: int = DEFAULT_REPORT_HOUR
    report_minute: int = DEFAULT_REPORT_MINUTE
    cooldown_seconds: int = DEFAULT_COOLDOWN_SECONDS
    # Overridable so the offline environment can serve fixture images
    # without the bot knowing it is not talking to the real APIs.
    cat_api_url: str = DEFAULT_CAT_API_URL
    dog_api_url: str = DEFAULT_DOG_API_URL
    gecko_api_url: str = DEFAULT_GECKO_API_URL
    commons_api_url: str = DEFAULT_COMMONS_API_URL
    # Where the /healthz readiness probe binds. Read here rather than in
    # __main__ so that .env governs it: __main__ runs before anything has
    # loaded a .env file, so an os.getenv() there would silently ignore it.
    bind_host: str = DEFAULT_BIND_HOST
    bind_port: int = DEFAULT_BIND_PORT
    # Set LOG_LEVEL=DEBUG to see why a frame was ignored. Without a knob,
    # the one log line that explains a silently-dropped message is
    # unreachable in production, which is how a scoping mismatch turns
    # into a bot that looks healthy and answers nothing.
    log_level: str = DEFAULT_LOG_LEVEL
    # Who may add this bot to a group. Empty means anyone. See signal_bot.trust.
    trusted_inviters: tuple[str, ...] = ()
    # Lowered in the e2e run so the deafness gate can be proven against the
    # real process rather than only in unit tests.
    silence_seconds: int = DEFAULT_SILENCE_SECONDS

    @property
    def is_pinned(self) -> bool:
        """True when GROUP_IDS restricts the bot to a fixed set of groups."""
        return bool(self.group_ids)

    def allows(self, group_id: str) -> bool:
        """True when a discovered group is permitted by the allowlist.

        An empty allowlist permits everything: the bot answers wherever it
        has been added. This is only the *permission* check -- whether the
        account is actually a member is decided by the live listing, in
        signal_bot.discovery.

        Args:
            group_id: A canonical group id from the group listing.

        Returns:
            True when the group may be watched.
        """
        if not self.group_ids:
            return True
        return group_id in build_alias_map(self.group_ids)

    @property
    def receive_url(self) -> str:
        """WebSocket URL the bot subscribes to for incoming envelopes."""
        parts = urlsplit(self.api_base_url)
        scheme = _WS_SCHEMES.get(parts.scheme, parts.scheme)
        path = f"/v1/receive/{self.phone_number}"
        query = "send_read_receipts=false"
        return urlunsplit((scheme, parts.netloc, path, query, ""))

    @property
    def send_url(self) -> str:
        """REST endpoint for outgoing messages and attachments."""
        return f"{self.api_base_url}/v2/send"

    @property
    def attachments_url(self) -> str:
        """REST endpoint listing server-side stored attachments."""
        return f"{self.api_base_url}/v1/attachments"

    @property
    def about_url(self) -> str:
        """REST endpoint reporting the container build and its mode."""
        return f"{self.api_base_url}/v1/about"

    @property
    def devices_url(self) -> str:
        """The one endpoint that fails when signal-cli cannot reach Signal.

        Verified on 2026-08-29, in both directions: 200 with a device list
        while healthy, and 400 "Failed to get linked devices: Closed
        unexpectedly" during the outage -- while /v1/groups and /v1/about
        kept answering 200 from local state and the container's own
        healthcheck stayed green. Local state is why the bot could look
        perfectly healthy for three hours with a dead connection.
        """
        return f"{self.api_base_url}/v1/devices/{self.phone_number}"

    @property
    def groups_url(self) -> str:
        """REST endpoint listing the groups this account knows about."""
        return f"{self.api_base_url}/v1/groups/{self.phone_number}"

    @property
    def image_api_urls(self) -> dict[str, str]:
        """Image-API endpoint per command name, as the commands want it."""
        return {"cat": self.cat_api_url, "dog": self.dog_api_url, "gecko": self.gecko_api_url}