diff --git a/src/infrastructure/telegram/bot-pool.ts b/src/infrastructure/telegram/bot-pool.ts index 405e695..bc6e01b 100644 --- a/src/infrastructure/telegram/bot-pool.ts +++ b/src/infrastructure/telegram/bot-pool.ts @@ -1,3 +1,4 @@ +import PQueue from 'p-queue'; import { Telegraf } from 'telegraf'; import type { ForwardResult, @@ -14,36 +15,11 @@ import { type TelegramMessageResult, } from './types'; -/** - * 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 => { - return new Promise((resolve) => setTimeout(resolve, ms)); -}; +const sleep = (ms: number): Promise => 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', @@ -71,119 +47,84 @@ 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 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): BotEntry | null { + let best: BotEntry | null = null; + let bestPending = Infinity; - /** - * 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( - action: (botInstance: Telegraf, botToken: string) => Promise, - retries = 5, - attemptedBots = 0, - ): Promise { - 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((_, 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); + for (const bot of this.bots) { + if (skipIndexes?.has(bot.index)) continue; + if (bot.rateLimitedUntil > Date.now()) continue; - if (match) { - const nextIndex = this.nextBotIndex; - const nextAttemptedBots = attemptedBots + 1; - - 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( + bot: BotEntry, + action: (instance: Telegraf, token: string) => Promise, + ): Promise { + return Promise.race([ + action(bot.instance, bot.token), + new Promise((_, 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 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, @@ -191,46 +132,100 @@ export class BotPool implements ITelegramService { fileType: string, ): Promise { let lastError: unknown; - let attempt = 0; + const attemptedIndexes = new Set(); + + // 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); - while (attempt <= MAX_TRANSIENT_RETRIES) { - attempt++; try { - 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); - const result = await this.executeWithBotRetry((activeBot) => { - const telegram = activeBot.telegram as unknown as Record; - return telegram[sendMethodName](config.storageChatId, filePayload, payload); + const tgResult = await this.executeBotAction( + bot, + (activeBot) => { + const telegram = activeBot.telegram as unknown as Record; + 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) { + // Bot was rate-limited — already marked, try next bot continue; } + // Non-transient — give up logger.error('Failed to forward file to storage', { fileName, error: errorStr, @@ -240,59 +235,45 @@ 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 { 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; 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})`, - { 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), }); @@ -300,7 +281,4 @@ export class BotPool implements ITelegramService { } } -/** - * Singleton BotPool instance initialised from application configuration. - */ export const botPool = new BotPool();