signal-bot/signal_bot/analytics/backfill_join.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Attaching a command name from the journal to a message that was one.

Split out of ``backfill.py`` so both stay under the 250-line cap. This half
is the part that can be wrong, so it is the part that refuses to guess.

The direction matters and is easy to get backwards. Iterating the journal
and looking for a message means resolving a display name to a uuid and a
group, which the journal does not know. Iterating ``message_events`` instead
means every row already carries the group, the uuid, the day and the real
hour, and the journal supplies the one field missing -- the command's name.
"""

from __future__ import annotations

from collections import Counter, defaultdict
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Final

from signal_bot.analytics.uses import (
    OUTCOME_COOLDOWN,
    OUTCOME_OK,
    SOURCE_JOURNAL,
    CommandEvent,
)
from signal_bot.stats.periods import weekday_of

if TYPE_CHECKING:
    import sqlite3
    from collections.abc import Iterable, Sequence

    from signal_bot.analytics.journal import JournalLine
    from signal_bot.store import Store

# How far apart a message and the log line about it may be and still be the
# same event.
#
# SYMMETRIC, and that is not obvious. The log line is written strictly after
# the message arrives, so an asymmetric [0, +W] window looks correct -- and
# throws away most of the data. Measured against the live database on
# 2026-08-31: of 299 counted commands, 194 have a journal line *earlier*
# than their own sent_at_ms, by up to 3.2 seconds, median -173 ms.
#
# The reason is that the two stamps come from two different clocks.
# ``sent_at_ms`` is Signal's envelope timestamp, set on the *sender's phone*;
# the journal stamp is this host's. Nothing keeps those in step, so their
# difference is skew and can have either sign.
#
# 15 seconds each way is far wider than the 3.2 s of skew actually seen and
# still narrower than a person's own repeat rate. Widening it cannot corrupt
# anything either way: two different commands inside one window are dropped
# as ambiguous rather than guessed at.
WINDOW_MS: Final = 15_000


@dataclass(slots=True)
class _Candidate:
    """One journal line, and whether the join has already spent it."""

    line: JournalLine
    used: bool = False


@dataclass(slots=True)
class Tally:
    """The running result of a pass."""

    events: list[CommandEvent] = field(default_factory=list)
    matched: Counter[tuple[str, str, str, str]] = field(default_factory=Counter)
    unmatched: int = 0
    ambiguous: int = 0


def _by_name(lines: Iterable[JournalLine]) -> dict[str, list[_Candidate]]:
    """Group journal lines by the display name they were logged for."""
    grouped: dict[str, list[_Candidate]] = defaultdict(list)
    for line in lines:
        grouped[line.display_name].append(_Candidate(line))
    for candidates in grouped.values():
        candidates.sort(key=lambda candidate: candidate.line.at_ms)
    return dict(grouped)


def _assign(
    rows: Sequence[sqlite3.Row], grouped: dict[str, list[_Candidate]]
) -> dict[int, _Candidate]:
    """Pair each counted command with the journal line most likely to be it.

    Closest pair first, globally, rather than best-for-each-row in turn.
    That difference is what makes two commands from one person seconds apart
    resolvable at all.

    Per-row greedy sees both lines inside the window, cannot tell which is
    which, and drops both. Global assignment can: the clock skew between a
    sender's phone and this host is *systematic*, so the true pairs share
    almost the same offset and are the two smallest distances available.

    Ties break on row order and then line order, so a re-run against the
    same inputs produces the same pairing. The backfill has to be
    re-runnable, and "mostly the same" is not that.
    """
    pairs: list[tuple[int, int, int]] = []
    for index, row in enumerate(rows):
        candidates = grouped.get(str(row["display_name"]), [])
        for position, candidate in enumerate(candidates):
            delta = candidate.line.at_ms - int(row["sent_at_ms"])
            if abs(delta) <= WINDOW_MS:
                pairs.append((abs(delta), index, position))
    pairs.sort()

    taken: dict[int, _Candidate] = {}
    for _distance, index, position in pairs:
        if index in taken:
            continue
        candidate = grouped[str(rows[index]["display_name"])][position]
        if candidate.used:
            continue
        candidate.used = True
        taken[index] = candidate
    return taken


def _in_window(row: sqlite3.Row, grouped: dict[str, list[_Candidate]]) -> bool:
    """Whether the journal held any line at all for this message."""
    sent_at_ms = int(row["sent_at_ms"])
    return any(
        abs(candidate.line.at_ms - sent_at_ms) <= WINDOW_MS
        for candidate in grouped.get(str(row["display_name"]), [])
    )


def from_journal(store: Store, lines: Iterable[JournalLine], floor: int) -> Tally:
    """Walk the counted commands and attach a name to each from the journal."""
    tally = Tally()
    grouped = _by_name(lines)
    rows = store.events.connection.execute(
        """
        SELECT group_id, uuid, display_name, sent_at_ms, day, local_hour
        FROM message_events
        WHERE is_command = 1 AND local_hour >= 0 AND sent_at_ms < ?
        ORDER BY sent_at_ms
        """,
        (floor,),
    ).fetchall()
    taken = _assign(rows, grouped)
    for index, row in enumerate(rows):
        chosen = taken.get(index)
        if chosen is None:
            # A message the journal cannot explain. Either it never held a
            # line for it -- vacuumed, or older than the log -- or every line
            # in its window went to a closer message, which is the honest
            # reading of "two of these happened and I have one record".
            # Dropped either way: a wrong attribution is worse than a gap,
            # because it is indistinguishable from data.
            if _in_window(row, grouped):
                tally.ambiguous += 1
            else:
                tally.unmatched += 1
            continue
        line = chosen.line
        outcome = OUTCOME_COOLDOWN if line.refused else OUTCOME_OK
        day = str(row["day"])
        tally.events.append(
            CommandEvent(
                group_id=str(row["group_id"]),
                user_uuid=str(row["uuid"]),
                command=line.command,
                outcome=outcome,
                invoked_at_ms=int(row["sent_at_ms"]),
                day=day,
                local_hour=int(row["local_hour"]),
                weekday=weekday_of(day),
                source=SOURCE_JOURNAL,
            )
        )
        if outcome == OUTCOME_OK:
            # Only successes are in daily_uses, so only successes may be
            # subtracted from it below.
            tally.matched[(str(row["group_id"]), str(row["uuid"]), line.command, day)] += 1
    return tally