signal-bot/signal_bot/groups.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Group ids: the three spellings, and the one that wins.

Split out of config.py because it is the whole reason the bot can watch more
than one group without answering in the wrong one. The REST API reports a
group as ``group.<base64>``, an envelope carries what that decodes to, and
only the first form can be sent to -- so every id is normalised here, once,
before anything downstream compares or addresses one.
"""

from __future__ import annotations

import base64
import binascii
from typing import TYPE_CHECKING, Final

if TYPE_CHECKING:
    from collections.abc import Iterable

GROUP_ID_PREFIX: Final = "group."


class ConfigError(RuntimeError):
    """Raised when the environment does not describe a runnable bot."""


def group_id_aliases(group_id: str) -> frozenset[str]:
    """Every spelling of one group id that can arrive in an envelope.

    The REST API reports group ids two different ways and never says so:
    ``/v1/groups`` returns ``group.<base64>`` (what the setup script writes
    into .env), while an incoming envelope's ``groupInfo.groupId`` carries
    the *internal* id -- which is exactly what that base64 decodes to. The
    two never compare equal, so a bot configured from the API's own output
    silently ignores every message in its own group. Verified live on
    2026-08-29: ``group.WTI5dVptbG...`` decodes to ``Y29uZmlnLWdy/...``, the id
    the same endpoint reports as ``internal_id``.

    Args:
        group_id: The configured group id, in either spelling.

    Returns:
        The configured id plus its decoded internal form, when it has one.
    """
    aliases = {group_id}
    if group_id.startswith(GROUP_ID_PREFIX):
        try:
            decoded = base64.b64decode(group_id.removeprefix(GROUP_ID_PREFIX), validate=True)
        except binascii.Error, ValueError:
            return frozenset(aliases)
        aliases.add(decoded.decode("utf-8", errors="replace"))
    else:
        aliases.add(canonical_group_id(group_id))
    return frozenset(aliases)


def canonical_group_id(group_id: str) -> str:
    """Return the ``group.<base64>`` spelling of a group id.

    One spelling has to win, because the same string is used for three
    different jobs: matching an incoming envelope, addressing a reply, and
    keying per-group state. The prefixed form wins because it is the only
    one ``/v2/send`` and the group REST paths accept -- the internal form
    contains ``/`` and ``+``. The internal id is itself base64 text, so the
    conversion is just the inverse of what group_id_aliases decodes.

    Args:
        group_id: A group id in either spelling.

    Returns:
        The id in ``group.<base64>`` form, unchanged if it already is one.
    """
    if group_id.startswith(GROUP_ID_PREFIX):
        return group_id
    return GROUP_ID_PREFIX + base64.b64encode(group_id.encode("utf-8")).decode("ascii")


def build_alias_map(group_ids: Iterable[str]) -> dict[str, str]:
    """Map every spelling of every configured group to its canonical id.

    Args:
        group_ids: Configured group ids, already canonical.

    Returns:
        A lookup from any spelling an envelope might carry to the one
        canonical id that scopes state and addresses replies.
    """
    mapping: dict[str, str] = {}
    for group_id in group_ids:
        for alias in group_id_aliases(group_id):
            mapping[alias] = group_id
    return mapping


def parse_group_ids(env: dict[str, str]) -> tuple[str, ...]:
    """Return the watched groups, canonical and de-duplicated.

    ``GROUP_IDS`` is the list form; ``GROUP_ID`` remains accepted as a
    one-element spelling so an existing .env keeps working. Setting both is
    an error rather than a precedence rule, because a precedence rule means
    silently ignoring one of two values the operator clearly meant.

    Args:
        env: The environment mapping to read.

    Returns:
        Canonical group ids, in the order they were listed.

    Raises:
        ConfigError: Neither key is set, both are set, or a removed key is.
    """
    if env.get("GROUP_ID_SEND", "").strip():
        msg = (
            "GROUP_ID_SEND has been removed. Replies and the daily report now "
            "go back to the group the message came from, which a fixed send "
            "target cannot express once the bot watches more than one group. "
            "Delete the line from .env."
        )
        raise ConfigError(msg)

    listed = env.get("GROUP_IDS", "").strip()
    single = env.get("GROUP_ID", "").strip()
    if listed and single:
        msg = "set GROUP_IDS or GROUP_ID, not both; GROUP_ID is the one-group spelling"
        raise ConfigError(msg)

    raw = listed or single
    if not raw:
        msg = (
            "GROUP_IDS is required and must not be empty. A linked Signal "
            "device sees every chat on the account, so the bot refuses to "
            "start without an explicit list of groups to scope itself to. "
            "Find them with: curl -s $API_BASE_URL/v1/groups/$PHONE_NUMBER"
        )
        raise ConfigError(msg)

    group_ids: list[str] = []
    seen: dict[str, str] = {}
    for part in raw.split(","):
        candidate = part.strip()
        if not candidate:
            continue
        canonical = canonical_group_id(candidate)
        aliases = group_id_aliases(canonical)
        clash = next((seen[alias] for alias in aliases if alias in seen), None)
        if clash is not None:
            msg = f"group {candidate!r} is listed twice, once as {clash!r}"
            raise ConfigError(msg)
        seen.update(dict.fromkeys(aliases, canonical))
        group_ids.append(canonical)

    if not group_ids:
        msg = f"GROUP_IDS lists no group ids, only separators: {raw!r}"
        raise ConfigError(msg)
    return tuple(group_ids)