signal-bot/signal_bot/registry_holder.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The one mutable box the command table lives in, so data changes need no restart.

``PackHolder`` made a *photo* visible the moment it was stored, and said out
loud what it would never do: add a command. A new person in the tagger, a
new alias, a command switched off from the management page -- every one of
those changes the registry itself, and until now every one of them meant
restarting the unit. Restarting is cheap and lossless (signal-cli replays
what arrived meanwhile), but it is a deploy step somebody has to remember,
and a 21:37 report can land in the middle of it.

So the registry is rebuilt, whole, from the same builder that made it at
startup, and swapped in one assignment. Never patched in place: a table
that is edited command by command can be observed half-edited, and a
duplicate trigger found on the third command of five leaves the first two
already live. Building a fresh table first means ``DuplicateTriggerError``
and ``ConfigError`` are raised before anything the dispatcher reads has
changed.

``reload`` never raises on bad *data*. The builder reads files people edit
by hand and pages write mid-session, and the only safe answer to a file
that does not load is the table that was already answering. A programming
error is not data and is not swallowed: it reaches the caller, which for
``POST /_reload`` means a 500 and a traceback in the journal.
"""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING

from signal_bot.groups import ConfigError
from signal_bot.registry import DuplicateTriggerError

if TYPE_CHECKING:
    from collections.abc import Callable

    from signal_bot.registry import CommandRegistry

logger = logging.getLogger(__name__)

# A `type` alias is evaluated lazily, so the TYPE_CHECKING-only names are fine.
type Builder = Callable[[], CommandRegistry]


class RegistryHolder:
    """A mutable reference to the command table the dispatcher reads through."""

    def __init__(self, build: Builder) -> None:
        """Build the first table now, loudly.

        Startup is the one time a bad table must *not* be survived: there is
        no previous one to fall back on, and ``Restart=always`` looping on a
        ConfigError is a clearer signal than a bot that answers nothing.
        """
        self._build = build
        self.registry = build()
        self.last_error = ""

    def reload(self) -> bool:
        """Rebuild the table from whatever is on disk now and publish it.

        Returns:
            True when the new table is live, False when the old one is still
            being served because the data no longer builds -- and then
            ``last_error`` says why, for whoever asked.
        """
        try:
            fresh = self._build()
        except (ConfigError, DuplicateTriggerError) as error:
            self.last_error = str(error)
            logger.exception("command registry no longer builds; keeping the previous one")
            return False
        self.registry = fresh
        self.last_error = ""
        logger.info("command registry rebuilt: %d commands", len(fresh.commands))
        return True