fix: prevent pool exhaustion that caused 4-hour duplicate check stall

The pg pool had max=5 connections shared between Prisma operations and
advisory locks. With 2 account locks held permanently and hash locks
from timed-out (but still running) background work, pool.connect()
would block forever — causing the Turnbase.7z stall.

- Increase pool max from 5 to 15 for headroom
- Add 30s connectionTimeoutMillis so pool.connect() throws instead of
  hanging forever when the pool is exhausted
- On startup, terminate zombie PostgreSQL sessions from previous worker
  instances that hold stale advisory locks

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

.gitignore

@@ -54,3 +54,4 @@ src/generated
 # temp files
 nul
 tmpclaude-*
+.worktrees/

docker-compose.yml

@@ -28,6 +28,8 @@ services:
       timeout: 5s
       retries: 3
       start_period: 60s
+    volumes:
+      - manual_uploads:/data/uploads
     restart: unless-stopped
     deploy:
       resources:
@@ -54,6 +56,7 @@ services:
     volumes:
       - tdlib_state:/data/tdlib
       - tmp_zips:/tmp/zips
+      - manual_uploads:/data/uploads
     depends_on:
       db:
         condition: service_healthy
@@ -121,6 +124,7 @@ volumes:
   tdlib_state:
   tdlib_bot_state:
   tmp_zips:
+  manual_uploads:
 
 networks:
   frontend:

docs/superpowers/specs/2026-05-02-worker-improvements-design.md

@@ -0,0 +1,184 @@
+# Worker Improvements Design
+
+**Date:** 2026-05-02  
+**Status:** Approved  
+**Scope:** Dragon's Stash Telegram ingestion worker
+
+## Problem Statement
+
+Three issues to address:
+
+1. **Double-uploads**: The same archive occasionally appears twice in the destination Telegram channel. Root causes: (a) the worker crashes between `uploadToChannel()` confirming success and `createPackageWithFiles()` writing to the DB — no DB record means `recoverIncompleteUploads()` can't detect the orphaned Telegram message, and the next cycle re-uploads; (b) two accounts scanning the same source channel can both pass the hash dedup check before either creates a DB record, racing to upload the same file.
+
+2. **Sequential account processing**: Both Telegram accounts are processed one after another via `withTdlibMutex`, even though TDLib fully supports multiple concurrent clients in the same process (each with separate `databaseDirectory` and `filesDirectory`). This halves throughput unnecessarily.
+
+3. **Premium upload limit not used**: The Premium account can upload up to 4 GB per file, but `MAX_UPLOAD_SIZE` is hardcoded at ~1,950 MB. This causes unnecessary file splitting and expensive repack operations for files that could upload directly.
+
+## Solution Overview
+
+Three targeted changes, no architectural overhaul:
+
+1. Two-phase DB write + hash advisory lock (fixes double-uploads)
+2. Remove TDLib mutex from the scheduler loop (enables parallel accounts)
+3. Per-account `maxUploadSize` from `getMe().is_premium` (enables 4 GB for Premium)
+
+---
+
+## Section 1: Double-Upload Fix
+
+### 1a. Two-Phase DB Write
+
+**Current flow:**
+```
+uploadToChannel() → preview download → metadata extraction → createPackageWithFiles()
+```
+
+If the worker crashes anywhere between upload confirmation and `createPackageWithFiles()`, no DB record exists. `recoverIncompleteUploads()` only checks packages with an existing `destMessageId` in the DB — it cannot find an orphaned Telegram message with no corresponding row.
+
+**New flow:**
+```
+uploadToChannel()
+  → createPackageStub()          ← minimal record, destMessageId set immediately
+  → preview download
+  → metadata extraction
+  → updatePackageWithMetadata()  ← adds file list, preview, creator, tags
+```
+
+`createPackageStub()` writes: `contentHash`, `fileName`, `fileSize`, `archiveType`, `sourceChannelId`, `sourceMessageId`, `destChannelId`, `destMessageId`, `isMultipart`, `partCount`, `ingestionRunId`. File list and preview are left empty.
+
+If the worker crashes after the stub is written:
+- `recoverIncompleteUploads()` finds the record (has `destMessageId`), verifies the Telegram message exists, keeps it.
+- Next cycle: `packageExistsByHash()` returns true → skips re-upload.
+- The stub has `fileCount = 0` and no file listing. The UI shows "metadata pending" rather than failing silently.
+
+Stubs with `fileCount = 0` are valid deliverable packages (the bot can still send the file). Backfilling metadata on stubs is out of scope for this change — the crash case is rare and the stub is functional.
+
+### 1b. Hash Advisory Lock
+
+**The race (two accounts, shared source channel):**
+```
+Worker A: packageExistsByHash(X) → false  (no record yet)
+Worker B: packageExistsByHash(X) → false  (no record yet)
+Worker A: uploads file → destMessageId_A
+Worker B: uploads file → destMessageId_B  ← duplicate Telegram message
+Worker A: createPackageStub() → succeeds  (contentHash @unique satisfied)
+Worker B: createPackageStub() → fails unique constraint on contentHash
+```
+Result: two Telegram messages, one DB record. Worker B's upload is wasted.
+
+**Fix:** Before calling `uploadToChannel()`, acquire a PostgreSQL session advisory lock keyed on the content hash:
+
+```sql
+SELECT pg_try_advisory_lock(hash_bigint)
+```
+
+Where `hash_bigint` is the first 8 bytes of the SHA-256 content hash interpreted as a signed bigint.
+
+- `pg_try_advisory_lock` is non-blocking. If another worker holds the lock (same file, shared channel), return `false` → treat as duplicate, skip.
+- After acquiring the lock, **re-run `packageExistsByHash()`** before uploading. This catches the case where another worker finished and released the lock between the first check and this one — without the re-check, the current worker would proceed to re-upload.
+- The lock is session-scoped: released automatically on DB session end. No manual cleanup needed on crash.
+- The lock is released explicitly after `createPackageStub()` completes (or on any error path).
+
+**Implementation location:** New helper `tryAcquireHashLock(contentHash)` / `releaseHashLock(contentHash)` in `worker/src/db/locks.ts`, reusing the existing DB client pattern.
+
+---
+
+## Section 2: Parallel Account Processing
+
+### Current Constraint
+
+`withTdlibMutex` in `scheduler.ts` serializes all TDLib operations across accounts. This was a conservative guard, but TDLib explicitly supports multiple concurrent clients in the same process provided each has its own `databaseDirectory` and `filesDirectory`.
+
+The codebase already satisfies this requirement:
+```typescript
+// worker/src/tdlib/client.ts
+const dbPath = path.join(config.tdlibStateDir, account.id);
+const client = createClient({
+  databaseDirectory: dbPath,
+  filesDirectory: path.join(dbPath, "files"),
+});
+```
+
+Each account gets `<TDLIB_STATE_DIR>/<account.id>/` — fully isolated.
+
+### Change
+
+Replace the sequential `for` loop in `scheduler.ts` with `Promise.allSettled()`:
+
+```typescript
+// Before
+for (const account of accounts) {
+  await withTdlibMutex(`ingest:${account.phone}`, () => runWorkerForAccount(account));
+}
+
+// After
+await Promise.allSettled(accounts.map((account) => runWorkerForAccount(account)));
+```
+
+The per-account PostgreSQL advisory lock in `db/locks.ts` already prevents any account from being processed twice simultaneously. `Promise.allSettled()` ensures one account's failure doesn't abort the other.
+
+The `withTdlibMutex` wrapper can be removed from the ingest path entirely. The auth path (`authenticateAccount`) should also be run in parallel but may remain guarded if TDLib auth flows have ordering dependencies — verify during implementation.
+
+**No Docker Compose changes needed.** Both accounts run in the same container.
+
+### Speed Limit Notifications
+
+TDLib fires `updateSpeedLimitNotification` when an account's upload or download speed is throttled (non-Premium accounts). Log this event at `warn` level in the client update handler so it's visible in logs without being actionable.
+
+---
+
+## Section 3: Per-Account Premium Upload Limit
+
+### Premium Detection
+
+After successful authentication, call `getMe()` and read `is_premium: bool` from the returned `user` object. Store this on `TelegramAccount.isPremium` (new boolean field, default `false`, updated on each successful auth).
+
+```typescript
+const me = await client.invoke({ _: 'getMe' }) as { is_premium?: boolean };
+await updateAccountPremiumStatus(account.id, me.is_premium ?? false);
+```
+
+### Upload Size Limits
+
+| Account type | `maxUploadSize` | Effect |
+|---|---|---|
+| Premium | 3,950 MB | Parts ≤ 3.95 GB upload as-is; repack only for parts >3.95 GB (extremely rare) |
+| Non-Premium | 1,950 MB | Current behavior unchanged |
+
+Pass `maxUploadSize` into `processOneArchiveSet()` as a parameter (currently hardcoded as `MAX_UPLOAD_SIZE` at `worker.ts:1023` and in `archive/split.ts`).
+
+The `hasOversizedPart` check and `byteLevelSplit` call both use this value, so the repack step is effectively eliminated for Premium accounts in practice — no separate "skip repack" flag needed.
+
+### Migration
+
+```prisma
+model TelegramAccount {
+  // ... existing fields
+  isPremium Boolean @default(false)
+}
+```
+
+One migration, one new query `updateAccountPremiumStatus(accountId, isPremium)`.
+
+---
+
+## Files to Change
+
+| File | Change |
+|---|---|
+| `prisma/schema.prisma` | Add `isPremium Boolean @default(false)` to `TelegramAccount` |
+| `worker/src/db/queries.ts` | Add `updateAccountPremiumStatus()`, `createPackageStub()`, `updatePackageWithMetadata()` |
+| `worker/src/db/locks.ts` | Add `tryAcquireHashLock()`, `releaseHashLock()` |
+| `worker/src/tdlib/client.ts` | Call `getMe()` after auth, return `isPremium` from `createTdlibClient()` |
+| `worker/src/worker.ts` | Two-phase write, hash lock acquire/release, pass `maxUploadSize` per account |
+| `worker/src/archive/split.ts` | Accept `maxPartSize` parameter instead of hardcoded constant |
+| `worker/src/scheduler.ts` | Replace sequential loop with `Promise.allSettled()`, remove `withTdlibMutex` from ingest path |
+
+---
+
+## What Is Explicitly Out of Scope
+
+- Backfilling metadata on stub records (rare crash case, functional without it)
+- Download pre-fetching / pipeline parallelism within one account
+- Two separate worker containers (single container is sufficient)
+- Bot or app changes (worker-only)

prisma/migrations/20260330150000_manual_uploads/migration.sql

@@ -0,0 +1,30 @@
+-- CreateEnum
+CREATE TYPE "ManualUploadStatus" AS ENUM ('PENDING', 'PROCESSING', 'COMPLETED', 'FAILED');
+
+-- CreateTable
+CREATE TABLE "manual_uploads" (
+    "id" TEXT NOT NULL,
+    "status" "ManualUploadStatus" NOT NULL DEFAULT 'PENDING',
+    "groupName" TEXT,
+    "userId" TEXT NOT NULL,
+    "errorMessage" TEXT,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "completedAt" TIMESTAMP(3),
+    CONSTRAINT "manual_uploads_pkey" PRIMARY KEY ("id")
+);
+
+CREATE TABLE "manual_upload_files" (
+    "id" TEXT NOT NULL,
+    "uploadId" TEXT NOT NULL,
+    "fileName" TEXT NOT NULL,
+    "filePath" TEXT NOT NULL,
+    "fileSize" BIGINT NOT NULL,
+    "packageId" TEXT,
+    CONSTRAINT "manual_upload_files_pkey" PRIMARY KEY ("id")
+);
+
+CREATE INDEX "manual_uploads_status_idx" ON "manual_uploads"("status");
+CREATE INDEX "manual_upload_files_uploadId_idx" ON "manual_upload_files"("uploadId");
+
+ALTER TABLE "manual_uploads" ADD CONSTRAINT "manual_uploads_userId_fkey" FOREIGN KEY ("userId") REFERENCES "User"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
+ALTER TABLE "manual_upload_files" ADD CONSTRAINT "manual_upload_files_uploadId_fkey" FOREIGN KEY ("uploadId") REFERENCES "manual_uploads"("id") ON DELETE CASCADE ON UPDATE CASCADE;

