signal-bot/signal_bot/registry.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Command lookup: trigger words to handlers.

Reimplements the idea behind the unmerged ``rules.py`` branch: triggers are
data, not a chain of if-statements, so aliases (including the Unicode-styled
ones people actually type) cost one tuple entry each.
"""

from __future__ import annotations

from collections.abc import Awaitable, Callable, Iterable
from dataclasses import dataclass

from signal_bot.context import CommandContext
from signal_bot.folding import fold

COMMAND_PREFIX = "!"


Handler = Callable[[CommandContext], Awaitable[None]]

# Turns the words after a trigger into whatever the handler is allowed to see.
# For the pack that means pseudonym keys and nothing else: the raw remainder
# stops here, so no handler, log line or command_uses row ever holds a name
# somebody typed. A command without one gets the words unchanged, which is
# what every command before this feature did.
ArgResolver = Callable[[tuple[str, ...]], tuple[str, ...]]


@dataclass(frozen=True, slots=True)
class Command:
    """A named command, every phrase that triggers it, and how to explain it."""

    name: str
    triggers: frozenset[str]
    handler: Handler
    # Group-facing Polish, one line, for !pomoc. Required at registration so
    # that a command cannot exist without one -- a parallel help table would
    # go stale exactly the way the README's command list did.
    description: str
    # The spelling !pomoc shows. It cannot come from `triggers`: those are
    # folded skeletons in an unordered set, so "!papiez" is all that survives
    # of "!papież". This is the first raw trigger instead, which is the
    # canonical Polish one for every entry in aliases.json.
    primary: str
    # The only groups this command exists in, or None for everywhere. Not a
    # gate inside the handler: the dispatcher treats a command that is not
    # allowed here as no command at all, so it is never answered, never
    # counted as a command and never logged by name in a group it does not
    # belong to. A silent handler would still leak all three.
    groups: frozenset[str] | None = None
    # Whether words after the trigger are arguments rather than nonsense.
    # Off by default, so "!cat blah" keeps resolving to nothing exactly as
    # it does today.
    takes_args: bool = False
    # Applied to those words before they reach CommandContext.args. The
    # scrubbing seam -- see ArgResolver.
    arg_resolver: ArgResolver | None = None

    def allowed_in(self, group_id: str) -> bool:
        """Whether this command exists in one group."""
        return self.groups is None or group_id in self.groups


class DuplicateTriggerError(ValueError):
    """Raised when two commands that can both run claim the same trigger word.

    "Can both run" is the whole test. Two albums scoped to different groups
    may each define "!alfa" -- they are different people, and no message can
    ever mean both, because a command a group is not scoped to is not a
    command there at all. A collision is only a collision where the group
    sets overlap, and then it is fatal at startup rather than a coin toss.
    """


class CommandRefusedError(Exception):
    """A gate declined the command, so nothing happened and nothing is recorded.

    The dispatcher records one use per invocation, and it cannot see inside a
    handler. Without this signal a *refused* run looked exactly like a
    successful one, which slid the cooldown window forward every time somebody
    was told to wait -- so waiting the stated time never helped.
    """


@dataclass(frozen=True, slots=True)
class Resolved:
    """A command a message fired, and the words that followed it."""

    command: Command
    args: tuple[str, ...]


class CommandRegistry:
    """Maps trigger words to commands."""

    def __init__(self) -> None:
        """Start with no commands registered."""
        # One trigger can be claimed by several commands, as long as no two
        # of them are ever live in the same group. See DuplicateTriggerError.
        self._by_trigger: dict[str, list[Command]] = {}
        self._commands: list[Command] = []

    def register(
        self,
        name: str,
        triggers: Iterable[str],
        handler: Handler,
        description: str,
        *,
        groups: frozenset[str] | None = None,
        takes_args: bool = False,
        arg_resolver: ArgResolver | None = None,
    ) -> Command:
        """Register one command under every one of its triggers.

        Raises:
            DuplicateTriggerError: A trigger is already claimed.
        """
        raw = list(triggers)
        normalised = frozenset(fold(t) for t in raw)
        for trigger in sorted(normalised):
            clash = next(
                (c for c in self._by_trigger.get(trigger, ()) if _overlap(c.groups, groups)),
                None,
            )
            if clash is not None:
                msg = f"trigger {trigger!r} is already registered by {clash.name!r}"
                raise DuplicateTriggerError(msg)
        command = Command(
            name=name,
            triggers=normalised,
            handler=handler,
            description=description,
            primary=raw[0],
            groups=groups,
            takes_args=takes_args,
            arg_resolver=arg_resolver,
        )
        for trigger in normalised:
            self._by_trigger.setdefault(trigger, []).append(command)
        self._commands.append(command)
        return command

    def _claimant(self, trigger: str, group_id: str | None) -> Command | None:
        """The command that owns one folded trigger, in one group.

        ``group_id`` None means "anywhere", which is what a caller asking
        "is this word a command at all?" wants -- pack_add uses it to refuse
        minting a label that already means something.
        """
        claimants = self._by_trigger.get(trigger, [])
        if group_id is None:
            return claimants[0] if claimants else None
        return next((c for c in claimants if c.allowed_in(group_id)), None)

    def resolve(self, text: str | None, group_id: str | None = None) -> Command | None:
        """Return the command a message body triggers in one group, or None.

        Without a group this answers "does anything claim this word", which
        is a different and deliberately broader question.
        """
        if not text:
            return None
        # Folded, not just lowercased: a Cyrillic "o" is a different
        # codepoint from a Latin one and used to resolve to nothing at all.
        candidate = fold(text)
        if not candidate.startswith(COMMAND_PREFIX):
            return None
        return self._claimant(candidate, group_id)

    def resolve_with_args(self, text: str | None, group_id: str) -> Resolved | None:
        """Resolve a message in one group, with anything that followed it.

        Two lookups, in this order. The whole text first, which is what every
        command has always been matched by and stays byte-for-byte the same
        rule -- so "!cat blah" still resolves to nothing. Only then the first
        word, and only for a command that says it takes arguments.

        A command the group is not allowed is not a near miss: it comes back
        as None, indistinguishable from a word nobody registered.
        """
        if not text:
            return None
        command = self.resolve(text, group_id)
        if command is not None:
            return Resolved(command, ())
        words = text.split()
        if not words:
            return None
        head = self._claimant(fold(words[0]), group_id)
        if head is None or not head.takes_args:
            return None
        tail = tuple(words[1:])
        # Applied here rather than in the handler, which is the whole point:
        # a resolver that runs later has already let the raw words past the
        # log line the dispatcher writes when it names the command.
        if head.arg_resolver is not None:
            tail = head.arg_resolver(tail)
        return Resolved(head, tail)

    @property
    def commands(self) -> list[Command]:
        """Every registered command, in registration order."""
        return list(self._commands)


def _overlap(one: frozenset[str] | None, other: frozenset[str] | None) -> bool:
    """Whether two group scopes can ever both be live for one message.

    None means "every group", so it overlaps with anything -- including
    another None.
    """
    if one is None or other is None:
        return True
    return not one.isdisjoint(other)