feat(worker): forum-topic scan-skip + getForumTopic short-circuit

Mirror of the non-forum guards from 1a4bc6f, scoped to forum topics
inside the topic loop:

  - Top-of-topic-loop recency/backoff skip
  - getForumTopic short-circuit after the SkippedPackage retry pass
  - upsertTopicScanState for end-of-scan persistence (both the
    archives-found path and the no-archives path)

Same trulyIdle definition throughout: no archives this scan, no
failures this scan, no retryable SkippedPackage rows pending. Topics
with chronic failures stay out of backoff because their counter
never increments.

For MPE specifically (1,086 forum topics), per-cycle searchChatMessages
calls drop from ~1,086 to roughly the count of topics with new
activity in the last 5 minutes — typically <50.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

docs/superpowers/specs/2026-05-26-channel-scan-skip-design.md

@@ -0,0 +1,353 @@
+# Channel-Scan Skip Optimization — Design
+
+**Goal:** stop the worker from re-scanning channels and forum topics that haven't changed since the last scan, especially on restart. Reduce the per-cycle API call count for the Model Printing Emporium channel (1,086 forum topics) from ~1,000+ to ~50.
+
+**Non-goals:**
+- Replacing polling with event-driven ingestion (`updateNewMessage`). That's a separate, larger design (Phase 2 in the original brainstorm).
+- Surfacing per-channel scan history in the UI (also a separate, observability-only design).
+
+**Architecture sketch:**
+Add three persisted columns to `AccountChannelMap` and `TopicProgress`, plus one runtime `getChat`/`getForumTopicInfo` lookup before each scan. The new state survives restarts because it's in PostgreSQL; the lookup is a cheap TDLib local-cache call. Failure-retry semantics (`d99a506` + `901f32f`) must be preserved — a channel sitting on retryable `SkippedPackage` rows is never considered idle.
+
+---
+
+## Problem statement
+
+### Today's behavior
+
+Every ingestion cycle the worker walks every linked source channel for every authenticated account. For each channel/topic it calls TDLib's `searchChatMessages` paginated from `lastProcessedMessageId`. Even when nothing has changed since the previous scan:
+
+- One `searchChatMessages` call (sometimes paginated) is still made
+- For Model Printing Emporium, that's ~1,086 calls per cycle (one per forum topic)
+- The 1-second `apiDelayMs` between pages multiplies the cost
+- Most calls return zero new messages — the work is wasted
+
+The cost is most acute right after a restart: the worker boots, runs recovery, then issues 1,000+ effectively-empty calls before any productive work happens.
+
+### What we already track
+
+- `AccountChannelMap.lastProcessedMessageId` — highest processed message ID (per non-forum channel, per account)
+- `TopicProgress.lastProcessedMessageId` — same per forum topic
+- Both are advanced incrementally per archive set (`77aeb4c`)
+- Both are pulled back below failed messages by the `SkippedPackage` retry pass (`901f32f`)
+
+### What we don't track and want to add
+
+- When was the last scan?
+- Did the last scan find any archives, OR is there outstanding retry work?
+- How many cycles in a row have been totally idle?
+
+These let us skip the scan entirely when nothing has changed.
+
+---
+
+## High-level approach
+
+Three guards at the top of the per-channel and per-topic processing loops:
+
+1. **DB-persistent "skip if recently scanned and truly idle"** — checks `lastScannedAt`, `lastScanFoundArchives`, and a `retryableSkippedCount` query. If all three say "nothing new, nothing failing", skip without any TDLib call.
+
+2. **Adaptive backoff for cold channels** — `consecutiveEmptyScans` counter. After it crosses a threshold, scan only every Nth cycle. Reset to 0 whenever the channel is "not idle".
+
+3. **`chat.last_message.id` short-circuit** — if (1) and (2) don't skip but the channel's last server-side message ID matches our watermark, skip the `searchChatMessages` paginated call. This runs after the existing `SkippedPackage` retry pass, which pulls the watermark back below failures, so it correctly forces a scan when retries are pending.
+
+The retry pass from `901f32f` is preserved untouched — it runs in front of these guards and adjusts the watermark, so retries always happen.
+
+---
+
+## Schema changes
+
+### `AccountChannelMap` (worker/src/db/schema.prisma)
+
+```prisma
+model AccountChannelMap {
+  // ... existing fields ...
+  lastScannedAt          DateTime?
+  lastScanFoundArchives  Boolean   @default(false)
+  consecutiveEmptyScans  Int       @default(0)
+}
+```
+
+### `TopicProgress`
+
+```prisma
+model TopicProgress {
+  // ... existing fields ...
+  lastScannedAt          DateTime?
+  lastScanFoundArchives  Boolean   @default(false)
+  consecutiveEmptyScans  Int       @default(0)
+}
+```
+
+### Migration
+
+```sql
+-- Both tables get the same three columns. Existing rows get defaults:
+--   lastScannedAt        = NULL  (next scan will populate)
+--   lastScanFoundArchives = false (safe default — will be overwritten by next scan)
+--   consecutiveEmptyScans = 0    (resets backoff for existing channels)
+
+ALTER TABLE "account_channel_map"
+  ADD COLUMN "lastScannedAt" TIMESTAMP(3),
+  ADD COLUMN "lastScanFoundArchives" BOOLEAN NOT NULL DEFAULT false,
+  ADD COLUMN "consecutiveEmptyScans" INTEGER NOT NULL DEFAULT 0;
+
+ALTER TABLE "topic_progress"
+  ADD COLUMN "lastScannedAt" TIMESTAMP(3),
+  ADD COLUMN "lastScanFoundArchives" BOOLEAN NOT NULL DEFAULT false,
+  ADD COLUMN "consecutiveEmptyScans" INTEGER NOT NULL DEFAULT 0;
+```
+
+NULL `lastScannedAt` means "never scanned" — every channel will be scanned the first cycle after deploy. Subsequent cycles benefit from the new fields.
+
+---
+
+## Configuration
+
+Two new env vars in `worker/src/util/config.ts`:
+
+```typescript
+/** Window in which a recent successful empty scan lets us skip. Default 5 min. */
+skipRecentScanWindowMs:
+  parseInt(process.env.WORKER_SKIP_RECENT_SCAN_WINDOW_MS ?? "300000", 10),
+
+/** After this many consecutive empty scans, channel enters backoff mode. */
+emptyScanBackoffThreshold:
+  parseInt(process.env.WORKER_EMPTY_SCAN_BACKOFF_THRESHOLD ?? "5", 10),
+
+/** Backoff factor — N means "scan every Nth cycle once in backoff". */
+emptyScanBackoffEveryNth:
+  parseInt(process.env.WORKER_EMPTY_SCAN_BACKOFF_EVERY_NTH ?? "5", 10),
+```
+
+All three are tunable per deployment without code changes.
+
+---
+
+## Decision logic per channel / topic
+
+The skip decision sits at the top of each channel/topic iteration in `runWorkerForAccount`. It runs BEFORE the existing `SkippedPackage` retry pass.
+
+```text
+For each channel (or topic):
+
+  1. Query retryableSkippedCount for this scope (already a query we do elsewhere)
+
+  2. If retryableSkippedCount > 0:
+       Force scan (don't skip — failures need retry)
+       Proceed to existing flow (retry pass → scan)
+
+  3. Else if lastScannedAt is NULL:
+       Force scan (we've never touched this)
+       Proceed to existing flow
+
+  4. Else if Date.now() - lastScannedAt.getTime() < skipRecentScanWindowMs
+          AND lastScanFoundArchives === false:
+       Skip — recently scanned and truly idle
+
+  5. Else if consecutiveEmptyScans >= emptyScanBackoffThreshold
+          AND (cycleCount % emptyScanBackoffEveryNth !== 0):
+       Skip — channel is cold, not its turn to scan
+
+  6. Else:
+       Run the existing flow:
+         a. SkippedPackage retry pass (901f32f) — may pull watermark back
+         b. NEW: getChat (or getForumTopicInfo) — if last_message.id <= watermark, skip
+         c. searchChatMessages scan
+```
+
+`cycleCount` is the global ingestion-cycle counter from `scheduler.ts`. It already increments per cycle.
+
+---
+
+## End-of-scan bookkeeping
+
+After every scan (whether it found archives or not), update the three new fields atomically with the existing watermark write:
+
+```typescript
+// "Truly idle" means: nothing new this scan AND nothing failed AND no leftover
+// retryable failures. The retry-pending check is critical — without it, a
+// scan that found no new archives but left SkippedPackage retries pending
+// would be marked idle and incorrectly skipped next cycle.
+const retryablePending = await getRetryableSkippedMessageIds({
+  accountId, sourceChannelId, topicId, cap: maxSkipAttempts,
+});
+const trulyIdle =
+  scanResult.archives.length === 0
+  && minFailedId === null
+  && retryablePending.length === 0;
+
+const newConsecutiveEmpty = trulyIdle
+  ? (prev.consecutiveEmptyScans ?? 0) + 1
+  : 0;
+
+await upsertChannelOrTopicScanState({
+  // ... existing watermark fields ...
+  lastScannedAt: new Date(),
+  lastScanFoundArchives: !trulyIdle,
+  consecutiveEmptyScans: newConsecutiveEmpty,
+});
+```
+
+The `consecutiveEmptyScans` counter resets to 0 the moment *anything* happens — archives found, archives failed, or unresolved retries pending. A channel with a chronically-failing archive (whose attemptCount is still below the cap) keeps the counter at 0 and never enters backoff.
+
+If a SkippedPackage hits `attemptCount === maxSkipAttempts`, it's no longer "retryable pending" (it's been given up on), so the counter increments correctly. Same for SkippedPackages that get deleted via the UI's "retry" button — the counter behaves correctly without special-casing.
+
+---
+
+## `getChat` / `getForumTopicInfo` short-circuit
+
+After the retry pass has finalized the effective watermark, but before `searchChatMessages`:
+
+```typescript
+// For non-forum channels:
+const chat = await client.invoke({ _: "getChat", chat_id: Number(channel.telegramId) });
+const channelLastMessageId = chat.last_message?.id;
+
+if (channelLastMessageId && BigInt(channelLastMessageId) <= effectiveWatermark) {
+  // Nothing new server-side — skip the paginated search entirely.
+  // Still update lastScannedAt / consecutiveEmptyScans so the recent-scan
+  // skip kicks in next cycle.
+  await persistScanState({ trulyIdle: true });
+  continue;
+}
+
+// For forum topics:
+const topicInfo = await client.invoke({
+  _: "getForumTopicInfo",
+  chat_id: Number(channel.telegramId),
+  message_thread_id: Number(topic.topicId),
+});
+const topicLastMessageId = topicInfo.info?.last_message_id;
+
+if (topicLastMessageId && BigInt(topicLastMessageId) <= effectiveWatermark) {
+  await persistScanState({ trulyIdle: true });
+  continue;
+}
+```
+
+`getChat` is served from TDLib's local cache (no network) for chats we've already loaded, which we do up front via `loadChats`. `getForumTopicInfo` is a single round-trip but much cheaper than a paginated `searchChatMessages` call.
+
+The comparison is `<=` because the watermark is the highest message we've fully processed — if the server's last is the same, we're caught up.
+
+This step is correct in the failure-retry case because the retry pass runs FIRST: if there were retryable failures, the retry pass pulled the watermark back below them, and `channelLastMessageId > effectiveWatermark` (since the failed message exists in TG), so we don't skip — we scan and re-pick-up the failure.
+
+---
+
+## Restart behavior
+
+The improvements compose for restart safety:
+
+| Scenario | Today | After this change |
+|---|---|---|
+| Restart 5 min after a clean cycle | ~2,000 API calls for MPE | ~10 calls (only retryable + truly-active topics) |
+| Restart 1 hour later (one missed cycle) | ~2,000 API calls | `getChat` per channel + scan only those where `last_message.id > watermark` (≈ 50 for MPE) |
+| Restart after long downtime (12h) | ~2,000 calls + lots of new content | `getChat` per channel, scan everything with new activity |
+
+The three new columns are in PostgreSQL — they survive container restarts directly. `consecutiveEmptyScans = 47` for a cold topic stays at 47 across restart, so backoff continues to apply.
+
+---
+
+## Edge cases and their handling
+
+### 1. Manual SkippedPackage retry via UI between cycles
+The UI's `retrySkippedPackageAction` lowers the watermark and deletes the SkippedPackage. Next cycle: `retryableSkippedCount === 0` (the row is gone), but the watermark is lower than `chat.last_message.id` (the retried message exists in TG). So step 6 in the decision tree triggers a scan via the `getChat` check. ✓
+
+### 2. SkippedPackage hits the attempt cap mid-cycle
+Once `attemptCount === maxSkipAttempts`, the row is no longer in `getRetryableSkippedMessageIds` results. The channel correctly becomes idle-eligible. The capped SkippedPackage stays in the table as "permanently failed (manual retry only)" — that's the existing behavior. ✓
+
+### 3. New SkippedPackage is created mid-cycle (e.g., an upload fails)
+At the end of that scan, `retryablePending` includes the new row → `trulyIdle = false` → `lastScanFoundArchives = true` → next cycle does NOT skip. ✓
+
+### 4. Channel/topic added after deploy
+New rows in `AccountChannelMap` / `TopicProgress` have `lastScannedAt = NULL`, so step 3 in the decision tree always triggers a scan. After the first scan, the fields are populated normally. ✓
+
+### 5. Clock skew / drift
+The `lastScannedAt < 5 min ago` check uses `Date.now() - lastScannedAt.getTime()`. Both are application-side clocks (Node.js + PostgreSQL `NOW()` at write). A few seconds of drift doesn't matter; an hour of clock jump (rare but possible) just means one cycle either skips or re-scans — recoverable.
+
+### 6. TDLib `getChat` returns stale data
+TDLib's local cache could theoretically be stale (e.g., the account hasn't received the latest update yet). If `channelLastMessageId` is stale (lower than server reality), we'd skip a scan that should have happened. Mitigation: the next cycle's `getChat` likely has fresh data; the watermark guards correctness (we don't lose data, we just process it one cycle later). Acceptable.
+
+### 7. `getForumTopicInfo` rate limit
+Calling it per-topic could add up for channels with 1000+ topics. Mitigation: skip-on-recent-scan (step 4) eliminates the call for most topics; only "stale-but-was-active" topics get the call. Worst case is ~50 calls per cycle for MPE, comfortably under the 30 req/sec global limit.
+
+### 8. Channel becomes a forum (or vice versa) between cycles
+Existing code handles this — `isChatForum` is rechecked each cycle and `setChannelForum` updates the DB. The new fields live on the same rows, so no extra handling needed.
+
+---
+
+## File-level changes
+
+### New / modified
+
+- `prisma/schema.prisma` — add the six new columns
+- `prisma/migrations/<timestamp>_channel_scan_state/migration.sql` — the ALTER TABLE
+- `worker/src/util/config.ts` — three new env vars
+- `worker/src/db/queries.ts` — new helpers:
+  - `getChannelScanState(mappingId)` and `getTopicScanState(topicProgressId)`
+  - `upsertChannelScanState(...)` and `upsertTopicScanState(...)`
+  - Both wrap the existing `updateLastProcessedMessage` / `upsertTopicProgress` so callers don't need to remember to update the new fields too.
+- `worker/src/worker.ts` — top-of-loop skip checks in both the forum and non-forum branches, plus end-of-scan state writes
+- `worker/src/tdlib/chats.ts` — small helper `getChatLastMessageId(client, chatId)` and `getForumTopicLastMessageId(client, chatId, topicId)` wrapping the TDLib calls with the existing `invokeWithTimeout` pattern
+
+### Untouched
+
+- `recovery.ts` — recovery is per-startup and one-shot; not affected
+- `scheduler.ts` — `cycleCount` is already there; just expose it where needed
+- The existing `SkippedPackage` retry pass logic in `runWorkerForAccount` is unchanged
+
+---
+
+## Testing plan
+
+The project has no automated tests, so verification is manual via Docker logs after deploy:
+
+1. **Build cleanly:** `docker compose up -d --build worker` — no migration errors
+2. **First cycle after deploy:** all channels scan (NULL `lastScannedAt`), all fields populated at end of cycle. Log lines confirm normal scan flow.
+3. **Second cycle 5 min later:**
+   - Check logs for `"Skipping recently-scanned idle channel"` — should appear for any channel/topic that was empty last cycle
+   - Total `searchChatMessages` calls per cycle should drop dramatically (compare to first cycle)
+4. **Failure-retry preservation:**
+   - Find a SkippedPackage with `attemptCount < cap`
+   - Run a cycle — confirm the channel/topic is NOT skipped (log says it's scanned)
+   - Confirm the SkippedPackage gets re-tried
+5. **Backoff:**
+   - Pick a cold channel, wait for it to scan 5+ cycles cleanly
+   - Confirm `consecutiveEmptyScans` climbs to 5+
+   - Confirm subsequent cycles skip it (only scan every 5th)
+6. **`getChat` short-circuit:**
+   - Pick an active channel
+   - Trigger an immediate cycle (UI button)
+   - If `last_message.id <= watermark`, expect log `"Channel caught up via getChat — skipping searchChatMessages"`
+7. **Restart safety:**
+   - Push the change, restart worker
+   - First cycle after restart should log multiple "Skipping recently-scanned idle channel" lines (because the DB state survived)
+   - Total cycle time should be a fraction of a baseline restart
+
+---
+
+## Risks and mitigations
+
+| Risk | Mitigation |
+|---|---|
+| Skip incorrectly applied → real failures never retried | Rule 1 (truly-idle includes `retryablePending === 0`) + dedicated test step 4 |
+| `getChat` returns stale data | Next cycle's `getChat` corrects it; watermark guards correctness (no data loss) |
+| `getForumTopicInfo` not available in TDLib 1.8.64 | Verify the method exists in the schema; fall back to scan if it throws |
+| Backoff applies during legitimate activity bursts | Counter resets to 0 the moment any archive is found OR any retry is pending |
+| Migration takes too long on the live DB | Both columns have NOT NULL defaults — Postgres can add them as fast metadata changes (no table rewrite) |
+
+---
+
+## What's explicitly NOT in this design
+
+To keep scope tight:
+- **Event-driven ingestion via `updateNewMessage`.** Bigger design, addressed separately. This design is compatible with it — when (D) lands, polling becomes a 4-hour safety net using these same skip rules.
+- **Per-channel scan history UI.** Observability layer; separate design.
+- **Surfacing the new counters in the admin dashboard.** Can come after the worker-side change is verified.
+- **Backfilling `consecutiveEmptyScans` from historical `IngestionRun` data.** Not worth it — it'll converge to the correct value within ~6 cycles.
+
+---
+
+## Open questions
+
+None — the failure-retry interaction was the main risk and is handled by Rule 1 + the existing retry pass.

prisma/migrations/20260526000000_channel_scan_state/migration.sql

@@ -0,0 +1,11 @@
+-- AlterTable: per-channel scan-state columns
+ALTER TABLE "account_channel_map"
+  ADD COLUMN "lastScannedAt" TIMESTAMP(3),
+  ADD COLUMN "lastScanFoundArchives" BOOLEAN NOT NULL DEFAULT false,
+  ADD COLUMN "consecutiveEmptyScans" INTEGER NOT NULL DEFAULT 0;
+
+-- AlterTable: per-topic scan-state columns (forum channels)
+ALTER TABLE "topic_progress"
+  ADD COLUMN "lastScannedAt" TIMESTAMP(3),
+  ADD COLUMN "lastScanFoundArchives" BOOLEAN NOT NULL DEFAULT false,
+  ADD COLUMN "consecutiveEmptyScans" INTEGER NOT NULL DEFAULT 0;

prisma/schema.prisma

@@ -450,6 +450,17 @@ model AccountChannelMap {
   channelId              String
   role                   ChannelRole @default(READER)
   lastProcessedMessageId BigInt?
+  /// When this channel was last scanned (any reason, including skipped scans
+  /// that bumped the timestamp). Used by the recency-skip guard.
+  lastScannedAt          DateTime?
+  /// True if the last scan found archives OR left retryable SkippedPackages
+  /// pending. Tracks "this channel has work I might need to revisit" — not
+  /// just "I uploaded something this cycle".
+  lastScanFoundArchives  Boolean     @default(false)
+  /// Number of consecutive cycles where this channel was trulyIdle (no
+  /// archives, no failures, no retryables). Drives the backoff that lets
+  /// cold channels skip cycles entirely.
+  consecutiveEmptyScans  Int         @default(0)
   createdAt              DateTime    @default(now())
 
   account        TelegramAccount @relation(fields: [accountId], references: [id], onDelete: Cascade)
@@ -587,6 +598,14 @@ model TopicProgress {
   topicId                BigInt
   topicName              String?
   lastProcessedMessageId BigInt?
+  /// When this topic was last scanned (any reason). Used by recency-skip.
+  lastScannedAt          DateTime?
+  /// True if the last scan found archives OR has retryable SkippedPackages
+  /// pending for this topic. See AccountChannelMap doc for details.
+  lastScanFoundArchives  Boolean   @default(false)
+  /// Number of consecutive cycles where this topic was trulyIdle. Drives
+  /// backoff for cold topics.
+  consecutiveEmptyScans  Int       @default(0)
 
   accountChannelMap AccountChannelMap @relation(fields: [accountChannelMapId], references: [id], onDelete: Cascade)
 

worker/src/db/queries.ts

@@ -455,6 +455,73 @@ export async function updateLastProcessedMessage(
   });
 }
 
+export interface ScanStateUpdate {
+  /** New watermark to persist. Use the same value the caller would have
+   *  passed to updateLastProcessedMessage / upsertTopicProgress. */
+  lastProcessedMessageId: bigint | null;
+  /** True if the scan found archives OR has retryable SkippedPackages
+   *  pending. The caller computes this via the trulyIdle formula. */
+  lastScanFoundArchives: boolean;
+  /** Pre-incremented value of consecutiveEmptyScans. Caller passes:
+   *    trulyIdle ? prev + 1 : 0
+   *  We do the arithmetic outside the helper so the helper stays a pure
+   *  setter — easier to reason about. */
+  consecutiveEmptyScans: number;
+}
+
+/**
+ * Atomically update an AccountChannelMap's watermark and scan-state fields.
+ * Replaces the older updateLastProcessedMessage for the post-scan write.
+ * Sets lastScannedAt = NOW() server-side.
+ */
+export async function upsertChannelScanState(
+  mappingId: string,
+  update: ScanStateUpdate
+) {
+  return db.accountChannelMap.update({
+    where: { id: mappingId },
+    data: {
+      lastProcessedMessageId: update.lastProcessedMessageId ?? undefined,
+      lastScannedAt: new Date(),
+      lastScanFoundArchives: update.lastScanFoundArchives,
+      consecutiveEmptyScans: update.consecutiveEmptyScans,
+    },
+  });
+}
+
+/**
+ * Atomically upsert a TopicProgress row with the new watermark + scan-state
+ * fields. Same semantics as upsertChannelScanState but for forum topics.
+ */
+export async function upsertTopicScanState(
+  accountChannelMapId: string,
+  topicId: bigint,
+  topicName: string | null,
+  update: ScanStateUpdate
+) {
+  return db.topicProgress.upsert({
+    where: {
+      accountChannelMapId_topicId: { accountChannelMapId, topicId },
+    },
+    create: {
+      accountChannelMapId,
+      topicId,
+      topicName,
+      lastProcessedMessageId: update.lastProcessedMessageId,
+      lastScannedAt: new Date(),
+      lastScanFoundArchives: update.lastScanFoundArchives,
+      consecutiveEmptyScans: update.consecutiveEmptyScans,
+    },
+    update: {
+      topicName,
+      lastProcessedMessageId: update.lastProcessedMessageId ?? undefined,
+      lastScannedAt: new Date(),
+      lastScanFoundArchives: update.lastScanFoundArchives,
+      consecutiveEmptyScans: update.consecutiveEmptyScans,
+    },
+  });
+}
+
 export async function markStaleRunsAsFailed() {
   return db.ingestionRun.updateMany({
     where: { status: "RUNNING" },

worker/src/scheduler.ts

@@ -19,6 +19,12 @@ let activeCyclePromise: Promise<void> | null = null;
  */
 const CYCLE_TIMEOUT_MS = (parseInt(process.env.WORKER_CYCLE_TIMEOUT_MINUTES ?? "240", 10)) * 60 * 1000;
 
+/** Read-only access to the current cycle counter for code that needs to
+ *  apply per-cycle modulo logic (e.g. the cold-channel backoff). */
+export function getCurrentCycle(): number {
+  return cycleCount;
+}
+
 /**
  * Run one ingestion cycle:
  * 1. Authenticate any PENDING accounts (triggers SMS code flow + auto-fetch channels)

worker/src/tdlib/chats.ts

@@ -307,3 +307,63 @@ export async function searchPublicChat(
 function sleep(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }
+
+/**
+ * Return the chat's server-side last_message.id from TDLib's local cache.
+ * Used by the channel-scan-skip guard to short-circuit a paginated
+ * searchChatMessages when nothing has changed since our watermark.
+ *
+ * Returns null when the chat has no last_message (empty channel) or the
+ * call fails — callers must treat null as "unknown" and run the scan.
+ */
+export async function getChannelLastMessageId(
+  client: Client,
+  chatId: bigint
+): Promise<bigint | null> {
+  try {
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const chat = (await client.invoke({
+      _: "getChat",
+      chat_id: Number(chatId),
+    })) as { last_message?: { id?: number } };
+    const id = chat.last_message?.id;
+    return id ? BigInt(id) : null;
+  } catch (err) {
+    log.debug({ err, chatId: chatId.toString() }, "getChannelLastMessageId failed");
+    return null;
+  }
+}
+
+/**
+ * Return the forum topic's last_message_id from TDLib. Same purpose as
+ * getChannelLastMessageId but scoped to a single topic in a forum
+ * supergroup. TDLib's `getForumTopic` returns a `forumTopic` whose `info`
+ * field contains the last_message_id.
+ *
+ * Returns null on failure or empty topic — caller treats as "unknown".
+ */
+export async function getForumTopicLastMessageId(
+  client: Client,
+  chatId: bigint,
+  topicId: bigint
+): Promise<bigint | null> {
+  try {
+    // TDLib 1.8.64 uses `forum_topic_id` (renamed from `message_thread_id`
+    // in the request) — consistent with the rest of the forum-topic API
+    // surface in this version.
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const topic = (await client.invoke({
+      _: "getForumTopic",
+      chat_id: Number(chatId),
+      forum_topic_id: Number(topicId),
+    })) as { last_message?: { id?: number }; info?: { last_message_id?: number } };
+    const id = topic.last_message?.id ?? topic.info?.last_message_id;
+    return id ? BigInt(id) : null;
+  } catch (err) {
+    log.debug(
+      { err, chatId: chatId.toString(), topicId: topicId.toString() },
+      "getForumTopicLastMessageId failed"
+    );
+    return null;
+  }
+}

worker/src/util/config.ts

@@ -24,4 +24,22 @@ export const config = {
    *  stops auto-retrying and lets the watermark advance past it. The user can
    *  manually retry via the UI to reset and try again. */
   maxSkipAttempts: parseInt(process.env.WORKER_MAX_SKIP_ATTEMPTS ?? "5", 10),
+  /** Window in which a recent successful empty scan lets us skip the next
+   *  scan entirely. Default 5 minutes. */
+  skipRecentScanWindowMs: parseInt(
+    process.env.WORKER_SKIP_RECENT_SCAN_WINDOW_MS ?? "300000",
+    10
+  ),
+  /** After this many consecutive empty scans, a channel/topic enters
+   *  backoff and is only scanned every Nth cycle. */
+  emptyScanBackoffThreshold: parseInt(
+    process.env.WORKER_EMPTY_SCAN_BACKOFF_THRESHOLD ?? "5",
+    10
+  ),
+  /** While in backoff, scan only every Nth cycle. Default 5 = scan every
+   *  fifth cycle = once every ~5 hours given the 60-min default interval. */
+  emptyScanBackoffEveryNth: parseInt(
+    process.env.WORKER_EMPTY_SCAN_BACKOFF_EVERY_NTH ?? "5",
+    10
+  ),
 } as const;

worker/src/worker.ts

@@ -35,10 +35,18 @@ import {
   findPackageByRemoteUniqueId,
   getRetryableSkippedMessageIds,
   updatePackageTopicContext,
+  upsertChannelScanState,
+  upsertTopicScanState,
 } from "./db/queries.js";
 import type { ActivityUpdate } from "./db/queries.js";
 import { createTdlibClient, closeTdlibClient } from "./tdlib/client.js";
-import { getAccountChats, joinChatByInviteLink } from "./tdlib/chats.js";
+import {
+  getAccountChats,
+  joinChatByInviteLink,
+  getChannelLastMessageId,
+  getForumTopicLastMessageId,
+} from "./tdlib/chats.js";
+import { getCurrentCycle } from "./scheduler.js";
 import { getChannelMessages, downloadFile, downloadPhotoThumbnail } from "./tdlib/download.js";
 import type { DownloadProgress, ChannelScanResult } from "./tdlib/download.js";
 import { isChatForum, getForumTopicList, getTopicMessages } from "./tdlib/topics.js";
@@ -590,6 +598,49 @@ export async function runWorkerForAccount(
                 }
               }
 
+              // ── Topic-scan-skip guard ──
+              // Same three-signal decision as the non-forum branch, but
+              // scoped to a single topic. Uses `progress` for the persisted
+              // scan-state fields (lastScannedAt etc).
+              try {
+                const retryableForTopic = await getRetryableSkippedMessageIds({
+                  accountId: account.id,
+                  sourceChannelId: channel.id,
+                  topicId: topic.topicId,
+                  cap: config.maxSkipAttempts,
+                });
+                if (retryableForTopic.length === 0 && progress?.lastScannedAt) {
+                  const sinceLastScanMs = Date.now() - progress.lastScannedAt.getTime();
+                  const withinRecencyWindow = sinceLastScanMs < config.skipRecentScanWindowMs;
+                  const inBackoff =
+                    (progress.consecutiveEmptyScans ?? 0) >= config.emptyScanBackoffThreshold;
+                  const backoffSkipsThisCycle =
+                    inBackoff && getCurrentCycle() % config.emptyScanBackoffEveryNth !== 0;
+
+                  if (
+                    (withinRecencyWindow && !progress.lastScanFoundArchives) ||
+                    backoffSkipsThisCycle
+                  ) {
+                    accountLog.debug(
+                      {
+                        channel: channel.title,
+                        topic: topic.name,
+                        sinceLastScanMs,
+                        consecutiveEmptyScans: progress.consecutiveEmptyScans,
+                        reason: withinRecencyWindow ? "recent-idle" : "backoff",
+                      },
+                      "Skipping topic — recently scanned and idle, or in backoff"
+                    );
+                    continue;
+                  }
+                }
+              } catch (skipErr) {
+                accountLog.warn(
+                  { err: skipErr, topic: topic.name },
+                  "Topic skip guard failed, proceeding with scan"
+                );
+              }
+
               // ── SkippedPackage retry pass ──
               // If we have failed messages in this topic with attemptCount
               // below the cap, pull the watermark back below the lowest of
@@ -640,6 +691,38 @@ export async function runWorkerForAccount(
                 ? ` (topic ${tIdx + 1}/${topics.length})`
                 : "";
 
+              // ── getForumTopic short-circuit ──
+              // After the retry pass has settled the effective watermark,
+              // ask TDLib for the topic's last_message_id. If it's <= our
+              // watermark, no new content — skip the paginated search.
+              const topicLastId = await getForumTopicLastMessageId(
+                client,
+                channel.telegramId,
+                topic.topicId
+              );
+              const effectiveTopicWatermark = progress?.lastProcessedMessageId ?? null;
+              if (
+                topicLastId !== null
+                && effectiveTopicWatermark !== null
+                && topicLastId <= effectiveTopicWatermark
+              ) {
+                accountLog.info(
+                  {
+                    channel: channel.title,
+                    topic: topic.name,
+                    topicLastId: topicLastId.toString(),
+                    watermark: effectiveTopicWatermark.toString(),
+                  },
+                  "Topic caught up via getForumTopic — skipping searchChatMessages"
+                );
+                await upsertTopicScanState(mapping.id, topic.topicId, topic.name, {
+                  lastProcessedMessageId: effectiveTopicWatermark,
+                  lastScanFoundArchives: false,
+                  consecutiveEmptyScans: (progress?.consecutiveEmptyScans ?? 0) + 1,
+                });
+                continue;
+              }
+
               await updateRunActivity(activeRunId, {
                 currentActivity: `Scanning "${topicLabel}"${topicProgress}`,
                 currentStep: "scanning",
@@ -677,14 +760,25 @@ export async function runWorkerForAccount(
                   { channelId: channel.id, topic: topic.name, totalScanned: scanResult.totalScanned },
                   "No new archives in topic"
                 );
-                // Still advance topic watermark so we don't re-scan these messages next cycle
+                // Still advance topic watermark so we don't re-scan these
+                // messages next cycle. Truly idle only when no retryable
+                // SkippedPackages are pending for this topic — chronically-
+                // failing archives must NOT push a topic into backoff.
+                const retryableTopicNoArchives = await getRetryableSkippedMessageIds({
+                  accountId: account.id,
+                  sourceChannelId: channel.id,
+                  topicId: topic.topicId,
+                  cap: config.maxSkipAttempts,
+                });
+                const topicTrulyIdleNoArchives = retryableTopicNoArchives.length === 0;
                 if (scanResult.maxScannedMessageId) {
-                  await upsertTopicProgress(
-                    mapping.id,
-                    topic.topicId,
-                    topic.name,
-                    scanResult.maxScannedMessageId
-                  );
+                  await upsertTopicScanState(mapping.id, topic.topicId, topic.name, {
+                    lastProcessedMessageId: scanResult.maxScannedMessageId,
+                    lastScanFoundArchives: !topicTrulyIdleNoArchives,
+                    consecutiveEmptyScans: topicTrulyIdleNoArchives
+                      ? (progress?.consecutiveEmptyScans ?? 0) + 1
+                      : 0,
+                  });
                 }
                 continue;
               }
@@ -725,13 +819,27 @@ export async function runWorkerForAccount(
               if (minFailedId !== null && topicWatermark !== null && topicWatermark >= minFailedId) {
                 topicWatermark = minFailedId - 1n;
               }
+              // trulyIdle: no archives this scan AND no failures AND no
+              // retryable pending. Same definition as the non-forum branch.
+              const retryableTopicPendingNow = await getRetryableSkippedMessageIds({
+                accountId: account.id,
+                sourceChannelId: channel.id,
+                topicId: topic.topicId,
+                cap: config.maxSkipAttempts,
+              });
+              const topicTrulyIdle =
+                scanResult.archives.length === 0
+                && minFailedId === null
+                && retryableTopicPendingNow.length === 0;
+              const newTopicConsecutive = topicTrulyIdle
+                ? (progress?.consecutiveEmptyScans ?? 0) + 1
+                : 0;
               if (topicWatermark !== null) {
-                await upsertTopicProgress(
-                  mapping.id,
-                  topic.topicId,
-                  topic.name,
-                  topicWatermark
-                );
+                await upsertTopicScanState(mapping.id, topic.topicId, topic.name, {
+                  lastProcessedMessageId: topicWatermark,
+                  lastScanFoundArchives: !topicTrulyIdle,
+                  consecutiveEmptyScans: newTopicConsecutive,
+                });
               }
             } catch (topicErr) {
               accountLog.warn(
@@ -741,6 +849,51 @@ export async function runWorkerForAccount(
             }
           }
         } else {
+          // ── Channel-scan-skip guard ──
+          // Before any TDLib call, decide whether this channel can be
+          // skipped entirely this cycle. Three signals (in order):
+          //   1. retryable SkippedPackages exist → MUST scan
+          //   2. lastScannedAt within window AND last scan was idle → skip
+          //   3. in backoff AND not the Nth cycle → skip
+          try {
+            const retryable = await getRetryableSkippedMessageIds({
+              accountId: account.id,
+              sourceChannelId: channel.id,
+              topicId: null,
+              cap: config.maxSkipAttempts,
+            });
+            if (retryable.length === 0 && mapping.lastScannedAt) {
+              const sinceLastScanMs = Date.now() - mapping.lastScannedAt.getTime();
+              const withinRecencyWindow = sinceLastScanMs < config.skipRecentScanWindowMs;
+              const inBackoff = mapping.consecutiveEmptyScans >= config.emptyScanBackoffThreshold;
+              const backoffSkipsThisCycle =
+                inBackoff && getCurrentCycle() % config.emptyScanBackoffEveryNth !== 0;
+
+              if (
+                (withinRecencyWindow && !mapping.lastScanFoundArchives) ||
+                backoffSkipsThisCycle
+              ) {
+                accountLog.debug(
+                  {
+                    channel: channel.title,
+                    sinceLastScanMs,
+                    consecutiveEmptyScans: mapping.consecutiveEmptyScans,
+                    reason: withinRecencyWindow ? "recent-idle" : "backoff",
+                  },
+                  "Skipping channel — recently scanned and idle, or in backoff"
+                );
+                continue;
+              }
+            }
+          } catch (skipErr) {
+            // Skip guard is best-effort. If the retryable query fails,
+            // fall through and do the normal scan.
+            accountLog.warn(
+              { err: skipErr, channel: channel.title },
+              "Skip guard failed, proceeding with scan"
+            );
+          }
+
           // ── Non-forum channel: flat scan (existing behavior) ──
           await updateRunActivity(activeRunId, {
             currentActivity: `Scanning "${channelLabel}" for new archives`,
@@ -797,6 +950,34 @@ export async function runWorkerForAccount(
             );
           }
 
+          // ── getChat short-circuit ──
+          // After the retry pass has settled the effective watermark, ask
+          // TDLib for the channel's last_message.id. If it's <= our watermark,
+          // no new content exists since last cycle — skip the paginated
+          // searchChatMessages entirely. Still update scan-state so the
+          // recent-scan skip can kick in next cycle.
+          const channelLastId = await getChannelLastMessageId(client, channel.telegramId);
+          if (
+            channelLastId !== null
+            && effectiveChannelWatermark !== null
+            && channelLastId <= effectiveChannelWatermark
+          ) {
+            accountLog.info(
+              {
+                channel: channel.title,
+                channelLastId: channelLastId.toString(),
+                watermark: effectiveChannelWatermark.toString(),
+              },
+              "Channel caught up via getChat — skipping searchChatMessages"
+            );
+            await upsertChannelScanState(mapping.id, {
+              lastProcessedMessageId: effectiveChannelWatermark,
+              lastScanFoundArchives: false,
+              consecutiveEmptyScans: (mapping.consecutiveEmptyScans ?? 0) + 1,
+            });
+            continue;
+          }
+
           const scanResult = await getChannelMessages(
             client,
             channel.telegramId,
@@ -817,10 +998,24 @@ export async function runWorkerForAccount(
 
           if (scanResult.archives.length === 0) {
             accountLog.info({ channelId: channel.id, title: channel.title, totalScanned: scanResult.totalScanned }, "No new archives in channel");
-            // Still advance watermark to highest scanned message so we don't
-            // re-scan these messages next cycle
+            // Truly idle requires no retryable SkippedPackages — a channel
+            // with a chronically-failing archive must NOT enter backoff just
+            // because no NEW archives showed up this scan.
+            const retryableNoArchives = await getRetryableSkippedMessageIds({
+              accountId: account.id,
+              sourceChannelId: channel.id,
+              topicId: null,
+              cap: config.maxSkipAttempts,
+            });
+            const channelTrulyIdleNoArchives = retryableNoArchives.length === 0;
             if (scanResult.maxScannedMessageId) {
-              await updateLastProcessedMessage(mapping.id, scanResult.maxScannedMessageId);
+              await upsertChannelScanState(mapping.id, {
+                lastProcessedMessageId: scanResult.maxScannedMessageId,
+                lastScanFoundArchives: !channelTrulyIdleNoArchives,
+                consecutiveEmptyScans: channelTrulyIdleNoArchives
+                  ? (mapping.consecutiveEmptyScans ?? 0) + 1
+                  : 0,
+              });
             }
             continue;
           }
@@ -856,8 +1051,29 @@ export async function runWorkerForAccount(
           if (minFailedId !== null && channelWatermark !== null && channelWatermark >= minFailedId) {
             channelWatermark = minFailedId - 1n;
           }
+          // trulyIdle: nothing new this scan AND nothing failed AND no
+          // retryable SkippedPackages pending. The retryable check matters —
+          // a chronically-failing archive should NEVER let the channel back
+          // off, even though zipsFound stays at 0 for it.
+          const retryablePendingNow = await getRetryableSkippedMessageIds({
+            accountId: account.id,
+            sourceChannelId: channel.id,
+            topicId: null,
+            cap: config.maxSkipAttempts,
+          });
+          const trulyIdle =
+            scanResult.archives.length === 0
+            && minFailedId === null
+            && retryablePendingNow.length === 0;
+          const newConsecutive = trulyIdle
+            ? (mapping.consecutiveEmptyScans ?? 0) + 1
+            : 0;
           if (channelWatermark !== null) {
-            await updateLastProcessedMessage(mapping.id, channelWatermark);
+            await upsertChannelScanState(mapping.id, {
+              lastProcessedMessageId: channelWatermark,
+              lastScanFoundArchives: !trulyIdle,
+              consecutiveEmptyScans: newConsecutive,
+            });
           }
         }
         } catch (channelErr) {
@@ -1681,28 +1897,56 @@ async function processOneArchiveSet(
       );
     }
 
-      // ── Pre-upload integrity test ──
-      // Catch broken/encrypted archives before we burn upload bandwidth.
+      // ── Pre-upload integrity test (advisory) ──
+      // Run unzip -t / unrar t / 7z t to look for corruption or encryption
+      // before we upload. This is ADVISORY only — failures get logged and
+      // emit a SystemNotification but never block upload, because:
       //
-      // Important nuance: ZIP multipart archives use byte-level chunk naming
-      // (`.zip.001`, `.zip.002`, ...). Individual chunks aren't valid ZIPs
-      // — the central directory only exists in the last chunk and unzip can't
-      // span the `.zip.001` naming convention. Testing the first chunk alone
-      // always fails with "no central directory found". Skip the test for
-      // those.
+      //  1. Multipart ZIPs (`.zip.001`, `.zip.002`, ...) aren't testable
+      //     chunk-by-chunk. Skip them entirely.
+      //  2. Large 7z archives can OOM-kill `7z t` (exit 137) during
+      //     decompression on memory-limited containers — that's a tool
+      //     limitation, not actual corruption.
+      //  3. p7zip can fail with unhelpful messages on newer 7z features.
       //
-      // RAR and 7z CLI tools auto-discover sibling parts when pointed at the
-      // first part, so `unrar t` / `7z t` work for multipart RAR/7z.
-      //
-      // Single-file archives (regardless of whether WE re-split them for
-      // upload size limits) are always testable on the original tempPaths[0]
-      // since that's the unsplit downloaded file.
+      // Hash verification + archive metadata parse already cover byte-level
+      // corruption and structural readability. The integrity test is a
+      // nice-to-have stronger signal; not worth losing uploads over false
+      // positives.
       const archType = archiveSet.type === "7Z" ? "SEVEN_Z" : archiveSet.type;
       const isMultipartZip = archType === "ZIP" && tempPaths.length > 1;
       if (!isMultipartZip) {
         const integrity = await testArchiveIntegrity(archType, tempPaths[0]);
         if (!integrity.ok) {
-          throw new Error(`Archive integrity check failed: ${integrity.reason}`);
+          // Detect encryption specifically — those won't extract for users
+          // even if we upload them. Surface clearly via notification but
+          // STILL proceed: the user can audit and decide what to do.
+          const isEncrypted = /encrypted/i.test(integrity.reason);
+          accountLog.warn(
+            { fileName: archiveName, reason: integrity.reason.slice(0, 200), isEncrypted },
+            "Archive integrity test failed — proceeding with upload anyway (advisory check)"
+          );
+          try {
+            await db.systemNotification.create({
+              data: {
+                type: isEncrypted ? "UPLOAD_FAILED" : "HASH_MISMATCH",
+                severity: "WARNING",
+                title: isEncrypted
+                  ? `Archive may be encrypted: ${archiveName}`
+                  : `Integrity test reported issues: ${archiveName}`,
+                message: integrity.reason.slice(0, 1000),
+                context: {
+                  fileName: archiveName,
+                  sourceChannelId: channel.id,
+                  sourceMessageId: Number(archiveSet.parts[0].id),
+                  archiveType: archType,
+                  advisory: true,
+                },
+              },
+            });
+          } catch {
+            // Best-effort notification
+          }
         }
       }