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

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

// 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

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

// 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:

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

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:

import { clearQueue, getQueueStats, waitForQueue } from './infrastructure/telegram/upload-queue';

Remove the drain block (lines 70-81):

  // 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

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:

  uploadConcurrency: number;

and:

  uploadConcurrency: parseNumber(process.env.UPLOAD_CONCURRENCY, 8),

• Step 2: Update chunked-storage.ts backpressure

Replace:

import { config } from '../env';
// ...
if (inFlight.size >= config.uploadConcurrency * 2) {

With:

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

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

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

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>"