From d58b2390efbce1b1b9e2778c0a6a926d6482f84d Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Mon, 18 May 2026 21:54:39 +0700 Subject: [PATCH] refactor: remove concurrency limit from Telegram upload queue and enhance media group handling --- CLAUDE.md | 3 +- src/bot.ts | 116 ------------------------------------- src/utils/telegramQueue.ts | 46 +-------------- test/telegram.test.ts | 33 ++++++++--- test/telegramQueue.test.ts | 15 ++--- test/upload.test.ts | 35 ++++++++--- 6 files changed, 61 insertions(+), 187 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 96166a9..d444ef8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -22,8 +22,7 @@ Default to using Bun instead of Node.js. - 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. -- Mendukung pengiriman batch upload (Media Group) di `src/bot.ts` dengan membungkus berkas yang memiliki `media_group_id` yang sama menggunakan debounce timer 600ms, lalu mengunggahnya secara sekaligus via `forwardMediaGroupToStorage` di `src/utils/telegram.ts`. +- Pengiriman berkas ke Telegram dieksekusi secara responsif dan paralel penuh tanpa batas konkurensi/antrian. ## Testing diff --git a/src/bot.ts b/src/bot.ts index 048f002..1565ff7 100644 --- a/src/bot.ts +++ b/src/bot.ts @@ -6,18 +6,6 @@ import { config } from './env'; import logger from './utils/logger'; import { forwardToStorage } from './utils/telegram'; -interface MediaGroupBufferItem { - ctx: any; - fileId: string; - fileSize: number; - mimeType: string; - fileName: string; - fileType: string; - fileUniqueId: string; -} - -const mediaGroupCache = new Map(); - export const startBot = async (): Promise> => { try { const bot = new Telegraf(config.botToken); @@ -88,110 +76,6 @@ export const startBot = async (): Promise> => { return ctx.reply(`File size exceeds ${maxSize / (1024 * 1024)}MB limit`); } - const mediaGroupId = ctx.message.media_group_id; - - if (mediaGroupId) { - if (!mediaGroupCache.has(mediaGroupId)) { - mediaGroupCache.set(mediaGroupId, { timer: null, items: [] }); - } - - const group = mediaGroupCache.get(mediaGroupId)!; - - if (group.timer) { - clearTimeout(group.timer); - } - - group.items.push({ - ctx, - fileId: file_id, - fileSize: file_size, - mimeType: mime_type, - fileName: fileName, - fileType: fileType, - fileUniqueId: fileObj.file_unique_id, - }); - - group.timer = setTimeout(async () => { - mediaGroupCache.delete(mediaGroupId); - - try { - const itemsToUpload: MediaGroupBufferItem[] = []; - const responses: string[] = []; - - for (const item of group.items) { - const existing = await db - .select() - .from(fileSchema) - .where(eq(fileSchema.telegramFileUniqueId, item.fileUniqueId)) - .limit(1); - - if (existing && existing.length > 0) { - const url = `${config.baseUrl}/f/${existing[0].publicId}`; - responses.push(`File *${item.fileName}* sudah diupload! 📎\nDownload: ${url}`); - } else { - itemsToUpload.push(item); - } - } - - if (itemsToUpload.length > 0) { - const uploadItems = itemsToUpload.map((item) => ({ - fileId: item.fileId, - fileName: item.fileName, - fileType: item.fileType, - })); - - const { forwardMediaGroupToStorage } = await import('./utils/telegram'); - const batchResult = await forwardMediaGroupToStorage(uploadItems); - - const dbInserts = []; - for (let i = 0; i < itemsToUpload.length; i++) { - const item = itemsToUpload[i]; - const publicId = nanoid(); - const uploaded = { - publicId, - telegramFileId: batchResult.telegramFileIds[i], - telegramFileUniqueId: batchResult.telegramFileUniqueIds[i], - storageChatId: config.storageChatId, - storageMessageId: batchResult.storageMessageId, - fileName: item.fileName, - mimeType: item.mimeType || 'application/octet-stream', - sizeBytes: item.fileSize, - fileType: item.fileType, - uploaderId: ctx.from.id, - createdAt: new Date(), - updatedAt: new Date(), - }; - - dbInserts.push(uploaded); - responses.push( - `File *${item.fileName}* berhasil diupload! 📎\nDownload: ${config.baseUrl}/f/${publicId}`, - ); - } - - if (dbInserts.length > 0) { - await db.insert(fileSchema).values(dbInserts); - } - } - - await ctx.reply(responses.join('\n\n')); - logger.info('Media group uploaded as batch', { - mediaGroupId, - totalFiles: group.items.length, - uploadedFiles: itemsToUpload.length, - uploader: ctx.from.id, - }); - } catch (error: any) { - logger.error('Failed to process media group batch', { - error: error.message, - mediaGroupId, - }); - await ctx.reply('❌ Gagal mengupload beberapa file di media group.'); - } - }, 600); - - return; - } - const existing = await db .select() .from(fileSchema) diff --git a/src/utils/telegramQueue.ts b/src/utils/telegramQueue.ts index 91e3712..2e51a54 100644 --- a/src/utils/telegramQueue.ts +++ b/src/utils/telegramQueue.ts @@ -1,47 +1,3 @@ -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); + return task(); }; diff --git a/test/telegram.test.ts b/test/telegram.test.ts index 10fe690..ea43b39 100644 --- a/test/telegram.test.ts +++ b/test/telegram.test.ts @@ -1,8 +1,27 @@ // @ts-nocheck -import { afterEach, beforeEach, describe, expect, it, mock, spyOn } from 'bun:test'; +import { afterEach, beforeAll, beforeEach, describe, expect, it, mock, spyOn } from 'bun:test'; import { config } from '../src/env'; import logger from '../src/utils/logger'; +let realPhotoBuffer: Buffer; + +beforeAll(async () => { + try { + const res = await fetch( + 'https://upload.wikimedia.org/wikipedia/commons/4/47/PNG_transparency_demonstration_1.png', + ); + if (!res.ok) throw new Error('Wikimedia download failed'); + const arrayBuffer = await res.arrayBuffer(); + realPhotoBuffer = Buffer.from(arrayBuffer); + } catch { + // Fallback 1x1px JPEG + realPhotoBuffer = Buffer.from( + 'ffd8ffe000104a46494600010101006000600000ffdb004300080606070605080707070909080a0c140d0c0b0b0c1912130f141d1a1f1e1d1a1c1c20242e2720222c231c1c2837292c30313434341f27393d38323c2e333432ffc0000b080001000101011100ffc4001f0000010501010110000000000000000000000102030405060708ffda000c03010002110311003f00a0ffd9', + 'hex', + ); + } +}); + // Mock Telegraf and fetch mock.module('telegraf', () => { return { @@ -66,8 +85,8 @@ describe('Telegram API Utilities', () => { describe('forwardToStorage', () => { it('should forward photo to storage chat and return file details', async () => { - const chunk = Buffer.from('fake photo data'); - const fileName = 'test_photo.jpg'; + const chunk = realPhotoBuffer; + const fileName = 'test_photo.png'; const result = await forwardToStorage(chunk, fileName, 'photo'); expect(result).toEqual({ @@ -104,8 +123,8 @@ describe('Telegram API Utilities', () => { const bot = getBot(); bot.telegram.sendPhoto = mock(() => Promise.reject(new Error('Telegram send failed'))); - const chunk = Buffer.from('fake photo data'); - const fileName = 'test_photo.jpg'; + const chunk = realPhotoBuffer; + const fileName = 'test_photo.png'; await expect(forwardToStorage(chunk, fileName, 'photo')).rejects.toThrow( 'Telegram send failed', @@ -130,8 +149,8 @@ describe('Telegram API Utilities', () => { }); }); - const chunk = Buffer.from('fake photo data'); - const fileName = 'test_photo.jpg'; + const chunk = realPhotoBuffer; + const fileName = 'test_photo.png'; const startTime = Date.now(); const result = await forwardToStorage(chunk, fileName, 'photo'); diff --git a/test/telegramQueue.test.ts b/test/telegramQueue.test.ts index 53fa322..7b6aba6 100644 --- a/test/telegramQueue.test.ts +++ b/test/telegramQueue.test.ts @@ -2,10 +2,9 @@ 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 () => { + it('should process tasks in parallel without limit', async () => { let activeTasks = 0; let maxActiveTasks = 0; - const executionOrder: number[] = []; const createTask = (id: number, delayMs: number) => { return async () => { @@ -16,24 +15,22 @@ describe('Telegram Queue', () => { 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)), + 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 2, so maximum active tasks at any time should be <= 2 - expect(maxActiveTasks).toBeLessThanOrEqual(2); + // Concurrency limit is removed, so active tasks should be able to reach 4 (fully parallel) + expect(maxActiveTasks).toBe(4); }); }); diff --git a/test/upload.test.ts b/test/upload.test.ts index 1c1e6e1..68c354f 100644 --- a/test/upload.test.ts +++ b/test/upload.test.ts @@ -1,5 +1,24 @@ // @ts-nocheck -import { afterAll, beforeEach, describe, expect, it, mock } from 'bun:test'; +import { afterAll, beforeAll, beforeEach, describe, expect, it, mock } from 'bun:test'; + +let realPhotoBuffer: Buffer; + +beforeAll(async () => { + try { + const res = await fetch( + 'https://upload.wikimedia.org/wikipedia/commons/4/47/PNG_transparency_demonstration_1.png', + ); + if (!res.ok) throw new Error('Wikimedia download failed'); + const arrayBuffer = await res.arrayBuffer(); + realPhotoBuffer = Buffer.from(arrayBuffer); + } catch { + // Fallback 1x1px JPEG + realPhotoBuffer = Buffer.from( + 'ffd8ffe000104a46494600010101006000600000ffdb004300080606070605080707070909080a0c140d0c0b0b0c1912130f141d1a1f1e1d1a1c1c20242e2720222c231c1c2837292c30313434341f27393d38323c2e333432ffc0000b080001000101011100ffc4001f0000010501010110000000000000000000000102030405060708ffda000c03010002110311003f00a0ffd9', + 'hex', + ); + } +}); // Mock db let mockSelectResult: any[] = []; @@ -96,8 +115,8 @@ describe('Upload Route Handler', () => { 'content-type': 'application/json', }, body: JSON.stringify({ - file: Buffer.from('hello world').toString('base64'), - fileName: 'test.txt', + file: realPhotoBuffer.toString('base64'), + fileName: 'test.png', }), }); @@ -108,8 +127,8 @@ describe('Upload Route Handler', () => { expect(body.public_id).toBe('mocked-nanoid-id'); expect(body.telegram_file_id).toBe('tg-file-id-123'); expect(body.telegram_file_unique_id).toBe('tg-unique-id-abc'); - expect(body.file_name).toBe('test.txt'); - expect(body.file_type).toBe('document'); + expect(body.file_name).toBe('test.png'); + expect(body.file_type).toBe('photo'); }); it('should reject JSON upload without file key', async () => { @@ -131,8 +150,8 @@ describe('Upload Route Handler', () => { it('should process multipart upload successfully', async () => { const formData = new FormData(); - const fileBlob = new Blob([Buffer.from('multipart hello')], { type: 'text/plain' }); - formData.append('file', fileBlob, 'test_multi.txt'); + const fileBlob = new Blob([realPhotoBuffer], { type: 'image/png' }); + formData.append('file', fileBlob, 'test_multi.png'); const req = new Request('http://localhost:3000/api/upload', { method: 'POST', @@ -143,7 +162,7 @@ describe('Upload Route Handler', () => { expect(res.status).toBe(200); const body = await res.json(); expect(body.public_id).toBe('mocked-nanoid-id'); - expect(body.file_name).toBe('test_multi.txt'); + expect(body.file_name).toBe('test_multi.png'); }); it('should deduplicate multipart upload if hash exists', async () => {