signal-bot/signal_bot/announce.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Telling a group that the bot has been upgraded.

A group is told about every release it has not seen, and a group that has
never been told anything has not seen any of them -- so the first
announcement starts at 0.0.0 "Swiety Piotr Apostol" rather than skipping
straight to whatever is current. Watching the ladder start at Peter is the
point of naming releases after popes at all.

That does mean a group added years from now would be handed the whole
backlog, so the message is capped at MAX_RELEASES and says how many older
versions it left out.

The version is recorded *before* the message is sent. `Restart=always` with
`RestartSec=5` means a crash between the two would otherwise re-announce on
every restart, and a missed announcement is far cheaper than a looping one.
"""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Final

from signal_bot.changelog import since

if TYPE_CHECKING:
    from collections.abc import Iterable

    from signal_bot.changelog import Release
    from signal_bot.store import Store
    from signal_bot.transport_base import Transport

logger = logging.getLogger(__name__)

ANNOUNCED_KEY: Final = "announced_version"

# How many releases one announcement may list before it becomes a wall.
MAX_RELEASES: Final = 10


def render(version: str, pope: str, releases: Iterable[Release]) -> str:
    """Return the message announcing a new version, in Polish.

    One release is rendered flat, because the header already names it. Several
    get a block each, so the group can see the ladder it is climbing.
    """
    shown = list(releases)
    header = f"Wersja {version} \u00ab{pope}\u00bb wydana"
    omitted = max(0, len(shown) - MAX_RELEASES)
    shown = shown[-MAX_RELEASES:]
    if len(shown) == 1:
        body = "\n".join(f"- {line}" for line in shown[0].lines)
        return f"{header}\nChangelog:\n{body}"
    blocks = [
        "\n".join(
            [f"{release.version} \u00ab{release.pope}\u00bb", *(f"- {x}" for x in release.lines)]
        )
        for release in shown
    ]
    tail = f"\n\n(pominieto {omitted} starszych wersji)" if omitted else ""
    return f"{header}\nChangelog:\n\n" + "\n\n".join(blocks) + tail


async def announce_to(
    transport: Transport,
    store: Store,
    group_id: str,
    releases: tuple[Release, ...],
    version: str,
    pope: str,
) -> None:
    """Announce the current version to one group, if it has not seen it."""
    previous = store.meta(group_id, ANNOUNCED_KEY)
    if previous == version:
        return
    # No record at all means the group has seen nothing, not that it should be
    # skipped: that is how Swiety Piotr Apostol gets announced rather than
    # silently assumed.
    pending = releases if previous is None else since(releases, previous)
    store.set_meta(group_id, ANNOUNCED_KEY, version)
    if not pending:
        logger.info("group %s recorded at %s with nothing to announce", group_id, version)
        return
    await transport.send_text(render(version, pope, pending), group_id)


async def announce_all(
    transport: Transport,
    store: Store,
    group_ids: Iterable[str],
    releases: tuple[Release, ...],
    version: str,
    pope: str,
) -> None:
    """Announce to every watched group; one failure must not silence the rest."""
    for group_id in group_ids:
        try:
            await announce_to(transport, store, group_id, releases, version, pope)
        except Exception:
            logger.exception("release announcement failed for group %s", group_id)