From 825e9695698d6c406fda99caea74099cc36851ed Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Mon, 18 May 2026 21:18:16 +0700 Subject: [PATCH] feat: implement multi-bot support for Telegram API with rate limit handling and add task queue for uploads --- .env.example | 1 + CLAUDE.md | 5 +- src/env.ts | 7 +++ src/utils/telegram.ts | 112 ++++++++++++++++++++++++------------- src/utils/telegramQueue.ts | 47 ++++++++++++++++ test/telegramQueue.test.ts | 39 +++++++++++++ 6 files changed, 170 insertions(+), 41 deletions(-) create mode 100644 src/utils/telegramQueue.ts create mode 100644 test/telegramQueue.test.ts diff --git a/.env.example b/.env.example index e238216..1b91890 100644 --- a/.env.example +++ b/.env.example @@ -1,4 +1,5 @@ BOT_TOKEN=isi_token_bot_telegram +ADDITIONAL_BOT_TOKENS=token_cadangan_1,token_cadangan_2,token_cadangan_3 STORAGE_CHANNEL_ID=-1001234567890 BASE_URL=https://tele.asepharyana.tech DATABASE_URL=postgresql://user:password@localhost:5432/telegram_uploader diff --git a/CLAUDE.md b/CLAUDE.md index 183c26f..30f8be8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -19,7 +19,10 @@ Default to using Bun instead of Node.js. - Prefer `Bun.file` over `node:fs`'s readFile/writeFile - Bun.$`ls` instead of execa. - Rate limiter lokal dinonaktifkan (`checkRateLimit` di `src/utils/rateLimit.ts` selalu mengembalikan `true`). -- Telegram API memiliki auto-retry otomatis jika mengembalikan error 429 (Too Many Requests) menggunakan helper `withRetry` di `src/utils/telegram.ts`. +- Telegram API memiliki auto-retry otomatis jika mengembalikan error 429 (Too Many Requests) menggunakan pool Telegraf multi-bot di `src/utils/telegram.ts`. +- Multi-bot dikonfigurasi melalui `ADDITIONAL_BOT_TOKENS` (koma terpisah) di `.env` yang digabung dengan `BOT_TOKEN` utama (total 4 bot). +- Menggunakan mekanisme rotasi instan jika ada bot yang terkena rate limit 429 sebelum memutuskan untuk sleep. +- Pengiriman berkas ke Telegram dikontrol oleh antrian in-memory (`TelegramQueue` di `src/utils/telegramQueue.ts`) dengan batas konkurensi = 2 untuk mencegah rate limit berlebih. ## Testing diff --git a/src/env.ts b/src/env.ts index 1823ae2..694d362 100644 --- a/src/env.ts +++ b/src/env.ts @@ -2,6 +2,7 @@ import logger from './utils/logger'; interface AppConfig { botToken: string; + additionalBotTokens: string[]; storageChatId: number; baseUrl: string; databaseUrl: string; @@ -31,6 +32,12 @@ if (missing.length > 0) { export const config: AppConfig = { botToken: process.env.BOT_TOKEN!, + additionalBotTokens: process.env.NODE_ENV === 'test' + ? [] + : (process.env.ADDITIONAL_BOT_TOKENS || '') + .split(',') + .map((t) => t.trim()) + .filter((t) => t !== ''), storageChatId: parseInt(process.env.STORAGE_CHANNEL_ID!, 10), baseUrl: process.env.BASE_URL!, databaseUrl: process.env.DATABASE_URL!, diff --git a/src/utils/telegram.ts b/src/utils/telegram.ts index 8c4f631..5a3a866 100644 --- a/src/utils/telegram.ts +++ b/src/utils/telegram.ts @@ -1,23 +1,48 @@ import { Telegraf } from 'telegraf'; import { config } from '../env'; import logger from './logger'; +import { enqueueUpload } from './telegramQueue'; -const bot = new Telegraf(config.botToken); +const botTokens = Array.from(new Set([config.botToken, ...config.additionalBotTokens])); + +const bots = botTokens.map((token) => new Telegraf(token)); const TELEGRAM_API_URL = `https://api.telegram.org/bot${config.botToken}/`; -const withRetry = async (fn: () => Promise, retries = 5): Promise => { +let currentBotIndex = 0; + +const executeWithBotRetry = async ( + action: (botInstance: Telegraf) => Promise, + retries = 5, + attemptedBots = 0, +): Promise => { + const currentBot = bots[currentBotIndex]; try { - return await fn(); + return await action(currentBot); } catch (error: any) { const errorStr = error.message || String(error); const match = errorStr.match(/retry after (\d+)/i); - if (match && retries > 0) { - const seconds = parseInt(match[1], 10); - logger.warn(`Telegram 429 Too Many Requests detected. Retrying after ${seconds} seconds...`, { - error: errorStr, - }); - await new Promise((resolve) => setTimeout(resolve, seconds * 1000)); - return withRetry(fn, retries - 1); + + if (match) { + // 429 rate limit hit! Rotate bot index instantly + const prevIndex = currentBotIndex; + currentBotIndex = (currentBotIndex + 1) % bots.length; + const nextIndex = currentBotIndex; + attemptedBots++; + + if (attemptedBots < bots.length) { + logger.info(`Bot Index ${prevIndex} hit 429. Instantly rotating to Bot Index ${nextIndex}...`); + return executeWithBotRetry(action, retries, attemptedBots); + } + + // If all bots in the pool have been tried and hit 429, sleep + 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 new Promise((resolve) => setTimeout(resolve, seconds * 1000)); + return executeWithBotRetry(action, retries - 1, 0); + } } throw error; } @@ -41,36 +66,43 @@ export const forwardToStorage = async ( fileType: string, ): Promise => { try { - const filePayload = { source: fileChunk, filename: fileName }; - let result: any; + const result: any = await enqueueUpload(async () => { + const filePayload = { source: fileChunk, filename: fileName }; + const uploadResult = await executeWithBotRetry((activeBot) => { + if (fileType === 'photo') { + return activeBot.telegram.sendPhoto(config.storageChatId, filePayload, { + caption: fileName, + }); + } else if (fileType === 'audio') { + return activeBot.telegram.sendAudio(config.storageChatId, filePayload, { + caption: fileName, + }); + } else if (fileType === 'video') { + return activeBot.telegram.sendVideo(config.storageChatId, filePayload, { + caption: fileName, + }); + } else if (fileType === 'voice') { + return activeBot.telegram.sendVoice(config.storageChatId, filePayload, { + caption: fileName, + }); + } else if (fileType === 'animation') { + return activeBot.telegram.sendAnimation(config.storageChatId, filePayload, { + caption: fileName, + }); + } else if (fileType === 'sticker') { + return activeBot.telegram.sendSticker(config.storageChatId, filePayload); + } else { + return activeBot.telegram.sendDocument(config.storageChatId, filePayload, { + caption: `📁 ${fileName}`, + }); + } + }); - if (fileType === 'photo') { - result = await withRetry(() => bot.telegram.sendPhoto(config.storageChatId, filePayload, { - caption: fileName, - })); - } else if (fileType === 'audio') { - result = await withRetry(() => bot.telegram.sendAudio(config.storageChatId, filePayload, { - caption: fileName, - })); - } else if (fileType === 'video') { - result = await withRetry(() => bot.telegram.sendVideo(config.storageChatId, filePayload, { - caption: fileName, - })); - } else if (fileType === 'voice') { - result = await withRetry(() => bot.telegram.sendVoice(config.storageChatId, filePayload, { - caption: fileName, - })); - } else if (fileType === 'animation') { - result = await withRetry(() => bot.telegram.sendAnimation(config.storageChatId, filePayload, { - caption: fileName, - })); - } else if (fileType === 'sticker') { - result = await withRetry(() => bot.telegram.sendSticker(config.storageChatId, filePayload)); - } else { - result = await withRetry(() => bot.telegram.sendDocument(config.storageChatId, filePayload, { - caption: `📁 ${fileName}`, - })); - } + // Advance round-robin index for next job + currentBotIndex = (currentBotIndex + 1) % bots.length; + + return uploadResult; + }); let uploadedFile: any; if (result.document) uploadedFile = result.document; @@ -131,4 +163,4 @@ export const getFileInfo = async ( } }; -export const getBot = (): Telegraf => bot; +export const getBot = (): Telegraf => bots[0]; diff --git a/src/utils/telegramQueue.ts b/src/utils/telegramQueue.ts new file mode 100644 index 0000000..91e3712 --- /dev/null +++ b/src/utils/telegramQueue.ts @@ -0,0 +1,47 @@ +type QueueTask = { + task: () => Promise; + resolve: (value: T | PromiseLike) => void; + reject: (reason?: any) => void; +}; + +class TelegramQueue { + private activeCount = 0; + private queue: QueueTask[] = []; + private concurrency: number; + + constructor(concurrency = 2) { + this.concurrency = concurrency; + } + + public enqueue(task: () => Promise): Promise { + return new Promise((resolve, reject) => { + this.queue.push({ task, resolve, reject }); + this.processNext(); + }); + } + + private async processNext(): Promise { + if (this.activeCount >= this.concurrency || this.queue.length === 0) { + return; + } + + const item = this.queue.shift()!; + this.activeCount++; + + try { + const result = await item.task(); + item.resolve(result); + } catch (error) { + item.reject(error); + } finally { + this.activeCount--; + this.processNext(); + } + } +} + +const telegramQueue = new TelegramQueue(2); + +export const enqueueUpload = (task: () => Promise): Promise => { + return telegramQueue.enqueue(task); +}; diff --git a/test/telegramQueue.test.ts b/test/telegramQueue.test.ts new file mode 100644 index 0000000..53fa322 --- /dev/null +++ b/test/telegramQueue.test.ts @@ -0,0 +1,39 @@ +import { describe, expect, it } from 'bun:test'; +import { enqueueUpload } from '../src/utils/telegramQueue'; + +describe('Telegram Queue', () => { + it('should process tasks in order and limit concurrency', async () => { + let activeTasks = 0; + let maxActiveTasks = 0; + const executionOrder: number[] = []; + + const createTask = (id: number, delayMs: number) => { + return async () => { + activeTasks++; + if (activeTasks > maxActiveTasks) { + maxActiveTasks = activeTasks; + } + + await new Promise((resolve) => setTimeout(resolve, delayMs)); + + executionOrder.push(id); + activeTasks--; + return id; + }; + }; + + // Enqueue 4 tasks with delays + const promises = [ + enqueueUpload(createTask(1, 50)), + enqueueUpload(createTask(2, 20)), + enqueueUpload(createTask(3, 10)), + enqueueUpload(createTask(4, 5)), + ]; + + const results = await Promise.all(promises); + + expect(results).toEqual([1, 2, 3, 4]); + // Concurrency limit is 2, so maximum active tasks at any time should be <= 2 + expect(maxActiveTasks).toBeLessThanOrEqual(2); + }); +});