refactor: remove concurrency limit from Telegram upload queue and enhance media group handling
This commit is contained in:
MythEclipse
@@ -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
|
||||
|
||||
|
||||
-116
@@ -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<string, { timer: any; items: MediaGroupBufferItem[] }>();
|
||||
|
||||
export const startBot = async (): Promise<Telegraf<Context>> => {
|
||||
try {
|
||||
const bot = new Telegraf(config.botToken);
|
||||
@@ -88,110 +76,6 @@ export const startBot = async (): Promise<Telegraf<Context>> => {
|
||||
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)
|
||||
|
||||
@@ -1,47 +1,3 @@
|
||||
type QueueTask<T> = {
|
||||
task: () => Promise<T>;
|
||||
resolve: (value: T | PromiseLike<T>) => void;
|
||||
reject: (reason?: any) => void;
|
||||
};
|
||||
|
||||
class TelegramQueue {
|
||||
private activeCount = 0;
|
||||
private queue: QueueTask<any>[] = [];
|
||||
private concurrency: number;
|
||||
|
||||
constructor(concurrency = 2) {
|
||||
this.concurrency = concurrency;
|
||||
}
|
||||
|
||||
public enqueue<T>(task: () => Promise<T>): Promise<T> {
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
this.queue.push({ task, resolve, reject });
|
||||
this.processNext();
|
||||
});
|
||||
}
|
||||
|
||||
private async processNext(): Promise<void> {
|
||||
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 = <T>(task: () => Promise<T>): Promise<T> => {
|
||||
return telegramQueue.enqueue(task);
|
||||
return task();
|
||||
};
|
||||
|
||||
+26
-7
@@ -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');
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
|
||||
+27
-8
@@ -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 () => {
|
||||
|
||||
Reference in New Issue
Block a user