Add spinner/countdown-bar to stats panel; backfill on widened days_back

Stats panel: a persistent spinner marks whichever target is currently
being scanned/reacted to, and jittered waits show a live-ticking
countdown bar instead of a static "waiting Xs" line. Both the status
line and countdown row are always rendered (blank when idle) so their
appearing/disappearing no longer shifts the table's height.

Backlog: channel_scan_state now tracks oldest_covered_at alongside
newest_id_seen. If days_back is widened after a channel already has a
resume checkpoint, the forward-only diff would never look back far
enough to notice — now the newly-exposed gap gets backfilled first.

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,101 @@ 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, resumably, in two phases.
 
-    First run: walk forward from `cutoff` (now - days_back) to the present.
+    Phase 1 finds every message from the friend in the window first (so the
-    Subsequent runs: only diff forward from the newest message already seen,
+    "found" count settles before any reacting starts, rather than climbing
-    since the days_back window only ever moves forward in time.
+    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.
+
+    The channel's resume state tracks not just newest_id_seen (for the
+    forward diff) but also oldest_covered_at — the earliest point in time
+    already fully scanned. If days_back has been widened since the last run
+    (cutoff is now older than oldest_covered_at), the newly-exposed gap
+    [cutoff, oldest_covered_at) is backfilled first — otherwise the forward-
+    only diff would silently never look back far enough to notice.
     """
     channel_id = getattr(channel, "id", None)
     if channel_id is None:
         return
+    chan_label = getattr(channel, "name", None) or target.name
+
+    if display is not None:
+        display.set_status(f"Checking {chan_label} for previously found, unreacted message(s)...")
+
+    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
+    oldest_covered_at = (
+        datetime.datetime.fromisoformat(state["oldest_covered_at"])
+        if state and state.get("oldest_covered_at") else None
+    )
 
-    after = discord.Object(id=newest_id_seen) if newest_id_seen else cutoff
+    async def scan_range(after, before, status_verb: str) -> None:
-
+        nonlocal newest_id_seen
-    checkpoint_every = 25
+        if display is not None:
-    seen_since_checkpoint = 0
+            display.set_status(f"{status_verb} {chan_label}...")
-
+        async for message in channel.history(after=after, before=before, oldest_first=True, limit=None):
-    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)
+            if message.author.id == friend_id:
-
+                found_messages.append(message)
-            seen_since_checkpoint += 1
-            if seen_since_checkpoint >= checkpoint_every:
-                seen_since_checkpoint = 0
                 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:
+                    display.set_status(f"{status_verb} {chan_label} — {len(found_messages)} found so far...")
+                    await display.refresh()
+
+    try:
+        if oldest_covered_at is None:
+            # Never scanned before: one pass across the whole configured window.
+            await scan_range(cutoff, None, "Fetching message history for")
+            oldest_covered_at = cutoff
+        else:
+            if cutoff < oldest_covered_at:
+                # days_back was widened since the last run — the forward-only
+                # diff below would never notice the newly-exposed older range,
+                # so backfill exactly that gap first.
+                await scan_range(cutoff, oldest_covered_at, "Backfilling older history in")
+                oldest_covered_at = cutoff
+            after = discord.Object(id=newest_id_seen) if newest_id_seen else cutoff
+            await scan_range(after, None, "Fetching new messages in")
 
         await asyncio.to_thread(
-            storage.set_scan_state, channel_id, target.target_id, newest_id_seen, True
+            storage.set_scan_state, channel_id, target.target_id, newest_id_seen, oldest_covered_at, 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 i, message in enumerate(found_messages, start=1):
+        if display is not None:
+            display.set_status(f"Reacting to message {i}/{len(found_messages)} in {chan_label}...")
+        await react_fn(message, target)

bot/discovery.py

@@ -10,16 +10,27 @@ log = logging.getLogger(__name__)
 @dataclass
 class Target:
     target_id: int
-    target_type: str  # "guild" | "group_dm"
+    target_type: str  # "guild" | "group_dm" | "dm"
     name: str
 
 
-async def discover_targets(client: discord.Client, friend_id: int, manual_guild_ids: Optional[List[int]] = None) -> List[Target]:
+async def resolve_friend(client: discord.Client, friend_id: int) -> discord.User:
+    """Fetch the target user object so we can display who the script is
+    actually reacting to (helps catch a wrong/typo'd friend_id early)."""
+    user = client.get_user(friend_id)
+    if user is None:
+        user = await client.fetch_user(friend_id)
+    return user
+
+
+async def discover_targets(
+    client: discord.Client, friend: discord.User, manual_guild_ids: Optional[List[int]] = None
+) -> List[Target]:
+    friend_id = friend.id
     targets: List[Target] = []
     guild_ids_seen = set()
 
     try:
-        friend = client.get_user(friend_id) or await client.fetch_user(friend_id)
         profile = await friend.profile(with_mutual_guilds=True, with_mutual_friends=False, with_mutual_friends_count=False)
         for mutual in (profile.mutual_guilds or []):
             guild = client.get_guild(mutual.id)
@@ -44,6 +55,12 @@ async def discover_targets(client: discord.Client, friend_id: int, manual_guild_
         targets.append(Target(guild_id, "guild", name))
         guild_ids_seen.add(guild_id)
 
+    try:
+        dm_channel = friend.dm_channel or await friend.create_dm()
+        targets.append(Target(dm_channel.id, "dm", f"DM with {friend}"))
+    except discord.HTTPException:
+        log.warning("Failed to open the 1:1 DM channel with %s", friend, exc_info=True)
+
     for channel in client.private_channels:
         if isinstance(channel, discord.GroupChannel) and any(u.id == friend_id for u in channel.recipients):
             name = channel.name or ", ".join(u.name for u in channel.recipients)

bot/display.py

@@ -1,35 +1,100 @@
 import asyncio
-from typing import List
+import time
+from typing import List, Optional
 
-from rich.console import Console
+from rich.console import Console, Group
 from rich.live import Live
 from rich.panel import Panel
+from rich.progress_bar import ProgressBar
+from rich.spinner import Spinner
 from rich.table import Table
+from rich.text import Text
 
 from bot.storage import Storage
 
+_TYPE_LABELS = {"guild": "server", "group_dm": "group", "dm": "DM"}
 
-def _render(stats: List[dict]) -> Panel:
+
+class CountdownBar:
+    """A Rich renderable that computes its own remaining time at render
+    time (same trick Spinner uses to animate for free under a Live display)
+    so it ticks down smoothly without us needing an explicit polling loop."""
+
+    def __init__(self, duration: float, label: str):
+        self.duration = max(duration, 0.001)
+        self.label = label
+        self.start_time = time.monotonic()
+
+    def __rich_console__(self, console, options):
+        elapsed = time.monotonic() - self.start_time
+        remaining = max(0.0, self.duration - elapsed)
+        bar = ProgressBar(total=self.duration, completed=min(elapsed, self.duration), width=30)
+        grid = Table.grid(padding=(0, 1))
+        grid.add_column()
+        grid.add_column(ratio=1)
+        grid.add_column(justify="right")
+        grid.add_row(Text(f"⏳ {self.label}", style="dim"), bar, Text(f"{remaining:0.1f}s", style="dim"))
+        yield grid
+
+
+def _render(
+    stats: List[dict],
+    friend_label: Optional[str] = None,
+    status: Optional[str] = None,
+    countdown: Optional[CountdownBar] = None,
+    active_spinner: Optional[Spinner] = None,
+    active_target_id: 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
+
+        name = row["name"] or row["target_id"]
+        if active_spinner is not None and active_target_id is not None and str(row["target_id"]) == str(active_target_id):
+            target_cell = _spinner_row(active_spinner, name)
+        else:
+            target_cell = Text(name)
+
+        bar = ProgressBar(total=max(found, reacted, 1), completed=reacted, width=None)
         table.add_row(
-            row["name"] or row["target_id"],
+            target_cell,
-            "server" if row["target_type"] == "guild" else "group",
+            _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)", "-", "-", "-", "-")
 
-    return Panel(table, title=f"Reaction stats — total: {total}", border_style="cyan")
+    # Status/countdown rows are always present (even blank) so the panel's
+    # height never changes as they toggle on/off — that jumping is jarring
+    # under a Live display that also has scrolling log output above it.
+    status_line = Text(f"▸ {status}", style="dim italic") if status else Text(" ")
+    countdown_line = countdown if countdown is not None else Text(" ")
+    content = Group(table, status_line, countdown_line)
+
+    who = f" for {friend_label}" if friend_label else ""
+    return Panel(content, title=f"Reaction stats{who} — total: {total_reacted}/{total_found}", border_style="cyan")
+
+
+def _spinner_row(spinner: Spinner, name: str):
+    grid = Table.grid(padding=(0, 1))
+    grid.add_column()
+    grid.add_column()
+    grid.add_row(spinner, Text(name))
+    return grid
 
 
 class StatsDisplay:
@@ -39,10 +104,16 @@ class StatsDisplay:
     def __init__(self, console: Console, storage: Storage):
         self.console = console
         self.storage = storage
+        self.friend_label: Optional[str] = None
+        self.status: str = "Starting..."
+        self.active_target_id: Optional[str] = None
+        self._spinner = Spinner("dots")
+        self._countdown: Optional[CountdownBar] = None
+        self._last_stats: List[dict] = []
         self._live = Live(
-            _render([]),
+            _render([], None, self.status),
             console=console,
-            refresh_per_second=4,
+            refresh_per_second=8,
             transient=False,
         )
 
@@ -53,10 +124,48 @@ class StatsDisplay:
     def __exit__(self, *exc_info):
         return self._live.__exit__(*exc_info)
 
+    def _redraw(self) -> None:
+        self._live.update(
+            _render(
+                self._last_stats,
+                self.friend_label,
+                self.status,
+                self._countdown,
+                self._spinner,
+                self.active_target_id,
+            )
+        )
+
+    def set_friend(self, label: str) -> None:
+        self.friend_label = label
+
+    def set_active_target(self, target_id: Optional[object]) -> None:
+        """Show a little spinner next to whichever target row is currently
+        being scanned/reacted to. Pass None when idle (e.g. once backlog
+        scanning finishes and we're just watching for live messages)."""
+        self.active_target_id = str(target_id) if target_id is not None else None
+        self._redraw()
+
+    def set_status(self, text: str) -> None:
+        """Update the "what's happening right now" line without hitting the
+        DB — cheap enough to call for every jittered wait / API call."""
+        self.status = text
+        self._redraw()
+
+    def start_countdown(self, duration: float, label: str) -> None:
+        """Show a live-ticking countdown bar for a jittered wait, replacing
+        the status line for the duration of the wait."""
+        self._countdown = CountdownBar(duration, label)
+        self._redraw()
+
+    def clear_countdown(self) -> None:
+        self._countdown = None
+        self._redraw()
+
     def refresh_sync(self) -> None:
-        stats = self.storage.get_stats()
+        self._last_stats = self.storage.get_stats()
-        self._live.update(_render(stats))
+        self._redraw()
 
     async def refresh(self) -> None:
-        stats = await asyncio.to_thread(self.storage.get_stats)
+        self._last_stats = await asyncio.to_thread(self.storage.get_stats)
-        self._live.update(_render(stats))
+        self._redraw()

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
     )
@@ -38,8 +46,15 @@ async def react_to_message(
     if already_reacted:
         return False
 
-    await asyncio.sleep(human_jitter())
+    chan_label = getattr(message.channel, "name", None) or target.name
+    delay = human_jitter()
+    if display is not None:
+        display.start_countdown(delay, f"reacting in {chan_label}")
+    await asyncio.sleep(delay)
 
+    if display is not None:
+        display.clear_countdown()
+        display.set_status(f"Sending reaction in {chan_label}...")
     try:
         await message.add_reaction(emoji)
     except discord.HTTPException:
@@ -57,7 +72,10 @@ async def react_to_message(
         message.created_at,
     )
     if inserted:
-        log.info("Reacted to message %s in %s [%s] (%s)", message.id, target.name, target.target_type, source)
+        log.info(
+            "Reacted to message in %s [%s] (%s): %s",
+            target.name, target.target_type, source, message.jump_url,
+        )
         if display is not None:
             await display.refresh()
     return inserted

bot/storage.py

@@ -6,7 +6,7 @@ from typing import Optional
 SCHEMA = """
 CREATE TABLE IF NOT EXISTS targets (
     target_id      TEXT PRIMARY KEY,
-    target_type    TEXT NOT NULL CHECK(target_type IN ('guild','group_dm')),
+    target_type    TEXT NOT NULL CHECK(target_type IN ('guild','group_dm','dm')),
     name           TEXT,
     first_seen_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
     reaction_count INTEGER NOT NULL DEFAULT 0
@@ -30,9 +30,19 @@ CREATE TABLE IF NOT EXISTS channel_scan_state (
     channel_id         TEXT PRIMARY KEY,
     target_id          TEXT NOT NULL,
     newest_id_seen     TEXT,
+    oldest_covered_at  TEXT,
     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);
 """
 
 
@@ -55,6 +65,37 @@ class Storage:
         if "message_created_at" not in columns:
             self._conn.execute("ALTER TABLE reactions_log ADD COLUMN message_created_at TEXT")
 
+        # Older DBs created before oldest_covered_at existed (widening days_back
+        # couldn't be detected/backfilled without it — see scan_channel_backlog).
+        scan_columns = {row["name"] for row in self._conn.execute("PRAGMA table_info(channel_scan_state)")}
+        if "oldest_covered_at" not in scan_columns:
+            self._conn.execute("ALTER TABLE channel_scan_state ADD COLUMN oldest_covered_at TEXT")
+
+        # 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_row = self._conn.execute(
+            "SELECT sql FROM sqlite_master WHERE type='table' AND name='targets'"
+        ).fetchone()
+        if targets_sql_row and "'dm'" not in targets_sql_row["sql"]:
+            self._conn.executescript(
+                """
+                ALTER TABLE targets RENAME TO targets_old;
+                CREATE TABLE targets (
+                    target_id      TEXT PRIMARY KEY,
+                    target_type    TEXT NOT NULL CHECK(target_type IN ('guild','group_dm','dm')),
+                    name           TEXT,
+                    first_seen_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
+                    reaction_count INTEGER NOT NULL DEFAULT 0
+                );
+                """
+            )
+            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:
             self._conn.execute(
@@ -66,6 +107,61 @@ class Storage:
                 (str(target_id), target_type, name),
             )
 
+    def prune_targets(self, keep_target_ids: list) -> int:
+        """Remove targets not present in the latest discovery run (e.g. a server
+        that's no longer mutual, or a stale entry from an earlier bug/test run).
+        reactions_log/channel_scan_state rows are left alone for history/undo."""
+        keep = [str(t) for t in keep_target_ids]
+        with self._lock, self._conn:
+            placeholders = ",".join("?" * len(keep)) if keep else None
+            if placeholders:
+                cur = self._conn.execute(
+                    f"DELETE FROM targets WHERE target_id NOT IN ({placeholders})", keep
+                )
+            else:
+                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(
@@ -95,6 +191,15 @@ class Storage:
                 )
             return inserted
 
+    def clear_scan_state(self) -> int:
+        """Wipe all per-channel resume points, forcing a full fresh backlog
+        scan on the next run (widening days_back alone no longer requires
+        this — see oldest_covered_at in scan_channel_backlog — but --undo
+        still wants a clean slate since it also removes the reactions log)."""
+        with self._lock, self._conn:
+            cur = self._conn.execute("DELETE FROM channel_scan_state")
+            return cur.rowcount
+
     def get_scan_state(self, channel_id) -> Optional[dict]:
         with self._lock:
             cur = self._conn.execute(
@@ -104,19 +209,25 @@ class Storage:
             row = cur.fetchone()
             return dict(row) if row else None
 
-    def set_scan_state(self, channel_id, target_id, newest_id_seen, backlog_complete: bool) -> None:
+    def set_scan_state(self, channel_id, target_id, newest_id_seen, oldest_covered_at, backlog_complete: bool) -> None:
         with self._lock, self._conn:
             self._conn.execute(
                 """
                 INSERT INTO channel_scan_state
-                    (channel_id, target_id, newest_id_seen, backlog_complete, last_scanned_at)
+                    (channel_id, target_id, newest_id_seen, oldest_covered_at, backlog_complete, last_scanned_at)
-                VALUES (?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
+                VALUES (?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
                 ON CONFLICT(channel_id) DO UPDATE SET
                     newest_id_seen = excluded.newest_id_seen,
+                    oldest_covered_at = excluded.oldest_covered_at,
                     backlog_complete = excluded.backlog_complete,
                     last_scanned_at = excluded.last_scanned_at
                 """,
-                (str(channel_id), str(target_id), str(newest_id_seen) if newest_id_seen else None, int(backlog_complete)),
+                (
+                    str(channel_id), str(target_id),
+                    str(newest_id_seen) if newest_id_seen else None,
+                    oldest_covered_at.isoformat() if oldest_covered_at else None,
+                    int(backlog_complete),
+                ),
             )
 
     def get_all_reactions(self) -> list:
@@ -143,9 +254,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
                 """

bot/undo.py

@@ -24,6 +24,10 @@ async def undo_all_reactions(client: discord.Client, storage: Storage, display:
         emoji = row["emoji"]
         should_clear_log = True
 
+        if display is not None:
+            display.set_active_target(row["target_id"])
+            display.set_status(f"Removing reaction from message {message_id} ({i}/{total})...")
+
         try:
             channel = client.get_channel(channel_id) or await client.fetch_channel(channel_id)
             message = await channel.fetch_message(message_id)
@@ -41,6 +45,23 @@ async def undo_all_reactions(client: discord.Client, storage: Storage, display:
             if display is not None:
                 await display.refresh()
 
-        await asyncio.sleep(human_jitter())
+        delay = human_jitter()
+        if display is not None:
+            display.start_countdown(delay, "next removal")
+        await asyncio.sleep(delay)
+        if display is not None:
+            display.clear_countdown()
 
-    log.info("Undo complete: removed %d/%d reaction(s)", removed, total)
+    if display is not None:
+        display.set_active_target(None)
+        display.set_status("Clearing channel resume points and found-message log...")
+    cleared = await asyncio.to_thread(storage.clear_scan_state)
+    await asyncio.to_thread(storage.reset_message_counts)
+    if display is not None:
+        display.set_status("Undo complete.")
+        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.",
+        removed, total, cleared,
+    )

main.py

@@ -10,7 +10,7 @@ from rich.logging import RichHandler
 from bot.backlog import scan_channel_backlog
 from bot.channels import get_scannable_channels
 from bot.config import load_config
-from bot.discovery import Target, discover_targets
+from bot.discovery import Target, discover_targets, resolve_friend
 from bot.display import StatsDisplay
 from bot.reactor import react_to_message
 from bot.storage import Storage
@@ -44,11 +44,31 @@ class ReactorClient(discord.Client):
         await self.close()
 
     async def _run_backlog_then_watch(self):
-        targets = await discover_targets(self, self.cfg.friend_id, self.cfg.manual_guild_ids)
+        self.display.set_status(f"Resolving friend id {self.cfg.friend_id}...")
+        try:
+            friend = await resolve_friend(self, self.cfg.friend_id)
+        except discord.NotFound:
+            log.error(
+                "friend_id %s does not resolve to any Discord user. Check config.json and fix it.",
+                self.cfg.friend_id,
+            )
+            self.display.set_status("Failed: friend_id does not resolve to a Discord user.")
+            return
+
+        log.info("Targeting friend: %s (%s)", friend, friend.id)
+        self.display.set_friend(str(friend))
+        self.display.set_status(f"Discovering mutual servers/groups with {friend}...")
+        await self.display.refresh()
+
+        targets = await discover_targets(self, friend, self.cfg.manual_guild_ids)
         log.info("Discovered %d mutual target(s): %s", len(targets), [t.name for t in targets])
 
         for t in targets:
             await asyncio.to_thread(self.storage.upsert_target, t.target_id, t.target_type, t.name)
+        pruned = await asyncio.to_thread(self.storage.prune_targets, [t.target_id for t in targets])
+        if pruned:
+            log.info("Pruned %d stale target(s) no longer mutual/discovered", pruned)
+        self.display.set_status(f"Starting backlog scan across {len(targets)} target(s)...")
         await self.display.refresh()
 
         cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=self.cfg.days_back)
@@ -56,23 +76,32 @@ 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:
-            if target.target_type == "group_dm":
+            self.display.set_active_target(target.target_id)
+            if target.target_type in ("group_dm", "dm"):
                 channel = self.get_channel(target.target_id)
                 if channel is None:
-                    log.warning("Group DM %s not found in cache, skipping", target.target_id)
+                    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,
+                    )
 
+        self.display.set_active_target(None)
         log.info("Backlog scan complete. Now watching live.")
+        self.display.set_status("Backlog scan complete — watching live for new messages...")
 
     async def on_message(self, message: discord.Message):
         if self.undo:
@@ -82,10 +111,14 @@ class ReactorClient(discord.Client):
             return
         if message.author.id != self.cfg.friend_id:
             return
+        chan_label = getattr(message.channel, "name", None) or target.name
+        self.display.set_active_target(target.target_id)
+        self.display.set_status(f"New message from friend in {chan_label}...")
         await react_to_message(
             self.storage, message, target, self.cfg.friend_id, self.cfg.emoji,
             source="live", display=self.display,
         )
+        self.display.set_active_target(None)
 
 
 def parse_args():
@@ -111,6 +144,7 @@ def main():
 
     storage = Storage(cfg.db_path)
     display = StatsDisplay(console, storage)
+    display.set_status("Connecting to Discord...")
 
     with display:
         client = ReactorClient(cfg, storage, display, undo=args.undo)