prisma/migrations/20260502000000_add_is_premium_to_telegram_account/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "telegram_accounts" ADD COLUMN "isPremium" BOOLEAN NOT NULL DEFAULT false;

prisma/schema.prisma

@@ -39,9 +39,10 @@ model User {
   settings     UserSettings?
   telegramLink TelegramLink?
   kickstarters Kickstarter[]
-  inviteCodes  InviteCode[]   @relation("InviteCreator")
-  usedInvite   InviteCode?    @relation("InviteUser", fields: [usedInviteId], references: [id], onDelete: SetNull)
-  usedInviteId String?
+  inviteCodes    InviteCode[]   @relation("InviteCreator")
+  usedInvite     InviteCode?    @relation("InviteUser", fields: [usedInviteId], references: [id], onDelete: SetNull)
+  usedInviteId   String?
+  manualUploads  ManualUpload[]
 }
 
 model Account {
@@ -405,6 +406,7 @@ model TelegramAccount {
   isActive    Boolean   @default(true)
   authState   AuthState @default(PENDING)
   authCode    String?
+  isPremium   Boolean   @default(false)
   lastSeenAt  DateTime?
   createdAt   DateTime  @default(now())
   updatedAt   DateTime  @updatedAt
@@ -865,3 +867,40 @@ model GroupingRule {
   @@index([sourceChannelId])
   @@map("grouping_rules")
 }
+
+enum ManualUploadStatus {
+  PENDING
+  PROCESSING
+  COMPLETED
+  FAILED
+}
+
+model ManualUpload {
+  id              String             @id @default(cuid())
+  status          ManualUploadStatus @default(PENDING)
+  groupName       String?            // Group name if multiple files
+  userId          String
+  errorMessage    String?
+  createdAt       DateTime           @default(now())
+  completedAt     DateTime?
+
+  files           ManualUploadFile[]
+  user            User               @relation(fields: [userId], references: [id])
+
+  @@index([status])
+  @@map("manual_uploads")
+}
+
+model ManualUploadFile {
+  id              String       @id @default(cuid())
+  uploadId        String
+  fileName        String
+  filePath        String       // Path on shared volume
+  fileSize        BigInt
+  packageId       String?      // Set after processing
+
+  upload          ManualUpload @relation(fields: [uploadId], references: [id], onDelete: Cascade)
+
+  @@index([uploadId])
+  @@map("manual_upload_files")
+}

src/app/(app)/stls/_components/stl-table.tsx

@@ -3,7 +3,8 @@
 import { useState, useCallback, useTransition, useMemo, useRef } from "react";
 import { useRouter, usePathname, useSearchParams } from "next/navigation";
 import { toast } from "sonner";
-import { Search, Layers } from "lucide-react";
+import { Search, Layers, Upload } from "lucide-react";
+import { UploadDialog } from "./upload-dialog";
 import { useDataTable } from "@/hooks/use-data-table";
 import {
   getPackageColumns,
@@ -106,6 +107,9 @@ export function StlTable({
   // Group merge state
   const [mergeSourceId, setMergeSourceId] = useState<string | null>(null);
 
+  // Upload dialog state
+  const [uploadOpen, setUploadOpen] = useState(false);
+
   const toggleGroup = useCallback((groupId: string) => {
     setExpandedGroups((prev) => {
       const next = new Set(prev);
@@ -497,6 +501,10 @@ export function StlTable({
               </Select>
             )}
             <DataTableViewOptions table={table} />
+            <Button variant="outline" size="sm" className="h-9" onClick={() => setUploadOpen(true)}>
+              <Upload className="mr-2 h-4 w-4" />
+              Upload Files
+            </Button>
             {selectedPackages.size >= 2 && (
               <Button
                 variant="outline"
@@ -587,6 +595,8 @@ export function StlTable({
         </DialogContent>
       </Dialog>
 
+      <UploadDialog open={uploadOpen} onOpenChange={setUploadOpen} />
+
       {/* Hidden file input for group preview upload (Task 12) */}
       <input
         ref={previewInputRef}

src/app/(app)/stls/_components/upload-dialog.tsx

@@ -0,0 +1,243 @@
+"use client";
+
+import { useState, useRef, useTransition, useEffect } from "react";
+import { Upload, File, X, Loader2, CheckCircle2, AlertCircle } from "lucide-react";
+import { toast } from "sonner";
+import { Button } from "@/components/ui/button";
+import { Input } from "@/components/ui/input";
+import { Label } from "@/components/ui/label";
+import {
+  Dialog,
+  DialogContent,
+  DialogDescription,
+  DialogFooter,
+  DialogHeader,
+  DialogTitle,
+} from "@/components/ui/dialog";
+
+interface UploadDialogProps {
+  open: boolean;
+  onOpenChange: (open: boolean) => void;
+}
+
+function formatSize(bytes: number): string {
+  if (bytes >= 1024 * 1024 * 1024) return `${(bytes / (1024 * 1024 * 1024)).toFixed(1)} GB`;
+  if (bytes >= 1024 * 1024) return `${(bytes / (1024 * 1024)).toFixed(0)} MB`;
+  return `${(bytes / 1024).toFixed(0)} KB`;
+}
+
+type UploadStatus = "idle" | "uploading" | "processing" | "done" | "error";
+
+export function UploadDialog({ open, onOpenChange }: UploadDialogProps) {
+  const [files, setFiles] = useState<File[]>([]);
+  const [groupName, setGroupName] = useState("");
+  const [status, setStatus] = useState<UploadStatus>("idle");
+  const [error, setError] = useState<string | null>(null);
+  const [isPending, startTransition] = useTransition();
+  const fileInputRef = useRef<HTMLInputElement>(null);
+  const pollRef = useRef<ReturnType<typeof setInterval> | null>(null);
+
+  useEffect(() => {
+    if (open) {
+      setFiles([]);
+      setGroupName("");
+      setStatus("idle");
+      setError(null);
+    }
+    return () => {
+      if (pollRef.current) clearInterval(pollRef.current);
+    };
+  }, [open]);
+
+  function handleFileChange(e: React.ChangeEvent<HTMLInputElement>) {
+    if (e.target.files) {
+      setFiles(Array.from(e.target.files));
+    }
+  }
+
+  function removeFile(index: number) {
+    setFiles((prev) => prev.filter((_, i) => i !== index));
+  }
+
+  function handleUpload() {
+    if (files.length === 0) return;
+
+    startTransition(async () => {
+      setStatus("uploading");
+      setError(null);
+
+      try {
+        const formData = new FormData();
+        for (const file of files) {
+          formData.append("files", file);
+        }
+        if (groupName.trim()) {
+          formData.append("groupName", groupName.trim());
+        }
+
+        const res = await fetch("/api/uploads", {
+          method: "POST",
+          body: formData,
+        });
+
+        const data = await res.json();
+        if (!res.ok) {
+          setStatus("error");
+          setError(data.error ?? "Upload failed");
+          return;
+        }
+
+        setStatus("processing");
+
+        // Poll for completion
+        pollRef.current = setInterval(async () => {
+          try {
+            const statusRes = await fetch(`/api/uploads/${data.uploadId}`);
+            const statusData = await statusRes.json();
+
+            if (statusData.status === "COMPLETED") {
+              setStatus("done");
+              toast.success(`${files.length} file(s) uploaded and indexed`);
+              if (pollRef.current) clearInterval(pollRef.current);
+            } else if (statusData.status === "FAILED") {
+              setStatus("error");
+              setError(statusData.errorMessage ?? "Processing failed");
+              if (pollRef.current) clearInterval(pollRef.current);
+            }
+          } catch {
+            // Keep polling
+          }
+        }, 3000);
+
+        // Stop polling after 10 minutes
+        setTimeout(() => {
+          if (pollRef.current) {
+            clearInterval(pollRef.current);
+            pollRef.current = null;
+            setStatus((s) => s === "processing" ? "done" : s);
+          }
+        }, 600_000);
+      } catch {
+        setStatus("error");
+        setError("Network error");
+      }
+    });
+  }
+
+  return (
+    <Dialog open={open} onOpenChange={onOpenChange}>
+      <DialogContent className="sm:max-w-lg">
+        <DialogHeader>
+          <DialogTitle>Upload Files</DialogTitle>
+          <DialogDescription>
+            Upload archive files to be processed and indexed. Multiple files will be automatically grouped.
+          </DialogDescription>
+        </DialogHeader>
+
+        {status === "idle" && (
+          <div className="space-y-4">
+            <div
+              className="border-2 border-dashed rounded-lg p-8 text-center cursor-pointer hover:border-primary/50 transition-colors"
+              onClick={() => fileInputRef.current?.click()}
+            >
+              <Upload className="h-8 w-8 mx-auto mb-2 text-muted-foreground" />
+              <p className="text-sm text-muted-foreground">
+                Click to select files or drag & drop
+              </p>
+              <p className="text-xs text-muted-foreground mt-1">
+                ZIP, RAR, 7Z files up to 4GB each
+              </p>
+              <input
+                ref={fileInputRef}
+                type="file"
+                multiple
+                accept=".zip,.rar,.7z,.pdf,.stl"
+                onChange={handleFileChange}
+                className="hidden"
+              />
+            </div>
+
+            {files.length > 0 && (
+              <div className="space-y-2">
+                {files.map((file, i) => (
+                  <div key={i} className="flex items-center gap-2 p-2 rounded bg-muted/30">
+                    <File className="h-4 w-4 shrink-0 text-muted-foreground" />
+                    <span className="text-sm flex-1 truncate">{file.name}</span>
+                    <span className="text-xs text-muted-foreground">{formatSize(file.size)}</span>
+                    <button onClick={() => removeFile(i)} className="p-0.5 hover:text-destructive">
+                      <X className="h-3.5 w-3.5" />
+                    </button>
+                  </div>
+                ))}
+              </div>
+            )}
+
+            {files.length > 1 && (
+              <div>
+                <Label htmlFor="groupName" className="text-sm">Group Name (optional)</Label>
+                <Input
+                  id="groupName"
+                  value={groupName}
+                  onChange={(e) => setGroupName(e.target.value)}
+                  placeholder="Auto-generated from filenames"
+                  className="mt-1"
+                />
+              </div>
+            )}
+          </div>
+        )}
+
+        {(status === "uploading" || status === "processing") && (
+          <div className="flex items-center gap-3 p-6 rounded-lg bg-muted/30 border">
+            <Loader2 className="h-6 w-6 animate-spin text-primary" />
+            <div>
+              <p className="text-sm font-medium">
+                {status === "uploading" ? "Uploading files..." : "Processing & uploading to Telegram..."}
+              </p>
+              <p className="text-xs text-muted-foreground mt-0.5">
+                {status === "uploading"
+                  ? "Sending files to server"
+                  : "Hashing, extracting metadata, uploading to destination channel"}
+              </p>
+            </div>
+          </div>
+        )}
+
+        {status === "done" && (
+          <div className="flex items-center gap-3 p-6 rounded-lg bg-green-500/10 border border-green-500/20">
+            <CheckCircle2 className="h-6 w-6 text-green-500" />
+            <div>
+              <p className="text-sm font-medium text-green-500">Upload complete!</p>
+              <p className="text-xs text-muted-foreground">Files have been indexed and uploaded to Telegram.</p>
+            </div>
+          </div>
+        )}
+
+        {status === "error" && (
+          <div className="flex items-center gap-3 p-6 rounded-lg bg-destructive/10 border border-destructive/20">
+            <AlertCircle className="h-6 w-6 text-destructive" />
+            <div>
+              <p className="text-sm font-medium text-destructive">Upload failed</p>
+              <p className="text-xs text-muted-foreground">{error}</p>
+            </div>
+          </div>
+        )}
+
+        <DialogFooter>
+          {status === "idle" && (
+            <>
+              <Button variant="outline" onClick={() => onOpenChange(false)}>Cancel</Button>
+              <Button onClick={handleUpload} disabled={files.length === 0 || isPending}>
+                <Upload className="h-4 w-4 mr-1" />
+                Upload {files.length > 0 ? `(${files.length})` : ""}
+              </Button>
+            </>
+          )}
+          {(status === "done" || status === "error") && (
+            <Button variant="outline" onClick={() => onOpenChange(false)}>Close</Button>
+          )}
+        </DialogFooter>
+      </DialogContent>
+    </Dialog>
+  );
+}

src/app/api/notifications/read/route.ts

@@ -3,6 +3,8 @@ import { auth } from "@/lib/auth";
 import {
   markNotificationRead,
   markAllNotificationsRead,
+  dismissNotification,
+  clearAllNotifications,
 } from "@/data/notification.queries";
 
 export const dynamic = "force-dynamic";
@@ -15,8 +17,13 @@ export async function POST(request: Request) {
 
   const body = await request.json().catch(() => ({}));
   const id = body.id as string | undefined;
+  const action = (body.action as string) ?? "read";
 
-  if (id) {
+  if (action === "dismiss" && id) {
+    await dismissNotification(id);
+  } else if (action === "clear") {
+    await clearAllNotifications();
+  } else if (id) {
     await markNotificationRead(id);
   } else {
     await markAllNotificationsRead();

src/app/api/uploads/[id]/route.ts

@@ -0,0 +1,43 @@
+import { NextResponse } from "next/server";
+import { auth } from "@/lib/auth";
+import { prisma } from "@/lib/prisma";
+
+export const dynamic = "force-dynamic";
+
+export async function GET(
+  _request: Request,
+  { params }: { params: Promise<{ id: string }> }
+) {
+  const session = await auth();
+  if (!session?.user?.id) {
+    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
+  }
+
+  const { id } = await params;
+
+  const upload = await prisma.manualUpload.findUnique({
+    where: { id },
+    include: {
+      files: {
+        select: { id: true, fileName: true, fileSize: true, packageId: true },
+      },
+    },
+  });
+
+  if (!upload || upload.userId !== session.user.id) {
+    return NextResponse.json({ error: "Not found" }, { status: 404 });
+  }
+
+  return NextResponse.json({
+    id: upload.id,
+    status: upload.status,
+    groupName: upload.groupName,
+    errorMessage: upload.errorMessage,
+    files: upload.files.map((f) => ({
+      ...f,
+      fileSize: f.fileSize.toString(),
+    })),
+    createdAt: upload.createdAt.toISOString(),
+    completedAt: upload.completedAt?.toISOString() ?? null,
+  });
+}

src/app/api/uploads/route.ts

@@ -0,0 +1,83 @@
+import { NextResponse } from "next/server";
+import { auth } from "@/lib/auth";
+import { prisma } from "@/lib/prisma";
+import { writeFile, mkdir } from "fs/promises";
+import path from "path";
+
+export const dynamic = "force-dynamic";
+
+const UPLOAD_DIR = process.env.UPLOAD_DIR ?? "/data/uploads";
+const MAX_FILE_SIZE = 4 * 1024 * 1024 * 1024; // 4GB per file
+
+export async function POST(request: Request) {
+  const session = await auth();
+  if (!session?.user?.id) {
+    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
+  }
+
+  try {
+    const formData = await request.formData();
+    const files = formData.getAll("files") as File[];
+    const groupName = formData.get("groupName") as string | null;
+
+    if (!files.length) {
+      return NextResponse.json({ error: "No files provided" }, { status: 400 });
+    }
+
+    // Create the upload record
+    const upload = await prisma.manualUpload.create({
+      data: {
+        userId: session.user.id,
+        groupName: groupName || (files.length > 1 ? files[0].name.replace(/\.[^.]+$/, "") : null),
+        status: "PENDING",
+      },
+    });
+
+    // Save files to shared volume
+    const uploadDir = path.join(UPLOAD_DIR, upload.id);
+    await mkdir(uploadDir, { recursive: true });
+
+    for (const file of files) {
+      if (file.size > MAX_FILE_SIZE) {
+        return NextResponse.json(
+          { error: `File "${file.name}" exceeds 4GB limit` },
+          { status: 400 }
+        );
+      }
+
+      const filePath = path.join(uploadDir, file.name);
+      const buffer = Buffer.from(await file.arrayBuffer());
+      await writeFile(filePath, buffer);
+
+      await prisma.manualUploadFile.create({
+        data: {
+          uploadId: upload.id,
+          fileName: file.name,
+          filePath,
+          fileSize: BigInt(file.size),
+        },
+      });
+    }
+
+    // Notify worker
+    try {
+      await prisma.$queryRawUnsafe(
+        `SELECT pg_notify('manual_upload', $1)`,
+        upload.id
+      );
+    } catch {
+      // Best-effort
+    }
+
+    return NextResponse.json({
+      uploadId: upload.id,
+      fileCount: files.length,
+      status: "PENDING",
+    });
+  } catch (err) {
+    return NextResponse.json(
+      { error: err instanceof Error ? err.message : "Upload failed" },
+      { status: 500 }
+    );
+  }
+}

src/components/layout/notification-bell.tsx

@@ -1,7 +1,7 @@
 "use client";
 
 import { useState, useEffect, useCallback } from "react";
-import { Bell, AlertTriangle, AlertCircle, Info, CheckCircle2 } from "lucide-react";
+import { Bell, AlertTriangle, AlertCircle, Info, CheckCircle2, X, Trash2 } from "lucide-react";
 import { Button } from "@/components/ui/button";
 import { Badge } from "@/components/ui/badge";
 import {
@@ -94,6 +94,34 @@ export function NotificationBell() {
     }
   }
 
+  async function handleDismiss(id: string) {
+    try {
+      await fetch("/api/notifications/read", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ id, action: "dismiss" }),
+      });
+      setNotifications((prev) => prev.filter((n) => n.id !== id));
+      setUnreadCount((c) => Math.max(0, c - 1));
+    } catch {
+      // Ignore
+    }
+  }
+
+  async function handleClearAll() {
+    try {
+      await fetch("/api/notifications/read", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ action: "clear" }),
+      });
+      setNotifications([]);
+      setUnreadCount(0);
+    } catch {
+      // Ignore
+    }
+  }
+
   async function handleRepair(notificationId: string) {
     try {
       const res = await fetch("/api/notifications/repair", {
@@ -141,16 +169,29 @@ export function NotificationBell() {
       <PopoverContent className="w-96 p-0" align="end">
         <div className="flex items-center justify-between border-b px-4 py-3">
           <h3 className="text-sm font-semibold">Notifications</h3>
-          {unreadCount > 0 && (
-            <Button
-              variant="ghost"
-              size="sm"
-              className="h-7 text-xs"
-              onClick={handleMarkAllRead}
-            >
-              Mark all read
-            </Button>
-          )}
+          <div className="flex items-center gap-1">
+            {unreadCount > 0 && (
+              <Button
+                variant="ghost"
+                size="sm"
+                className="h-7 text-xs"
+                onClick={handleMarkAllRead}
+              >
+                Mark all read
+              </Button>
+            )}
+            {notifications.length > 0 && (
+              <Button
+                variant="ghost"
+                size="sm"
+                className="h-7 text-xs text-muted-foreground"
+                onClick={handleClearAll}
+              >
+                <Trash2 className="h-3 w-3 mr-1" />
+                Clear
+              </Button>
+            )}
+          </div>
         </div>
         <ScrollArea className="max-h-[400px]">
           {notifications.length === 0 ? (
@@ -187,6 +228,13 @@ export function NotificationBell() {
                         {!n.isRead && (
                           <span className="h-2 w-2 rounded-full bg-primary shrink-0" />
                         )}
+                        <button
+                          className="ml-auto shrink-0 p-0.5 rounded hover:bg-muted text-muted-foreground hover:text-foreground"
+                          onClick={(e) => { e.stopPropagation(); handleDismiss(n.id); }}
+                          title="Dismiss"
+                        >
+                          <X className="h-3 w-3" />
+                        </button>
                       </div>
                       <p className="text-xs text-muted-foreground line-clamp-2 mt-0.5">
                         {n.message}

src/data/notification.queries.ts

@@ -35,3 +35,11 @@ export async function markAllNotificationsRead() {
     data: { isRead: true },
   });
 }
+
+export async function dismissNotification(id: string) {
+  return prisma.systemNotification.delete({ where: { id } });
+}
+
+export async function clearAllNotifications() {
+  return prisma.systemNotification.deleteMany({});
+}

worker/src/archive/split.ts

@@ -18,20 +18,22 @@ const log = childLogger("split");
 const MAX_PART_SIZE = BigInt(config.maxPartSizeMB) * 1024n * 1024n;
 
 /**
- * Split a file into ≤2GB parts using byte-level splitting.
- * Returns paths to the split parts. If the file is already ≤2GB, returns the original path.
+ * Split a file into parts using byte-level splitting.
+ * Returns paths to the split parts. If the file fits in one part, returns the original path.
+ * Pass maxPartSize to override the global default (e.g., 3950 MiB for Premium accounts).
  */
-export async function byteLevelSplit(filePath: string): Promise<string[]> {
+export async function byteLevelSplit(filePath: string, maxPartSize?: bigint): Promise<string[]> {
+  const effectiveMax = maxPartSize ?? MAX_PART_SIZE;
   const stats = await stat(filePath);
   const fileSize = BigInt(stats.size);
 
-  if (fileSize <= MAX_PART_SIZE) {
+  if (fileSize <= effectiveMax) {
     return [filePath];
   }
 
   const dir = path.dirname(filePath);
   const baseName = path.basename(filePath);
-  const partSize = Number(MAX_PART_SIZE);
+  const partSize = Number(effectiveMax);
   const totalParts = Math.ceil(Number(fileSize) / partSize);
   const parts: string[] = [];
 

worker/src/audit.ts

@@ -38,7 +38,9 @@ export async function runIntegrityAudit(): Promise<{ checked: number; issues: nu
 
   for (const pkg of multipartPackages) {
     const actualParts = pkg.destMessageIds.length;
-    if (actualParts > 0 && actualParts !== pkg.partCount) {
+    // Only flag when we have >1 stored IDs but count doesn't match.
+    // Packages with exactly 1 ID are legacy (backfilled from single destMessageId) — not actionable.
+    if (actualParts > 1 && actualParts !== pkg.partCount) {
       issues++;
 
       // Check if we already have a notification for this

worker/src/db/client.ts

@@ -5,7 +5,14 @@ import { config } from "../util/config.js";
 
 const pool = new pg.Pool({
   connectionString: config.databaseUrl,
-  max: 5,
+  // Pool needs headroom for: 2 account advisory locks (held for entire cycle),
+  // up to 2 concurrent hash locks, plus Prisma operations from both accounts.
+  // Previously max=5 caused pool exhaustion and indefinite hangs.
+  max: 15,
+  // Prevent pool.connect() from blocking forever when pool is exhausted.
+  // Throws an error after 30s so the operation can fail and retry instead of
+  // silently hanging for hours (as happened with the Turnbase.7z stall).
+  connectionTimeoutMillis: 30_000,
 });
 
 const adapter = new PrismaPg(pool);

worker/src/db/locks.ts

@@ -79,3 +79,66 @@ export async function releaseLock(accountId: string): Promise<void> {
     client.release();
   }
 }
+
+/**
+ * Derive a lock ID for a content hash. Prefixes with "hash:" so the resulting
+ * 32-bit integer does not collide with account advisory lock IDs.
+ */
+function contentHashToLockId(contentHash: string): number {
+  return hashToLockId(`hash:${contentHash}`);
+}
+
+/**
+ * Acquire a per-content-hash advisory lock before uploading.
+ * Prevents two concurrent workers from uploading the same archive
+ * when both scan a shared source channel.
+ *
+ * Returns true if acquired (proceed with upload).
+ * Returns false if already held (another worker is handling this archive — skip).
+ *
+ * MUST be released via releaseHashLock() after createPackageStub() completes,
+ * including on all error paths (use try/finally).
+ */
+export async function tryAcquireHashLock(contentHash: string): Promise<boolean> {
+  const lockId = contentHashToLockId(contentHash);
+  const client = await pool.connect();
+  try {
+    const result = await client.query<{ pg_try_advisory_lock: boolean }>(
+      "SELECT pg_try_advisory_lock($1)",
+      [lockId]
+    );
+    const acquired = result.rows[0]?.pg_try_advisory_lock ?? false;
+    if (acquired) {
+      heldConnections.set(`hash:${contentHash}`, client);
+      log.debug({ hash: contentHash.slice(0, 16), lockId }, "Hash lock acquired");
+      return true;
+    } else {
+      client.release();
+      log.debug({ hash: contentHash.slice(0, 16), lockId }, "Hash lock held by another worker — skipping");
+      return false;
+    }
+  } catch (err) {
+    client.release();
+    throw err;
+  }
+}
+
+/**
+ * Release the per-content-hash advisory lock.
+ * Call after createPackageStub() completes (or on any error path).
+ */
+export async function releaseHashLock(contentHash: string): Promise<void> {
+  const lockId = contentHashToLockId(contentHash);
+  const client = heldConnections.get(`hash:${contentHash}`);
+  if (!client) {
+    log.warn({ hash: contentHash.slice(0, 16) }, "No held connection for hash lock release");
+    return;
+  }
+  try {
+    await client.query("SELECT pg_advisory_unlock($1)", [lockId]);
+    log.debug({ hash: contentHash.slice(0, 16) }, "Hash lock released");
+  } finally {
+    heldConnections.delete(`hash:${contentHash}`);
+    client.release();
+  }
+}

worker/src/db/queries.ts

@@ -74,6 +74,105 @@ export async function getUploadedPackageByHash(contentHash: string) {
   });
 }
 
+export interface CreatePackageStubInput {
+  contentHash: string;
+  fileName: string;
+  fileSize: bigint;
+  archiveType: ArchiveType;
+  sourceChannelId: string;
+  sourceMessageId: bigint;
+  sourceTopicId?: bigint | null;
+  destChannelId: string;
+  destMessageId: bigint;
+  destMessageIds: bigint[];
+  isMultipart: boolean;
+  partCount: number;
+  ingestionRunId: string;
+  creator?: string | null;
+  tags?: string[];
+}
+
+/**
+ * Write a minimal Package record immediately after Telegram confirms the upload.
+ * Call this before preview/metadata extraction so recoverIncompleteUploads() can
+ * detect and verify the package if the worker crashes mid-metadata.
+ *
+ * Follow with updatePackageWithMetadata() once file entries and preview are ready.
+ */
+export async function createPackageStub(
+  input: CreatePackageStubInput
+): Promise<{ id: string }> {
+  const pkg = await db.package.create({
+    data: {
+      contentHash: input.contentHash,
+      fileName: input.fileName,
+      fileSize: input.fileSize,
+      archiveType: input.archiveType,
+      sourceChannelId: input.sourceChannelId,
+      sourceMessageId: input.sourceMessageId,
+      sourceTopicId: input.sourceTopicId ?? undefined,
+      destChannelId: input.destChannelId,
+      destMessageId: input.destMessageId,
+      destMessageIds: input.destMessageIds,
+      isMultipart: input.isMultipart,
+      partCount: input.partCount,
+      fileCount: 0,
+      ingestionRunId: input.ingestionRunId,
+      creator: input.creator ?? undefined,
+      tags: input.tags?.length ? input.tags : undefined,
+    },
+    select: { id: true },
+  });
+
+  try {
+    await db.$queryRawUnsafe(
+      `SELECT pg_notify('new_package', $1)`,
+      JSON.stringify({
+        packageId: pkg.id,
+        fileName: input.fileName,
+        creator: input.creator ?? null,
+        tags: input.tags ?? [],
+      })
+    );
+  } catch {
+    // Best-effort
+  }
+
+  return pkg;
+}
+
+/**
+ * Update a stub Package with file entries and preview after metadata extraction.
+ * Called as Phase 2 of the two-phase write after createPackageStub().
+ */
+export async function updatePackageWithMetadata(
+  packageId: string,
+  input: {
+    files: {
+      path: string;
+      fileName: string;
+      extension: string | null;
+      compressedSize: bigint;
+      uncompressedSize: bigint;
+      crc32: string | null;
+    }[];
+    previewData?: Buffer | null;
+    previewMsgId?: bigint | null;
+  }
+): Promise<void> {
+  await db.package.update({
+    where: { id: packageId },
+    data: {
+      fileCount: input.files.length,
+      previewData: input.previewData ? new Uint8Array(input.previewData) : undefined,
+      previewMsgId: input.previewMsgId ?? undefined,
+      files: {
+        create: input.files,
+      },
+    },
+  });
+}
+
 /**
  * Check if a package already exists for a given source message ID
  * AND was successfully uploaded to the destination (destMessageId is set).
@@ -308,6 +407,16 @@ export async function updateAccountAuthState(
   });
 }
 
+export async function updateAccountPremiumStatus(
+  accountId: string,
+  isPremium: boolean
+): Promise<void> {
+  await db.telegramAccount.update({
+    where: { id: accountId },
+    data: { isPremium },
+  });
+}
+
 export async function getAccountAuthCode(accountId: string) {
   const account = await db.telegramAccount.findUnique({
     where: { id: accountId },

worker/src/extract-listener.ts

@@ -101,16 +101,14 @@ export async function processExtractRequest(requestId: string): Promise<void> {
   try {
     await mkdir(tempDir, { recursive: true });
 
-    // Wrap the entire TDLib session in the mutex so no other TDLib
-    // operation can run concurrently (TDLib is single-session).
-    await withTdlibMutex("extract", async () => {
-      const accounts = await getActiveAccounts();
-      if (accounts.length === 0) {
-        throw new Error("No authenticated Telegram accounts available");
-      }
+    const accounts = await getActiveAccounts();
+    if (accounts.length === 0) {
+      throw new Error("No authenticated Telegram accounts available");
+    }
+    const account = accounts[0];
 
-      const account = accounts[0];
-      const client = await createTdlibClient({ id: account.id, phone: account.phone });
+    await withTdlibMutex(account.phone, "extract", async () => {
+      const { client } = await createTdlibClient({ id: account.id, phone: account.phone });
 
       try {
         // Load chat list so TDLib can find the dest channel

worker/src/fetch-listener.ts

@@ -5,6 +5,7 @@ import { withTdlibMutex } from "./util/mutex.js";
 import { processFetchRequest } from "./worker.js";
 import { processExtractRequest } from "./extract-listener.js";
 import { rebuildPackageDatabase } from "./rebuild.js";
+import { processManualUpload } from "./manual-upload.js";
 import { generateInviteLink, createSupergroup, searchPublicChat } from "./tdlib/chats.js";
 import { createTdlibClient, closeTdlibClient } from "./tdlib/client.js";
 import { triggerImmediateCycle } from "./scheduler.js";
@@ -13,6 +14,7 @@ import {
   getGlobalSetting,
   setGlobalSetting,
   getActiveAccounts,
+  getChannelFetchRequest,
   upsertChannel,
   ensureAccountChannelLink,
   updateFetchRequestStatus,
@@ -55,6 +57,7 @@ async function connectListener(): Promise<void> {
     await pgClient.query("LISTEN join_channel");
     await pgClient.query("LISTEN archive_extract");
     await pgClient.query("LISTEN rebuild_packages");
+    await pgClient.query("LISTEN manual_upload");
 
     pgClient.on("notification", (msg) => {
       if (msg.channel === "channel_fetch" && msg.payload) {
@@ -71,6 +74,8 @@ async function connectListener(): Promise<void> {
         handleArchiveExtract(msg.payload);
       } else if (msg.channel === "rebuild_packages" && msg.payload) {
         handleRebuildPackages(msg.payload);
+      } else if (msg.channel === "manual_upload" && msg.payload) {
+        handleManualUpload(msg.payload);
       }
     });
 
@@ -96,7 +101,7 @@ async function connectListener(): Promise<void> {
       }
     });
 
-    log.info("Fetch listener started (channel_fetch, generate_invite, create_destination, ingestion_trigger, join_channel, archive_extract, rebuild_packages)");
+    log.info("Fetch listener started (channel_fetch, generate_invite, create_destination, ingestion_trigger, join_channel, archive_extract, rebuild_packages, manual_upload)");
   } catch (err) {
     log.error({ err }, "Failed to start fetch listener — retrying");
     scheduleReconnect();
@@ -129,7 +134,9 @@ let fetchQueue: Promise<void> = Promise.resolve();
 function handleChannelFetch(requestId: string): void {
   fetchQueue = fetchQueue.then(async () => {
     try {
-      await withTdlibMutex("fetch-channels", () =>
+      const request = await getChannelFetchRequest(requestId);
+      const key = request?.account?.phone ?? "global";
+      await withTdlibMutex(key, "fetch-channels", () =>
         processFetchRequest(requestId)
       );
     } catch (err) {
@@ -143,22 +150,20 @@ function handleChannelFetch(requestId: string): void {
 function handleGenerateInvite(channelId: string): void {
   fetchQueue = fetchQueue.then(async () => {
     try {
-      await withTdlibMutex("generate-invite", async () => {
+      const accounts = await getActiveAccounts();
+      if (accounts.length === 0) {
+        log.warn("No authenticated accounts to generate invite link");
+        return;
+      }
+      const account = accounts[0];
+      await withTdlibMutex(account.phone, "generate-invite", async () => {
         const destChannel = await getGlobalDestinationChannel();
         if (!destChannel || destChannel.id !== channelId) {
           log.warn({ channelId }, "Destination channel mismatch, skipping invite generation");
           return;
         }
 
-        // Use the first available authenticated account to generate the link
-        const accounts = await getActiveAccounts();
-        if (accounts.length === 0) {
-          log.warn("No authenticated accounts to generate invite link");
-          return;
-        }
-
-        const account = accounts[0];
-        const client = await createTdlibClient({ id: account.id, phone: account.phone });
+        const { client } = await createTdlibClient({ id: account.id, phone: account.phone });
 
         try {
           const link = await generateInviteLink(client, destChannel.telegramId);
@@ -183,7 +188,13 @@ function handleCreateDestination(payload: string): void {
       const parsed = JSON.parse(payload) as { requestId: string; title: string };
       requestId = parsed.requestId;
 
-      await withTdlibMutex("create-destination", async () => {
+      const accounts = await getActiveAccounts();
+      if (accounts.length === 0) {
+        throw new Error("No authenticated accounts available to create the group");
+      }
+      const account = accounts[0];
+
+      await withTdlibMutex(account.phone, "create-destination", async () => {
         const { db } = await import("./db/client.js");
 
         // Mark the request as in-progress
@@ -192,14 +203,7 @@ function handleCreateDestination(payload: string): void {
           data: { status: "IN_PROGRESS" },
         });
 
-        // Use the first available authenticated account
-        const accounts = await getActiveAccounts();
-        if (accounts.length === 0) {
-          throw new Error("No authenticated accounts available to create the group");
-        }
-
-        const account = accounts[0];
-        const client = await createTdlibClient({ id: account.id, phone: account.phone });
+        const { client } = await createTdlibClient({ id: account.id, phone: account.phone });
 
         try {
           // Create the supergroup via TDLib
@@ -324,16 +328,16 @@ function handleJoinChannel(payload: string): void {
       const parsed = JSON.parse(payload) as { requestId: string; input: string; accountId: string };
       requestId = parsed.requestId;
 
-      await withTdlibMutex("join-channel", async () => {
+      const accounts = await getActiveAccounts();
+      const account = accounts.find((a) => a.id === parsed.accountId) ?? accounts[0];
+      if (!account) {
+        throw new Error("No authenticated accounts available");
+      }
+
+      await withTdlibMutex(account.phone, "join-channel", async () => {
         await updateFetchRequestStatus(requestId!, "IN_PROGRESS");
 
-        const accounts = await getActiveAccounts();
-        const account = accounts.find((a) => a.id === parsed.accountId) ?? accounts[0];
-        if (!account) {
-          throw new Error("No authenticated accounts available");
-        }
-
-        const client = await createTdlibClient({ id: account.id, phone: account.phone });
+        const { client } = await createTdlibClient({ id: account.id, phone: account.phone });
 
         try {
           const linkInfo = parseTelegramInput(parsed.input);
@@ -503,7 +507,12 @@ function handleIngestionTrigger(): void {
 function handleRebuildPackages(requestId: string): void {
   fetchQueue = fetchQueue.then(async () => {
     try {
-      await withTdlibMutex("rebuild-packages", () =>
+      const accounts = await getActiveAccounts();
+      if (accounts.length === 0) {
+        log.warn("No authenticated accounts to rebuild packages");
+        return;
+      }
+      await withTdlibMutex(accounts[0].phone, "rebuild-packages", () =>
         rebuildPackageDatabase(requestId)
       );
     } catch (err) {
@@ -511,3 +520,11 @@ function handleRebuildPackages(requestId: string): void {
     }
   });
 }
+
+// ── Manual upload handler ──
+
+function handleManualUpload(uploadId: string): void {
+  fetchQueue = fetchQueue
+    .then(() => processManualUpload(uploadId))
+    .catch((err) => log.error({ err, uploadId }, "Manual upload processing failed"));
+}

worker/src/index.ts

@@ -27,6 +27,33 @@ async function main(): Promise<void> {
   await cleanupTempDir();
   await markStaleRunsAsFailed();
 
+  // Release any advisory locks orphaned by a previous worker instance.
+  // When Docker kills a container, PostgreSQL may keep the session alive
+  // (zombie connections), holding advisory locks that block the new worker.
+  try {
+    const result = await pool.query(`
+      SELECT pid, state, left(query, 80) as query, age(clock_timestamp(), state_change) as idle_time
+      FROM pg_stat_activity
+      WHERE datname = current_database()
+        AND pid != pg_backend_pid()
+        AND state = 'idle'
+        AND query LIKE '%pg_try_advisory_lock%'
+        AND state_change < clock_timestamp() - interval '5 minutes'
+    `);
+    for (const row of result.rows) {
+      log.warn(
+        { pid: row.pid, idleTime: row.idle_time, query: row.query },
+        "Terminating stale advisory lock session from previous worker"
+      );
+      await pool.query("SELECT pg_terminate_backend($1)", [row.pid]);
+    }
+    if (result.rows.length > 0) {
+      log.info({ terminated: result.rows.length }, "Cleaned up stale advisory lock sessions");
+    }
+  } catch (err) {
+    log.warn({ err }, "Failed to clean up stale advisory locks (non-fatal)");
+  }
+
   // Verify destination messages exist for all "uploaded" packages.
   // Resets any packages whose dest message is missing so they get re-processed.
   await recoverIncompleteUploads();

worker/src/manual-upload.ts

@@ -0,0 +1,211 @@
+import path from "path";
+import { rm } from "fs/promises";
+import { db } from "./db/client.js";
+import { childLogger } from "./util/logger.js";
+import { config } from "./util/config.js";
+import { hashParts } from "./archive/hash.js";
+import { byteLevelSplit } from "./archive/split.js";
+import { uploadToChannel } from "./upload/channel.js";
+import { createTdlibClient, closeTdlibClient } from "./tdlib/client.js";
+import { readZipCentralDirectory } from "./archive/zip-reader.js";
+import { readRarContents } from "./archive/rar-reader.js";
+import { read7zContents } from "./archive/sevenz-reader.js";
+import { getActiveAccounts } from "./db/queries.js";
+
+const log = childLogger("manual-upload");
+
+export async function processManualUpload(uploadId: string): Promise<void> {
+  log.info({ uploadId }, "Processing manual upload");
+
+  const upload = await db.manualUpload.findUnique({
+    where: { id: uploadId },
+    include: { files: true },
+  });
+
+  if (!upload || upload.status !== "PENDING") {
+    log.warn({ uploadId }, "Manual upload not found or not pending");
+    return;
+  }
+
+  await db.manualUpload.update({
+    where: { id: uploadId },
+    data: { status: "PROCESSING" },
+  });
+
+  try {
+    // Get destination channel
+    const destSetting = await db.globalSetting.findUnique({
+      where: { key: "destination_channel_id" },
+    });
+    if (!destSetting) throw new Error("No destination channel configured");
+
+    const destChannel = await db.telegramChannel.findFirst({
+      where: { id: destSetting.value, type: "DESTINATION", isActive: true },
+    });
+    if (!destChannel) throw new Error("Destination channel not found or inactive");
+
+    // Get a TDLib client (use first active account)
+    const accounts = await getActiveAccounts();
+    const account = accounts[0];
+    if (!account) throw new Error("No authenticated Telegram account available");
+
+    const { client } = await createTdlibClient({ id: account.id, phone: account.phone });
+
+    try {
+      const packageIds: string[] = [];
+
+      for (const file of upload.files) {
+        try {
+          const filePath = file.filePath;
+          const fileName = file.fileName;
+          const fileSize = file.fileSize;
+
+          log.info({ fileName, fileSize: Number(fileSize) }, "Processing file");
+
+          // Determine archive type
+          let archiveType: "ZIP" | "RAR" | "SEVEN_Z" | "DOCUMENT" = "DOCUMENT";
+          const ext = fileName.toLowerCase();
+          if (ext.endsWith(".zip")) archiveType = "ZIP";
+          else if (ext.endsWith(".rar")) archiveType = "RAR";
+          else if (ext.endsWith(".7z")) archiveType = "SEVEN_Z";
+
+          // Hash the file
+          const contentHash = await hashParts([filePath]);
+
+          // Check for duplicates
+          const existing = await db.package.findFirst({
+            where: { contentHash, destMessageId: { not: null } },
+            select: { id: true },
+          });
+
+          if (existing) {
+            log.info({ fileName, contentHash }, "Duplicate file, skipping upload");
+            await db.manualUploadFile.update({
+              where: { id: file.id },
+              data: { packageId: existing.id },
+            });
+            packageIds.push(existing.id);
+            continue;
+          }
+
+          // Read archive metadata
+          let entries: {
+            path: string;
+            fileName: string;
+            extension: string | null;
+            compressedSize: bigint;
+            uncompressedSize: bigint;
+            crc32: string | null;
+          }[] = [];
+          try {
+            if (archiveType === "ZIP") entries = await readZipCentralDirectory([filePath]);
+            else if (archiveType === "RAR") entries = await readRarContents(filePath);
+            else if (archiveType === "SEVEN_Z") entries = await read7zContents(filePath);
+          } catch {
+            log.debug({ fileName }, "Could not read archive metadata");
+          }
+
+          // Split if needed
+          const MAX_UPLOAD_SIZE = BigInt(config.maxPartSizeMB) * 1024n * 1024n;
+          let uploadPaths = [filePath];
+          if (fileSize > MAX_UPLOAD_SIZE) {
+            uploadPaths = await byteLevelSplit(filePath);
+          }
+
+          // Upload to Telegram
+          const destResult = await uploadToChannel(
+            client,
+            destChannel.telegramId,
+            uploadPaths
+          );
+
+          // Create package record
+          const pkg = await db.package.create({
+            data: {
+              contentHash,
+              fileName,
+              fileSize,
+              archiveType,
+              sourceChannelId: destChannel.id,
+              sourceMessageId: destResult.messageId,
+              destChannelId: destChannel.id,
+              destMessageId: destResult.messageId,
+              destMessageIds: destResult.messageIds,
+              isMultipart: uploadPaths.length > 1,
+              partCount: uploadPaths.length,
+              fileCount: entries.length,
+              files: entries.length > 0 ? { create: entries } : undefined,
+            },
+          });
+
+          await db.manualUploadFile.update({
+            where: { id: file.id },
+            data: { packageId: pkg.id },
+          });
+
+          packageIds.push(pkg.id);
+          log.info({ fileName, packageId: pkg.id }, "File processed and uploaded");
+
+          // Clean up split files (but not the original)
+          if (uploadPaths.length > 1) {
+            for (const splitPath of uploadPaths) {
+              if (splitPath !== filePath) {
+                await rm(splitPath, { force: true }).catch(() => {});
+              }
+            }
+          }
+        } catch (fileErr) {
+          log.error({ err: fileErr, fileName: file.fileName }, "Failed to process file");
+        }
+      }
+
+      // Group packages if multiple files
+      if (packageIds.length >= 2) {
+        const groupName =
+          upload.groupName ?? upload.files[0].fileName.replace(/\.[^.]+$/, "");
+        const group = await db.packageGroup.create({
+          data: {
+            name: groupName,
+            sourceChannelId: destChannel.id,
+            groupingSource: "MANUAL",
+          },
+        });
+        await db.package.updateMany({
+          where: { id: { in: packageIds } },
+          data: { packageGroupId: group.id },
+        });
+        log.info(
+          { groupId: group.id, groupName, packageCount: packageIds.length },
+          "Created group for uploaded files"
+        );
+      }
+
+      await db.manualUpload.update({
+        where: { id: uploadId },
+        data: { status: "COMPLETED", completedAt: new Date() },
+      });
+
+      log.info(
+        { uploadId, fileCount: upload.files.length, packageCount: packageIds.length },
+        "Manual upload completed"
+      );
+    } finally {
+      await closeTdlibClient(client);
+    }
+  } catch (err) {
+    const message = err instanceof Error ? err.message : String(err);
+    log.error({ err, uploadId }, "Manual upload failed");
+    await db.manualUpload.update({
+      where: { id: uploadId },
+      data: { status: "FAILED", errorMessage: message },
+    });
+  }
+
+  // Clean up uploaded files
+  try {
+    const uploadDir = path.join("/data/uploads", uploadId);
+    await rm(uploadDir, { recursive: true, force: true });
+  } catch {
+    // Best-effort cleanup
+  }
+}

worker/src/rebuild.ts

@@ -63,7 +63,7 @@ export async function rebuildPackageDatabase(
     }
 
     const account = accounts[0];
-    const client = await createTdlibClient({
+    const { client } = await createTdlibClient({
       id: account.id,
       phone: account.phone,
     });

worker/src/recovery.ts

@@ -63,7 +63,7 @@ export async function recoverIncompleteUploads(): Promise<void> {
   let client: Client | undefined;
 
   try {
-    client = await createTdlibClient({ id: account.id, phone: account.phone });
+    ({ client } = await createTdlibClient({ id: account.id, phone: account.phone }));
 
     // Load the chat list so TDLib can resolve chat IDs
     try {

worker/src/scheduler.ts

@@ -1,6 +1,6 @@
 import { config } from "./util/config.js";
 import { childLogger } from "./util/logger.js";
-import { withTdlibMutex } from "./util/mutex.js";
+import { withTdlibMutex, forceReleaseMutex } from "./util/mutex.js";
 import { getActiveAccounts, getPendingAccounts } from "./db/queries.js";
 import { runWorkerForAccount, authenticateAccount } from "./worker.js";
 import { runIntegrityAudit } from "./audit.js";
@@ -24,8 +24,8 @@ const CYCLE_TIMEOUT_MS = (parseInt(process.env.WORKER_CYCLE_TIMEOUT_MINUTES ?? "
  * 1. Authenticate any PENDING accounts (triggers SMS code flow + auto-fetch channels)
  * 2. Process all active AUTHENTICATED accounts for ingestion
  *
- * All TDLib operations are wrapped in the mutex to ensure only one client
- * runs at a time (also shared with the fetch listener for on-demand requests).
+ * Each account's TDLib operations are wrapped in a per-key mutex so different
+ * accounts run concurrently while the same account is still serialized.
  *
  * The cycle has a configurable timeout (WORKER_CYCLE_TIMEOUT_MINUTES, default 4h).
  * Once the timeout elapses, no new accounts will be started but any in-progress
@@ -55,7 +55,7 @@ async function runCycle(): Promise<void> {
           log.warn("Cycle timeout reached during authentication phase, stopping");
           break;
         }
-        await withTdlibMutex(`auth:${account.phone}`, () =>
+        await withTdlibMutex(account.phone, `auth:${account.phone}`, () =>
           authenticateAccount(account)
         );
       }
@@ -71,17 +71,38 @@ async function runCycle(): Promise<void> {
 
     log.info({ accountCount: accounts.length }, "Processing accounts");
 
-    for (const account of accounts) {
-      if (Date.now() - cycleStart > CYCLE_TIMEOUT_MS) {
-        log.warn(
-          { elapsed: Math.round((Date.now() - cycleStart) / 60_000), timeoutMinutes: CYCLE_TIMEOUT_MS / 60_000 },
-          "Cycle timeout reached, skipping remaining accounts"
+    const results = await Promise.allSettled(
+      accounts.map((account) => {
+        let timer: ReturnType<typeof setTimeout>;
+        return Promise.race([
+          withTdlibMutex(account.phone, `ingest:${account.phone}`, () =>
+            runWorkerForAccount(account)
+          ),
+          new Promise<never>((_, reject) => {
+            timer = setTimeout(
+              () => reject(new Error(`Account ${account.phone} ingestion timed out after ${CYCLE_TIMEOUT_MS / 60_000}min`)),
+              CYCLE_TIMEOUT_MS
+            );
+          }),
+        ]).finally(() => clearTimeout(timer));
+      })
+    );
+
+    for (let i = 0; i < results.length; i++) {
+      if (results[i].status === "rejected") {
+        const reason = (results[i] as PromiseRejectedResult).reason;
+        log.error(
+          { phone: accounts[i].phone, err: reason },
+          "Account ingestion failed"
         );
-        break;
+        // If the cycle timed out, force-release the mutex so the next cycle
+        // (or other operations like fetch-channels) can proceed immediately
+        // instead of waiting 30 minutes for the mutex timeout.
+        const errMsg = reason instanceof Error ? reason.message : String(reason);
+        if (errMsg.includes("timed out") || errMsg.includes("mutex wait timeout")) {
+          forceReleaseMutex(accounts[i].phone);
+        }
       }
-      await withTdlibMutex(`ingest:${account.phone}`, () =>
-        runWorkerForAccount(account)
-      );
     }
 
     log.info(

worker/src/tdlib/client.ts

@@ -6,6 +6,7 @@ import { childLogger } from "../util/logger.js";
 import {
   updateAccountAuthState,
   getAccountAuthCode,
+  updateAccountPremiumStatus,
 } from "../db/queries.js";
 
 const log = childLogger("tdlib-client");
@@ -27,7 +28,7 @@ interface AccountConfig {
  */
 export async function createTdlibClient(
   account: AccountConfig
-): Promise<Client> {
+): Promise<{ client: Client; isPremium: boolean }> {
   const dbPath = path.join(config.tdlibStateDir, account.id);
 
   const client = createClient({
@@ -78,7 +79,30 @@ export async function createTdlibClient(
 
     await updateAccountAuthState(account.id, "AUTHENTICATED");
     log.info({ accountId: account.id }, "TDLib client authenticated");
-    return client;
+
+    let isPremium = false;
+    try {
+      const me = await client.invoke({ _: "getMe" }) as { is_premium?: boolean };
+      isPremium = me.is_premium ?? false;
+      await updateAccountPremiumStatus(account.id, isPremium);
+      log.info({ accountId: account.id, isPremium }, "Account Premium status detected");
+    } catch (err) {
+      log.warn({ err, accountId: account.id }, "Could not detect Premium status, defaulting to false");
+    }
+
+    client.on("update", (update: unknown) => {
+      const u = update as { _?: string; is_upload?: boolean };
+      if (u?._ === "updateSpeedLimitNotification") {
+        log.warn(
+          { accountId: account.id, isUpload: u.is_upload },
+          u.is_upload
+            ? "Upload speed limited by Telegram (account is not Premium)"
+            : "Download speed limited by Telegram (account is not Premium)"
+        );
+      }
+    });
+
+    return { client, isPremium };
   } catch (err) {
     log.error({ err, accountId: account.id }, "TDLib authentication failed");
     await updateAccountAuthState(account.id, "EXPIRED");

worker/src/tdlib/download.ts

@@ -79,6 +79,8 @@ export interface ChannelScanResult {
   archives: TelegramMessage[];
   photos: TelegramPhoto[];
   totalScanned: number;
+  /** Highest message ID seen during scan (for watermark, even when no archives found). */
+  maxScannedMessageId: bigint | null;
 }
 
 export type ScanProgressCallback = (messagesScanned: number) => void;
@@ -158,6 +160,7 @@ export async function getChannelMessages(
   const archives: TelegramMessage[] = [];
   const photos: TelegramPhoto[] = [];
   const boundary = lastProcessedMessageId ? Number(lastProcessedMessageId) : null;
+  let maxScannedMessageId: bigint | null = null;
 
   // Open the chat so TDLib can access it
   try {
@@ -204,6 +207,12 @@ export async function getChannelMessages(
 
       totalScanned += result.messages.length;
 
+      // Track highest message ID (first message in batch = newest, since results are newest-first)
+      const batchMaxId = BigInt(result.messages[0].id);
+      if (maxScannedMessageId === null || batchMaxId > maxScannedMessageId) {
+        maxScannedMessageId = batchMaxId;
+      }
+
       for (const msg of result.messages) {
         // Check for archive documents
         const doc = msg.content?.document;
@@ -246,6 +255,11 @@ export async function getChannelMessages(
       fromMessageId = result.messages[result.messages.length - 1].id;
       if (result.messages.length < Math.min(limit, 100)) break;
 
+      // Early exit: searchChatMessages returns newest-first. Once the oldest
+      // message on this page is at or below the boundary, all remaining pages
+      // are even older — no new messages exist, stop scanning immediately.
+      if (boundary && fromMessageId <= boundary) break;
+
       await sleep(config.apiDelayMs);
     }
   }
@@ -266,6 +280,7 @@ export async function getChannelMessages(
     archives: archives.reverse(),
     photos: photos.reverse(),
     totalScanned,
+    maxScannedMessageId,
   };
 }
 

worker/src/tdlib/topics.ts

@@ -178,6 +178,7 @@ export async function getTopicMessages(
   const archives: TelegramMessage[] = [];
   const photos: TelegramPhoto[] = [];
   const boundary = lastProcessedMessageId ? Number(lastProcessedMessageId) : null;
+  let maxScannedMessageId: bigint | null = null;
 
   let currentFromId = 0;
   let totalScanned = 0;
@@ -239,6 +240,12 @@ export async function getTopicMessages(
 
     totalScanned += result.messages.length;
 
+    // Track highest message ID (first message = newest, since results are newest-first)
+    const batchMaxId = BigInt(result.messages[0].id);
+    if (maxScannedMessageId === null || batchMaxId > maxScannedMessageId) {
+      maxScannedMessageId = batchMaxId;
+    }
+
     for (const msg of result.messages) {
       // Check for archive documents
       const doc = msg.content?.document;
@@ -302,6 +309,7 @@ export async function getTopicMessages(
     archives: archives.reverse(),
     photos: photos.reverse(),
     totalScanned,
+    maxScannedMessageId,
   };
 }
 

worker/src/upload/channel.ts

@@ -7,6 +7,18 @@ import { withFloodWait, extractFloodWaitSeconds } from "../util/retry.js";
 
 const log = childLogger("upload");
 
+/**
+ * Custom error class to distinguish upload stalls from other errors.
+ * When consecutive stalls occur, the caller can use this signal to
+ * recreate the TDLib client (whose event stream may have degraded).
+ */
+export class UploadStallError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = "UploadStallError";
+  }
+}
+
 export interface UploadResult {
   messageId: bigint;
   messageIds: bigint[];
@@ -109,13 +121,21 @@ async function sendWithRetry(
 
       // Stall or timeout — retry with a cooldown
       const errMsg = err instanceof Error ? err.message : "";
-      if ((errMsg.includes("stalled") || errMsg.includes("timed out")) && !isLastAttempt) {
-        log.warn(
-          { fileName, attempt: attempt + 1, maxRetries: MAX_UPLOAD_RETRIES },
-          "Upload stalled/timed out — retrying"
+      if (errMsg.includes("stalled") || errMsg.includes("timed out")) {
+        if (!isLastAttempt) {
+          log.warn(
+            { fileName, attempt: attempt + 1, maxRetries: MAX_UPLOAD_RETRIES },
+            "Upload stalled/timed out — retrying"
+          );
+          await sleep(10_000);
+          continue;
+        }
+        // All stall retries exhausted — throw UploadStallError so the caller
+        // knows the TDLib client's event stream is likely degraded and can
+        // recreate the client before continuing.
+        throw new UploadStallError(
+          `Upload stalled after ${MAX_UPLOAD_RETRIES} retries for ${fileName}`
         );
-        await sleep(10_000);
-        continue;
       }
 
       throw err;
@@ -166,8 +186,10 @@ async function sendAndWaitForUpload(
       }
     }, timeoutMs);
 
-    // Stall detection: no progress for 5 minutes after upload started → reject
-    const STALL_TIMEOUT_MS = 5 * 60_000;
+    // Stall detection: no progress for 3 minutes after upload started → reject
+    // (reduced from 5min — once data is fully sent, confirmation should arrive quickly;
+    //  a 3min silence strongly indicates a degraded TDLib event stream)
+    const STALL_TIMEOUT_MS = 3 * 60_000;
     const stallChecker = setInterval(() => {
       if (settled || !uploadStarted) return;
       const stallMs = Date.now() - lastProgressTime;

worker/src/util/mutex.ts

@@ -2,39 +2,66 @@ import { childLogger } from "./logger.js";
 
 const log = childLogger("mutex");
 
-let locked = false;
-let holder = "";
-const queue: Array<{ resolve: () => void; reject: (err: Error) => void; label: string }> = [];
-
-/**
- * Maximum time to wait for the TDLib mutex (ms).
- * If the mutex is not available within this time, the operation is rejected.
- * Default: 30 minutes (long enough for large downloads, short enough to detect hangs).
- */
 const MUTEX_WAIT_TIMEOUT_MS = 30 * 60 * 1000;
 
+const locks = new Map<string, boolean>();
+const holders = new Map<string, string>();
+const queues = new Map<
+  string,
+  Array<{ resolve: () => void; reject: (err: Error) => void; label: string }>
+>();
+
 /**
- * Ensures only one TDLib client runs at a time across the entire worker process.
- * Both the scheduler (auth, ingestion) and the fetch listener acquire this
- * before creating any TDLib client.
+ * Force-release a stuck mutex.
+ * This should only be called when the holder is known to be stuck (e.g. after
+ * a cycle timeout). It releases the lock and lets the next queued waiter proceed.
+ */
+export function forceReleaseMutex(key: string): void {
+  if (!locks.has(key)) return;
+
+  const holder = holders.get(key);
+  log.warn({ key, holder }, "Force-releasing stuck TDLib mutex");
+
+  locks.delete(key);
+  holders.delete(key);
+  const next = queues.get(key)?.shift();
+  if (next) {
+    log.info({ key, next: next.label }, "TDLib mutex force-released to next waiter");
+    next.resolve();
+  } else {
+    queues.delete(key);
+    log.info({ key }, "TDLib mutex force-released (no waiters)");
+  }
+}
+
+/**
+ * Ensures only one TDLib operation runs at a time FOR THE SAME KEY.
+ * Different keys run concurrently — this allows two accounts to ingest in parallel
+ * while still preventing concurrent use of the same account's TDLib state dir.
  *
- * Includes a wait timeout to prevent indefinite blocking if the current holder hangs.
+ * key:   the account phone number for account-specific ops (auth, ingest),
+ *        or 'global' for ops that don't belong to a specific account.
+ * label: human-readable name for logging.
  */
 export async function withTdlibMutex<T>(
+  key: string,
   label: string,
   fn: () => Promise<T>
 ): Promise<T> {
-  if (locked) {
-    log.info({ waiting: label, holder }, "Waiting for TDLib mutex");
+  if (locks.get(key)) {
+    log.info({ waiting: label, key, holder: holders.get(key) }, "Waiting for TDLib mutex");
     await new Promise<void>((resolve, reject) => {
       const timer = setTimeout(() => {
-        const idx = queue.indexOf(entry);
+        const q = queues.get(key) ?? [];
+        const idx = q.indexOf(entry);
         if (idx !== -1) {
-          queue.splice(idx, 1);
-          reject(new Error(
-            `TDLib mutex wait timeout after ${MUTEX_WAIT_TIMEOUT_MS / 60_000}min ` +
-            `(waiting: ${label}, holder: ${holder})`
-          ));
+          q.splice(idx, 1);
+          reject(
+            new Error(
+              `TDLib mutex wait timeout after ${MUTEX_WAIT_TIMEOUT_MS / 60_000}min ` +
+                `(waiting: ${label}, key: ${key}, holder: ${holders.get(key)})`
+            )
+          );
         }
       }, MUTEX_WAIT_TIMEOUT_MS);
 
@@ -46,25 +73,28 @@ export async function withTdlibMutex<T>(
         reject,
         label,
       };
-      queue.push(entry);
+
+      if (!queues.has(key)) queues.set(key, []);
+      queues.get(key)!.push(entry);
     });
   }
 
-  locked = true;
-  holder = label;
-  log.debug({ label }, "TDLib mutex acquired");
+  locks.set(key, true);
+  holders.set(key, label);
+  log.debug({ key, label }, "TDLib mutex acquired");
 
   try {
     return await fn();
   } finally {
-    locked = false;
-    holder = "";
-    const next = queue.shift();
+    locks.delete(key);
+    holders.delete(key);
+    const next = queues.get(key)?.shift();
     if (next) {
-      log.debug({ next: next.label }, "TDLib mutex releasing to next waiter");
+      log.debug({ key, next: next.label }, "TDLib mutex releasing to next waiter");
       next.resolve();
     } else {
-      log.debug({ label }, "TDLib mutex released");
+      queues.delete(key);
+      log.debug({ key, label }, "TDLib mutex released");
     }
   }
 }

worker/src/worker.ts

@@ -2,13 +2,14 @@ import path from "path";
 import { unlink, readdir, mkdir, rm } from "fs/promises";
 import { config } from "./util/config.js";
 import { childLogger } from "./util/logger.js";
-import { tryAcquireLock, releaseLock } from "./db/locks.js";
+import { tryAcquireLock, releaseLock, tryAcquireHashLock, releaseHashLock } from "./db/locks.js";
 import {
   getSourceChannelMappings,
   getGlobalDestinationChannel,
   packageExistsByHash,
   packageExistsBySourceMessage,
-  createPackageWithFiles,
+  createPackageStub,
+  updatePackageWithMetadata,
   createIngestionRun,
   completeIngestionRun,
   failIngestionRun,
@@ -46,7 +47,7 @@ import { readZipCentralDirectory } from "./archive/zip-reader.js";
 import { readRarContents } from "./archive/rar-reader.js";
 import { read7zContents } from "./archive/sevenz-reader.js";
 import { byteLevelSplit, concatenateFiles } from "./archive/split.js";
-import { uploadToChannel } from "./upload/channel.js";
+import { uploadToChannel, UploadStallError } from "./upload/channel.js";
 import { processAlbumGroups, processRuleBasedGroups, processTimeWindowGroups, processPatternGroups, processCreatorGroups, processZipPathGroups, processReplyChainGroups, processCaptionGroups, detectGroupingConflicts, type IndexedPackageRef } from "./grouping.js";
 import { db } from "./db/client.js";
 import type { TelegramAccount, TelegramChannel } from "@prisma/client";
@@ -73,10 +74,10 @@ export async function authenticateAccount(
 
   let client: Client | undefined;
   try {
-    client = await createTdlibClient({
+    client = (await createTdlibClient({
       id: account.id,
       phone: account.phone,
-    });
+    })).client;
     aLog.info("Authentication successful");
 
     // Auto-fetch channels and create a fetch request result
@@ -131,7 +132,7 @@ export async function processFetchRequest(requestId: string): Promise<void> {
   await updateFetchRequestStatus(requestId, "IN_PROGRESS");
   aLog.info({ accountId: request.accountId }, "Processing fetch request");
 
-  const client = await createTdlibClient({
+  const { client } = await createTdlibClient({
     id: request.account.id,
     phone: request.account.phone,
   });
@@ -285,6 +286,7 @@ interface PipelineContext {
   client: Client;
   runId: string;
   accountId: string;
+  accountPhone: string;
   channelTitle: string;
   channel: TelegramChannel;
   destChannelTelegramId: bigint;
@@ -301,6 +303,9 @@ interface PipelineContext {
   /** Forum topic ID (null for non-forum). */
   sourceTopicId: bigint | null;
   accountLog: ReturnType<typeof childLogger>;
+  maxUploadSize: bigint;
+  /** How many consecutive upload stalls have occurred (resets on success). */
+  consecutiveStalls: number;
 }
 
 /**
@@ -336,10 +341,14 @@ export async function runWorkerForAccount(
       currentStep: "connecting",
     });
 
-    const client = await createTdlibClient({
+    // Use let so the client can be replaced on TDLib recreation after stalls
+    let { client, isPremium } = await createTdlibClient({
       id: account.id,
       phone: account.phone,
     });
+    const maxUploadSize = isPremium
+      ? 3950n * 1024n * 1024n
+      : BigInt(config.maxPartSizeMB) * 1024n * 1024n;
 
     // Load all chats into TDLib's local cache using loadChats (the recommended API).
     // Without this, getChat/searchChatMessages fail with "Chat not found".
@@ -443,6 +452,7 @@ export async function runWorkerForAccount(
           client,
           runId: activeRunId,
           accountId: account.id,
+          accountPhone: account.phone,
           channelTitle: channel.title,
           channel,
           destChannelTelegramId: destChannel.telegramId,
@@ -452,6 +462,8 @@ export async function runWorkerForAccount(
           topicCreator: null,
           sourceTopicId: null,
           accountLog,
+          maxUploadSize,
+          consecutiveStalls: 0,
         };
 
         if (forum) {
@@ -526,6 +538,15 @@ 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
+                if (scanResult.maxScannedMessageId) {
+                  await upsertTopicProgress(
+                    mapping.id,
+                    topic.topicId,
+                    topic.name,
+                    scanResult.maxScannedMessageId
+                  );
+                }
                 continue;
               }
 
@@ -540,14 +561,17 @@ export async function runWorkerForAccount(
               pipelineCtx.channelTitle = `${channel.title} › ${topic.name}`;
 
               const maxProcessedId = await processArchiveSets(pipelineCtx, scanResult, run.id, progress?.lastProcessedMessageId);
+              // Sync client back in case it was recreated during upload stall recovery
+              client = pipelineCtx.client;
 
-              // Only advance progress to the highest successfully processed message
-              if (maxProcessedId) {
+              // Advance progress: use archive watermark if available, fall back to scan watermark
+              const topicWatermark = maxProcessedId ?? scanResult.maxScannedMessageId;
+              if (topicWatermark) {
                 await upsertTopicProgress(
                   mapping.id,
                   topic.topicId,
                   topic.name,
-                  maxProcessedId
+                  topicWatermark
                 );
               }
             } catch (topicErr) {
@@ -597,6 +621,11 @@ 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
+            if (scanResult.maxScannedMessageId) {
+              await updateLastProcessedMessage(mapping.id, scanResult.maxScannedMessageId);
+            }
             continue;
           }
 
@@ -611,10 +640,13 @@ export async function runWorkerForAccount(
           pipelineCtx.channelTitle = channel.title;
 
           const maxProcessedId = await processArchiveSets(pipelineCtx, scanResult, run.id, mapping.lastProcessedMessageId);
+          // Sync client back in case it was recreated during upload stall recovery
+          client = pipelineCtx.client;
 
-          // Only advance progress to the highest successfully processed message
-          if (maxProcessedId) {
-            await updateLastProcessedMessage(mapping.id, maxProcessedId);
+          // Advance progress: use archive watermark if available, fall back to scan watermark
+          const channelWatermark = maxProcessedId ?? scanResult.maxScannedMessageId;
+          if (channelWatermark) {
+            await updateLastProcessedMessage(mapping.id, channelWatermark);
           }
         }
         } catch (channelErr) {
@@ -754,12 +786,68 @@ async function processArchiveSets(
       if (setMaxId > (maxProcessedId ?? 0n)) {
         maxProcessedId = setMaxId;
       }
+
+      // Reset stall counter on any successful upload
+      ctx.consecutiveStalls = 0;
     } catch (setErr) {
       // If a set fails, do NOT advance the watermark past it
       accountLog.warn(
         { err: setErr, baseName: archiveSets[setIdx].baseName },
         "Archive set failed, watermark will not advance past this set"
       );
+
+      // ── TDLib client recreation on repeated upload stalls ──
+      // When the TDLib event stream degrades, uploads complete (bytes sent)
+      // but confirmations never arrive. Retrying with the same broken client
+      // is futile. Recreate the client to get a fresh connection.
+      if (setErr instanceof UploadStallError) {
+        ctx.consecutiveStalls++;
+        accountLog.warn(
+          { consecutiveStalls: ctx.consecutiveStalls },
+          "Upload stall detected — TDLib event stream may be degraded"
+        );
+
+        // After 1 stalled set (= 3 failed retry attempts already), recreate the client
+        if (ctx.consecutiveStalls >= 1) {
+          accountLog.info("Recreating TDLib client after consecutive upload stalls");
+          try {
+            await closeTdlibClient(ctx.client);
+          } catch (closeErr) {
+            accountLog.warn({ err: closeErr }, "Error closing stale TDLib client");
+          }
+
+          try {
+            const { client: newClient } = await createTdlibClient({
+              id: ctx.accountId,
+              phone: ctx.accountPhone,
+            });
+            ctx.client = newClient;
+
+            // Reload chats so the new client can access channels
+            try {
+              for (let page = 0; page < 500; page++) {
+                await newClient.invoke({
+                  _: "loadChats",
+                  chat_list: { _: "chatListMain" },
+                  limit: 100,
+                });
+              }
+            } catch {
+              // 404 = all loaded (expected)
+            }
+
+            ctx.consecutiveStalls = 0;
+            accountLog.info("TDLib client recreated successfully — continuing ingestion");
+          } catch (recreateErr) {
+            accountLog.error(
+              { err: recreateErr },
+              "Failed to recreate TDLib client — aborting remaining uploads"
+            );
+            break;
+          }
+        }
+      }
+
       // Record the failure for visibility in the UI
       try {
         const archiveSet = archiveSets[setIdx];
@@ -1027,6 +1115,35 @@ async function processOneArchiveSet(
       return null;
     }
 
+    // ── Hash lock: prevent concurrent workers racing on shared-channel archives ──
+    const hashLockAcquired = await tryAcquireHashLock(contentHash);
+    if (!hashLockAcquired) {
+      counters.zipsDuplicate++;
+      accountLog.info(
+        { fileName: archiveName, hash: contentHash.slice(0, 16) },
+        "Hash lock held by another worker — skipping concurrent duplicate"
+      );
+      return null;
+    }
+
+    let entries: { path: string; fileName: string; extension: string | null; compressedSize: bigint; uncompressedSize: bigint; crc32: string | null }[] = [];
+    let creator: string | null = null;
+    const tags: string[] = [];
+    let stub: { id: string } | null = null;
+
+    try {
+    // Re-check after acquiring lock: another worker may have finished between
+    // the first check above and this point.
+    const existsAfterLock = await packageExistsByHash(contentHash);
+    if (existsAfterLock) {
+      counters.zipsDuplicate++;
+      accountLog.debug(
+        { fileName: archiveName, hash: contentHash.slice(0, 16) },
+        "Duplicate detected after acquiring hash lock — skipping"
+      );
+      return null;
+    }
+
     // ── Reading metadata ──
     await updateRunActivity(runId, {
       currentActivity: `Reading file list from ${archiveName}`,
@@ -1037,7 +1154,6 @@ async function processOneArchiveSet(
       totalFiles: totalSets,
     });
 
-    let entries: { path: string; fileName: string; extension: string | null; compressedSize: bigint; uncompressedSize: bigint; crc32: string | null }[] = [];
     try {
       if (archiveSet.type === "ZIP") {
         entries = await readZipCentralDirectory(tempPaths);
@@ -1069,7 +1185,7 @@ async function processOneArchiveSet(
       (sum, p) => sum + p.fileSize,
       0n
     );
-    const MAX_UPLOAD_SIZE = BigInt(config.maxPartSizeMB) * 1024n * 1024n;
+    const MAX_UPLOAD_SIZE = ctx.maxUploadSize;
     const hasOversizedPart = archiveSet.parts.some((p) => p.fileSize > MAX_UPLOAD_SIZE);
 
     if (hasOversizedPart) {
@@ -1084,7 +1200,7 @@ async function processOneArchiveSet(
       });
       const concatPath = path.join(setDir, `${archiveSet.baseName}.concat`);
       await concatenateFiles(tempPaths, concatPath);
-      splitPaths = await byteLevelSplit(concatPath);
+      splitPaths = await byteLevelSplit(concatPath, ctx.maxUploadSize);
       uploadPaths = splitPaths;
       // Clean up the concat intermediate file
       await unlink(concatPath).catch(() => {});
@@ -1098,7 +1214,7 @@ async function processOneArchiveSet(
         currentFileNum: setIdx + 1,
         totalFiles: totalSets,
       });
-      splitPaths = await byteLevelSplit(tempPaths[0]);
+      splitPaths = await byteLevelSplit(tempPaths[0], ctx.maxUploadSize);
       uploadPaths = splitPaths;
     }
 
@@ -1139,72 +1255,112 @@ async function processOneArchiveSet(
       );
     }
 
-    // ── Uploading ──
-    // Check if a prior run already uploaded this file (orphaned upload scenario:
-    // file reached Telegram but DB write failed or worker crashed before indexing)
-    const existingUpload = await getUploadedPackageByHash(contentHash);
-    let destResult: { messageId: bigint; messageIds: bigint[] };
+      // ── Uploading ──
+      // Check if a prior run already uploaded this file (orphaned upload scenario:
+      // file reached Telegram but DB write failed or worker crashed before indexing)
+      const existingUpload = await getUploadedPackageByHash(contentHash);
+      let destResult: { messageId: bigint; messageIds: bigint[] };
 
-    if (existingUpload && existingUpload.destMessageId) {
-      accountLog.info(
-        { fileName: archiveName, destMessageId: Number(existingUpload.destMessageId) },
-        "Reusing existing upload (file already on destination channel)"
-      );
-      destResult = {
-        messageId: existingUpload.destMessageId,
-        messageIds: existingUpload.destMessageIds?.length
-          ? (existingUpload.destMessageIds as bigint[])
-          : [existingUpload.destMessageId],
-      };
-    } else {
-      const uploadLabel = uploadPaths.length > 1
-        ? ` (${uploadPaths.length} parts)`
-        : "";
-      await updateRunActivity(runId, {
-        currentActivity: `Uploading ${archiveName} to archive channel${uploadLabel}`,
-        currentStep: "uploading",
-        currentChannel: channelTitle,
-        currentFile: archiveName,
-        currentFileNum: setIdx + 1,
-        totalFiles: totalSets,
+      if (existingUpload && existingUpload.destMessageId) {
+        accountLog.info(
+          { fileName: archiveName, destMessageId: Number(existingUpload.destMessageId) },
+          "Reusing existing upload (file already on destination channel)"
+        );
+        destResult = {
+          messageId: existingUpload.destMessageId,
+          messageIds: existingUpload.destMessageIds?.length
+            ? (existingUpload.destMessageIds as bigint[])
+            : [existingUpload.destMessageId],
+        };
+      } else {
+        const uploadLabel = uploadPaths.length > 1
+          ? ` (${uploadPaths.length} parts)`
+          : "";
+        await updateRunActivity(runId, {
+          currentActivity: `Uploading ${archiveName} to archive channel${uploadLabel}`,
+          currentStep: "uploading",
+          currentChannel: channelTitle,
+          currentFile: archiveName,
+          currentFileNum: setIdx + 1,
+          totalFiles: totalSets,
+        });
+
+        destResult = await uploadToChannel(
+          client,
+          destChannelTelegramId,
+          uploadPaths
+        );
+      }
+
+      // ── Post-upload integrity check ──
+      // Verify the files on disk still match before we index
+      if (uploadPaths.length > 0 && !existingUpload) {
+        try {
+          const postUploadHash = await hashParts(uploadPaths);
+          if (splitPaths.length > 0) {
+            // Split files — hash should match the split hash (already verified above)
+            // No additional check needed since we verified split hash = original hash
+          } else if (postUploadHash !== contentHash) {
+            accountLog.error(
+              { fileName: archiveName, originalHash: contentHash, postUploadHash },
+              "Hash changed between hashing and upload — possible disk corruption"
+            );
+            await db.systemNotification.create({
+              data: {
+                type: "HASH_MISMATCH",
+                severity: "ERROR",
+                title: `Post-upload hash mismatch: ${archiveName}`,
+                message: `Hash changed between download and upload. Original: ${contentHash.slice(0, 16)}…, post-upload: ${postUploadHash.slice(0, 16)}…`,
+                context: { fileName: archiveName, originalHash: contentHash, postUploadHash, sourceChannelId: channel.id },
+              },
+            });
+          }
+        } catch {
+          // Best-effort — don't fail the ingestion
+        }
+      }
+
+      // ── Phase 1: Stub record — persisted immediately after upload ──
+      await deleteOrphanedPackageByHash(contentHash);
+
+      creator =
+        topicCreator ??
+        extractCreatorFromFileName(archiveName) ??
+        extractCreatorFromChannelTitle(channelTitle) ??
+        null;
+
+      if (channel.category) {
+        tags.push(channel.category);
+      }
+
+      stub = await createPackageStub({
+        contentHash,
+        fileName: archiveName,
+        fileSize: totalSize,
+        archiveType: archiveSet.type === "7Z" ? "SEVEN_Z" : archiveSet.type,
+        sourceChannelId: channel.id,
+        sourceMessageId: archiveSet.parts[0].id,
+        sourceTopicId,
+        destChannelId,
+        destMessageId: destResult.messageId,
+        destMessageIds: destResult.messageIds,
+        isMultipart: archiveSet.parts.length > 1 || uploadPaths.length > 1,
+        partCount: uploadPaths.length,
+        ingestionRunId,
+        creator,
+        tags,
       });
 
-      destResult = await uploadToChannel(
-        client,
-        destChannelTelegramId,
-        uploadPaths
-      );
+      counters.zipsIngested++;
+      await deleteSkippedPackage(channel.id, archiveSet.parts[0].id);
+    } finally {
+      await releaseHashLock(contentHash);
     }
 
-    // ── Post-upload integrity check ──
-    // Verify the files on disk still match before we index
-    if (uploadPaths.length > 0 && !existingUpload) {
-      try {
-        const postUploadHash = await hashParts(uploadPaths);
-        if (splitPaths.length > 0) {
-          // Split files — hash should match the split hash (already verified above)
-          // No additional check needed since we verified split hash = original hash
-        } else if (postUploadHash !== contentHash) {
-          accountLog.error(
-            { fileName: archiveName, originalHash: contentHash, postUploadHash },
-            "Hash changed between hashing and upload — possible disk corruption"
-          );
-          await db.systemNotification.create({
-            data: {
-              type: "HASH_MISMATCH",
-              severity: "ERROR",
-              title: `Post-upload hash mismatch: ${archiveName}`,
-              message: `Hash changed between download and upload. Original: ${contentHash.slice(0, 16)}…, post-upload: ${postUploadHash.slice(0, 16)}…`,
-              context: { fileName: archiveName, originalHash: contentHash, postUploadHash, sourceChannelId: channel.id },
-            },
-          });
-        }
-      } catch {
-        // Best-effort — don't fail the ingestion
-      }
-    }
+    if (!stub) return null;
 
     // ── Preview thumbnail ──
+    // (moved here from before stub creation — lock is released, preview doesn't need it)
     let previewData: Buffer | null = null;
     let previewMsgId: bigint | null = null;
     const matchedPhoto = previewMatches.get(archiveSet.baseName);
@@ -1218,8 +1374,6 @@ async function processOneArchiveSet(
         totalFiles: totalSets,
       });
       previewData = await downloadPhotoThumbnail(client, matchedPhoto.fileId);
-      // Only set previewMsgId if we actually got the image data —
-      // otherwise the UI thinks there's a preview but the API returns 404
       if (previewData) {
         previewMsgId = matchedPhoto.id;
       }
@@ -1242,13 +1396,7 @@ async function processOneArchiveSet(
       }
     }
 
-    // ── Resolve creator: topic name > filename extraction > channel title > null ──
-    const creator = topicCreator
-      ?? extractCreatorFromFileName(archiveName)
-      ?? extractCreatorFromChannelTitle(channelTitle)
-      ?? null;
-
-    // ── Indexing ──
+    // ── Phase 2: Update stub with file entries and preview ──
     await updateRunActivity(runId, {
       currentActivity: `Saving metadata for ${archiveName} (${entries.length} files)`,
       currentStep: "indexing",
@@ -1258,43 +1406,12 @@ async function processOneArchiveSet(
       totalFiles: totalSets,
     });
 
-    // Clean up any orphaned record (same hash but no dest upload) before creating
-    await deleteOrphanedPackageByHash(contentHash);
-
-    // Auto-inherit source channel category as initial tag
-    const tags: string[] = [];
-    if (channel.category) {
-      tags.push(channel.category);
-    }
-
-    const pkg = await createPackageWithFiles({
-      contentHash,
-      fileName: archiveName,
-      fileSize: totalSize,
-      archiveType: archiveSet.type === "7Z" ? "SEVEN_Z" : archiveSet.type,
-      sourceChannelId: channel.id,
-      sourceMessageId: archiveSet.parts[0].id,
-      sourceTopicId,
-      destChannelId,
-      destMessageId: destResult.messageId,
-      destMessageIds: destResult.messageIds,
-      isMultipart:
-        archiveSet.parts.length > 1 || uploadPaths.length > 1,
-      partCount: uploadPaths.length,
-      ingestionRunId,
-      creator,
-      tags,
+    await updatePackageWithMetadata(stub.id, {
+      files: entries,
       previewData,
       previewMsgId,
-      sourceCaption: archiveSet.parts[0].caption ?? null,
-      replyToMessageId: archiveSet.parts[0].replyToMessageId ?? null,
-      files: entries,
     });
 
-    counters.zipsIngested++;
-    // Clean up any prior skip record for this archive
-    await deleteSkippedPackage(channel.id, archiveSet.parts[0].id);
-
     await updateRunActivity(runId, {
       currentActivity: `Ingested ${archiveName} (${entries.length} files indexed)`,
       currentStep: "complete",
@@ -1310,7 +1427,7 @@ async function processOneArchiveSet(
       "Archive ingested"
     );
 
-    return pkg.id;
+    return stub.id;
   } finally {
     // ALWAYS delete temp files and the set directory
     await deleteFiles([...tempPaths, ...splitPaths]);