From ab0afa149a68dc529b554e8659fa74916ff3dd86 Mon Sep 17 00:00:00 2001 From: Minz Date: Sat, 1 Aug 2026 18:46:28 +0200 Subject: [PATCH] Two-phase backlog scan with self-healing found-count tracking Split each channel's backlog scan into a find phase (settles the found count first) and a react phase, so the progress display no longer grows both numbers in lockstep. Found messages are logged in a dedicated, message_id-deduped table and recovered across restarts independent of the resume checkpoint, so nothing found is silently abandoned if the script stops mid-react. Also drops the separately-maintained messages_found counter, which proved prone to drift under repeated interruptions/rescans, in favor of computing it live from the dedup table - self-healing regardless of how many times a channel gets rescanned. --- bot/backlog.py | 56 ++++++++++++++++++++++++++++------------- bot/display.py | 21 +++++++++++----- bot/reactor.py | 8 ++++++ bot/storage.py | 68 ++++++++++++++++++++++++++++++++++++++++++++++---- bot/undo.py | 3 +++ main.py | 12 ++++++--- 6 files changed, 136 insertions(+), 32 deletions(-) diff --git a/bot/backlog.py b/bot/backlog.py index db28346..7bc28c9 100644 --- a/bot/backlog.py +++ b/bot/backlog.py @@ -1,11 +1,12 @@ import asyncio import datetime import logging -from typing import Awaitable, Callable +from typing import Awaitable, Callable, Optional import discord from bot.discovery import Target +from bot.display import StatsDisplay from bot.storage import Storage log = logging.getLogger(__name__) @@ -18,45 +19,64 @@ async def scan_channel_backlog( channel: discord.abc.Messageable, target: Target, friend_id: int, + emoji: str, cutoff: datetime.datetime, react_fn: ReactFn, + display: Optional[StatsDisplay] = None, ) -> None: - """Scan a channel's history once, resumably. + """Scan a channel's history once, resumably, in two phases. - First run: walk forward from `cutoff` (now - days_back) to the present. - Subsequent runs: only diff forward from the newest message already seen, - since the days_back window only ever moves forward in time. + Phase 1 finds every message from the friend in the window first (so the + "found" count settles before any reacting starts, rather than climbing + in lockstep with "reacted"), permanently logging each one (deduped by + message ID) so re-scanning it later never inflates the count. Phase 2 + then reacts to each one in order. + + Before Phase 1 even starts, any message previously logged as found for + this channel but not yet reacted to (e.g. the script was stopped midway + through Phase 2 last time) is recovered from that same log and queued + up again — this is independent of the newest_id_seen checkpoint below, + so nothing found is ever silently abandoned even if the checkpoint has + already moved past it. """ channel_id = getattr(channel, "id", None) if channel_id is None: return + found_messages = [] + pending_ids = await asyncio.to_thread(storage.get_pending_found_message_ids, channel_id, emoji) + for message_id in pending_ids: + try: + found_messages.append(await channel.fetch_message(int(message_id))) + except discord.NotFound: + pass # message (or its channel access) is gone since it was found + except discord.HTTPException: + log.exception("Failed to refetch pending message %s in channel %s", message_id, channel_id) + state = await asyncio.to_thread(storage.get_scan_state, channel_id) newest_id_seen = int(state["newest_id_seen"]) if state and state.get("newest_id_seen") else None - after = discord.Object(id=newest_id_seen) if newest_id_seen else cutoff - checkpoint_every = 25 - seen_since_checkpoint = 0 - try: async for message in channel.history(after=after, oldest_first=True, limit=None): if newest_id_seen is None or message.id > newest_id_seen: newest_id_seen = message.id - await react_fn(message, target) - - seen_since_checkpoint += 1 - if seen_since_checkpoint >= checkpoint_every: - seen_since_checkpoint = 0 + if message.author.id == friend_id: + found_messages.append(message) await asyncio.to_thread( - storage.set_scan_state, channel_id, target.target_id, newest_id_seen, False + storage.record_message_found, message.id, channel_id, target.target_id, friend_id ) + if display is not None: + await display.refresh() - await asyncio.to_thread( - storage.set_scan_state, channel_id, target.target_id, newest_id_seen, True - ) + await asyncio.to_thread(storage.set_scan_state, channel_id, target.target_id, newest_id_seen, True) except discord.Forbidden: log.warning("No access to channel %s (%s), skipping", channel_id, getattr(channel, "name", "")) + return except discord.HTTPException: log.exception("HTTP error scanning channel %s", channel_id) + return + + for message in found_messages: + await react_fn(message, target) diff --git a/bot/display.py b/bot/display.py index c921dc2..cdfa529 100644 --- a/bot/display.py +++ b/bot/display.py @@ -4,6 +4,7 @@ from typing import List, Optional from rich.console import Console from rich.live import Live from rich.panel import Panel +from rich.progress_bar import ProgressBar from rich.table import Table from bot.storage import Storage @@ -15,24 +16,32 @@ def _render(stats: List[dict], friend_label: Optional[str] = None) -> Panel: table = Table(expand=True) table.add_column("Target") table.add_column("Type") - table.add_column("Reactions", justify="right") + table.add_column("Progress", ratio=1) + table.add_column("Reacted/Found", justify="right") table.add_column("Last reaction (UTC)") - total = 0 + total_reacted = 0 + total_found = 0 for row in stats: - total += row["reaction_count"] + reacted = row["reaction_count"] + found = max(row["messages_found"], reacted) # found can never be less than reacted + total_reacted += reacted + total_found += found + + bar = ProgressBar(total=max(found, reacted, 1), completed=reacted, width=None) table.add_row( row["name"] or row["target_id"], _TYPE_LABELS.get(row["target_type"], row["target_type"]), - str(row["reaction_count"]), + bar, + f"{reacted}/{found}", row["last_reaction_at"] or "-", ) if not stats: - table.add_row("(no targets discovered yet)", "-", "-", "-") + table.add_row("(no targets discovered yet)", "-", "-", "-", "-") who = f" for {friend_label}" if friend_label else "" - return Panel(table, title=f"Reaction stats{who} — total: {total}", border_style="cyan") + return Panel(table, title=f"Reaction stats{who} — total: {total_reacted}/{total_found}", border_style="cyan") class StatsDisplay: diff --git a/bot/reactor.py b/bot/reactor.py index a46096d..219e24a 100644 --- a/bot/reactor.py +++ b/bot/reactor.py @@ -26,10 +26,18 @@ async def react_to_message( emoji: str, source: str, display: Optional[StatsDisplay] = None, + count_found: bool = True, ) -> bool: if message.author.id != friend_id: return False + if count_found: + await asyncio.to_thread( + storage.record_message_found, message.id, message.channel.id, target.target_id, friend_id + ) + if display is not None: + await display.refresh() + already_reacted = any( reaction.me and str(reaction.emoji) == emoji for reaction in message.reactions ) diff --git a/bot/storage.py b/bot/storage.py index c3f5361..16be0e2 100644 --- a/bot/storage.py +++ b/bot/storage.py @@ -33,6 +33,15 @@ CREATE TABLE IF NOT EXISTS channel_scan_state ( backlog_complete INTEGER NOT NULL DEFAULT 0, last_scanned_at TEXT ); + +CREATE TABLE IF NOT EXISTS found_messages ( + message_id TEXT PRIMARY KEY, + channel_id TEXT NOT NULL, + target_id TEXT NOT NULL, + author_id TEXT NOT NULL, + found_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) +); +CREATE INDEX IF NOT EXISTS idx_found_messages_channel ON found_messages(channel_id); """ @@ -58,10 +67,10 @@ class Storage: # Older DBs created before 'dm' targets (1:1 DMs) existed have a CHECK # constraint that only allows 'guild'/'group_dm' — rebuild the table # with the wider constraint, preserving all existing rows. - targets_sql = self._conn.execute( + targets_sql_row = self._conn.execute( "SELECT sql FROM sqlite_master WHERE type='table' AND name='targets'" ).fetchone() - if targets_sql and "'dm'" not in targets_sql["sql"]: + if targets_sql_row and "'dm'" not in targets_sql_row["sql"]: self._conn.executescript( """ ALTER TABLE targets RENAME TO targets_old; @@ -72,10 +81,13 @@ class Storage: first_seen_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), reaction_count INTEGER NOT NULL DEFAULT 0 ); - INSERT INTO targets SELECT * FROM targets_old; - DROP TABLE targets_old; """ ) + old_columns = {row["name"] for row in self._conn.execute("PRAGMA table_info(targets_old)")} + common = [c for c in ("target_id", "target_type", "name", "first_seen_at", "reaction_count") if c in old_columns] + col_list = ", ".join(common) + self._conn.execute(f"INSERT INTO targets ({col_list}) SELECT {col_list} FROM targets_old") + self._conn.execute("DROP TABLE targets_old") def upsert_target(self, target_id, target_type: str, name: str) -> None: with self._lock, self._conn: @@ -103,6 +115,46 @@ class Storage: cur = self._conn.execute("DELETE FROM targets") return cur.rowcount + def record_message_found(self, message_id, channel_id, target_id, author_id) -> bool: + """Log that a friend message was found, deduped by message_id (the + PRIMARY KEY) so rescanning the same message ever after (e.g. after an + interrupted run) is a no-op. This table is the sole source of truth + for the "found" count — see get_stats() — precisely so that count can + never drift out of sync the way a separately-maintained counter can.""" + with self._lock, self._conn: + cur = self._conn.execute( + """ + INSERT OR IGNORE INTO found_messages (message_id, channel_id, target_id, author_id) + VALUES (?, ?, ?, ?) + """, + (str(message_id), str(channel_id), str(target_id), str(author_id)), + ) + return cur.rowcount > 0 + + def get_pending_found_message_ids(self, channel_id, emoji: str) -> list: + """Messages already logged as found for this channel that don't have + this emoji's reaction logged yet — lets a channel's reacting phase + resume across a restart even if the resume checkpoint already moved + past them.""" + with self._lock: + cur = self._conn.execute( + """ + SELECT f.message_id FROM found_messages f + WHERE f.channel_id = ? + AND NOT EXISTS ( + SELECT 1 FROM reactions_log r + WHERE r.message_id = f.message_id AND r.emoji = ? + ) + ORDER BY f.message_id + """, + (str(channel_id), emoji), + ) + return [row["message_id"] for row in cur.fetchall()] + + def reset_message_counts(self) -> None: + with self._lock, self._conn: + self._conn.execute("DELETE FROM found_messages") + def has_reacted(self, message_id, emoji: str) -> bool: with self._lock: cur = self._conn.execute( @@ -190,9 +242,15 @@ class Storage: cur = self._conn.execute( """ SELECT t.target_id, t.target_type, t.name, t.reaction_count, - COUNT(r.id) AS logged_reactions, MAX(r.message_created_at) AS last_reaction_at + COALESCE(fm.found_count, 0) AS messages_found, + MAX(r.message_created_at) AS last_reaction_at FROM targets t LEFT JOIN reactions_log r ON r.target_id = t.target_id + LEFT JOIN ( + SELECT target_id, COUNT(*) AS found_count + FROM found_messages + GROUP BY target_id + ) fm ON fm.target_id = t.target_id GROUP BY t.target_id ORDER BY t.reaction_count DESC """ diff --git a/bot/undo.py b/bot/undo.py index b3f3bb6..3d12d3c 100644 --- a/bot/undo.py +++ b/bot/undo.py @@ -44,6 +44,9 @@ async def undo_all_reactions(client: discord.Client, storage: Storage, display: await asyncio.sleep(human_jitter()) cleared = await asyncio.to_thread(storage.clear_scan_state) + await asyncio.to_thread(storage.reset_message_counts) + if display is not None: + await display.refresh() log.info( "Undo complete: removed %d/%d reaction(s), cleared %d channel resume point(s) — " "the next normal run will do a full fresh backlog scan.", diff --git a/main.py b/main.py index c860f0a..18b7f8e 100644 --- a/main.py +++ b/main.py @@ -72,7 +72,7 @@ class ReactorClient(discord.Client): async def react_fn(message: discord.Message, target: Target) -> bool: return await react_to_message( self.storage, message, target, self.cfg.friend_id, self.cfg.emoji, - source="backlog", display=self.display, + source="backlog", display=self.display, count_found=False, ) for target in targets: @@ -82,11 +82,17 @@ class ReactorClient(discord.Client): log.warning("Channel %s (%s) not found in cache, skipping", target.target_id, target.name) continue self.known_channel_targets[channel.id] = target - await scan_channel_backlog(self.storage, channel, target, self.cfg.friend_id, cutoff, react_fn) + await scan_channel_backlog( + self.storage, channel, target, self.cfg.friend_id, self.cfg.emoji, + cutoff, react_fn, display=self.display, + ) else: for channel in get_scannable_channels(self, target.target_id): self.known_channel_targets[channel.id] = target - await scan_channel_backlog(self.storage, channel, target, self.cfg.friend_id, cutoff, react_fn) + await scan_channel_backlog( + self.storage, channel, target, self.cfg.friend_id, self.cfg.emoji, + cutoff, react_fn, display=self.display, + ) log.info("Backlog scan complete. Now watching live.")