diff --git a/.gitignore b/.gitignore index 2d274e5..07eb103 100644 --- a/.gitignore +++ b/.gitignore @@ -5,7 +5,7 @@ node_modules out dist *.tgz - +.codegraph # code coverage coverage *.lcov diff --git a/src/routes/files.ts b/src/routes/files.ts index a76d96b..79fe3d1 100644 --- a/src/routes/files.ts +++ b/src/routes/files.ts @@ -1,9 +1,9 @@ import { findFileByPublicId } from '../db/files'; +import { fileInfoCache } from '../utils/cache'; import { formatCreatedAt, getErrorMessage } from '../utils/file'; import logger from '../utils/logger'; import { checkRateLimit } from '../utils/rateLimit'; import { getBot } from '../utils/telegram'; -import { fileInfoCache } from '../utils/cache'; type RequestWithParams = Request & { params?: { diff --git a/src/routes/upload.ts b/src/routes/upload.ts index 31ed1b8..71b8aa4 100644 --- a/src/routes/upload.ts +++ b/src/routes/upload.ts @@ -1,4 +1,4 @@ -import { createReadStream } from 'node:fs'; +import { createReadStream, createWriteStream } from 'node:fs'; import { unlink } from 'node:fs/promises'; import { nanoid } from 'nanoid'; import { db, files as fileSchema } from '../db'; @@ -43,16 +43,118 @@ const normalizeFileType = (mimeType: string, fileName: string): string => { return fileType === 'application' ? 'document' : fileType; }; -const performUpload = async ( - fileBuffer: Buffer, - fileName: string, - mimeType: string, -): Promise => { +const JSON_UPLOAD_LIMIT_BYTES = 50 * 1024 * 1024; +const SIGNATURE_BYTES = 16; + +type PreparedUpload = { + tempPath: string; + fileHash: string; + sizeBytes: number; + signatureBuffer: Buffer; +}; + +const cleanupTempFile = async (tempPath: string): Promise => { + try { + await unlink(tempPath); + } catch (err) { + logger.warn('Failed to cleanup temp file', { tempPath, error: getErrorMessage(err) }); + } +}; + +const streamFileToTemp = async (file: File): Promise => { + const tempPath = `/tmp/teleuploader-${nanoid()}`; + const writer = createWriteStream(tempPath); + const hasher = new Bun.CryptoHasher('sha256'); + const reader = file.stream().getReader(); + const signatureChunks: Buffer[] = []; + let signatureBytes = 0; + let sizeBytes = 0; + + const writeChunk = async (chunk: Buffer): Promise => { + if (!writer.write(chunk)) { + await new Promise((resolve, reject) => { + writer.once('drain', resolve); + writer.once('error', reject); + }); + } + }; + + const finishWriter = async (): Promise => { + await new Promise((resolve, reject) => { + writer.end(() => resolve()); + writer.once('error', reject); + }); + }; + + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + const chunk = Buffer.from(value); + sizeBytes += chunk.byteLength; + hasher.update(chunk); + await writeChunk(chunk); + + if (signatureBytes < SIGNATURE_BYTES) { + const remaining = SIGNATURE_BYTES - signatureBytes; + const signatureChunk = chunk.subarray(0, remaining); + signatureChunks.push(signatureChunk); + signatureBytes += signatureChunk.byteLength; + } + } + + await finishWriter(); + + return { + tempPath, + fileHash: hasher.digest('hex'), + sizeBytes, + signatureBuffer: Buffer.concat(signatureChunks, signatureBytes), + }; + } catch (error) { + writer.destroy(); + await cleanupTempFile(tempPath); + throw error; + } finally { + reader.releaseLock(); + } +}; + +const writeBufferToTemp = async (fileBuffer: Buffer, fileHash: string): Promise => { const tempPath = `/tmp/teleuploader-${nanoid()}`; try { await Bun.write(tempPath, fileBuffer); - const fileStream = createReadStream(tempPath); - const result = await forwardToStorage(fileStream, fileName, getFileType(mimeType, fileName)); + return { + tempPath, + fileHash, + sizeBytes: fileBuffer.byteLength, + signatureBuffer: fileBuffer.subarray(0, SIGNATURE_BYTES), + }; + } catch (error) { + await cleanupTempFile(tempPath); + throw error; + } +}; + +const closeFileStream = async (fileStream: ReturnType): Promise => { + if (fileStream.closed) return; + + await new Promise((resolve) => { + fileStream.once('close', resolve); + fileStream.destroy(); + }); +}; + +const performUpload = async ( + prepared: PreparedUpload, + fileName: string, + mimeType: string, + fileType: string, +): Promise => { + const fileStream = createReadStream(prepared.tempPath); + try { + const result = await forwardToStorage(fileStream, fileName, fileType); return { publicId: nanoid(), @@ -62,21 +164,16 @@ const performUpload = async ( storageMessageId: result.storageMessageId, fileName, mimeType: mimeType || 'application/octet-stream', - sizeBytes: fileBuffer.byteLength, - fileType: getFileType(mimeType, fileName), + sizeBytes: prepared.sizeBytes, + fileType, uploaderId: 0, - fileHash: computeHash(fileBuffer), + fileHash: prepared.fileHash, createdAt: new Date(), updatedAt: new Date(), }; } finally { - setTimeout(async () => { - try { - await unlink(tempPath); - } catch (err) { - logger.warn('Failed to cleanup temp file', { tempPath, error: getErrorMessage(err) }); - } - }, 500); + await closeFileStream(fileStream); + await cleanupTempFile(prepared.tempPath); } }; @@ -112,28 +209,28 @@ const handleMultipartUpload = async (req: Request): Promise => { return Response.json({ error: 'No file provided' }, { status: 400 }); } - const fileBytes = await file.arrayBuffer(); - const fileBuffer = Buffer.from(fileBytes); - const hash = computeHash(fileBuffer); + const prepared = await streamFileToTemp(file); - const existingFile = await findFileByHash(hash); + const existingFile = await findFileByHash(prepared.fileHash); if (existingFile) { + await cleanupTempFile(prepared.tempPath); return Response.json(buildUploadResponse(existingFile, config.baseUrl), { status: 200 }); } const rawMimeType = file.type || extractMimeType({}, req) || 'application/octet-stream'; const { fileName: finalFileName, mimeType } = ensureExtension( fileName, - fileBuffer, + prepared.signatureBuffer, rawMimeType, ); const fileType = getFileType(mimeType, finalFileName); - if (!checkFileSize(fileBuffer.byteLength, fileType)) { + if (!checkFileSize(prepared.sizeBytes, fileType)) { + await cleanupTempFile(prepared.tempPath); return Response.json({ error: `File size exceeds ${fileType} limit` }, { status: 400 }); } - const uploaded = await performUpload(fileBuffer, finalFileName, mimeType); + const uploaded = await performUpload(prepared, finalFileName, mimeType, fileType); await db.insert(fileSchema).values(uploaded); return Response.json(buildUploadResponse(uploaded, config.baseUrl), { status: 200 }); @@ -156,6 +253,17 @@ const handleJSONUpload = async (req: Request): Promise => { } const { base64Data, mimeType: rawMimeType } = parseBase64File(file); + const estimatedSizeBytes = Math.floor((base64Data.length * 3) / 4); + if (estimatedSizeBytes > JSON_UPLOAD_LIMIT_BYTES) { + return Response.json( + { + error: + 'JSON base64 uploads are limited to 50MB. Use multipart/form-data for larger files', + }, + { status: 400 }, + ); + } + const fileBytes = Buffer.from(base64Data, 'base64'); const hash = computeHash(fileBytes); @@ -171,7 +279,8 @@ const handleJSONUpload = async (req: Request): Promise => { return Response.json({ error: `File size exceeds ${fileType} limit` }, { status: 400 }); } - const uploaded = await performUpload(fileBytes, finalFileName, mimeType); + const prepared = await writeBufferToTemp(fileBytes, hash); + const uploaded = await performUpload(prepared, finalFileName, mimeType, fileType); await db.insert(fileSchema).values(uploaded); return Response.json(buildUploadResponse(uploaded, config.baseUrl), { status: 200 }); diff --git a/src/utils/cache.ts b/src/utils/cache.ts index df272dd..a12bffd 100644 --- a/src/utils/cache.ts +++ b/src/utils/cache.ts @@ -71,11 +71,14 @@ export const fileInfoCache = new Cache<{ }>(3600); // Cleanup expired cache entries every 5 minutes -setInterval(() => { - const removed = fileInfoCache.cleanup(); - if (removed > 0) { - console.log(`Cleaned up ${removed} expired cache entries`); - } -}, 5 * 60 * 1000); +setInterval( + () => { + const removed = fileInfoCache.cleanup(); + if (removed > 0) { + console.log(`Cleaned up ${removed} expired cache entries`); + } + }, + 5 * 60 * 1000, +); export { Cache }; diff --git a/src/utils/metrics.ts b/src/utils/metrics.ts index 9a88153..1358675 100644 --- a/src/utils/metrics.ts +++ b/src/utils/metrics.ts @@ -109,14 +109,17 @@ class MetricsCollector { export const metricsCollector = new MetricsCollector(); // Log metrics every 5 minutes -setInterval(() => { - const snapshot = metricsCollector.getSnapshot(); - logger.info('Metrics snapshot', { - uploadLatency: snapshot.uploadLatency, - uploadThroughput: snapshot.uploadThroughput.toFixed(2), - errorRate: snapshot.errorRate.toFixed(2), - cacheHitRate: snapshot.cacheHitRate.toFixed(2), - }); -}, 5 * 60 * 1000); +setInterval( + () => { + const snapshot = metricsCollector.getSnapshot(); + logger.info('Metrics snapshot', { + uploadLatency: snapshot.uploadLatency, + uploadThroughput: snapshot.uploadThroughput.toFixed(2), + errorRate: snapshot.errorRate.toFixed(2), + cacheHitRate: snapshot.cacheHitRate.toFixed(2), + }); + }, + 5 * 60 * 1000, +); export { MetricsCollector }; diff --git a/src/utils/telegram.ts b/src/utils/telegram.ts index 982c59e..ff892b1 100644 --- a/src/utils/telegram.ts +++ b/src/utils/telegram.ts @@ -7,12 +7,12 @@ const botTokens = Array.from(new Set([config.botToken, ...config.additionalBotTo const bots = botTokens.map((token) => new Telegraf(token)); -let currentBotIndex = 0; +let nextBotIndex = 0; -const rotateBot = (): { previousIndex: number; nextIndex: number } => { - const previousIndex = currentBotIndex; - currentBotIndex = (currentBotIndex + 1) % bots.length; - return { previousIndex, nextIndex: currentBotIndex }; +const claimBotIndex = (): number => { + const botIndex = nextBotIndex; + nextBotIndex = (nextBotIndex + 1) % bots.length; + return botIndex; }; const sleep = (seconds: number): Promise => { @@ -24,7 +24,8 @@ const executeWithBotRetry = async ( retries = 5, attemptedBots = 0, ): Promise => { - const currentBot = bots[currentBotIndex]; + const botIndex = claimBotIndex(); + const currentBot = bots[botIndex]; try { return await action(currentBot); } catch (error: unknown) { @@ -32,17 +33,16 @@ const executeWithBotRetry = async ( const match = errorStr.match(/retry after (\d+)/i); if (match) { - const { previousIndex, nextIndex } = rotateBot(); - attemptedBots++; + const nextIndex = nextBotIndex; + const nextAttemptedBots = attemptedBots + 1; - if (attemptedBots < bots.length) { + if (nextAttemptedBots < bots.length) { logger.info( - `Bot Index ${previousIndex} hit 429. Instantly rotating to Bot Index ${nextIndex}...`, + `Bot Index ${botIndex} hit 429. Instantly rotating to Bot Index ${nextIndex}...`, ); - return executeWithBotRetry(action, retries, attemptedBots); + return executeWithBotRetry(action, retries, nextAttemptedBots); } - // 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...`, { @@ -159,13 +159,10 @@ export const forwardToStorage = async ( const sendMethod = sendMethodMap[fileType] || 'sendDocument'; const payload = buildSendPayload(fileType, fileName); - const uploadResult = await executeWithBotRetry((activeBot) => { + return executeWithBotRetry((activeBot) => { const telegram = activeBot.telegram as unknown as Record; return telegram[sendMethod](config.storageChatId, filePayload, payload); }); - - currentBotIndex = (currentBotIndex + 1) % bots.length; - return uploadResult; }); const uploadedFile = extractUploadedFile(result, fileType); @@ -202,16 +199,13 @@ export const forwardMediaGroupToStorage = async ( const result = await enqueueUpload(async (): Promise => { const mediaGroup = buildMediaGroup(items); - const uploadResult = await executeWithBotRetry((activeBot) => { + return executeWithBotRetry((activeBot) => { const sendMediaGroup = activeBot.telegram.sendMediaGroup as unknown as ( chatId: number, media: MediaGroupPayloadItem[], ) => Promise; return sendMediaGroup(config.storageChatId, mediaGroup); }); - - currentBotIndex = (currentBotIndex + 1) % bots.length; - return uploadResult; }); const messages = Array.isArray(result) ? result : [result]; @@ -239,9 +233,7 @@ export const forwardMediaGroupToStorage = async ( } }; -export const getFileInfo = async ( - telegramFileId: string, -): Promise => { +export const getFileInfo = async (telegramFileId: string): Promise => { try { const result = await executeWithBotRetry((activeBot) => activeBot.telegram.getFile(telegramFileId), @@ -261,6 +253,6 @@ export const getFileInfo = async ( } }; -export const getBot = (): Telegraf => bots[currentBotIndex]; +export const getBot = (): Telegraf => bots[nextBotIndex]; -export const getCurrentBotIndex = (): number => currentBotIndex; +export const getCurrentBotIndex = (): number => nextBotIndex; diff --git a/src/utils/telegramQueue.ts b/src/utils/telegramQueue.ts index 18f9882..d689559 100644 --- a/src/utils/telegramQueue.ts +++ b/src/utils/telegramQueue.ts @@ -1,14 +1,8 @@ import PQueue from 'p-queue'; import logger from './logger'; -// Create queue with concurrency limit matching bot pool size -// Concurrency: 4-8 uploads in parallel -// Interval: 1 second window for rate limiting -// IntervalCap: Max 10 tasks per second const uploadQueue = new PQueue({ - concurrency: 4, - interval: 1000, - intervalCap: 10, + concurrency: Number.POSITIVE_INFINITY, }); // Monitor queue events