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/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
                 """