fix(core): resolve wrong file_id error and zip stream memory leaks
- Refactored Telegram bot pool iteration to handle 'wrong file_id' properly by verifying against all active bots - Upgraded TelegramFileInfo interface to track owner bot_token - Patched MaxListenersExceededWarning memory leak in zip.ts stream writes with events.once and async iteration
This commit is contained in:
+2
-2
@@ -16,7 +16,7 @@ type RequestWithParams = Request & {
|
||||
|
||||
const getTelegramFileInfo = async (telegramFileId: string, public_id: string) => {
|
||||
const cacheKey = `file_info_${telegramFileId}`;
|
||||
let fileInfo = fileInfoCache.get(cacheKey);
|
||||
let fileInfo = fileInfoCache.get(cacheKey) as any;
|
||||
|
||||
if (!fileInfo) {
|
||||
fileInfo = await getFileInfo(telegramFileId);
|
||||
@@ -26,7 +26,7 @@ const getTelegramFileInfo = async (telegramFileId: string, public_id: string) =>
|
||||
logger.debug('File info from cache', { public_id, cacheKey });
|
||||
}
|
||||
|
||||
return fileInfo;
|
||||
return fileInfo as { file_size: number; mime_type: string; file_path: string; bot_token: string };
|
||||
};
|
||||
|
||||
const buildTelegramFileUrl = (filePath: string, botToken: string): string =>
|
||||
|
||||
+28
-20
@@ -241,27 +241,35 @@ export const forwardMediaGroupToStorage = async (
|
||||
};
|
||||
|
||||
export const getFileInfo = async (telegramFileId: string): Promise<TelegramFileInfo> => {
|
||||
try {
|
||||
const { result, botToken } = await executeWithBotRetry<FileInfoResult>(
|
||||
async (activeBot, activeToken) => ({
|
||||
result: await activeBot.telegram.getFile(telegramFileId),
|
||||
botToken: activeToken,
|
||||
}),
|
||||
);
|
||||
|
||||
const fileData = result as unknown as TelegramFileInfo;
|
||||
return {
|
||||
file_size: fileData.file_size || 0,
|
||||
mime_type: fileData.mime_type || 'application/octet-stream',
|
||||
file_path: fileData.file_path || '',
|
||||
bot_token: botToken,
|
||||
};
|
||||
} catch (error: unknown) {
|
||||
logger.error('Failed to get file info', {
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
throw error;
|
||||
let lastError: unknown;
|
||||
for (const activeBot of bots) {
|
||||
try {
|
||||
const result = await activeBot.telegram.getFile(telegramFileId);
|
||||
const fileData = result as unknown as Omit<TelegramFileInfo, 'bot_token'>;
|
||||
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,
|
||||
};
|
||||
} catch (error: unknown) {
|
||||
lastError = error;
|
||||
const errorStr = error instanceof Error ? error.message : String(error);
|
||||
if (
|
||||
errorStr.includes('wrong file_id') ||
|
||||
errorStr.includes('file is temporarily unavailable') ||
|
||||
errorStr.includes('retry after')
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
logger.error('Failed to get file info from any bot', {
|
||||
error: lastError instanceof Error ? lastError.message : String(lastError),
|
||||
});
|
||||
throw lastError;
|
||||
};
|
||||
|
||||
export const getBot = (): Telegraf => bots[nextBotIndex];
|
||||
|
||||
+13
-26
@@ -1,4 +1,6 @@
|
||||
import { once } from 'node:events';
|
||||
import { createReadStream, createWriteStream } from 'node:fs';
|
||||
import { finished } from 'node:stream/promises';
|
||||
import { open, stat } from 'node:fs/promises';
|
||||
import { basename } from 'node:path';
|
||||
import { nanoid } from 'nanoid';
|
||||
@@ -65,18 +67,13 @@ const writeChunk = async (
|
||||
chunk: Buffer,
|
||||
): Promise<void> => {
|
||||
if (!writer.write(chunk)) {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
writer.once('drain', resolve);
|
||||
writer.once('error', reject);
|
||||
});
|
||||
await once(writer, 'drain');
|
||||
}
|
||||
};
|
||||
|
||||
const finishWriter = async (writer: ReturnType<typeof createWriteStream>): Promise<void> => {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
writer.end(() => resolve());
|
||||
writer.once('error', reject);
|
||||
});
|
||||
writer.end();
|
||||
await finished(writer);
|
||||
};
|
||||
|
||||
export const sanitizeZipEntryName = (fileName: string, usedNames = new Set<string>()): string => {
|
||||
@@ -102,16 +99,10 @@ export const sanitizeZipEntryName = (fileName: string, usedNames = new Set<strin
|
||||
|
||||
const calculateFileCrc32 = async (tempPath: string): Promise<number> => {
|
||||
let crc = 0xffffffff;
|
||||
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const reader = createReadStream(tempPath);
|
||||
reader.on('data', (chunk: Buffer) => {
|
||||
crc = updateCrc32(crc, chunk);
|
||||
});
|
||||
reader.once('end', resolve);
|
||||
reader.once('error', reject);
|
||||
});
|
||||
|
||||
const reader = createReadStream(tempPath);
|
||||
for await (const chunk of reader) {
|
||||
crc = updateCrc32(crc, chunk as Buffer);
|
||||
}
|
||||
return (crc ^ 0xffffffff) >>> 0;
|
||||
};
|
||||
|
||||
@@ -154,14 +145,10 @@ export const createZip = async (files: ZipInputFile[]): Promise<CreatedZip> => {
|
||||
]);
|
||||
|
||||
await writeHashed(localHeader);
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const reader = createReadStream(file.tempPath);
|
||||
reader.on('data', (chunk: Buffer) => {
|
||||
void writeHashed(chunk).catch(reject);
|
||||
});
|
||||
reader.once('end', resolve);
|
||||
reader.once('error', reject);
|
||||
});
|
||||
const reader = createReadStream(file.tempPath);
|
||||
for await (const chunk of reader) {
|
||||
await writeHashed(chunk as Buffer);
|
||||
}
|
||||
|
||||
entries.push({
|
||||
fileName: file.fileName,
|
||||
|
||||
Reference in New Issue
Block a user