Fix multiple issues

bot/src/send-listener.ts

@@ -12,29 +12,78 @@ import { copyMessageToUser, sendTextMessage, sendPhotoMessage } from "./tdlib/cl
 const log = childLogger("send-listener");
 
 let pgClient: pg.PoolClient | null = null;
+let stopped = false;
+
+/** Delay (ms) before attempting to reconnect after a connection loss. */
+const RECONNECT_DELAY_MS = 5_000;
 
 /**
  * Start listening for pg_notify signals:
  *   - `bot_send` — payload = requestId → send a package to a user
  *   - `new_package` — payload = JSON { packageId, fileName, creator } → notify subscribers
+ *
+ * If the underlying connection is lost, the listener automatically reconnects
+ * so that pg_notify signals are never silently dropped.
  */
 export async function startSendListener(): Promise<void> {
-  pgClient = await pool.connect();
+  stopped = false;
-  await pgClient.query("LISTEN bot_send");
+  await connectListener();
-  await pgClient.query("LISTEN new_package");
+}
 
-  pgClient.on("notification", (msg) => {
+async function connectListener(): Promise<void> {
-    if (msg.channel === "bot_send" && msg.payload) {
+  try {
-      handleBotSend(msg.payload);
+    pgClient = await pool.connect();
-    } else if (msg.channel === "new_package" && msg.payload) {
+    await pgClient.query("LISTEN bot_send");
-      handleNewPackage(msg.payload);
+    await pgClient.query("LISTEN new_package");
+
+    pgClient.on("notification", (msg) => {
+      if (msg.channel === "bot_send" && msg.payload) {
+        handleBotSend(msg.payload);
+      } else if (msg.channel === "new_package" && msg.payload) {
+        handleNewPackage(msg.payload);
+      }
+    });
+
+    // Reconnect automatically when the connection ends unexpectedly
+    pgClient.on("end", () => {
+      if (!stopped) {
+        log.warn("Send listener connection lost — reconnecting");
+        pgClient = null;
+        scheduleReconnect();
+      }
+    });
+
+    pgClient.on("error", (err) => {
+      log.error({ err }, "Send listener connection error");
+      if (!stopped && pgClient) {
+        try {
+          pgClient.release(true);
+        } catch (releaseErr) {
+          log.debug({ err: releaseErr }, "Failed to release pg client after error");
+        }
+        pgClient = null;
+        scheduleReconnect();
+      }
+    });
+
+    log.info("Send listener started (bot_send, new_package)");
+  } catch (err) {
+    log.error({ err }, "Failed to start send listener — retrying");
+    scheduleReconnect();
+  }
+}
+
+function scheduleReconnect(): void {
+  if (stopped) return;
+  setTimeout(() => {
+    if (!stopped) {
+      connectListener();
     }
-  });
+  }, RECONNECT_DELAY_MS);
-
-  log.info("Send listener started (bot_send, new_package)");
 }
 
 export function stopSendListener(): void {
+  stopped = true;
   if (pgClient) {
     pgClient.release();
     pgClient = null;

bot/src/tdlib/client.ts

@@ -33,7 +33,7 @@ export async function createBotClient(): Promise<tdl.Client> {
 
   await client.login(() => ({
     type: "bot",
-    token: config.botToken,
+    getToken: () => Promise.resolve(config.botToken),
   }));
 
   log.info("Bot client authenticated successfully");
@@ -54,7 +54,10 @@ export async function closeBotClient(): Promise<void> {
 
 /**
  * Forward a message from a channel to a user's DM.
- * Uses copyMessage to make it appear as sent by the bot.
+ * Uses forwardMessages with send_copy to make it appear as sent by the bot.
+ *
+ * The fromChatId is the TDLib chat ID stored in the DB — already in the correct
+ * format (negative for supergroups/channels, e.g. -1001234567890).
  */
 export async function copyMessageToUser(
   fromChatId: bigint,
@@ -63,14 +66,10 @@ export async function copyMessageToUser(
 ): Promise<void> {
   if (!client) throw new Error("Bot client not initialized");
 
-  // TDLib uses negative chat IDs for channels/supergroups
-  // The telegramId from the DB is the raw Telegram ID; for channels it needs -100 prefix
-  const fromChatIdNum = Number(-100n * 1n) + Number(fromChatId);
-
   await client.invoke({
     _: "forwardMessages",
     chat_id: Number(toUserId),
-    from_chat_id: Number(fromChatId) > 0 ? -Number(fromChatId) : Number(fromChatId),
+    from_chat_id: Number(fromChatId),
     message_ids: [Number(messageId)],
     send_copy: true,
     remove_caption: false,

src/app/(app)/telegram/_components/channels-tab.tsx

@@ -94,7 +94,7 @@ export function ChannelsTab({ channels, globalDestination, accounts }: ChannelsT
 
   return (
     <div className="space-y-4">
-      <DestinationCard destination={globalDestination} />
+      <DestinationCard destination={globalDestination} channels={channels} />
 
       <div className="flex items-center gap-2">
         <Button

src/app/(app)/telegram/_components/destination-card.tsx

@@ -1,9 +1,9 @@
 "use client";
 
 import { useState, useEffect, useTransition } from "react";
-import { Database, AlertTriangle, Link2, Plus, Loader2 } from "lucide-react";
+import { Database, AlertTriangle, Link2, Plus, Loader2, ArrowRight } from "lucide-react";
 import { toast } from "sonner";
-import { createDestinationViaWorker } from "../actions";
+import { createDestinationViaWorker, setGlobalDestination } from "../actions";
 import { Card, CardContent } from "@/components/ui/card";
 import { Badge } from "@/components/ui/badge";
 import { Button } from "@/components/ui/button";
@@ -17,10 +17,19 @@ import {
   DialogTitle,
   DialogFooter,
 } from "@/components/ui/dialog";
-import type { GlobalDestination } from "@/lib/telegram/admin-queries";
+import {
+  Select,
+  SelectContent,
+  SelectItem,
+  SelectTrigger,
+  SelectValue,
+} from "@/components/ui/select";
+import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs";
+import type { GlobalDestination, ChannelRow } from "@/lib/telegram/admin-queries";
 
 interface DestinationCardProps {
   destination: GlobalDestination;
+  channels?: ChannelRow[];
 }
 
 type CreateState =
@@ -29,11 +38,17 @@ type CreateState =
   | { phase: "done"; title: string; telegramId: string }
   | { phase: "error"; message: string };
 
-export function DestinationCard({ destination }: DestinationCardProps) {
+export function DestinationCard({ destination, channels = [] }: DestinationCardProps) {
   const [isPending, startTransition] = useTransition();
   const [createOpen, setCreateOpen] = useState(false);
   const [title, setTitle] = useState("dragonsstash db");
   const [createState, setCreateState] = useState<CreateState>({ phase: "idle" });
+  const [selectedChannelId, setSelectedChannelId] = useState<string>("");
+
+  // Channels that can be assigned as destination (SOURCE channels only, exclude current destination)
+  const assignableChannels = channels.filter(
+    (c) => c.type === "SOURCE" && c.id !== destination?.id
+  );
 
   // Poll for worker result when creating
   useEffect(() => {
@@ -103,6 +118,21 @@ export function DestinationCard({ destination }: DestinationCardProps) {
     });
   };
 
+  const handleAssignExisting = () => {
+    if (!selectedChannelId) return;
+
+    startTransition(async () => {
+      const result = await setGlobalDestination(selectedChannelId);
+      if (result.success) {
+        toast.success("Channel set as destination!");
+        setCreateOpen(false);
+        setSelectedChannelId("");
+      } else {
+        toast.error(result.error ?? "Failed to set destination");
+      }
+    });
+  };
+
   const handleOpenChange = (open: boolean) => {
     setCreateOpen(open);
     if (!open) {
@@ -110,6 +140,7 @@ export function DestinationCard({ destination }: DestinationCardProps) {
       if (createState.phase !== "creating") {
         setCreateState({ phase: "idle" });
       }
+      setSelectedChannelId("");
     }
   };
 
@@ -132,19 +163,23 @@ export function DestinationCard({ destination }: DestinationCardProps) {
             </div>
             <Button size="sm" onClick={() => setCreateOpen(true)}>
               <Plus className="mr-2 h-3.5 w-3.5" />
-              Create Destination
+              Set Destination
             </Button>
           </CardContent>
         </Card>
 
-        <CreateDestinationDialog
+        <DestinationDialog
           open={createOpen}
           onOpenChange={handleOpenChange}
           title={title}
           setTitle={setTitle}
-          onSubmit={handleCreate}
+          onSubmitCreate={handleCreate}
           createState={createState}
           isPending={isPending}
+          assignableChannels={assignableChannels}
+          selectedChannelId={selectedChannelId}
+          setSelectedChannelId={setSelectedChannelId}
+          onSubmitAssign={handleAssignExisting}
         />
       </>
     );
@@ -187,46 +222,59 @@ export function DestinationCard({ destination }: DestinationCardProps) {
         </CardContent>
       </Card>
 
-      <CreateDestinationDialog
+      <DestinationDialog
         open={createOpen}
         onOpenChange={handleOpenChange}
         title={title}
         setTitle={setTitle}
-        onSubmit={handleCreate}
+        onSubmitCreate={handleCreate}
         createState={createState}
         isPending={isPending}
+        assignableChannels={assignableChannels}
+        selectedChannelId={selectedChannelId}
+        setSelectedChannelId={setSelectedChannelId}
+        onSubmitAssign={handleAssignExisting}
       />
     </>
   );
 }
 
-function CreateDestinationDialog({
+function DestinationDialog({
   open,
   onOpenChange,
   title,
   setTitle,
-  onSubmit,
+  onSubmitCreate,
   createState,
   isPending,
+  assignableChannels,
+  selectedChannelId,
+  setSelectedChannelId,
+  onSubmitAssign,
 }: {
   open: boolean;
   onOpenChange: (open: boolean) => void;
   title: string;
   setTitle: (v: string) => void;
-  onSubmit: () => void;
+  onSubmitCreate: () => void;
   createState: CreateState;
   isPending: boolean;
+  assignableChannels: ChannelRow[];
+  selectedChannelId: string;
+  setSelectedChannelId: (v: string) => void;
+  onSubmitAssign: () => void;
 }) {
   const isCreating = createState.phase === "creating";
+  const hasAssignable = assignableChannels.length > 0;
 
   return (
     <Dialog open={open} onOpenChange={onOpenChange}>
       <DialogContent className="sm:max-w-md">
         <DialogHeader>
-          <DialogTitle>Create Destination Channel</DialogTitle>
+          <DialogTitle>Set Destination Channel</DialogTitle>
           <DialogDescription>
-            A private Telegram group will be created automatically using one of
+            Choose an existing channel or create a new private group. All
-            your authenticated accounts. All accounts will write archives here.
+            accounts will write archives to this destination.
           </DialogDescription>
         </DialogHeader>
 
@@ -241,46 +289,111 @@ function CreateDestinationDialog({
             </p>
           </div>
         ) : (
-          <div className="space-y-4">
+          <Tabs defaultValue={hasAssignable ? "existing" : "create"} className="w-full">
-            {createState.phase === "error" && (
+            <TabsList className="grid w-full grid-cols-2">
-              <div className="rounded-md border border-destructive/50 bg-destructive/10 p-3">
+              <TabsTrigger value="existing" disabled={!hasAssignable}>
-                <p className="text-sm text-destructive">{createState.message}</p>
+                <ArrowRight className="mr-1.5 h-3.5 w-3.5" />
+                Use Existing
+              </TabsTrigger>
+              <TabsTrigger value="create">
+                <Plus className="mr-1.5 h-3.5 w-3.5" />
+                Create New
+              </TabsTrigger>
+            </TabsList>
+
+            <TabsContent value="existing" className="space-y-4 pt-2">
+              {createState.phase === "error" && (
+                <div className="rounded-md border border-destructive/50 bg-destructive/10 p-3">
+                  <p className="text-sm text-destructive">{createState.message}</p>
+                </div>
+              )}
+
+              <div className="space-y-2">
+                <Label>Select Channel</Label>
+                <Select
+                  value={selectedChannelId}
+                  onValueChange={setSelectedChannelId}
+                >
+                  <SelectTrigger>
+                    <SelectValue placeholder="Pick a channel..." />
+                  </SelectTrigger>
+                  <SelectContent>
+                    {assignableChannels.map((ch) => (
+                      <SelectItem key={ch.id} value={ch.id}>
+                        {ch.title}{" "}
+                        <span className="text-muted-foreground text-xs">
+                          ({ch.telegramId})
+                        </span>
+                      </SelectItem>
+                    ))}
+                  </SelectContent>
+                </Select>
+                <p className="text-xs text-muted-foreground">
+                  The selected channel will become the destination. All accounts
+                  will be linked as writers automatically.
+                </p>
               </div>
-            )}
 
-            <div className="space-y-2">
+              <DialogFooter>
-              <Label htmlFor="dest-title">Group Name</Label>
+                <Button
-              <Input
+                  variant="outline"
-                id="dest-title"
+                  onClick={() => onOpenChange(false)}
-                placeholder="e.g. dragonsstash db"
+                >
-                value={title}
+                  Cancel
-                onChange={(e) => setTitle(e.target.value)}
+                </Button>
-              />
+                <Button
-              <p className="text-xs text-muted-foreground">
+                  onClick={onSubmitAssign}
-                This will be the name of the Telegram group. You can rename it later in Telegram.
+                  disabled={isPending || !selectedChannelId}
-              </p>
+                >
-            </div>
+                  {isPending && (
-          </div>
+                    <Loader2 className="mr-2 h-4 w-4 animate-spin" />
+                  )}
+                  Set as Destination
+                </Button>
+              </DialogFooter>
+            </TabsContent>
+
+            <TabsContent value="create" className="space-y-4 pt-2">
+              {createState.phase === "error" && (
+                <div className="rounded-md border border-destructive/50 bg-destructive/10 p-3">
+                  <p className="text-sm text-destructive">{createState.message}</p>
+                </div>
+              )}
+
+              <div className="space-y-2">
+                <Label htmlFor="dest-title">Group Name</Label>
+                <Input
+                  id="dest-title"
+                  placeholder="e.g. dragonsstash db"
+                  value={title}
+                  onChange={(e) => setTitle(e.target.value)}
+                />
+                <p className="text-xs text-muted-foreground">
+                  A new private Telegram group will be created using one of your
+                  authenticated accounts. You can rename it later in Telegram.
+                </p>
+              </div>
+
+              <DialogFooter>
+                <Button
+                  variant="outline"
+                  onClick={() => onOpenChange(false)}
+                >
+                  Cancel
+                </Button>
+                <Button
+                  onClick={onSubmitCreate}
+                  disabled={isPending || !title.trim()}
+                >
+                  {isPending && (
+                    <Loader2 className="mr-2 h-4 w-4 animate-spin" />
+                  )}
+                  Create Group
+                </Button>
+              </DialogFooter>
+            </TabsContent>
+          </Tabs>
         )}
-
-        <DialogFooter>
-          <Button
-            variant="outline"
-            onClick={() => onOpenChange(false)}
-            disabled={isCreating}
-          >
-            Cancel
-          </Button>
-          <Button
-            onClick={onSubmit}
-            disabled={isPending || isCreating || !title.trim()}
-          >
-            {(isPending || isCreating) && (
-              <Loader2 className="mr-2 h-4 w-4 animate-spin" />
-            )}
-            Create Group
-          </Button>
-        </DialogFooter>
       </DialogContent>
     </Dialog>
   );

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

@@ -270,6 +270,13 @@ export async function setChannelType(
   if (!existing) return { success: false, error: "Channel not found" };
 
   try {
+    if (type === "DESTINATION") {
+      // Setting as destination: use the full global destination logic
+      // so it updates the global settings key, creates WRITER links, etc.
+      return await setGlobalDestination(id);
+    }
+
+    // Setting as SOURCE — just change the type
     await prisma.telegramChannel.update({
       where: { id },
       data: { type },

worker/src/db/locks.ts

@@ -1,8 +1,16 @@
+import type pg from "pg";
 import { pool } from "./client.js";
 import { childLogger } from "../util/logger.js";
 
 const log = childLogger("locks");
 
+/**
+ * Holds the pooled connection for each active advisory lock.
+ * Session-level advisory locks are tied to the specific PostgreSQL connection,
+ * so we MUST keep the same connection checked out for the entire lock duration.
+ */
+const heldConnections = new Map<string, pg.PoolClient>();
+
 /**
  * Derive a stable 32-bit integer lock ID from an account ID string.
  * PostgreSQL advisory locks use bigint, but we use 32-bit for safety.
@@ -20,6 +28,9 @@ function hashToLockId(accountId: string): number {
 /**
  * Try to acquire a PostgreSQL advisory lock for an account.
  * Returns true if acquired, false if already held by another session.
+ *
+ * IMPORTANT: The pooled connection is kept checked out for the duration
+ * of the lock. You MUST call releaseLock() when done to return it to the pool.
  */
 export async function tryAcquireLock(accountId: string): Promise<boolean> {
   const lockId = hashToLockId(accountId);
@@ -31,26 +42,40 @@ export async function tryAcquireLock(accountId: string): Promise<boolean> {
     );
     const acquired = result.rows[0]?.pg_try_advisory_lock ?? false;
     if (acquired) {
+      // Keep the connection checked out — lock is tied to this connection
+      heldConnections.set(accountId, client);
       log.debug({ accountId, lockId }, "Advisory lock acquired");
+      return true;
     } else {
+      // Lock not acquired — release the connection back to the pool
+      client.release();
       log.debug({ accountId, lockId }, "Advisory lock already held");
+      return false;
     }
-    return acquired;
+  } catch (err) {
-  } finally {
     client.release();
+    throw err;
   }
 }
 
 /**
  * Release the advisory lock for an account.
+ * Uses the SAME connection that acquired the lock, then returns it to the pool.
  */
 export async function releaseLock(accountId: string): Promise<void> {
   const lockId = hashToLockId(accountId);
-  const client = await pool.connect();
+  const client = heldConnections.get(accountId);
+
+  if (!client) {
+    log.warn({ accountId, lockId }, "No held connection for lock release — lock may have already been released");
+    return;
+  }
+
   try {
     await client.query("SELECT pg_advisory_unlock($1)", [lockId]);
     log.debug({ accountId, lockId }, "Advisory lock released");
   } finally {
+    heldConnections.delete(accountId);
     client.release();
   }
 }

worker/src/tdlib/chats.ts

@@ -1,6 +1,7 @@
 import type { Client } from "tdl";
 import { childLogger } from "../util/logger.js";
 import { config } from "../util/config.js";
+import { withFloodWait } from "../util/retry.js";
 
 const log = childLogger("chats");
 
@@ -29,11 +30,14 @@ export async function getAccountChats(
 
   while (hasMore) {
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    const result = (await client.invoke({
+    const result = (await withFloodWait(
-      _: "getChats",
+      () => client.invoke({
-      chat_list: { _: "chatListMain" },
+        _: "getChats",
-      limit: 100,
+        chat_list: { _: "chatListMain" },
-    })) as { chat_ids: number[] };
+        limit: 100,
+      }),
+      "getChats"
+    )) as { chat_ids: number[] };
 
     if (!result.chat_ids || result.chat_ids.length === 0) {
       break;
@@ -42,10 +46,13 @@ export async function getAccountChats(
     for (const chatId of result.chat_ids) {
       try {
         // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        const chat = (await client.invoke({
+        const chat = (await withFloodWait(
-          _: "getChat",
+          () => client.invoke({
-          chat_id: chatId,
+            _: "getChat",
-        })) as any;
+            chat_id: chatId,
+          }),
+          "getChat"
+        )) as any;
 
         const chatType = chat.type?._;
         let type: TelegramChatInfo["type"] = "other";
@@ -55,10 +62,13 @@ export async function getAccountChats(
           // Get supergroup details to check if it's a channel or group
           try {
             // eslint-disable-next-line @typescript-eslint/no-explicit-any
-            const sg = (await client.invoke({
+            const sg = (await withFloodWait(
-              _: "getSupergroup",
+              () => client.invoke({
-              supergroup_id: chat.type.supergroup_id,
+                _: "getSupergroup",
-            })) as any;
+                supergroup_id: chat.type.supergroup_id,
+              }),
+              "getSupergroup"
+            )) as any;
 
             type = sg.is_channel ? "channel" : "supergroup";
             isForum = sg.is_forum ?? false;
@@ -109,12 +119,15 @@ export async function generateInviteLink(
   chatId: bigint
 ): Promise<string> {
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  const result = (await client.invoke({
+  const result = (await withFloodWait(
-    _: "createChatInviteLink",
+    () => client.invoke({
-    chat_id: Number(chatId),
+      _: "createChatInviteLink",
-    name: "DragonsStash Auto-Join",
+      chat_id: Number(chatId),
-    creates_join_request: false,
+      name: "DragonsStash Auto-Join",
-  })) as any;
+      creates_join_request: false,
+    }),
+    "createChatInviteLink"
+  )) as any;
 
   const link = result.invite_link as string;
   log.info({ chatId: chatId.toString(), link }, "Generated invite link");
@@ -130,13 +143,16 @@ export async function createSupergroup(
   title: string
 ): Promise<{ chatId: bigint; title: string }> {
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  const result = (await client.invoke({
+  const result = (await withFloodWait(
-    _: "createNewSupergroupChat",
+    () => client.invoke({
-    title,
+      _: "createNewSupergroupChat",
-    is_forum: false,
+      title,
-    is_channel: false,
+      is_forum: false,
-    description: "DragonsStash archive destination — all accounts write here",
+      is_channel: false,
-  })) as any;
+      description: "DragonsStash archive destination — all accounts write here",
+    }),
+    "createNewSupergroupChat"
+  )) as any;
 
   const chatId = BigInt(result.id);
   log.info({ chatId: chatId.toString(), title }, "Created new supergroup");
@@ -150,10 +166,13 @@ export async function joinChatByInviteLink(
   client: Client,
   inviteLink: string
 ): Promise<void> {
-  await client.invoke({
+  await withFloodWait(
-    _: "joinChatByInviteLink",
+    () => client.invoke({
-    invite_link: inviteLink,
+      _: "joinChatByInviteLink",
-  });
+      invite_link: inviteLink,
+    }),
+    "joinChatByInviteLink"
+  );
   log.info({ inviteLink }, "Joined chat by invite link");
 }
 

worker/src/tdlib/download.ts

@@ -2,6 +2,7 @@ import type { Client } from "tdl";
 import { readFile, rename, copyFile, unlink, stat } from "fs/promises";
 import { config } from "../util/config.js";
 import { childLogger } from "../util/logger.js";
+import { withFloodWait } from "../util/retry.js";
 import { isArchiveAttachment } from "../archive/detect.js";
 import type { TelegramMessage } from "../archive/multipart.js";
 import type { TelegramPhoto } from "../preview/match.js";
@@ -78,8 +79,12 @@ export interface ChannelScanResult {
 export type ScanProgressCallback = (messagesScanned: number) => void;
 
 /**
- * Invoke a TDLib method with a timeout to prevent indefinite hangs.
+ * Invoke a TDLib method with a timeout to prevent indefinite hangs,
+ * and automatic retry on FLOOD_WAIT rate-limit errors.
+ *
  * If TDLib does not respond within the timeout, the promise rejects.
+ * If Telegram returns a rate limit error, sleeps for the required
+ * duration and retries (up to maxRetries times).
  */
 export async function invokeWithTimeout<T>(
   client: Client,
@@ -87,32 +92,40 @@ export async function invokeWithTimeout<T>(
   request: Record<string, any>,
   timeoutMs = INVOKE_TIMEOUT_MS
 ): Promise<T> {
-  return new Promise<T>((resolve, reject) => {
+  return withFloodWait(
-    let settled = false;
+    () =>
+      new Promise<T>((resolve, reject) => {
+        let settled = false;
 
-    const timer = setTimeout(() => {
+        const timer = setTimeout(() => {
-      if (!settled) {
+          if (!settled) {
-        settled = true;
+            settled = true;
-        reject(new Error(`TDLib invoke timed out after ${timeoutMs}ms for ${request._}`));
+            reject(
-      }
+              new Error(
-    }, timeoutMs);
+                `TDLib invoke timed out after ${timeoutMs}ms for ${request._}`
+              )
+            );
+          }
+        }, timeoutMs);
 
-    (client.invoke(request) as Promise<T>)
+        (client.invoke(request) as Promise<T>)
-      .then((result) => {
+          .then((result) => {
-        if (!settled) {
+            if (!settled) {
-          settled = true;
+              settled = true;
-          clearTimeout(timer);
+              clearTimeout(timer);
-          resolve(result);
+              resolve(result);
-        }
+            }
-      })
+          })
-      .catch((err) => {
+          .catch((err) => {
-        if (!settled) {
+            if (!settled) {
-          settled = true;
+              settled = true;
-          clearTimeout(timer);
+              clearTimeout(timer);
-          reject(err);
+              reject(err);
-        }
+            }
-      });
+          });
-  });
+      }),
+    `TDLib:${request._}`
+  );
 }
 
 /**
@@ -415,15 +428,20 @@ export async function downloadFile(
     client.on("update", handleUpdate);
 
     // Start async download (non-blocking — progress via updateFile events)
-    client
+    // Wrapped in withFloodWait: if the initial invoke is rate-limited,
-      .invoke({
+    // it will sleep and retry before the download event loop begins.
-        _: "downloadFile",
+    withFloodWait(
-        file_id: numericId,
+      () =>
-        priority: 32,
+        client.invoke({
-        offset: 0,
+          _: "downloadFile",
-        limit: 0,
+          file_id: numericId,
-        synchronous: false,
+          priority: 32,
-      })
+          offset: 0,
+          limit: 0,
+          synchronous: false,
+        }),
+      `downloadFile:${fileName}`
+    )
       .then((result: unknown) => {
         // If the file was already cached locally, invoke returns immediately
         const file = result as TdFile | undefined;

worker/src/upload/channel.ts

@@ -3,6 +3,7 @@ import { stat } from "fs/promises";
 import type { Client } from "tdl";
 import { config } from "../util/config.js";
 import { childLogger } from "../util/logger.js";
+import { withFloodWait } from "../util/retry.js";
 
 const log = childLogger("upload");
 
@@ -84,24 +85,29 @@ async function sendAndWaitForUpload(
   fileName: string,
   fileSizeMB: number
 ): Promise<bigint> {
-  // Send the message — this returns a temporary message immediately
+  // Send the message — this returns a temporary message immediately.
-  const tempMsg = (await client.invoke({
+  // Wrapped in withFloodWait to handle Telegram rate limits on upload.
-    _: "sendMessage",
+  const tempMsg = (await withFloodWait(
-    chat_id: Number(chatId),
+    () =>
-    input_message_content: {
+      client.invoke({
-      _: "inputMessageDocument",
+        _: "sendMessage",
-      document: {
+        chat_id: Number(chatId),
-        _: "inputFileLocal",
+        input_message_content: {
-        path: filePath,
+          _: "inputMessageDocument",
-      },
+          document: {
-      caption: caption
+            _: "inputFileLocal",
-        ? {
+            path: filePath,
-            _: "formattedText",
+          },
-            text: caption,
+          caption: caption
-          }
+            ? {
-        : undefined,
+                _: "formattedText",
-    },
+                text: caption,
-  })) as { id: number };
+              }
+            : undefined,
+        },
+      }),
+    "sendMessage:upload"
+  )) as { id: number };
 
   const tempMsgId = tempMsg.id;
 

worker/src/util/retry.ts

@@ -0,0 +1,109 @@
+import { childLogger } from "./logger.js";
+import { config } from "./config.js";
+
+const log = childLogger("retry");
+
+/**
+ * Extract the FLOOD_WAIT duration (in seconds) from a TDLib error.
+ *
+ * TDLib errors for rate limiting look like:
+ *   - Error message: "Too Many Requests: retry after 30"
+ *   - Error message: "FLOOD_WAIT_30"
+ *   - Error code: 429
+ */
+export function extractFloodWaitSeconds(err: unknown): number | null {
+  if (!err || typeof err !== "object") return null;
+
+  const message = (err as { message?: string }).message ?? "";
+  const code = (err as { code?: number }).code;
+
+  // Match "FLOOD_WAIT_<seconds>" pattern
+  const floodMatch = message.match(/FLOOD_WAIT_(\d+)/i);
+  if (floodMatch) {
+    return parseInt(floodMatch[1], 10);
+  }
+
+  // Match "retry after <seconds>" pattern (from Telegram HTTP API style errors)
+  const retryMatch = message.match(/retry after (\d+)/i);
+  if (retryMatch) {
+    return parseInt(retryMatch[1], 10);
+  }
+
+  // If error code is 429 but no explicit wait time, default to 30 seconds
+  if (code === 429) {
+    return 30;
+  }
+
+  return null;
+}
+
+/**
+ * Sleep for a given number of milliseconds, with a descriptive log message.
+ */
+function sleepMs(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+/**
+ * Wraps a TDLib invoke operation with FLOOD_WAIT-aware retry logic.
+ *
+ * When Telegram returns a rate limit error (FLOOD_WAIT / 429), this:
+ *   1. Extracts the required wait time from the error
+ *   2. Logs a warning with the wait duration
+ *   3. Sleeps for the required duration + small jitter
+ *   4. Retries the operation (up to maxRetries times)
+ *
+ * Non-rate-limit errors are re-thrown immediately.
+ *
+ * Usage:
+ *   const result = await withFloodWait(() => client.invoke({ ... }));
+ */
+export async function withFloodWait<T>(
+  fn: () => Promise<T>,
+  context?: string,
+  maxRetries?: number
+): Promise<T> {
+  const limit = maxRetries ?? config.maxRetries;
+  let lastError: unknown;
+
+  for (let attempt = 0; attempt <= limit; attempt++) {
+    try {
+      return await fn();
+    } catch (err) {
+      lastError = err;
+      const waitSeconds = extractFloodWaitSeconds(err);
+
+      if (waitSeconds === null) {
+        // Not a rate limit error — re-throw immediately
+        throw err;
+      }
+
+      if (attempt >= limit) {
+        log.error(
+          { context, attempt, waitSeconds },
+          "Rate limit exceeded max retries — giving up"
+        );
+        throw err;
+      }
+
+      // Add small jitter (1–5 seconds) to avoid multiple clients retrying simultaneously
+      const jitter = 1000 + Math.random() * 4000;
+      const totalWaitMs = waitSeconds * 1000 + jitter;
+
+      log.warn(
+        {
+          context,
+          attempt: attempt + 1,
+          maxRetries: limit,
+          waitSeconds,
+          totalWaitMs: Math.round(totalWaitMs),
+        },
+        `Rate-limited by Telegram — sleeping ${waitSeconds}s before retry`
+      );
+
+      await sleepMs(totalWaitMs);
+    }
+  }
+
+  throw lastError;
+}

worker/src/worker.ts

@@ -716,6 +716,29 @@ async function processOneArchiveSet(
     return;
   }
 
+  // ── Size guard: skip archives that exceed WORKER_MAX_ZIP_SIZE_MB ──
+  const totalArchiveSize = archiveSet.parts.reduce((sum, p) => sum + p.fileSize, 0n);
+  const maxSizeBytes = BigInt(config.maxZipSizeMB) * 1024n * 1024n;
+  if (totalArchiveSize > maxSizeBytes) {
+    accountLog.warn(
+      {
+        fileName: archiveName,
+        totalSizeMB: Number(totalArchiveSize / (1024n * 1024n)),
+        maxSizeMB: config.maxZipSizeMB,
+      },
+      "Archive exceeds max size limit, skipping"
+    );
+    await updateRunActivity(runId, {
+      currentActivity: `Skipped ${archiveName} (exceeds ${config.maxZipSizeMB}MB limit)`,
+      currentStep: "skipping",
+      currentChannel: channelTitle,
+      currentFile: archiveName,
+      currentFileNum: setIdx + 1,
+      totalFiles: totalSets,
+    });
+    return;
+  }
+
   const tempPaths: string[] = [];
   let splitPaths: string[] = [];