harden(telegram): reject non-positive topic id in disableActiveTopic

BigInt("") / BigInt("0") don't throw; guard against them since the
action is exported and admin-callable independent of the UI button.

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

prisma/migrations/20260622120000_add_run_current_topic/migration.sql

@@ -0,0 +1,6 @@
+-- AlterTable: track the forum topic currently being processed on the live run
+-- so the worker status panel can offer a "skip & disable topic" action.
+-- Additive, nullable — no data change for existing rows.
+ALTER TABLE "ingestion_runs"
+  ADD COLUMN "currentTopicId" BIGINT,
+  ADD COLUMN "currentAccountChannelMapId" TEXT;

prisma/schema.prisma

@@ -582,6 +582,8 @@ model IngestionRun {
   totalBytes      BigInt?         // Total size of current download
   downloadPercent Int?            // 0-100
   lastActivityAt  DateTime?       // When activity was last updated
+  currentTopicId             BigInt?  // Forum topic currently being processed (live "skip topic")
+  currentAccountChannelMapId String?  // AccountChannelMap owning the topic being processed
 
   account  TelegramAccount @relation(fields: [accountId], references: [id])
   packages Package[]

src/app/(app)/telegram/_components/worker-status-panel.tsx

@@ -9,13 +9,14 @@ import {
   Radio,
   AlertTriangle,
   RefreshCw,
+  SkipForward,
 } from "lucide-react";
 import { Card, CardContent } from "@/components/ui/card";
 import { Badge } from "@/components/ui/badge";
 import { Button } from "@/components/ui/button";
 import { cn } from "@/lib/utils";
 import { toast } from "sonner";
-import { triggerIngestion } from "../actions";
+import { triggerIngestion, disableActiveTopic } from "../actions";
 import type { IngestionAccountStatus } from "@/lib/telegram/types";
 
 interface WorkerStatusPanelProps {
@@ -218,6 +219,24 @@ function RunningStatus({
 }: {
   run: NonNullable<IngestionAccountStatus["currentRun"]>;
 }) {
+  const [isDisabling, startDisable] = useTransition();
+
+  const handleSkipTopic = () => {
+    const acmId = run.currentAccountChannelMapId;
+    const topicId = run.currentTopicId;
+    if (!acmId || !topicId) return;
+    startDisable(async () => {
+      const result = await disableActiveTopic(acmId, topicId);
+      if (result.success) {
+        toast.success(
+          "Topic disabled — the current file finishes, then the rest is skipped"
+        );
+      } else {
+        toast.error(result.error ?? "Failed to disable topic");
+      }
+    });
+  };
+
   return (
     <div className="space-y-2">
       <div className="flex items-center gap-2">
@@ -273,6 +292,27 @@ function RunningStatus({
           </span>
         )}
       </div>
+
+      {/* Skip & disable the topic currently being processed */}
+      {run.currentTopicId && run.currentAccountChannelMapId && (
+        <div className="pl-6 pt-1">
+          <Button
+            variant="outline"
+            size="sm"
+            className="h-7 gap-1.5 text-xs"
+            onClick={handleSkipTopic}
+            disabled={isDisabling}
+            title="Finish the current file, skip the rest of this topic, and don't fetch it in future runs"
+          >
+            {isDisabling ? (
+              <Loader2 className="h-3 w-3 animate-spin" />
+            ) : (
+              <SkipForward className="h-3 w-3" />
+            )}
+            Skip &amp; disable this topic
+          </Button>
+        </div>
+      )}
     </div>
   );
 }

src/app/(app)/telegram/actions.ts

@@ -462,6 +462,54 @@ export async function setTopicFetchEnabled(
   }
 }
 
+/**
+ * Disable the topic currently being processed by the worker, identified by its
+ * account-channel mapping + Telegram topic id (as exposed on the live run
+ * status). Upserts so a disabled row exists even if the worker hasn't persisted
+ * the topic yet. The worker honours this live: it finishes the in-flight file
+ * then skips the rest of the topic, and future runs skip it entirely.
+ */
+export async function disableActiveTopic(
+  accountChannelMapId: string,
+  topicId: string
+): Promise<ActionResult> {
+  const admin = await requireAdmin();
+  if (!admin.success) return admin;
+
+  let topicIdBig: bigint;
+  try {
+    topicIdBig = BigInt(topicId);
+  } catch {
+    return { success: false, error: "Invalid topic id" };
+  }
+  // BigInt("") / BigInt("0") don't throw — reject non-positive ids explicitly
+  // since this action is exported and admin-callable outside the UI button.
+  if (topicIdBig <= BigInt(0)) {
+    return { success: false, error: "Invalid topic id" };
+  }
+
+  try {
+    await prisma.topicProgress.upsert({
+      where: {
+        accountChannelMapId_topicId: {
+          accountChannelMapId,
+          topicId: topicIdBig,
+        },
+      },
+      create: {
+        accountChannelMapId,
+        topicId: topicIdBig,
+        fetchEnabled: false,
+      },
+      update: { fetchEnabled: false },
+    });
+    revalidatePath(REVALIDATE_PATH);
+    return { success: true, data: undefined };
+  } catch {
+    return { success: false, error: "Failed to disable topic" };
+  }
+}
+
 // ── Account-Channel link actions ──
 
 export async function linkChannel(

src/lib/telegram/queries.ts

@@ -550,6 +550,8 @@ export async function getIngestionStatus(): Promise<IngestionAccountStatus[]> {
             totalBytes: currentRun.totalBytes?.toString() ?? null,
             downloadPercent: currentRun.downloadPercent,
             lastActivityAt: currentRun.lastActivityAt?.toISOString() ?? null,
+            currentTopicId: currentRun.currentTopicId?.toString() ?? null,
+            currentAccountChannelMapId: currentRun.currentAccountChannelMapId,
           }
         : null,
     });

src/lib/telegram/types.ts

@@ -123,5 +123,7 @@ export interface IngestionAccountStatus {
     totalBytes: string | null; // BigInt serialized as string
     downloadPercent: number | null;
     lastActivityAt: string | null;
+    currentTopicId: string | null; // BigInt serialized as string
+    currentAccountChannelMapId: string | null;
   } | null;
 }

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,