feat: add support for batch uploads of media groups to Telegram with debounce handling
This commit is contained in:
@@ -23,6 +23,7 @@ Default to using Bun instead of Node.js.
|
|||||||
- Multi-bot dikonfigurasi melalui `ADDITIONAL_BOT_TOKENS` (koma terpisah) di `.env` yang digabung dengan `BOT_TOKEN` utama (total 4 bot).
|
- 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.
|
- 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.
|
- 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`.
|
||||||
|
|
||||||
## Testing
|
## Testing
|
||||||
|
|
||||||
|
|||||||
+116
@@ -6,6 +6,18 @@ import { config } from './env';
|
|||||||
import logger from './utils/logger';
|
import logger from './utils/logger';
|
||||||
import { forwardToStorage } from './utils/telegram';
|
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>> => {
|
export const startBot = async (): Promise<Telegraf<Context>> => {
|
||||||
try {
|
try {
|
||||||
const bot = new Telegraf(config.botToken);
|
const bot = new Telegraf(config.botToken);
|
||||||
@@ -76,6 +88,110 @@ export const startBot = async (): Promise<Telegraf<Context>> => {
|
|||||||
return ctx.reply(`File size exceeds ${maxSize / (1024 * 1024)}MB limit`);
|
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
|
const existing = await db
|
||||||
.select()
|
.select()
|
||||||
.from(fileSchema)
|
.from(fileSchema)
|
||||||
|
|||||||
@@ -130,6 +130,79 @@ export const forwardToStorage = async (
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
export interface MediaGroupItem {
|
||||||
|
fileId: string;
|
||||||
|
fileName: string;
|
||||||
|
fileType: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
export const forwardMediaGroupToStorage = async (
|
||||||
|
items: MediaGroupItem[],
|
||||||
|
): Promise<{
|
||||||
|
storageMessageId: number;
|
||||||
|
telegramFileIds: string[];
|
||||||
|
telegramFileUniqueIds: string[];
|
||||||
|
}> => {
|
||||||
|
try {
|
||||||
|
const result: any = await enqueueUpload(async () => {
|
||||||
|
const mediaGroup: any = items.map((item) => {
|
||||||
|
let type: 'photo' | 'video' | 'audio' | 'document' = 'document';
|
||||||
|
if (item.fileType === 'photo') type = 'photo';
|
||||||
|
else if (item.fileType === 'video') type = 'video';
|
||||||
|
else if (item.fileType === 'audio') type = 'audio';
|
||||||
|
|
||||||
|
return {
|
||||||
|
type,
|
||||||
|
media: item.fileId,
|
||||||
|
caption: item.fileName,
|
||||||
|
};
|
||||||
|
});
|
||||||
|
|
||||||
|
const uploadResult = await executeWithBotRetry((activeBot) => {
|
||||||
|
return activeBot.telegram.sendMediaGroup(config.storageChatId, mediaGroup);
|
||||||
|
});
|
||||||
|
|
||||||
|
currentBotIndex = (currentBotIndex + 1) % bots.length;
|
||||||
|
|
||||||
|
return uploadResult;
|
||||||
|
});
|
||||||
|
|
||||||
|
const messages = Array.isArray(result) ? result : [result];
|
||||||
|
const storageMessageId = messages[0]?.message_id || 0;
|
||||||
|
|
||||||
|
const telegramFileIds: string[] = [];
|
||||||
|
const telegramFileUniqueIds: string[] = [];
|
||||||
|
|
||||||
|
for (let i = 0; i < messages.length; i++) {
|
||||||
|
const msg = messages[i];
|
||||||
|
const fileType = items[i]?.fileType || 'document';
|
||||||
|
let uploadedFile: any;
|
||||||
|
|
||||||
|
if (msg.document) uploadedFile = msg.document;
|
||||||
|
else if (msg.photo) uploadedFile = msg.photo?.slice(-1)[0];
|
||||||
|
else if (msg.video) uploadedFile = msg.video;
|
||||||
|
else if (msg.audio) uploadedFile = msg.audio;
|
||||||
|
else if (msg.voice) uploadedFile = msg.voice;
|
||||||
|
else if (msg.animation) uploadedFile = msg.animation;
|
||||||
|
else if (msg.sticker) uploadedFile = msg.sticker;
|
||||||
|
else if (msg.video_note) uploadedFile = msg.video_note;
|
||||||
|
else uploadedFile = msg[fileType];
|
||||||
|
|
||||||
|
telegramFileIds.push(uploadedFile?.file_id || '');
|
||||||
|
telegramFileUniqueIds.push(uploadedFile?.file_unique_id || '');
|
||||||
|
}
|
||||||
|
|
||||||
|
return {
|
||||||
|
storageMessageId,
|
||||||
|
telegramFileIds,
|
||||||
|
telegramFileUniqueIds,
|
||||||
|
};
|
||||||
|
} catch (error: any) {
|
||||||
|
logger.error('Failed to forward media group to storage', { error: error.message });
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
export const getFileInfo = async (
|
export const getFileInfo = async (
|
||||||
telegramFileId: string,
|
telegramFileId: string,
|
||||||
telegramFileUniqueId: string,
|
telegramFileUniqueId: string,
|
||||||
|
|||||||
Reference in New Issue
Block a user