feat: enhance file upload process with temporary file handling, improved error management, and metrics logging

.gitignore

@@ -5,7 +5,7 @@ node_modules
 out
 dist
 *.tgz
-
+.codegraph
 # code coverage
 coverage
 *.lcov

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?: {

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<UploadedFile> => {
+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<void> => {
+  try {
+    await unlink(tempPath);
+  } catch (err) {
+    logger.warn('Failed to cleanup temp file', { tempPath, error: getErrorMessage(err) });
+  }
+};
+
+const streamFileToTemp = async (file: File): Promise<PreparedUpload> => {
+  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<void> => {
+    if (!writer.write(chunk)) {
+      await new Promise<void>((resolve, reject) => {
+        writer.once('drain', resolve);
+        writer.once('error', reject);
+      });
+    }
+  };
+
+  const finishWriter = async (): Promise<void> => {
+    await new Promise<void>((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<PreparedUpload> => {
   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<typeof createReadStream>): Promise<void> => {
+  if (fileStream.closed) return;
+
+  await new Promise<void>((resolve) => {
+    fileStream.once('close', resolve);
+    fileStream.destroy();
+  });
+};
+
+const performUpload = async (
+  prepared: PreparedUpload,
+  fileName: string,
+  mimeType: string,
+  fileType: string,
+): Promise<UploadedFile> => {
+  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<Response> => {
       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<Response> => {
     }
 
     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<Response> => {
       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 });

src/utils/cache.ts

@@ -71,11 +71,14 @@ export const fileInfoCache = new Cache<{
 }>(3600);
 
 // Cleanup expired cache entries every 5 minutes
-setInterval(() => {
+setInterval(
+  () => {
     const removed = fileInfoCache.cleanup();
     if (removed > 0) {
       console.log(`Cleaned up ${removed} expired cache entries`);
     }
-}, 5 * 60 * 1000);
+  },
+  5 * 60 * 1000,
+);
 
 export { Cache };

src/utils/metrics.ts

@@ -109,7 +109,8 @@ class MetricsCollector {
 export const metricsCollector = new MetricsCollector();
 
 // Log metrics every 5 minutes
-setInterval(() => {
+setInterval(
+  () => {
     const snapshot = metricsCollector.getSnapshot();
     logger.info('Metrics snapshot', {
       uploadLatency: snapshot.uploadLatency,
@@ -117,6 +118,8 @@ setInterval(() => {
       errorRate: snapshot.errorRate.toFixed(2),
       cacheHitRate: snapshot.cacheHitRate.toFixed(2),
     });
-}, 5 * 60 * 1000);
+  },
+  5 * 60 * 1000,
+);
 
 export { MetricsCollector };

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<void> => {
@@ -24,7 +24,8 @@ const executeWithBotRetry = async <T>(
   retries = 5,
   attemptedBots = 0,
 ): Promise<T> => {
-  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 <T>(
     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<string, SendMethod>;
         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<TelegramMessageResult[]> => {
       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<TelegramMessageResult[]>;
         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<TelegramFileInfo> => {
+export const getFileInfo = async (telegramFileId: string): Promise<TelegramFileInfo> => {
   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;

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