feat(telegram): add "skip & disable topic" on the live worker panel

Track the forum topic currently being processed on IngestionRun
(currentTopicId + currentAccountChannelMapId; additive migration) and
expose it on the live status. processArchiveSets gains an optional
shouldStop callback polled before each archive set; the forum branch
passes a live isTopicFetchEnabled check, so disabling a topic mid-run
lets the in-flight file finish, then skips the rest of that topic.

A new disableActiveTopic server action sets the topic's fetchEnabled
false (upsert), and the worker status panel shows a "Skip & disable
this topic" button while a topic is being processed. Future runs skip
the topic via the existing live per-topic read.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

worker/src/db/queries.ts

@@ -369,6 +369,8 @@ export interface ActivityUpdate {
   downloadedBytes?: bigint | null;
   totalBytes?: bigint | null;
   downloadPercent?: number | null;
+  currentTopicId?: bigint | null;
+  currentAccountChannelMapId?: string | null;
   messagesScanned?: number;
   zipsFound?: number;
   zipsDuplicate?: number;
@@ -396,6 +398,10 @@ export async function updateRunActivity(
       ...(activity.zipsFound !== undefined && { zipsFound: activity.zipsFound }),
       ...(activity.zipsDuplicate !== undefined && { zipsDuplicate: activity.zipsDuplicate }),
       ...(activity.zipsIngested !== undefined && { zipsIngested: activity.zipsIngested }),
+      ...(activity.currentTopicId !== undefined && { currentTopicId: activity.currentTopicId }),
+      ...(activity.currentAccountChannelMapId !== undefined && {
+        currentAccountChannelMapId: activity.currentAccountChannelMapId,
+      }),
     },
   });
 }
@@ -410,6 +416,8 @@ const CLEAR_ACTIVITY = {
   downloadedBytes: null,
   totalBytes: null,
   downloadPercent: null,
+  currentTopicId: null,
+  currentAccountChannelMapId: null,
   lastActivityAt: new Date(),
 };
 

worker/src/worker.ts

@@ -538,6 +538,8 @@ export async function runWorkerForAccount(
             currentActivity: `Enumerating topics in "${channelLabel}"`,
             currentStep: "scanning",
             currentChannel: channelLabel,
+            currentTopicId: null,
+            currentAccountChannelMapId: null,
             currentFile: null,
             currentFileNum: null,
             totalFiles: null,
@@ -750,6 +752,8 @@ export async function runWorkerForAccount(
                 currentActivity: `Scanning "${topicLabel}"${topicProgress}`,
                 currentStep: "scanning",
                 currentChannel: channelLabel,
+                currentTopicId: topic.topicId,
+                currentAccountChannelMapId: mapping.id,
                 currentFile: null,
                 currentFileNum: null,
                 totalFiles: null,
@@ -830,7 +834,11 @@ export async function runWorkerForAccount(
                     topic.name,
                     messageId
                   );
-                }
+                },
+                // shouldStop: re-read the live fetch flag before each archive
+                // set. A mid-run "disable topic" lets the current file finish,
+                // then skips the rest of this topic's archives.
+                async () => !(await isTopicFetchEnabled(mapping.id, topic.topicId))
               );
               // Sync client back in case it was recreated during upload stall recovery
               client = pipelineCtx.client;
@@ -922,6 +930,8 @@ export async function runWorkerForAccount(
             currentActivity: `Scanning "${channelLabel}" for new archives`,
             currentStep: "scanning",
             currentChannel: channelLabel,
+            currentTopicId: null,
+            currentAccountChannelMapId: null,
             currentFile: null,
             currentFileNum: null,
             totalFiles: null,
@@ -1203,7 +1213,12 @@ async function processArchiveSets(
    *  below any failed message ID in this scan). Used by the caller to
    *  advance the channel/topic watermark incrementally — otherwise a long
    *  scan that gets killed by worker restart loses all progress. */
-  onWatermarkAdvance?: (messageId: bigint) => Promise<void>
+  onWatermarkAdvance?: (messageId: bigint) => Promise<void>,
+  /** Optional cancellation check, polled before each archive set. When it
+   *  resolves true, processing stops after the set currently in flight (that
+   *  one completes; remaining sets in this scan are skipped). Used by the
+   *  forum branch to honour a mid-run "disable topic". */
+  shouldStop?: () => Promise<boolean>
 ): Promise<{ maxProcessedId: bigint | null; minFailedId: bigint | null }> {
   const { client, runId, channelTitle, channel, throttled, counters, accountLog } = ctx;
 
@@ -1281,6 +1296,16 @@ async function processArchiveSets(
   const indexedPackageRefs: IndexedPackageRef[] = [];
 
   for (let setIdx = 0; setIdx < archiveSets.length; setIdx++) {
+    // Cooperative cancellation: if the caller signals stop (e.g. the topic was
+    // disabled mid-run), skip the remaining archive sets in this scan. The set
+    // processed in the previous iteration has already completed.
+    if (shouldStop && (await shouldStop())) {
+      accountLog.info(
+        { channel: channelTitle, processed: setIdx, total: archiveSets.length },
+        "Stop signal received (topic disabled) — skipping remaining archive sets in this scan"
+      );
+      break;
+    }
     try {
       const packageId = await processOneArchiveSet(
         ctx,