signal-bot/signal_bot/stats/streaks.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Who holds the crown, and for how long.

A tie shares the crown and keeps every tied streak alive. That is a decision,
not an oversight: breaking somebody's forty-day run on a technicality is the
kind of thing that starts an argument in a group chat, and "we were both first"
is what the people involved would say anyway.

Streaks count *recorded* days, not calendar days. A day nobody spoke does not
end a run -- there was no crown to lose -- and neither does a day the bot was
offline for, which it could not have judged.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from collections.abc import Iterable

    from signal_bot.stats.events import EventStore

DAILY_COUNTS = """
    SELECT day, uuid, display_name, COUNT(*) AS messages
    FROM message_events
    WHERE group_id = ? AND day <> ?
    GROUP BY day, uuid
    ORDER BY day ASC, messages DESC, display_name ASC
"""


@dataclass(frozen=True, slots=True)
class Streak:
    """A run of consecutive recorded days holding the crown."""

    uuid: str
    display_name: str
    length: int
    last_day: str


def daily_winners(
    events: EventStore,
    group_id: str,
    *,
    exclude: str | None = None,
) -> dict[str, list[tuple[str, str]]]:
    """Who led each recorded day, as (uuid, display_name), ties included."""
    rows = events.connection.execute(DAILY_COUNTS, (group_id, exclude or "")).fetchall()
    best: dict[str, int] = {}
    winners: dict[str, list[tuple[str, str]]] = {}
    for row in rows:
        day, count = row["day"], row["messages"]
        # Ordered messages-first, so the day's leader is always the row that
        # creates the entry and everything after it either ties or loses.
        if day not in best:
            best[day] = count
            winners[day] = [(row["uuid"], row["display_name"])]
        elif count == best[day]:
            winners[day].append((row["uuid"], row["display_name"]))
    return winners


def current_streaks(winners: dict[str, list[tuple[str, str]]]) -> list[Streak]:
    """Everyone whose run reaches the most recent recorded day.

    Sorted longest first, then by name, so two people on the same run are
    reported in a stable order rather than whichever the dict happened to
    yield.
    """
    if not winners:
        return []
    days = sorted(winners)
    latest = days[-1]
    holders = dict(winners[latest])

    lengths = dict.fromkeys(holders, 0)
    running = set(holders)
    for day in reversed(days):
        today = {uuid for uuid, _ in winners[day]}
        for uuid in sorted(running):
            if uuid in today:
                lengths[uuid] += 1
            else:
                # Their run ended on this day. What they had already
                # accumulated is kept -- dropping them from the tally, as an
                # earlier spelling did, reported every streak as zero.
                running.discard(uuid)
        if not running:
            break

    return sorted(
        (Streak(uuid, holders[uuid], length, latest) for uuid, length in lengths.items()),
        key=lambda s: (-s.length, s.display_name),
    )


def longest_ever(winners: dict[str, list[tuple[str, str]]]) -> Streak | None:
    """The longest crown run anybody has ever put together."""
    running: dict[str, int] = {}
    names: dict[str, str] = {}
    best: Streak | None = None

    for day in sorted(winners):
        today = dict(winners[day])
        names.update(today)
        running = {uuid: running.get(uuid, 0) + 1 for uuid in today}
        for uuid, length in running.items():
            if (
                best is None
                or length > best.length
                or (length == best.length and names[uuid] < best.display_name)
            ):
                best = Streak(uuid, names[uuid], length, day)
    return best


def held_days(winners: dict[str, list[tuple[str, str]]], uuid: str) -> int:
    """How many days somebody has held the crown in total, ties included."""
    return sum(1 for holders in winners.values() if any(u == uuid for u, _ in holders))


def names_of(holders: Iterable[tuple[str, str]]) -> list[str]:
    """Display names of a day's crown holders, alphabetical for stability."""
    return sorted(name for _, name in holders)