refactor: implement per-bot queue architecture

- Each bot has its own PQueue with concurrency=1
- selectBot() assigns uploads to least-loaded available bot
- 429 rate limits are tracked per-bot with cooldown timers
- Failed uploads retry on next available bot
- Removed global upload-queue.ts and uploadConcurrency config
- Updated ITelegramService interface

docs/superpowers/plans/2026-07-29-per-bot-queue.md

@@ -0,0 +1,576 @@
+# Per-Bot Queue Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Replace global PQueue with per-bot queues (concurrency=1 per bot) to eliminate 429 collisions and improve rate-limit safety.
+
+**Architecture:** Each bot token gets its own PQueue with concurrency=1. Uploads are assigned to the least-loaded available bot via `selectBot()`. On 429, the bot is marked rate-limited and the upload retries on the next available bot. The global `upload-queue.ts` is removed; `uploadConcurrency` config is replaced by `botCount * perBotConcurrency`.
+
+**Tech Stack:** TypeScript, PQueue, Telegraf
+
+## Global Constraints
+
+- Use `Bun` runtime, not Node.js
+- Follow existing code style (Biome lint)
+- Each bot queue has concurrency=1 (no two uploads hit same bot simultaneously)
+- `selectBot()` picks bot with lowest pending queue count, skipping rate-limited bots
+- Remove `uploadConcurrency` from config; derive effective concurrency from bot count
+- Remove `upload-queue.ts` entirely
+- Remove `enqueueUpload` from `ITelegramService` interface
+
+---
+## File Structure
+
+### Files to Modify
+| File | Change |
+|------|--------|
+| `src/infrastructure/telegram/bot-pool.ts` | BotEntry array, selectBot(), per-bot queues, retry logic |
+| `src/domain/ports/telegram-service.ts` | Remove `enqueueUpload` from interface |
+| `src/env.ts` | Remove `uploadConcurrency` field |
+| `src/index.ts` | Remove upload-queue import and usage |
+| `src/utils/chunked-storage.ts` | Replace `config.uploadConcurrency` with bot count |
+
+### Files to Delete
+| File | Reason |
+|------|--------|
+| `src/infrastructure/telegram/upload-queue.ts` | Global queue replaced by per-bot queues |
+| `test/telegramQueue.test.ts` | Tests for deleted module |
+
+### Files Not Changed
+| File | Reason |
+|------|--------|
+| `test/bot.test.ts` | Only uses ITelegramService interface (via `forwardToStorage`) |
+| `src/infrastructure/telegram/chunked-storage.ts` | Uses ITelegramService interface, not BotPool directly |
+
+---
+
+### Task 1: Refresh the spec & plan files after compaction
+
+Due to context compaction, re-read the current spec and plan files to ensure accuracy before implementing.
+
+- [ ] **Step 1: Re-read the spec**
+
+Read: `docs/superpowers/specs/2026-07-29-per-bot-queue-design.md`
+
+- [ ] **Step 2: Re-read key implementation files**
+
+Read: `src/infrastructure/telegram/bot-pool.ts`, `src/env.ts`, `src/utils/chunked-storage.ts`
+
+### Task 2: Refactor ITelegramService interface
+
+Remove `enqueueUpload` from the interface — BotPool handles queueing internally now.
+
+**Files:**
+- Modify: `src/domain/ports/telegram-service.ts`
+
+- [ ] **Step 1: Remove `enqueueUpload` from interface**
+
+```typescript
+// src/domain/ports/telegram-service.ts — remove entire section:
+  /**
+   * Enqueue a task for sequential upload execution.
+   *
+   * Ensures only one Telegram upload runs at a time to avoid
+   * rate limits and resource contention.
+   *
+   * @param task - An async function performing the upload.
+   * @returns The result of the task.
+   */
+  enqueueUpload<T>(task: () => Promise<T>): Promise<T>;
+```
+
+- [ ] **Step 2: Run lint to verify**
+
+Run: `bunx biome check src/domain/ports/telegram-service.ts`
+Expected: No errors.
+
+- [ ] **Step 3: Commit**
+
+```bash
+git add src/domain/ports/telegram-service.ts
+git commit -m "refactor: remove enqueueUpload from ITelegramService
+
+Per-bot queue handles queueing internally.
+
+Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>"
+```
+
+### Task 3: Refactor BotPool with per-bot queues
+
+The core of the redesign. Replace `claimBotIndex()` round-robin with per-bot PQueue instances and `selectBot()` for least-loaded assignment.
+
+**Files:**
+- Modify: `src/infrastructure/telegram/bot-pool.ts`
+
+**Interfaces:**
+- Consumes: `ITelegramService` (no `enqueueUpload` method)
+- Produces: `botPool` singleton with per-bot queues, `selectBot()`, per-bot rate-limit tracking
+
+- [ ] **Step 1: Write test file for per-bot queue behavior**
+
+```typescript
+// test/bot-pool.test.ts
+import { beforeEach, describe, expect, it, mock, spyOn } from 'bun:test';
+
+// We'll test the BotEntry queue behavior and selectBot logic
+```
+
+- [ ] **Step 2: Implement BotEntry type and BotPool refactor**
+
+Replace the class body:
+
+```typescript
+import PQueue from 'p-queue';
+import { Telegraf } from 'telegraf';
+import type {
+  ForwardResult,
+  ITelegramService,
+  TelegramFileInfo,
+} from '../../domain/ports/telegram-service';
+import { config } from '../../env';
+import logger from '../../shared/logger/index';
+import {
+  buildSendPayload,
+  extractUploadedFile,
+  type SendMethod,
+  sendMethodMap,
+  type TelegramMessageResult,
+} from './types';
+
+const sleep = (ms: number): Promise<void> => 
+  new Promise((resolve) => setTimeout(resolve, ms));
+
+const isTransientError = (error: unknown): boolean => {
+  const str = error instanceof Error ? error.message : String(error);
+  const transientPatterns = [
+    'timeout', 'Timed out', 'etimedout', 'econnrefused', 'econnreset',
+    'ECONNREFUSED', 'ECONNRESET', 'ETIMEDOUT', '5xx', '502', '503', '504',
+    'Bad Gateway', 'Service Unavailable', 'Gateway Timeout', 'socket hang up',
+    'socket closed', 'fetch failed', 'network error', 'network timeout',
+    'API closed', 'read ECONNRESET', 'write EPIPE',
+  ];
+  return transientPatterns.some((p) => str.toLowerCase().includes(p.toLowerCase()));
+};
+
+const MAX_TRANSIENT_RETRIES = 3;
+const TELEGRAM_API_TIMEOUT_MS = 120_000;
+const PER_BOT_CONCURRENCY = 1;
+
+interface BotEntry {
+  index: number;
+  token: string;
+  instance: Telegraf;
+  queue: PQueue;
+  rateLimitedUntil: number; // 0 = not rate-limited
+}
+
+export class BotPool implements ITelegramService {
+  private readonly bots: BotEntry[] = [];
+
+  constructor() {
+    const tokens = Array.from(new Set([config.botToken, ...config.additionalBotTokens]));
+    this.bots = tokens.map((token, index) => ({
+      index,
+      token,
+      instance: new Telegraf(token),
+      queue: new PQueue({ concurrency: PER_BOT_CONCURRENCY }),
+      rateLimitedUntil: 0,
+    }));
+  }
+
+  /** Number of bots in the pool */
+  get size(): number {
+    return this.bots.length;
+  }
+
+  /**
+   * Select the bot with the fewest pending tasks that isn't rate-limited
+   * or in the skip set.
+   */
+  private selectBot(skipIndexes?: Set<number>): BotEntry | null {
+    let best: BotEntry | null = null;
+    let bestPending = Infinity;
+
+    for (const bot of this.bots) {
+      if (skipIndexes?.has(bot.index)) continue;
+      if (bot.rateLimitedUntil > Date.now()) continue;
+
+      const pending = bot.queue.pending + bot.queue.size;
+      if (pending < bestPending) {
+        bestPending = pending;
+        best = bot;
+      }
+    }
+
+    return best;
+  }
+
+  /**
+   * Execute a Telegram API action on a specific bot entry.
+   * Wraps with timeout.
+   */
+  private async executeBotAction<T>(
+    bot: BotEntry,
+    action: (instance: Telegraf, token: string) => Promise<T>,
+  ): Promise<T> {
+    return Promise.race([
+      action(bot.instance, bot.token),
+      new Promise<never>((_, reject) =>
+        setTimeout(
+          () => reject(new Error(`Telegram API timeout after ${TELEGRAM_API_TIMEOUT_MS}ms`)),
+          TELEGRAM_API_TIMEOUT_MS,
+        ),
+      ),
+    ]);
+  }
+
+  /**
+   * Forward a file chunk to the configured Telegram storage chat.
+   *
+   * The upload is submitted to the least-loaded bot's queue. If the bot
+   * returns 429, it is marked rate-limited and the upload retries on the
+   * next available bot. If all bots are rate-limited, sleeps before retrying.
+   */
+  async forwardToStorage(
+    fileChunk: unknown,
+    fileName: string,
+    fileType: string,
+  ): Promise<ForwardResult> {
+    let lastError: unknown;
+    const attemptedIndexes = new Set<number>();
+    let transientAttempts = 0;
+
+    // Outer retry loop — up to 10 attempts across all bots
+    for (let attempt = 0; attempt < 10; attempt++) {
+      const bot = this.selectBot(attemptedIndexes);
+
+      if (!bot) {
+        // No available bots — either all rate-limited or all attempted
+        if (attemptedIndexes.size > 0) {
+          // All non-rate-limited bots were tried and failed — wait & reset
+          logger.warn('All available bots exhausted, sleeping 5s before retry');
+          await sleep(5000 + Math.random() * 1000);
+          attemptedIndexes.clear();
+          continue;
+        }
+        // All bots rate-limited — wait for the shortest cooldown
+        const earliestCooldown = Math.min(
+          ...this.bots.map((b) => b.rateLimitedUntil || Infinity),
+        );
+        const waitMs = Math.max(1000, earliestCooldown - Date.now() + 500);
+        logger.warn('All bots rate-limited, waiting', { waitMs });
+        await sleep(waitMs);
+        attemptedIndexes.clear();
+        continue;
+      }
+
+      attemptedIndexes.add(bot.index);
+
+      try {
+        const result = await bot.queue.add(async () => {
+          // Inner transient retry loop inside the queue
+          for (let innerRetry = 0; innerRetry <= MAX_TRANSIENT_RETRIES; innerRetry++) {
+            try {
+              const filePayload = { source: fileChunk, filename: fileName };
+              const sendMethodName = sendMethodMap[fileType] || 'sendDocument';
+              const payload = buildSendPayload(fileType, fileName);
+
+              const tgResult = await this.executeBotAction<TelegramMessageResult>(
+                bot,
+                (activeBot) => {
+                  const telegram = activeBot.telegram as unknown as Record<string, SendMethod>;
+                  return telegram[sendMethodName](config.storageChatId, filePayload, payload);
+                },
+              );
+
+              const uploadedFile = extractUploadedFile(tgResult, fileType);
+              return {
+                telegramFileId: uploadedFile?.file_id || '',
+                telegramFileUniqueId: uploadedFile?.file_unique_id || '',
+                storageMessageId: tgResult.message_id,
+              };
+            } catch (error: unknown) {
+              const errorStr = error instanceof Error ? error.message : String(error);
+              const retryAfterMatch = errorStr.match(/retry after (\d+)/i);
+
+              if (retryAfterMatch) {
+                // 429 — mark bot rate-limited, throw to outer loop for retry on different bot
+                const seconds = parseInt(retryAfterMatch[1], 10);
+                bot.rateLimitedUntil = Date.now() + seconds * 1000;
+                logger.info(`Bot #${bot.index} rate-limited for ${seconds}s`, { fileName, attempt });
+                throw error; // caught by outer retry loop
+              }
+
+              if (innerRetry < MAX_TRANSIENT_RETRIES && isTransientError(error)) {
+                const backoffMs = Math.min(1000 * 2 ** innerRetry, 10_000);
+                logger.warn(
+                  `Transient error on bot #${bot.index}, retrying (${innerRetry + 1}/${MAX_TRANSIENT_RETRIES})`,
+                  { fileName, error: errorStr, backoffMs },
+                );
+                await sleep(backoffMs);
+                continue;
+              }
+
+              throw error; // non-transient — propagate
+            }
+          }
+
+          throw new Error(`Exhausted transient retries on bot #${bot.index}`);
+        });
+
+        logger.info('File forwarded to storage', { fileName, message: result.storageMessageId });
+        return result;
+      } catch (error: unknown) {
+        lastError = error;
+        const errorStr = error instanceof Error ? error.message : String(error);
+        const retryAfterMatch = errorStr.match(/retry after (\d+)/i);
+
+        if (retryAfterMatch) {
+          // Bot was rate-limited — already marked, try next bot
+          continue;
+        }
+
+        // Transient error at the queue level (timeout, 5xx)
+        if (transientAttempts < MAX_TRANSIENT_RETRIES && isTransientError(error)) {
+          transientAttempts++;
+          continue;
+        }
+
+        // Non-transient — give up
+        logger.error('Failed to forward file to storage', {
+          fileName,
+          error: errorStr,
+          attempt,
+        });
+        throw error;
+      }
+    }
+
+    throw lastError || new Error('Failed to forward file after all retries');
+  }
+
+  /** Get total effective concurrency across all bots */
+  getEffectiveConcurrency(): number {
+    return this.bots.length * PER_BOT_CONCURRENCY;
+  }
+
+  async getFileInfo(telegramFileId: string): Promise<TelegramFileInfo> {
+    let lastError: unknown;
+    for (const bot of this.bots) {
+      for (let retry = 0; retry <= MAX_TRANSIENT_RETRIES; retry++) {
+        try {
+          const result = await bot.instance.telegram.getFile(telegramFileId);
+          const fileData = result as unknown as Omit<TelegramFileInfo, 'bot_token'>;
+          return {
+            file_size: fileData.file_size || 0,
+            mime_type: fileData.mime_type || 'application/octet-stream',
+            file_path: fileData.file_path || '',
+            bot_token: bot.token,
+          };
+        } catch (error: unknown) {
+          lastError = error;
+          const errorStr = error instanceof Error ? error.message : String(error);
+          if (
+            errorStr.includes('wrong file_id') ||
+            errorStr.includes('file is temporarily unavailable')
+          ) {
+            break;
+          }
+          if (retry < MAX_TRANSIENT_RETRIES && isTransientError(error)) {
+            const backoffMs = Math.min(1000 * 2 ** (retry + 1), 5_000);
+            await sleep(backoffMs);
+            continue;
+          }
+          break;
+        }
+      }
+    }
+    logger.error('Failed to get file info from any bot', {
+      error: lastError instanceof Error ? lastError.message : String(lastError),
+    });
+    throw lastError;
+  }
+}
+
+export const botPool = new BotPool();
+```
+
+- [ ] **Step 3: Run lint**
+
+Run: `bunx biome check src/infrastructure/telegram/bot-pool.ts`
+Expected: No errors.
+
+- [ ] **Step 4: Run existing test suite**
+
+Run: `bun test test/bot.test.ts`
+Expected: All tests pass.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add src/infrastructure/telegram/bot-pool.ts
+git commit -m "refactor: per-bot queue with selectBot() and rate-limit tracking
+
+Each bot has its own PQueue (concurrency=1). Uploads are assigned to
+the least-loaded available bot. On 429, the bot is marked rate-limited
+and the upload retries on the next available bot.
+
+Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>"
+```
+
+### Task 4: Remove global upload queue
+
+Delete the global queue, its test, and all references to it from index.ts.
+
+**Files:**
+- Delete: `src/infrastructure/telegram/upload-queue.ts`
+- Modify: `src/index.ts` (lines 4, 71-81)
+- Delete: `test/telegramQueue.test.ts`
+
+- [ ] **Step 1: Delete upload-queue.ts**
+
+Run: `rm src/infrastructure/telegram/upload-queue.ts`
+
+- [ ] **Step 2: Delete the test file**
+
+Run: `rm test/telegramQueue.test.ts`
+
+- [ ] **Step 3: Update index.ts — remove upload-queue import and shutdown drain logic**
+
+Remove line:
+```typescript
+import { clearQueue, getQueueStats, waitForQueue } from './infrastructure/telegram/upload-queue';
+```
+
+Remove the drain block (lines 70-81):
+```typescript
+  // Drain pending upload queue with a timeout
+  const { pending, size } = getQueueStats();
+  if (pending > 0 || size > 0) {
+    logger.info('Draining upload queue', { pending, size });
+    const drainTimeout = setTimeout(() => {
+      logger.warn('Upload queue drain timeout — clearing remaining tasks');
+      clearQueue();
+    }, 30_000);
+    await waitForQueue();
+    clearTimeout(drainTimeout);
+    logger.info('Upload queue drained');
+  }
+```
+
+- [ ] **Step 4: Run lint**
+
+Run: `bunx biome check src/index.ts`
+Expected: No errors.
+
+- [ ] **Step 5: Run tests**
+
+Run: `bun test`
+Expected: All tests pass (some may be skipped due to missing queue test).
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add src/infrastructure/telegram/upload-queue.ts test/telegramQueue.test.ts src/index.ts
+git commit -m "refactor: remove global upload queue
+
+Per-bot queues now handle concontrol internally.
+
+Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>"
+```
+
+### Task 5: Update env.ts and chunked-storage backpressure
+
+Remove `uploadConcurrency` from config and update chunked-storage to derive effective concurrency from bot pool.
+
+**Files:**
+- Modify: `src/env.ts`
+- Modify: `src/utils/chunked-storage.ts`
+
+- [ ] **Step 1: Remove `uploadConcurrency` from env.ts**
+
+Remove:
+```typescript
+  uploadConcurrency: number;
+```
+and:
+```typescript
+  uploadConcurrency: parseNumber(process.env.UPLOAD_CONCURRENCY, 8),
+```
+
+- [ ] **Step 2: Update chunked-storage.ts backpressure**
+
+Replace:
+```typescript
+import { config } from '../env';
+// ...
+if (inFlight.size >= config.uploadConcurrency * 2) {
+```
+With:
+```typescript
+import { botPool } from '../infrastructure/telegram/bot-pool';
+// ...
+if (inFlight.size >= botPool.getEffectiveConcurrency()) {
+```
+(Use effective concurrency * 2 for backpressure, or just use effective concurrency as the limit.)
+
+Actually let me think about this more carefully. The backpressure in chunked-storage:
+```
+if (inFlight.size >= config.uploadConcurrency * 2) {
+  await Promise.race(inFlight);
+  await new Promise((resolve) => setTimeout(resolve, 0));
+}
+```
+
+This limits the number of in-flight chunks per file. With `uploadConcurrency: 16`, it was 32. Now with effective concurrency of 6 (6 bots), it would be 12. That's fine as backpressure — it prevents too many chunks from being in memory at once.
+
+Let me use `botPool.getEffectiveConcurrency() * 2` to keep the same multiplier.
+
+- [ ] **Step 3: Run lint and tests**
+
+```bash
+bunx biome check src/env.ts src/utils/chunked-storage.ts
+bun test test/chunked-storage.test.ts
+```
+
+Expected: All checks pass.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add src/env.ts src/utils/chunked-storage.ts
+git commit -m "refactor: remove uploadConcurrency from config
+
+Effective concurrency derived from bot pool size. Chunked-storage
+backpressure now uses botPool.getEffectiveConcurrency().
+
+Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>"
+```
+
+### Task 6: Full integration test
+
+- [ ] **Step 1: Run the full test suite**
+
+Run: `bun test`
+Expected: All tests pass.
+
+- [ ] **Step 2: Run lint**
+
+Run: `bunx biome check src test`
+Expected: No errors.
+
+- [ ] **Step 3: Create summary commit**
+
+```bash
+git add -A
+git commit -m "refactor: implement per-bot queue architecture
+
+- Each bot has its own PQueue with concurrency=1
+- selectBot() assigns uploads to least-loaded available bot
+- 429 rate limits are tracked per-bot with cooldown timers
+- Failed uploads retry on next available bot
+- Removed global upload-queue.ts and uploadConcurrency config
+- Updated ITelegramService interface
+
+Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>"
+```

docs/superpowers/specs/2026-07-29-per-bot-queue-design.md

@@ -0,0 +1,125 @@
+# Per-Bot Queue: Rate-Limit Safe Telegram Upload
+
+**Date:** 2026-07-29
+**Status:** Approved Design
+
+## Problem
+
+Telegram Bot API rate-limits each bot to approximately 1-2 concurrent uploads. When multiple upload chunks hit the same bot simultaneously, Telegram returns HTTP 429 (Too Many Requests), causing delays of 30-60 seconds per retry. Under Docker push load, these cumulative delays trigger Gitea client timeouts and `500 Internal Server Error`.
+
+The current architecture uses a **global PQueue** with `concurrency=N` where each task picks a bot via round-robin (`claimBotIndex()`). This means two concurrent tasks can both land on the same bot index (after wrap-around), causing 429 collisions.
+
+## Solution: Per-Bot Queue
+
+Each bot has its own PQueue with `concurrency=1`. Uploads are assigned to the bot with the fewest pending tasks. If a bot rate-limits, the upload moves to the next available bot.
+
+### Architecture
+
+```
+┌────────────────────────────────────────────────────────┐
+│                      BotPool                           │
+│                                                        │
+│  ┌──────────────────────────────────────────────────┐  │
+│  │ BotEntry[0]    token=b1  queue=PQueue(conc=1)    │  │
+│  │                rateLimitedUntil=0                  │  │
+│  ├──────────────────────────────────────────────────┤  │
+│  │ BotEntry[1]    token=b2  queue=PQueue(conc=1)    │  │
+│  │                rateLimitedUntil=0                  │  │
+│  ├──────────────────────────────────────────────────┤  │
+│  │ ... up to N bots                                   │  │
+│  └──────────────────────────────────────────────────┘  │
+│                                                        │
+│  selectBot(skip?): number                              │
+│    └─ bot dengan pending queue paling sedikit           │
+│       dan tidak sedang rate-limited                     │
+│                                                        │
+│  forwardToStorage(file): ForwardResult                 │
+│    └─ retry loop: selectBot → queue.add → handle 429   │
+└────────────────────────────────────────────────────────┘
+```
+
+### BotEntry Structure
+
+```typescript
+interface BotEntry {
+  index: number;
+  token: string;
+  instance: Telegraf;
+  queue: PQueue;            // concurrency: 1
+  rateLimitedUntil: number;  // epoch ms, 0 = not limited
+}
+```
+
+### Data Flow: Upload
+
+```
+forwardToStorage(fileChunk, fileName, fileType)
+  │
+  ├─ MAX_RETRIES loop (attempt up to all bots)
+  │   │
+  │   ├─ selectBot(attemptedIndexes)
+  │   │   ├─ Filter out rate-limited bots (rateLimitedUntil > Date.now())
+  │   │   ├─ Filter out already-attempted bots
+  │   │   ├─ If none available:
+  │   │   │   ├─ Wait MIN_SLEEP_MS (5000ms)
+  │   │   │   ├─ Reset rate-limited timers (clear attemptedIndexes)
+  │   │   │   └─ Retry selectBot
+  │   │   └─ Return bot with smallest queue.pending count
+  │   │
+  │   ├─ attemptedIndexes.add(selectedBot)
+  │   │
+  │   ├─ result = await bots[selectedBot].queue.add(() =>
+  │   │     executeTelegramCall(bot, fileChunk, fileName)
+  │   │   )
+  │   │   │
+  │   │   ├─ ✅ Success → return ForwardResult
+  │   │   │
+  │   │   └─ ❌ Error
+  │   │       ├─ 429 → markRateLimited(bot, retryAfter)
+  │   │       │        → continue to next bot in retry loop
+  │   │       ├─ Transient (timeout, 5xx) → continue
+  │   │       └─ Non-transient → throw (propagate up)
+  │   │
+  │   └─ Attempt counter exhausted → throw lastError
+  │
+  └─ Sorted part tracking (for chunked uploads)
+```
+
+### Key Design Decisions
+
+1. **Concurrency=1 per bot**: Guarantees no two Telegram API calls compete for the same bot token. With 6 bots, effective concurrency = 6.
+
+2. **Least-loaded assignment**: `selectBot()` picks the bot with the fewest queued + pending tasks. This naturally load-balances even when some bots are slower.
+
+3. **Rate-limit isolation**: When bot A hits 429, only bot A's queue is paused. Other 5 bots continue serving uploads uninterrupted.
+
+4. **Per-bot rate-limit timer**: `rateLimitedUntil` prevents re-selecting a recently-429'd bot until its cooldown expires.
+
+5. **No global PQueue**: The old `upload-queue.ts` is removed. Each bot owns its queue, eliminating the global backpressure problem.
+
+### Changes by File
+
+| File | Action |
+|------|--------|
+| `src/infrastructure/telegram/bot-pool.ts` | **Major refactor**: BotEntry array, selectBot(), per-bot queues, retry loop |
+| `src/infrastructure/telegram/upload-queue.ts` | **Delete**: No longer needed |
+| `src/domain/ports/telegram-service.ts` | **Remove** `enqueueUpload<T>(task: () => Promise<T>): Promise<T>` from interface |
+| `src/utils/chunked-storage.ts` | **No changes** — only uses `forwardToStorage()` |
+| `src/env.ts` | **Remove** `uploadConcurrency` config (no longer needed) |
+
+### Error Handling
+
+- **429 per bot**: Mark bot rate-limited, move to next. Clear timer after `retryAfter` seconds.
+- **All bots 429**: Sleep 5 seconds with jitter, then retry from bot 0.
+- **Transient errors** (timeout, 5xx, connection reset): Retry on same bot (inside its queue), then on next bot.
+- **Non-transient errors** (4xx other than 429, wrong file_id, auth errors): Propagate immediately.
+- **MAX_RETRIES**: 10 attempts across all bots before giving up.
+
+### Testing
+
+- Unit: `selectBot()` returns bot with fewest pending tasks
+- Unit: `selectBot()` skips rate-limited bots
+- Unit: 429 on bot 0 → retries on bot 1 → succeeds
+- Unit: All bots rate-limited → sleeps → retries → succeeds
+- Unit: Per-bot queue has concurrency=1 (two tasks to same bot queue sequentially)
+- Integration: Forward a real file through the per-bot pool

src/domain/ports/telegram-service.ts

@@ -50,15 +50,4 @@ export interface ITelegramService {
    * @returns Metadata including size, MIME type, download path, and bot token.
    */
   getFileInfo(telegramFileId: string): Promise<TelegramFileInfo>;
-
-  /**
-   * Enqueue a task for sequential upload execution.
-   *
-   * Ensures only one Telegram upload runs at a time to avoid
-   * rate limits and resource contention.
-   *
-   * @param task - An async function performing the upload.
-   * @returns The result of the task.
-   */
-  enqueueUpload<T>(task: () => Promise<T>): Promise<T>;
 }

src/env.ts

@@ -12,7 +12,6 @@ interface AppConfig {
   rateLimitWindowMs: number;
   rateLimitMaxRequests: number;
   trustProxy: boolean;
-  uploadConcurrency: number;
   batchMaxItems: number;
   batchMaxSizeBytes: number;
   maxRequestBodyBytes: number;
@@ -105,7 +104,6 @@ export const config: AppConfig = {
   rateLimitWindowMs: parseNumber(process.env.RATE_LIMIT_WINDOW_MS, 60000),
   rateLimitMaxRequests: parseNumber(process.env.RATE_LIMIT_MAX_REQUESTS, 150),
   trustProxy: process.env.TRUST_PROXY === 'true',
-  uploadConcurrency: parseNumber(process.env.UPLOAD_CONCURRENCY, 8),
   batchMaxItems: parseNumber(process.env.BATCH_MAX_ITEMS, 20),
   batchMaxSizeBytes: parseNumber(process.env.BATCH_MAX_SIZE_BYTES, 500 * 1024 * 1024),
   maxRequestBodyBytes: parseNumber(process.env.MAX_REQUEST_BODY_BYTES, 2 * 1024 * 1024 * 1024),

src/index.ts

@@ -1,7 +1,6 @@
 import { serve } from 'bun';
 import { config } from './config/index';
 import { fileInfoCache } from './infrastructure/cache/index';
-import { clearQueue, getQueueStats, waitForQueue } from './infrastructure/telegram/upload-queue';
 import { startBot } from './interfaces/bot/handler';
 import { handleS3Request } from './interfaces/http/controllers/s3-controller';
 import { cleanupRateLimitCache } from './interfaces/http/middleware/rate-limit';
@@ -67,19 +66,6 @@ const gracefulShutdown = async (signal: string): Promise<void> => {
   logger.info('Closing HTTP server — no new requests accepted');
   server.stop();
 
-  // Drain pending upload queue with a timeout
-  const { pending, size } = getQueueStats();
-  if (pending > 0 || size > 0) {
-    logger.info('Draining upload queue', { pending, size });
-    const drainTimeout = setTimeout(() => {
-      logger.warn('Upload queue drain timeout — clearing remaining tasks');
-      clearQueue();
-    }, 30_000);
-    await waitForQueue();
-    clearTimeout(drainTimeout);
-    logger.info('Upload queue drained');
-  }
-
   logger.info('Stopping Telegram bot');
   bot.stop(signal);
 

src/infrastructure/telegram/bot-pool.ts

@@ -1,3 +1,4 @@
+import PQueue from 'p-queue';
 import { Telegraf } from 'telegraf';
 import type {
   ForwardResult,
@@ -13,38 +14,12 @@ import {
   sendMethodMap,
   type TelegramMessageResult,
 } from './types';
-import { enqueueUpload } from './upload-queue';
 
-/**
- * Sleep for a given number of milliseconds.
- *
- * Used as a backoff mechanism when all bots in the pool are rate-limited
- * or when retrying transient Telegram API errors.
- *
- * @param ms - Number of milliseconds to sleep.
- * @returns A promise that resolves after the specified delay.
- */
-const sleep = (ms: number): Promise<void> => {
-  return new Promise((resolve) => setTimeout(resolve, ms));
-};
+const sleep = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms));
 
-/**
- * Determines whether an error from the Telegram API is likely transient
- * and worth retrying.
- *
- * Transient telegrams errors include: network timeouts, 5xx server errors,
- * and "Too Many Requests" (429) which is already handled by bot rotation
- * but is also transient at the network level.
- *
- * @param error - The caught error object.
- * @returns True if the error is likely transient and worth retrying.
- */
 const isTransientError = (error: unknown): boolean => {
   const str = error instanceof Error ? error.message : String(error);
   const transientPatterns = [
-    // 'retry after' is deliberately omitted — 429 is handled by
-    // executeWithBotRetry at a deeper layer. Including it here would
-    // cause double-retry (up to 96 attempts per chunk).
     'timeout',
     'Timed out',
     'etimedout',
@@ -72,120 +47,87 @@ const isTransientError = (error: unknown): boolean => {
   return transientPatterns.some((p) => str.toLowerCase().includes(p.toLowerCase()));
 };
 
-/**
- * Maximum number of retries for transient Telegram API errors
- * before giving up and propagating the error to the caller.
- */
 const MAX_TRANSIENT_RETRIES = 3;
-
-/**
- * Timeout in milliseconds for individual Telegram API calls.
- * 120 seconds to accommodate large document uploads.
- */
+const MAX_OUTER_RETRIES = 10;
 const TELEGRAM_API_TIMEOUT_MS = 120_000;
+const PER_BOT_CONCURRENCY = 1;
+
+interface BotEntry {
+  index: number;
+  token: string;
+  instance: Telegraf;
+  queue: PQueue;
+  rateLimitedUntil: number; // 0 = not rate-limited
+}
 
-/**
- * Manages a pool of Telegram bots with automatic rotation and rate-limit handling.
- *
- * Distributes uploads across multiple bot tokens to maximise throughput.
- * When a bot receives a 429 (rate-limit) error, the pool instantly rotates
- * to the next available bot. If all bots are rate-limited, a coordinated
- * sleep is performed before retrying.
- *
- * Implements the {@link ITelegramService} contract.
- */
 export class BotPool implements ITelegramService {
-  private readonly bots: Telegraf[];
-  private readonly botTokens: string[];
-  private nextBotIndex = 0;
+  private readonly bots: BotEntry[] = [];
 
-  /** Create a new BotPool from the application configuration. */
   constructor() {
-    this.botTokens = Array.from(new Set([config.botToken, ...config.additionalBotTokens]));
-    this.bots = this.botTokens.map((token) => new Telegraf(token));
+    const tokens = Array.from(new Set([config.botToken, ...config.additionalBotTokens]));
+    this.bots = tokens.map((token, index) => ({
+      index,
+      token,
+      instance: new Telegraf(token),
+      queue: new PQueue({ concurrency: PER_BOT_CONCURRENCY }),
+      rateLimitedUntil: 0,
+    }));
+  }
+
+  /** Number of bots in the pool */
+  get size(): number {
+    return this.bots.length;
   }
 
   /**
-   * Claim the next bot index using round-robin rotation.
-   *
-   * @returns The index of the selected bot.
+   * Select the bot with the fewest pending tasks that isn't rate-limited
+   * or in the skip set.
    */
-  private claimBotIndex(): number {
-    const botIndex = this.nextBotIndex;
-    this.nextBotIndex = (this.nextBotIndex + 1) % this.bots.length;
-    return botIndex;
-  }
+  private selectBot(skipIndexes?: Set<number>): BotEntry | null {
+    if (this.bots.length === 0) return null;
 
-  /**
-   * Execute a Telegram API action with automatic retry and bot rotation.
-   *
-   * On 429 errors the pool either:
-   * 1. Rotates to the next bot immediately (if another bot is available), or
-   * 2. Sleeps for the required duration after all bots are exhausted, then retries.
-   *
-   * @param action - The action to execute on a bot instance.
-   * @param retries - Number of full-pool retry cycles remaining.
-   * @param attemptedBots - Number of bots attempted in the current cycle.
-   * @returns The result of the action.
-   */
-  private async executeWithBotRetry<T>(
-    action: (botInstance: Telegraf, botToken: string) => Promise<T>,
-    retries = 5,
-    attemptedBots = 0,
-  ): Promise<T> {
-    const botIndex = this.claimBotIndex();
-    const currentBot = this.bots[botIndex];
-    const currentToken = this.botTokens[botIndex];
-    try {
-      // Add timeout to prevent hung API calls from occupying queue slots
-      const result = await Promise.race([
-        action(currentBot, currentToken),
-        new Promise<never>((_, reject) =>
-          setTimeout(
-            () => reject(new Error(`Telegram API timeout after ${TELEGRAM_API_TIMEOUT_MS}ms`)),
-            TELEGRAM_API_TIMEOUT_MS,
-          ),
-        ),
-      ]);
-      return result;
-    } catch (error: unknown) {
-      const errorStr = error instanceof Error ? error.message : String(error);
-      const match = errorStr.match(/retry after (\d+)/i);
+    let best: BotEntry | null = null;
+    let bestPending = Infinity;
 
-      if (match) {
-        const nextIndex = this.nextBotIndex;
-        const nextAttemptedBots = attemptedBots + 1;
+    for (const bot of this.bots) {
+      if (skipIndexes?.has(bot.index)) continue;
+      if (bot.rateLimitedUntil > Date.now()) continue;
 
-        if (nextAttemptedBots < this.bots.length) {
-          logger.info(
-            `Bot Index ${botIndex} hit 429. Instantly rotating to Bot Index ${nextIndex}...`,
-          );
-          return this.executeWithBotRetry(action, retries, nextAttemptedBots);
-        }
-
-        if (retries > 0) {
-          const seconds = parseInt(match[1], 10);
-          logger.warn(`All bots in the pool are rate-limited. Sleeping for ${seconds} seconds...`, {
-            error: errorStr,
-          });
-          await sleep(seconds);
-          return this.executeWithBotRetry(action, retries - 1, 0);
-        }
+      const pending = bot.queue.pending + bot.queue.size;
+      if (pending < bestPending) {
+        bestPending = pending;
+        best = bot;
       }
-      throw error;
     }
+
+    return best;
+  }
+
+  /**
+   * Execute a Telegram API action on a specific bot entry.
+   * Wraps with timeout.
+   */
+  private async executeBotAction<T>(
+    bot: BotEntry,
+    action: (instance: Telegraf, token: string) => Promise<T>,
+  ): Promise<T> {
+    return Promise.race([
+      action(bot.instance, bot.token),
+      new Promise<never>((_, reject) =>
+        setTimeout(
+          () => reject(new Error(`Telegram API timeout after ${TELEGRAM_API_TIMEOUT_MS}ms`)),
+          TELEGRAM_API_TIMEOUT_MS,
+        ),
+      ),
+    ]);
   }
 
   /**
    * Forward a file chunk to the configured Telegram storage chat.
    *
-   * The upload is queued (via {@link enqueueUpload}) and executed with
-   * automatic bot rotation on rate-limit errors.
-   *
-   * @param fileChunk - The file data (ReadStream, Buffer, or file path).
-   * @param fileName - The original file name.
-   * @param fileType - The file type classification (e.g. "document", "photo").
-   * @returns The Telegram identifiers of the stored file.
+   * The upload is submitted to the least-loaded bot's queue. If the bot
+   * returns 429, it is marked rate-limited and the upload retries on the
+   * next available bot. If all bots are rate-limited, sleeps before retrying.
    */
   async forwardToStorage(
     fileChunk: unknown,
@@ -193,48 +135,113 @@ export class BotPool implements ITelegramService {
     fileType: string,
   ): Promise<ForwardResult> {
     let lastError: unknown;
-    let attempt = 0;
+    const attemptedIndexes = new Set<number>();
+    let transientAttempts = 0;
+
+    // Outer retry loop — up to MAX_OUTER_RETRIES attempts across all bots
+    for (let attempt = 0; attempt < MAX_OUTER_RETRIES; attempt++) {
+      const bot = this.selectBot(attemptedIndexes);
+
+      if (!bot) {
+        // No available bots — either all rate-limited or all attempted
+        if (attemptedIndexes.size > 0) {
+          // All non-rate-limited bots were tried and failed — wait & reset
+          logger.warn('All available bots exhausted, sleeping 5s before retry');
+          await sleep(5000 + Math.random() * 1000);
+          attemptedIndexes.clear();
+          continue;
+        }
+        // All bots rate-limited — wait for the shortest cooldown
+        const earliestCooldown = Math.min(...this.bots.map((b) => b.rateLimitedUntil || Infinity));
+        const waitMs = Math.max(1000, earliestCooldown - Date.now() + 500);
+        logger.warn('All bots rate-limited, waiting', { waitMs });
+        await sleep(waitMs);
+        attemptedIndexes.clear();
+        continue;
+      }
+
+      attemptedIndexes.add(bot.index);
 
-    while (attempt <= MAX_TRANSIENT_RETRIES) {
-      attempt++;
       try {
-        const result = await this.enqueueUpload<TelegramMessageResult>(async () => {
-          const filePayload = { source: fileChunk, filename: fileName };
-          const sendMethodName = sendMethodMap[fileType] || 'sendDocument';
-          const payload = buildSendPayload(fileType, fileName);
+        const result = await bot.queue.add(async () => {
+          // Inner transient retry loop inside the queue
+          for (let innerRetry = 0; innerRetry <= MAX_TRANSIENT_RETRIES; innerRetry++) {
+            try {
+              const filePayload = { source: fileChunk, filename: fileName };
+              const sendMethodName = sendMethodMap[fileType] || 'sendDocument';
+              const payload = buildSendPayload(fileType, fileName);
 
-          return this.executeWithBotRetry<TelegramMessageResult>((activeBot) => {
-            const telegram = activeBot.telegram as unknown as Record<string, SendMethod>;
-            return telegram[sendMethodName](config.storageChatId, filePayload, payload);
-          });
+              const tgResult = await this.executeBotAction<TelegramMessageResult>(
+                bot,
+                (activeBot) => {
+                  const telegram = activeBot.telegram as unknown as Record<string, SendMethod>;
+                  return telegram[sendMethodName](config.storageChatId, filePayload, payload);
+                },
+              );
+
+              const uploadedFile = extractUploadedFile(tgResult, fileType);
+              return {
+                telegramFileId: uploadedFile?.file_id || '',
+                telegramFileUniqueId: uploadedFile?.file_unique_id || '',
+                storageMessageId: tgResult.message_id,
+              };
+            } catch (error: unknown) {
+              const errorStr = error instanceof Error ? error.message : String(error);
+              const retryAfterMatch = errorStr.match(/retry after (\d+)/i);
+
+              if (retryAfterMatch) {
+                // 429 — mark bot rate-limited, throw to outer loop for retry on different bot
+                const seconds = parseInt(retryAfterMatch[1], 10);
+                bot.rateLimitedUntil = Date.now() + seconds * 1000;
+                logger.info(`Bot #${bot.index} rate-limited for ${seconds}s`, {
+                  fileName,
+                  attempt,
+                });
+                throw error; // caught by outer retry loop
+              }
+
+              if (innerRetry < MAX_TRANSIENT_RETRIES && isTransientError(error)) {
+                const backoffMs = Math.min(1000 * 2 ** innerRetry, 10_000);
+                logger.warn(
+                  `Transient error on bot #${bot.index}, retrying (${innerRetry + 1}/${MAX_TRANSIENT_RETRIES})`,
+                  { fileName, error: errorStr, backoffMs },
+                );
+                await sleep(backoffMs);
+                continue;
+              }
+
+              throw error; // non-transient — propagate
+            }
+          }
+
+          throw new Error(`Exhausted transient retries on bot #${bot.index}`);
         });
 
-        const uploadedFile = extractUploadedFile(result, fileType);
-        logger.info('File forwarded to storage', { fileName, message: result.message_id });
-
-        return {
-          telegramFileId: uploadedFile?.file_id || '',
-          telegramFileUniqueId: uploadedFile?.file_unique_id || '',
-          storageMessageId: result.message_id,
-        };
+        logger.info('File forwarded to storage', { fileName, message: result.storageMessageId });
+        return result;
       } catch (error: unknown) {
         lastError = error;
         const errorStr = error instanceof Error ? error.message : String(error);
+        const retryAfterMatch = errorStr.match(/retry after (\d+)/i);
 
-        if (attempt <= MAX_TRANSIENT_RETRIES && isTransientError(error)) {
-          const backoffMs = Math.min(1000 * 2 ** attempt, 10_000);
-          logger.warn(
-            `Transient error forwarding file, retrying (${attempt}/${MAX_TRANSIENT_RETRIES})`,
-            {
-              fileName,
-              error: errorStr,
-              backoffMs,
-            },
-          );
-          await sleep(backoffMs);
+        if (retryAfterMatch) {
+          // 429 catch in outer block: serves as a safety net for errors that
+          // contain "retry after N" wording but were rethrown from the inner
+          // queue task's fallback path (e.g., non-429 errors with similar text).
+          logger.warn('Retry-after pattern caught in outer loop (safety net)', {
+            fileName,
+            error: errorStr,
+          });
           continue;
         }
 
+        // Transient error at the queue level — retry on next bot
+        if (transientAttempts < MAX_TRANSIENT_RETRIES && isTransientError(error)) {
+          transientAttempts++;
+          continue;
+        }
+
+        // Non-transient — give up
         logger.error('Failed to forward file to storage', {
           fileName,
           error: errorStr,
@@ -244,80 +251,54 @@ export class BotPool implements ITelegramService {
       }
     }
 
-    // Should not reach here — last iteration throws above
-    throw lastError;
+    throw lastError || new Error('Failed to forward file after all retries');
+  }
+
+  /** Get total effective concurrency across all bots */
+  getEffectiveConcurrency(): number {
+    return this.bots.length * PER_BOT_CONCURRENCY;
   }
 
-  /**
-   * Retrieve file metadata from Telegram by file ID.
-   *
-   * Tries all configured bots sequentially; returns info from the first
-   * bot that can retrieve the file. Errors indicating the file belongs
-   * to a different bot are silently skipped.
-   *
-   * @param telegramFileId - The Telegram file_id to look up.
-   * @returns Metadata including size, MIME type, download path, and bot token.
-   */
   async getFileInfo(telegramFileId: string): Promise<TelegramFileInfo> {
     let lastError: unknown;
-    for (const activeBot of this.bots) {
+    for (const bot of this.bots) {
       for (let retry = 0; retry <= MAX_TRANSIENT_RETRIES; retry++) {
         try {
-          const result = await activeBot.telegram.getFile(telegramFileId);
+          const result = await bot.instance.telegram.getFile(telegramFileId);
           const fileData = result as unknown as Omit<TelegramFileInfo, 'bot_token'>;
           return {
             file_size: fileData.file_size || 0,
             mime_type: fileData.mime_type || 'application/octet-stream',
             file_path: fileData.file_path || '',
-            bot_token: activeBot.telegram.token,
+            bot_token: bot.token,
           };
         } catch (error: unknown) {
           lastError = error;
           const errorStr = error instanceof Error ? error.message : String(error);
-          // Belongs to a different bot — skip to next bot immediately
           if (
             errorStr.includes('wrong file_id') ||
             errorStr.includes('file is temporarily unavailable')
           ) {
-            break; // skip to next bot
+            break;
           }
-          // Transient — retry on the same bot
           if (retry < MAX_TRANSIENT_RETRIES && isTransientError(error)) {
             const backoffMs = Math.min(1000 * 2 ** (retry + 1), 5_000);
             logger.warn(
-              `Transient error getting file info, retrying bot ${activeBot.telegram.token.slice(0, 8)}... (${retry + 1}/${MAX_TRANSIENT_RETRIES})`,
+              `Transient error getting file info, retrying bot ${bot.token.slice(0, 8)}... (${retry + 1}/${MAX_TRANSIENT_RETRIES})`,
               { telegramFileId, error: errorStr, backoffMs },
             );
             await sleep(backoffMs);
             continue;
           }
-          // Non-transient or exhausted retries — try next bot
           break;
         }
       }
     }
-
     logger.error('Failed to get file info from any bot', {
       error: lastError instanceof Error ? lastError.message : String(lastError),
     });
     throw lastError;
   }
-
-  /**
-   * Enqueue a task for sequential upload execution.
-   *
-   * Delegates to the shared upload queue to ensure only a limited number
-   * of Telegram uploads run concurrently.
-   *
-   * @param task - An async function performing the upload.
-   * @returns The result of the task.
-   */
-  enqueueUpload<T>(task: () => Promise<T>): Promise<T> {
-    return enqueueUpload(task);
-  }
 }
 
-/**
- * Singleton BotPool instance initialised from application configuration.
- */
 export const botPool = new BotPool();

src/infrastructure/telegram/upload-queue.ts

@@ -1,100 +0,0 @@
-import PQueue from 'p-queue';
-import { config } from '../../env';
-import logger from '../../shared/logger/index';
-
-/**
- * Maximum number of pending (queued + in-flight) upload tasks before
- * new submissions are rejected. Prevents unbounded memory growth when
- * Telegram is slow or unavailable.
- */
-const MAX_QUEUE_PENDING = 1000;
-
-/**
- * P-queue instance for serialising Telegram upload tasks.
- *
- * Concurrency is governed by {@link config.uploadConcurrency}.
- * Built-in logging emits warnings when the queue grows beyond 5 pending items.
- */
-const uploadQueue = new PQueue({
-  concurrency: config.uploadConcurrency,
-});
-
-/* Monitor queue growth and emit warnings for large backlogs */
-uploadQueue.on('add', () => {
-  const stats = getQueueStats();
-  if (stats.size > 5) {
-    logger.warn('Upload queue building up', {
-      pending: stats.pending,
-      size: stats.size,
-      max: MAX_QUEUE_PENDING,
-    });
-  }
-});
-
-uploadQueue.on('next', () => {
-  const stats = getQueueStats();
-  logger.debug('Processing next upload', { pending: stats.pending, size: stats.size });
-});
-
-/**
- * Enqueue an upload task to be executed by the queue.
- *
- * Tasks are executed in FIFO order, subject to the concurrency limit.
- *
- * @param task - An async function representing the upload operation.
- * @returns A promise that resolves with the task's result.
- */
-export const enqueueUpload = <T>(task: () => Promise<T>): Promise<T> => {
-  const stats = getQueueStats();
-  if (stats.pending + stats.size > MAX_QUEUE_PENDING) {
-    return Promise.reject(
-      new Error(
-        `Upload queue full (${stats.pending + stats.size} pending, max ${MAX_QUEUE_PENDING})`,
-      ),
-    );
-  }
-  return uploadQueue.add(task);
-};
-
-/**
- * Get current queue statistics.
- *
- * @returns An object with `pending` (actively executing) and `size` (waiting) counts.
- */
-export const getQueueStats = (): { pending: number; size: number } => ({
-  pending: uploadQueue.pending,
-  size: uploadQueue.size,
-});
-
-/**
- * Get the number of items waiting in the queue (not yet started).
- *
- * @returns The number of queued items.
- */
-export const getQueueSize = (): number => uploadQueue.size;
-
-/**
- * Get the number of items currently being processed.
- *
- * @returns The number of pending (in-flight) items.
- */
-export const getPendingCount = (): number => uploadQueue.pending;
-
-/**
- * Clear all pending items and wait for in-flight ones to finish.
- *
- * @returns A promise that resolves when the queue is idle after clearing.
- */
-export const clearQueue = async (): Promise<void> => {
-  uploadQueue.clear();
-  await uploadQueue.onIdle();
-};
-
-/**
- * Wait for the queue to become idle (all tasks finished).
- *
- * @returns A promise that resolves when no tasks are pending or in-flight.
- */
-export const waitForQueue = async (): Promise<void> => {
-  await uploadQueue.onIdle();
-};

src/utils/chunked-storage.ts

@@ -125,7 +125,7 @@ export const uploadFileInTelegramChunks = async (input: {
 
     // Backpressure: if too many chunks are in-flight, wait for one to
     // finish before reading more — prevents unbounded memory growth.
-    if (inFlight.size >= config.uploadConcurrency * 2) {
+    if (inFlight.size >= botPool.getEffectiveConcurrency() * 2) {
       await Promise.race(inFlight);
       // Yield microtask to let .finally() run and remove from inFlight
       await new Promise((resolve) => setTimeout(resolve, 0));

test/bot-pool.test.ts

@@ -0,0 +1,131 @@
+import { afterEach, beforeEach, describe, expect, it, mock } from 'bun:test';
+
+process.env.BOT_TOKEN = 'bot1:token';
+process.env.ADDITIONAL_BOT_TOKENS = 'bot2:token,bot3:token';
+process.env.STORAGE_CHANNEL_ID = '-1001234567890';
+process.env.BASE_URL = 'https://example.com';
+process.env.DATABASE_URL = 'sqlite://test.db';
+process.env.PORT = '3000';
+
+// Track mock queue instances for per-bot assertions
+const queueInstances: Array<{
+  concurrency: number;
+  add: ReturnType<typeof mock>;
+  pending: number;
+  size: number;
+}> = [];
+
+// Mock PQueue so we can verify concurrency
+const mockAdd = mock(function addFn(this: any, fn: () => Promise<any>) {
+  return Promise.resolve().then(() => fn());
+});
+
+mock.module('p-queue', () => {
+  return {
+    default: mock(function MockQueue(this: any, opts?: { concurrency?: number }) {
+      const instance = {
+        concurrency: opts?.concurrency ?? 1,
+        add: mockAdd,
+        pending: 0,
+        size: 0,
+      };
+      queueInstances.push(instance);
+      return instance;
+    }),
+  };
+});
+
+// Mock Telegraf — use a class so `new Telegraf(token)` works correctly
+const mockTelegramInstances: Record<
+  string,
+  {
+    token: string;
+    sendDocument: ReturnType<typeof mock>;
+    sendPhoto: ReturnType<typeof mock>;
+    getFile: ReturnType<typeof mock>;
+  }
+> = {};
+
+class MockTelegraf {
+  token: string;
+  telegram: {
+    token: string;
+    sendDocument: ReturnType<typeof mock>;
+    sendPhoto: ReturnType<typeof mock>;
+    getFile: ReturnType<typeof mock>;
+  };
+
+  constructor(token: string) {
+    this.token = token;
+    this.telegram = {
+      token,
+      sendDocument: mock(() =>
+        Promise.resolve({
+          message_id: 1,
+          document: { file_id: `file_${token}`, file_unique_id: `uniq_${token}` },
+        }),
+      ),
+      sendPhoto: mock(() =>
+        Promise.resolve({
+          message_id: 1,
+          photo: [{ file_id: `photo_${token}`, file_unique_id: `photo_uniq_${token}` }],
+        }),
+      ),
+      getFile: mock(() =>
+        Promise.resolve({ file_size: 100, mime_type: 'text/plain', file_path: 'path' }),
+      ),
+    };
+    mockTelegramInstances[token] = this.telegram;
+  }
+}
+
+mock.module('telegraf', () => ({
+  Telegraf: MockTelegraf,
+}));
+
+describe('BotPool', () => {
+  let BotPool: typeof import('../src/infrastructure/telegram/bot-pool').BotPool;
+  let botPool: import('../src/infrastructure/telegram/bot-pool').BotPool;
+
+  beforeEach(async () => {
+    mockAdd.mockClear();
+    queueInstances.length = 0;
+    for (const token of Object.keys(mockTelegramInstances)) {
+      const tg = mockTelegramInstances[token];
+      if (tg) {
+        tg.sendDocument?.mockClear();
+        tg.getFile?.mockClear();
+      }
+    }
+    const mod = await import('../src/infrastructure/telegram/bot-pool');
+    BotPool = mod.BotPool;
+    botPool = new BotPool();
+  });
+
+  afterEach(() => {
+    // No module cache cleanup needed — Bun handles import caching correctly
+  });
+
+  it('should have correct bot count', () => {
+    expect(botPool.size).toBe(3);
+  });
+
+  it('should have correct effective concurrency', () => {
+    // 3 bots * 1 concurrency per bot
+    expect(botPool.getEffectiveConcurrency()).toBe(3);
+  });
+
+  it('should forward files through the queue', async () => {
+    const result = await botPool.forwardToStorage(Buffer.from('test data'), 'test.txt', 'document');
+    expect(result.telegramFileId).toBeDefined();
+    expect(result.storageMessageId).toBeGreaterThan(0);
+  });
+
+  it('should use per-bot queues with concurrency=1', () => {
+    // Each bot gets its own PQueue instance with concurrency=1
+    expect(queueInstances.length).toBe(3);
+    for (const qi of queueInstances) {
+      expect(qi.concurrency).toBe(1);
+    }
+  });
+});

test/telegramQueue.test.ts

@@ -1,36 +0,0 @@
-import { describe, expect, it } from 'bun:test';
-import { enqueueUpload } from '../src/infrastructure/telegram/upload-queue';
-
-describe('Telegram Queue', () => {
-  it('should process tasks in parallel without limit', async () => {
-    let activeTasks = 0;
-    let maxActiveTasks = 0;
-
-    const createTask = (id: number, delayMs: number) => {
-      return async () => {
-        activeTasks++;
-        if (activeTasks > maxActiveTasks) {
-          maxActiveTasks = activeTasks;
-        }
-
-        await new Promise((resolve) => setTimeout(resolve, delayMs));
-
-        activeTasks--;
-        return id;
-      };
-    };
-
-    const promises = [
-      enqueueUpload(createTask(1, 50)),
-      enqueueUpload(createTask(2, 50)),
-      enqueueUpload(createTask(3, 50)),
-      enqueueUpload(createTask(4, 50)),
-    ];
-
-    const results = await Promise.all(promises);
-
-    expect(results).toEqual([1, 2, 3, 4]);
-    // Concurrency limit is removed, so active tasks should be able to reach 4 (fully parallel)
-    expect(maxActiveTasks).toBe(4);
-  });
-});