[PATCH RFC 02/11] review: track message counts for all revisions of a series

[Christian Brauner <brauner@kernel.org>]

The series table only carries message counts for the tracked revision,
so new mail landing on an older version's thread is invisible.  Give
the revisions catalog its own message_count/seen_message_count/
last_update_check/last_activity_at columns (schema v11, backfilled
from series rows including archived upgrade leftovers), stitch reads
through a LEFT JOIN so live series counts win while archived rows
never shadow the catalog, and add update_revision_message_counts() --
an incremental per-revision poller (newest-first, zero writes when
quiet, rethreaded revisions summed across their per-patch threads,
thread mbox cached as a git blob).

refresh_message_count() and sync_seen_from_unseen_count() now fall
back to the catalog row when the (change_id, revision) pair is not
actively tracked, so the thread viewer's badge sync works for any
revision.

Assisted-by: LLM
Signed-off-by: Christian Brauner (Amutable) <brauner@kernel.org>
---
 src/b4/review/tracking.py | 566 ++++++++++++++++++++++++++++++++++++++++------
 1 file changed, 493 insertions(+), 73 deletions(-)

diff --git a/src/b4/review/tracking.py b/src/b4/review/tracking.py
index 0d10dff..c176c46 100644
--- a/src/b4/review/tracking.py
+++ b/src/b4/review/tracking.py
@@ -26,7 +26,7 @@ logger = b4.logger
 REVIEW_METADATA_DIR = 'b4-review'
 REVIEW_METADATA_FILE = 'metadata.json'
 
-SCHEMA_VERSION = 10
+SCHEMA_VERSION = 11
 
 SERIES_PATCHES_DDL = """
 CREATE TABLE IF NOT EXISTS series_patches (
@@ -38,12 +38,7 @@ CREATE TABLE IF NOT EXISTS series_patches (
     PRIMARY KEY (change_id, revision, position)
 )"""
 
-SCHEMA_SQL = (
-    """
-CREATE TABLE IF NOT EXISTS schema_version (
-    version INTEGER PRIMARY KEY
-);
-
+SERIES_DDL = """
 CREATE TABLE IF NOT EXISTS series (
     track_id INTEGER PRIMARY KEY,
     change_id TEXT NOT NULL,
@@ -68,7 +63,16 @@ CREATE TABLE IF NOT EXISTS series (
     target_branch TEXT,
     is_rethreaded INTEGER DEFAULT 0,
     UNIQUE (change_id, revision)
+)"""
+
+SCHEMA_SQL = (
+    """
+CREATE TABLE IF NOT EXISTS schema_version (
+    version INTEGER PRIMARY KEY
 );
+"""
+    + SERIES_DDL
+    + """;
 
 CREATE TABLE IF NOT EXISTS revisions (
     change_id   TEXT NOT NULL,
@@ -81,6 +85,10 @@ CREATE TABLE IF NOT EXISTS revisions (
     fingerprint TEXT,
     source      TEXT DEFAULT 'heuristic',
     is_rethreaded INTEGER DEFAULT 0,
+    message_count INT,
+    seen_message_count INT,
+    last_update_check TEXT,
+    last_activity_at TEXT,
     PRIMARY KEY (change_id, revision)
 );
 
@@ -187,6 +195,76 @@ def _migrate_db_if_needed(conn: sqlite3.Connection) -> None:
             conn.execute(
                 'ALTER TABLE revisions ADD COLUMN is_rethreaded INTEGER DEFAULT 0'
             )
+    if version < 11:
+        # Per-revision unread tracking: the series table only covers the
+        # tracked revision, so non-tracked versions get their message
+        # counts and new-mail detection from the revisions catalog.
+        # The stitched per-revision read joins series, so make sure the
+        # table exists (mirrors the v8 treatment of revisions).
+        conn.execute(SERIES_DDL)
+        existing = {row[1] for row in conn.execute('PRAGMA table_info(revisions)')}
+        for coldef in (
+            'message_count INT',
+            'seen_message_count INT',
+            'last_update_check TEXT',
+            'last_activity_at TEXT',
+        ):
+            if coldef.split()[0] not in existing:
+                conn.execute(f'ALTER TABLE revisions ADD COLUMN {coldef}')
+        # Backfill from the series table -- but only when it carries the
+        # full column set (a degenerate/absent series table has nothing
+        # worth backfilling from).
+        series_cols = {row[1] for row in conn.execute('PRAGMA table_info(series)')}
+        needed = {
+            'change_id',
+            'revision',
+            'message_id',
+            'subject',
+            'added_at',
+            'sent_at',
+            'fingerprint',
+            'is_rethreaded',
+            'message_count',
+            'seen_message_count',
+            'last_update_check',
+            'last_activity_at',
+        }
+        if needed <= series_cols:
+            # Every tracked series (archived rows included) needs a catalog
+            # row so its per-revision counts have somewhere to live.
+            conn.execute(
+                'INSERT OR IGNORE INTO revisions'
+                ' (change_id, revision, message_id, subject, found_at,'
+                '  fingerprint, source, is_rethreaded,'
+                '  message_count, seen_message_count, last_update_check,'
+                '  last_activity_at)'
+                ' SELECT change_id, revision, message_id, subject,'
+                "  COALESCE(added_at, sent_at), fingerprint, 'heuristic',"
+                '  COALESCE(is_rethreaded, 0),'
+                '  message_count, seen_message_count, last_update_check,'
+                '  last_activity_at'
+                " FROM series WHERE message_id IS NOT NULL AND message_id != ''"
+            )
+            # Seed counts on pre-existing catalog rows from any matching
+            # series row -- the only historical data available.
+            conn.execute(
+                'UPDATE revisions SET'
+                '  message_count = (SELECT s.message_count FROM series s'
+                '   WHERE s.change_id = revisions.change_id'
+                '   AND s.revision = revisions.revision),'
+                '  seen_message_count = (SELECT s.seen_message_count FROM series s'
+                '   WHERE s.change_id = revisions.change_id'
+                '   AND s.revision = revisions.revision),'
+                '  last_update_check = (SELECT s.last_update_check FROM series s'
+                '   WHERE s.change_id = revisions.change_id'
+                '   AND s.revision = revisions.revision),'
+                '  last_activity_at = (SELECT s.last_activity_at FROM series s'
+                '   WHERE s.change_id = revisions.change_id'
+                '   AND s.revision = revisions.revision)'
+                ' WHERE message_count IS NULL AND EXISTS (SELECT 1 FROM series s'
+                '  WHERE s.change_id = revisions.change_id'
+                '  AND s.revision = revisions.revision)'
+            )
     conn.execute('UPDATE schema_version SET version = ?', (SCHEMA_VERSION,))
     conn.commit()
 
@@ -1035,18 +1113,34 @@ _REVISION_COLS = (
     'fingerprint',
     'source',
     'is_rethreaded',
+    'message_count',
+    'seen_message_count',
+    'last_update_check',
+    'last_activity_at',
 )
 
+# Counts for the revision a series row actively tracks live on `series`;
+# the catalog columns cover every other revision.  COALESCE stitches the
+# two so readers get one consistent per-revision view.  Archived series
+# rows are upgrade leftovers whose stale counts must not shadow the
+# catalog's.
 _REVISION_SELECT = (
-    'SELECT change_id, revision, message_id, subject, link, found_at,'
-    ' thread_blob, fingerprint, source, is_rethreaded FROM revisions'
+    'SELECT r.change_id, r.revision, r.message_id, r.subject, r.link,'
+    ' r.found_at, r.thread_blob, r.fingerprint, r.source, r.is_rethreaded,'
+    ' COALESCE(s.message_count, r.message_count),'
+    ' COALESCE(s.seen_message_count, r.seen_message_count),'
+    ' COALESCE(s.last_update_check, r.last_update_check),'
+    ' COALESCE(s.last_activity_at, r.last_activity_at)'
+    ' FROM revisions r LEFT JOIN series s'
+    ' ON s.change_id = r.change_id AND s.revision = r.revision'
+    " AND s.status != 'archived'"
 )
 
 
 def get_revisions(conn: sqlite3.Connection, change_id: str) -> list[dict[str, Any]]:
     """Return all known revisions for a change_id, ordered ascending."""
     cursor = conn.execute(
-        _REVISION_SELECT + ' WHERE change_id = ? ORDER BY revision ASC',
+        _REVISION_SELECT + ' WHERE r.change_id = ? ORDER BY r.revision ASC',
         (change_id,),
     )
     return [dict(zip(_REVISION_COLS, row)) for row in cursor.fetchall()]
@@ -1064,7 +1158,7 @@ def find_revision_by_fingerprint(
     if not fingerprint:
         return None
     row = conn.execute(
-        _REVISION_SELECT + ' WHERE fingerprint = ? LIMIT 1',
+        _REVISION_SELECT + ' WHERE r.fingerprint = ? LIMIT 1',
         (fingerprint,),
     ).fetchone()
     if row is None:
@@ -1475,26 +1569,34 @@ def get_all_revisions_grouped(
     conn: sqlite3.Connection,
 ) -> dict[str, list[dict[str, Any]]]:
     """Return {change_id: [rev_dicts]} for all change_ids, ordered ascending."""
-    cols = (
-        'change_id',
-        'revision',
-        'message_id',
-        'subject',
-        'link',
-        'found_at',
-        'thread_blob',
-    )
-    cursor = conn.execute(
-        'SELECT change_id, revision, message_id, subject, link, found_at, thread_blob '
-        'FROM revisions ORDER BY change_id, revision ASC'
-    )
+    cursor = conn.execute(_REVISION_SELECT + ' ORDER BY r.change_id, r.revision ASC')
     result: dict[str, list[dict[str, Any]]] = {}
     for row in cursor.fetchall():
-        entry: dict[str, Any] = dict(zip(cols, row))
-        result.setdefault(row[0], []).append(entry)
+        entry: dict[str, Any] = dict(zip(_REVISION_COLS, row))
+        result.setdefault(entry['change_id'], []).append(entry)
     return result
 
 
+def summarize_revision_unseen(
+    revisions: List[Dict[str, Any]], tracked_revision: Optional[int] = None
+) -> int:
+    """Sum unseen messages across catalog revisions, excluding the tracked one.
+
+    Revisions whose thread was never fetched (NULL count) contribute
+    nothing.  The tracked revision is excluded because its unread state
+    is already surfaced through the series row badge.
+    """
+    total = 0
+    for rev in revisions:
+        if tracked_revision is not None and rev.get('revision') == tracked_revision:
+            continue
+        count = rev.get('message_count')
+        if count is None:
+            continue
+        total += max(0, int(count) - int(rev.get('seen_message_count') or 0))
+    return total
+
+
 def update_attestation(
     identifier: str, change_id: str, revision: int, attestation: Optional[str]
 ) -> None:
@@ -1925,6 +2027,45 @@ def _fetch_new_since(
         return None
 
 
+def _write_mbox_blob(topdir: str, msgs: List[Any]) -> Optional[str]:
+    """Serialize msgs to mboxrd and write as a git blob; return the SHA."""
+    import io
+
+    buf = io.BytesIO()
+    b4.save_mboxrd_mbox(msgs, buf)
+    mbox_bytes = buf.getvalue()
+    if not mbox_bytes:
+        return None
+
+    ecode, out = b4.git_run_command(
+        topdir, ['hash-object', '-w', '--stdin'], stdin=mbox_bytes
+    )
+    if ecode != 0:
+        return None
+    return str(out.strip())
+
+
+def _store_revision_thread_blob(
+    conn: sqlite3.Connection,
+    topdir: str,
+    change_id: str,
+    revision: int,
+    msgs: List[Any],
+) -> Optional[str]:
+    """Cache a catalog revision's thread mbox as a git blob.
+
+    Unlike _store_thread_blob this records the SHA in the revisions
+    catalog only -- the review branch tracking ref belongs to the
+    tracked revision.
+    """
+    blob_sha = _write_mbox_blob(topdir, msgs)
+    if blob_sha is None:
+        logger.debug('Could not store thread blob for %s v%d', change_id, revision)
+        return None
+    set_revision_thread_blob(conn, change_id, revision, blob_sha)
+    return blob_sha
+
+
 def _store_thread_blob(topdir: str, change_id: str, msgs: List[Any]) -> Optional[str]:
     """Serialize msgs to mboxrd and write as a git blob; update tracking commit.
 
@@ -1936,24 +2077,12 @@ def _store_thread_blob(topdir: str, change_id: str, msgs: List[Any]) -> Optional
     """
     # Local import first — avoids circular deps AND prevents UnboundLocalError
     # that would occur if `import b4.review` appeared after a `b4.xxx` call.
-    import io
-
     import b4.review as _b4_review
 
-    buf = io.BytesIO()
-    b4.save_mboxrd_mbox(msgs, buf)
-    mbox_bytes = buf.getvalue()
-    if not mbox_bytes:
-        logger.debug('No bytes to store for thread blob for %s', change_id)
-        return None
-
-    ecode, out = b4.git_run_command(
-        topdir, ['hash-object', '-w', '--stdin'], stdin=mbox_bytes
-    )
-    if ecode != 0:
+    blob_sha = _write_mbox_blob(topdir, msgs)
+    if blob_sha is None:
         logger.debug('Could not write thread blob for %s', change_id)
         return None
-    blob_sha = out.strip()
 
     branch_name = f'b4/review/{change_id}'
     if b4.git_branch_exists(topdir, branch_name):
@@ -2423,6 +2552,228 @@ def update_message_counts(
     return {'updated': updated, 'errors': errors}
 
 
+def _fetch_revision_thread_msgs(
+    identifier: str,
+    conn: sqlite3.Connection,
+    change_id: str,
+    rev: Dict[str, Any],
+) -> Optional[List[Any]]:
+    """Fetch the full thread for a catalog revision.
+
+    Rethreaded revisions reassemble from their per-patch message-ids;
+    plain revisions fetch the single thread mbox.  Returns None on
+    failure or when offline.
+    """
+    revision = int(rev['revision'])
+    if rev.get('is_rethreaded'):
+        patches = get_series_patches(conn, change_id, revision)
+        if sum(1 for p in patches if p.get('position', 0) > 0) >= 2:
+            import b4.review as _b4_review
+
+            series_dict = {
+                'message_id': rev.get('message_id', ''),
+                'change_id': change_id,
+                'revision': revision,
+                'is_rethreaded': True,
+            }
+            try:
+                return _b4_review.retrieve_series_messages(series_dict, identifier)
+            except Exception as ex:
+                logger.debug(
+                    'Could not reassemble rethreaded v%d of %s: %s',
+                    revision,
+                    change_id,
+                    ex,
+                )
+                return None
+    message_id = str(rev.get('message_id') or '')
+    if not message_id:
+        return None
+    mbox_bytes = _fetch_thread_mbox_bytes(message_id)
+    if mbox_bytes is None:
+        return None
+    return b4.split_and_dedupe_pi_results(mbox_bytes)
+
+
+def _update_one_revision_count(
+    identifier: str,
+    conn: sqlite3.Connection,
+    topdir: Optional[str],
+    change_id: str,
+    rev: Dict[str, Any],
+    now: str,
+) -> Optional[bool]:
+    """Update message counts for a single non-tracked catalog revision.
+
+    Returns True when the database changed, False when there was nothing
+    new (zero writes, keeping the DB mtime stable), None on fetch failure.
+    """
+    revision = int(rev['revision'])
+
+    if rev.get('message_count') is None or not rev.get('last_update_check'):
+        # ── First fetch: full thread download ────────────────────────────
+        msgs = _fetch_revision_thread_msgs(identifier, conn, change_id, rev)
+        if not msgs:
+            return None
+        count = len(msgs)
+        conn.execute(
+            'UPDATE revisions SET message_count = ?, seen_message_count = ?,'
+            ' last_update_check = ?, last_activity_at = ?'
+            ' WHERE change_id = ? AND revision = ?',
+            (count, count, now, _latest_date_from_msgs(msgs), change_id, revision),
+        )
+        conn.commit()
+        if topdir:
+            _store_revision_thread_blob(conn, topdir, change_id, revision, msgs)
+        return True
+
+    # ── Incremental: query for messages since last check ─────────────────
+    # A rethreaded revision is N separate threads, one per patch — sum the
+    # per-thread queries.  A reply CC'd into several patch threads can be
+    # counted more than once; the next full refresh corrects the total.
+    msgids = [str(rev.get('message_id') or '')]
+    if rev.get('is_rethreaded'):
+        patches = get_series_patches(conn, change_id, revision)
+        patch_ids = [p['message_id'] for p in patches if p.get('position', 0) > 0]
+        if patch_ids:
+            msgids = patch_ids
+
+    new_count = 0
+    new_activity: Optional[str] = None
+    for msgid in msgids:
+        if not msgid:
+            continue
+        result = _fetch_new_since(msgid, str(rev['last_update_check']))
+        if result is None:
+            return None
+        count, activity = result
+        new_count += count
+        if activity and (new_activity is None or activity > new_activity):
+            new_activity = activity
+
+    if new_count == 0:
+        return False
+
+    if topdir:
+        # New mail arrived — refresh the cached blob and take the exact
+        # count from the full refetch rather than trusting the increment.
+        msgs = _fetch_revision_thread_msgs(identifier, conn, change_id, rev)
+        if msgs:
+            conn.execute(
+                'UPDATE revisions SET message_count = ?, last_update_check = ?,'
+                ' last_activity_at = COALESCE(?, last_activity_at)'
+                ' WHERE change_id = ? AND revision = ?',
+                (
+                    len(msgs),
+                    now,
+                    _latest_date_from_msgs(msgs),
+                    change_id,
+                    revision,
+                ),
+            )
+            conn.commit()
+            _store_revision_thread_blob(conn, topdir, change_id, revision, msgs)
+            return True
+
+    conn.execute(
+        'UPDATE revisions SET message_count = message_count + ?,'
+        ' last_update_check = ?,'
+        ' last_activity_at = COALESCE(?, last_activity_at)'
+        ' WHERE change_id = ? AND revision = ?',
+        (new_count, now, new_activity, change_id, revision),
+    )
+    conn.commit()
+    return True
+
+
+def update_revision_message_counts(
+    identifier: str,
+    series_list: List[Dict[str, Any]],
+    topdir: Optional[str] = None,
+    max_revisions_per_series: Optional[int] = None,
+) -> Dict[str, int]:
+    """Fetch and store thread message counts for non-tracked revisions.
+
+    The series machinery owns the tracked revision's counts; this covers
+    every *other* revision in the catalog so new mail landing on an old
+    version's thread is still noticed:
+
+    - **First fetch** (``message_count IS NULL``): downloads the full
+      thread and stores the count, with ``seen_message_count`` initialised
+      to the same value so no badge appears until *new* activity arrives.
+      When *topdir* is given the mbox is cached as a git blob.
+    - **Incremental**: queries for messages newer than
+      ``last_update_check``.  An empty response produces **zero database
+      writes**, keeping the DB mtime stable and suppressing spurious list
+      reloads in the TUI.  New mail bumps the count and refreshes the
+      cached blob.
+
+    Revisions are polled newest-first, optionally capped by
+    *max_revisions_per_series*.
+
+    Returns ``{'updated': n, 'errors': n}`` where *updated* counts
+    revisions whose counts actually changed.
+    """
+    updated = 0
+    errors = 0
+    skip_statuses = frozenset(('archived', 'accepted', 'thanked', 'snoozed'))
+    now = datetime.datetime.now(datetime.timezone.utc).isoformat()
+
+    try:
+        conn = get_db(identifier)
+    except FileNotFoundError:
+        return {'updated': 0, 'errors': 0}
+
+    for series in series_list:
+        if series.get('status') in skip_statuses:
+            continue
+        change_id = series.get('change_id', '')
+        if not change_id:
+            continue
+        tracked_rev = int(series.get('revision') or 1)
+
+        # A catalog row for the tracked revision is not guaranteed (manual
+        # linking can record only newer versions); backfill it so the
+        # revision keeps its counts once it stops being tracked.  Guarded
+        # by an existence check to keep quiet runs write-free.
+        if series.get('message_id'):
+            row = conn.execute(
+                'SELECT 1 FROM revisions WHERE change_id = ? AND revision = ?',
+                (change_id, tracked_rev),
+            ).fetchone()
+            if row is None:
+                add_revision(
+                    conn,
+                    change_id,
+                    tracked_rev,
+                    str(series['message_id']),
+                    subject=series.get('subject'),
+                    fingerprint=series.get('fingerprint'),
+                    is_rethreaded=bool(series.get('is_rethreaded')),
+                )
+
+        polled = 0
+        for rev in reversed(get_revisions(conn, change_id)):
+            if int(rev['revision']) == tracked_rev:
+                continue
+            if (
+                max_revisions_per_series is not None
+                and polled >= max_revisions_per_series
+            ):
+                break
+            polled += 1
+            result = _update_one_revision_count(
+                identifier, conn, topdir, change_id, rev, now
+            )
+            if result is None:
+                errors += 1
+            elif result:
+                updated += 1
+
+    conn.close()
+    return {'updated': updated, 'errors': errors}
+
+
 def mark_all_messages_seen(
     conn: sqlite3.Connection, change_id: str, revision: int
 ) -> None:
@@ -2435,49 +2786,92 @@ def mark_all_messages_seen(
     conn.commit()
 
 
-def sync_seen_from_unseen_count(
-    identifier: str, change_id: str, revision: int, unseen_count: int
-) -> bool:
-    """Sync seen_message_count so the unread badge matches the messages DB.
+def mark_all_revision_messages_seen(
+    conn: sqlite3.Connection, change_id: str, revision: int
+) -> None:
+    """Clear the unread delta on a catalog revision (revisions table)."""
+    conn.execute(
+        'UPDATE revisions SET seen_message_count = message_count'
+        ' WHERE change_id = ? AND revision = ? AND message_count IS NOT NULL',
+        (change_id, revision),
+    )
+    conn.commit()
 
-    Sets ``seen_message_count = message_count - unseen_count``, clamped
-    to [0, message_count].  Only writes when the value actually changes.
 
-    Returns True if the database was updated, False otherwise.
+def _row_guard(table: str) -> str:
+    """WHERE-clause suffix scoping series lookups to non-archived rows.
+
+    Archived series rows are upgrade leftovers; per-revision reads and
+    writes must resolve to the revisions catalog instead of them.
     """
-    try:
-        conn = get_db(identifier)
-    except FileNotFoundError:
-        return False
+    return " AND status != 'archived'" if table == 'series' else ''
+
 
+def _apply_seen_sync(
+    conn: sqlite3.Connection,
+    table: str,
+    change_id: str,
+    revision: int,
+    unseen_count: int,
+) -> Optional[bool]:
+    """Apply a seen-count sync against one table.
+
+    Returns None when no row matches (caller should try the next table),
+    False when nothing changed, True when the database was updated.
+    """
+    guard = _row_guard(table)
     row = conn.execute(
-        'SELECT message_count, seen_message_count FROM series'
-        ' WHERE change_id = ? AND revision = ?',
+        f'SELECT message_count, seen_message_count FROM {table}'
+        f' WHERE change_id = ? AND revision = ?{guard}',
         (change_id, revision),
     ).fetchone()
     if row is None:
-        conn.close()
-        return False
+        return None
 
     fc = row['message_count']
     if fc is None:
-        conn.close()
         return False
 
     new_seen = max(0, min(fc, fc - unseen_count))
     if new_seen == row['seen_message_count']:
-        conn.close()
         return False
 
     conn.execute(
-        'UPDATE series SET seen_message_count = ? WHERE change_id = ? AND revision = ?',
+        f'UPDATE {table} SET seen_message_count = ?'
+        f' WHERE change_id = ? AND revision = ?{guard}',
         (new_seen, change_id, revision),
     )
     conn.commit()
-    conn.close()
     return True
 
 
+def sync_seen_from_unseen_count(
+    identifier: str, change_id: str, revision: int, unseen_count: int
+) -> bool:
+    """Sync seen_message_count so the unread badge matches the messages DB.
+
+    Sets ``seen_message_count = message_count - unseen_count``, clamped
+    to [0, message_count].  Only writes when the value actually changes.
+
+    The series row (the tracked revision) is preferred; when the
+    (change_id, revision) pair is not actively tracked the sync applies
+    to the revisions catalog row instead, so viewing an old version's
+    thread keeps its badge accurate.
+
+    Returns True if the database was updated, False otherwise.
+    """
+    try:
+        conn = get_db(identifier)
+    except FileNotFoundError:
+        return False
+
+    result = _apply_seen_sync(conn, 'series', change_id, revision, unseen_count)
+    if result is None:
+        result = _apply_seen_sync(conn, 'revisions', change_id, revision, unseen_count)
+    conn.close()
+    return bool(result)
+
+
 def refresh_message_count(
     identifier: str, change_id: str, revision: int, total_messages: int
 ) -> bool:
@@ -2496,6 +2890,10 @@ def refresh_message_count(
     Only writes to the database when the count differs from the stored
     value, keeping the DB mtime stable when nothing changed.
 
+    The series row (the tracked revision) is preferred; when the
+    (change_id, revision) pair is not actively tracked the count applies
+    to the revisions catalog row instead.
+
     Returns True if the database was updated, False otherwise.
     """
     now = datetime.datetime.now(datetime.timezone.utc).isoformat()
@@ -2504,29 +2902,52 @@ def refresh_message_count(
     except FileNotFoundError:
         return False
 
+    result = _apply_message_count(
+        conn, 'series', change_id, revision, total_messages, now
+    )
+    if result is None:
+        result = _apply_message_count(
+            conn, 'revisions', change_id, revision, total_messages, now
+        )
+    conn.close()
+    return bool(result)
+
+
+def _apply_message_count(
+    conn: sqlite3.Connection,
+    table: str,
+    change_id: str,
+    revision: int,
+    total_messages: int,
+    now: str,
+) -> Optional[bool]:
+    """Apply a fresh total message count against one table.
+
+    Returns None when no row matches (caller should try the next table),
+    False when nothing changed, True when the database was updated.
+    """
+    guard = _row_guard(table)
     row = conn.execute(
-        'SELECT message_count, seen_message_count FROM series'
-        ' WHERE change_id = ? AND revision = ?',
+        f'SELECT message_count, seen_message_count FROM {table}'
+        f' WHERE change_id = ? AND revision = ?{guard}',
         (change_id, revision),
     ).fetchone()
     if row is None:
-        conn.close()
-        return False
+        return None
 
     count = total_messages
     old_count = row['message_count']
 
     if old_count is not None and count == old_count:
         # Nothing changed — skip the write to keep the DB mtime stable.
-        conn.close()
         return False
 
     if old_count is None:
         # First fetch: initialise both counts equally (no badge).
         conn.execute(
-            'UPDATE series SET message_count = ?, seen_message_count = ?,'
-            '  last_update_check = ?'
-            ' WHERE change_id = ? AND revision = ?',
+            f'UPDATE {table} SET message_count = ?, seen_message_count = ?,'
+            f'  last_update_check = ?'
+            f' WHERE change_id = ? AND revision = ?{guard}',
             (count, count, now, change_id, revision),
         )
     else:
@@ -2535,20 +2956,19 @@ def refresh_message_count(
         seen = row['seen_message_count']
         if seen is not None and seen > count:
             conn.execute(
-                'UPDATE series SET message_count = ?, seen_message_count = ?,'
-                '  last_update_check = ?'
-                ' WHERE change_id = ? AND revision = ?',
+                f'UPDATE {table} SET message_count = ?, seen_message_count = ?,'
+                f'  last_update_check = ?'
+                f' WHERE change_id = ? AND revision = ?{guard}',
                 (count, count, now, change_id, revision),
             )
         else:
             conn.execute(
-                'UPDATE series SET message_count = ?, last_update_check = ?'
-                ' WHERE change_id = ? AND revision = ?',
+                f'UPDATE {table} SET message_count = ?, last_update_check = ?'
+                f' WHERE change_id = ? AND revision = ?{guard}',
                 (count, now, change_id, revision),
             )
 
     conn.commit()
-    conn.close()
     return True
 
 

-- 
2.53.0