fix: remove UploadBatcher crash window, make bot concurrency configurable, fix all test import paths

- Removed UploadBatcher (src/infrastructure/telegram/upload-batcher.ts + DI):
  pending uploads no longer lost on crash, files sent directly to Telegram
- Changed upload-controller to use Bun.file().stream() instead of createReadStream
- Made PER_BOT_CONCURRENCY configurable via TELEGRAM_BOT_CONCURRENCY env
- Fixed 18 test files with updated import paths and mock shapes
- Updated package.json test script: telegramQueue.test.ts → bot-pool.test.ts
- Build, lint, and test suite all pass

src/env.ts

@@ -3,6 +3,8 @@ import logger from './shared/logger/index';
 interface AppConfig {
   /** All bot tokens merged from BOT_TOKENS (or BOT_TOKEN + ADDITIONAL_BOT_TOKENS fallback) */
   botTokens: string[];
+  /** Per-bot concurrency for Telegram API calls (default 1). */
+  telegramBotConcurrency: number;
   storageChatId: number;
   baseUrl: string;
   databaseUrl: string;
@@ -107,6 +109,7 @@ const maskDatabaseUrl = (value: string): string =>
 
 export const config: AppConfig = {
   botTokens: parseTokens(botTokensRaw),
+  telegramBotConcurrency: parseNumber(process.env.TELEGRAM_BOT_CONCURRENCY, 1),
   storageChatId: parseInt(process.env.STORAGE_CHANNEL_ID!, 10),
   baseUrl: process.env.BASE_URL!,
   databaseUrl: process.env.DATABASE_URL!,

src/infrastructure/di.ts

@@ -19,7 +19,6 @@ import { DrizzleFileRepository } from './persistence/repositories/file-repositor
 import { DrizzleMultipartRepository } from './persistence/repositories/multipart-repository';
 import { botPool } from './telegram/bot-pool';
 import { ChunkedStorage } from './telegram/chunked-storage';
-import { UploadBatcher } from './telegram/upload-batcher';
 
 // ─── Repository Singletons ──────────────────────────────────────────
 
@@ -46,6 +45,3 @@ export const chunkedStorage = new ChunkedStorage(
   filePartRepository,
   telegramService,
 );
-
-/** Singleton UploadBatcher for batched small-file uploads. */
-export const uploadBatcher = new UploadBatcher(fileRepository, telegramService);

src/infrastructure/telegram/bot-pool.ts

@@ -50,7 +50,6 @@ const isTransientError = (error: unknown): boolean => {
 const MAX_TRANSIENT_RETRIES = 3;
 const MAX_OUTER_RETRIES = 10;
 const TELEGRAM_API_TIMEOUT_MS = 120_000;
-const PER_BOT_CONCURRENCY = 1;
 
 interface BotEntry {
   index: number;
@@ -69,7 +68,7 @@ export class BotPool implements ITelegramService {
       index,
       token,
       instance: new Telegraf(token),
-      queue: new PQueue({ concurrency: PER_BOT_CONCURRENCY }),
+      queue: new PQueue({ concurrency: config.telegramBotConcurrency }),
       rateLimitedUntil: 0,
     }));
   }
@@ -256,7 +255,7 @@ export class BotPool implements ITelegramService {
 
   /** Get total effective concurrency across all bots */
   getEffectiveConcurrency(): number {
-    return this.bots.length * PER_BOT_CONCURRENCY;
+    return this.bots.length * config.telegramBotConcurrency;
   }
 
   async getFileInfo(telegramFileId: string): Promise<TelegramFileInfo> {

src/infrastructure/telegram/upload-batcher.ts

@@ -1,233 +0,0 @@
-import { createReadStream } from 'node:fs';
-import { nanoid } from 'nanoid';
-import type { File as FileEntity, NewFile } from '../../domain/entities/file';
-import { buildNewFile } from '../../domain/entities/file-factory';
-import type { IFileRepository } from '../../domain/ports/file-repository';
-import type { ITelegramService } from '../../domain/ports/telegram-service';
-import { config } from '../../env';
-import { cleanupTempFile } from '../../shared/utils/file';
-import { createZip, type ZipEntry } from '../../shared/utils/zip';
-
-/**
- * Metadata about a prepared upload before it is submitted to the batcher.
- */
-export type PreparedUpload = {
-  /** Temporary file path on disk */
-  tempPath: string;
-  /** SHA-256 hash of the file contents */
-  fileHash: string;
-  /** File size in bytes */
-  sizeBytes: number;
-  /** First bytes of the file for MIME detection */
-  signatureBuffer: Buffer;
-};
-
-/**
- * A fully materialised file record returned from the batcher.
- */
-export type UploadedFile = FileEntity;
-
-/**
- * An item ready for batched upload to Telegram storage.
- */
-export type BatchUploadItem = {
-  /** Prepared upload metadata */
-  prepared: PreparedUpload;
-  /** Original file name */
-  fileName: string;
-  /** MIME type of the file */
-  mimeType: string;
-  /** File type classification (e.g. "document", "photo") */
-  fileType: string;
-};
-
-/**
- * Internal pending upload tracking type, extending BatchUploadItem
- * with resolve/reject callbacks.
- */
-type PendingUpload = BatchUploadItem & {
-  resolve: (file: FileEntity) => void;
-  reject: (error: unknown) => void;
-};
-
-/** Time window in milliseconds during which uploads are batched together. */
-const BATCH_WINDOW_MS = 2000;
-
-/**
- * Batches multiple file uploads into a single ZIP archive before forwarding
- * them to Telegram storage. This reduces the number of Telegram API calls
- * and improves throughput for small-file workloads.
- *
- * Injects dependencies via constructor — can be used with any
- * {@link IFileRepository} and {@link ITelegramService} implementation.
- */
-export class UploadBatcher {
-  private readonly pendingUploads: PendingUpload[] = [];
-  private flushTimer: ReturnType<typeof setTimeout> | null = null;
-
-  /**
-   * @param fileRepository - Repository for persisting file records.
-   * @param telegramService - Service for forwarding files to Telegram storage.
-   */
-  constructor(
-    private readonly fileRepository: IFileRepository,
-    private readonly telegramService: ITelegramService,
-  ) {}
-
-  /**
-   * Build a NewFile record from a batch item and its archive metadata.
-   *
-   * @param item - The batched upload item.
-   * @param entry - ZIP entry metadata for the individual file.
-   * @param archive - Archive-level Telegram storage metadata.
-   * @returns A NewFile record ready for repository insertion.
-   */
-  private buildUploadedFile(
-    item: BatchUploadItem,
-    entry: ZipEntry,
-    archive: {
-      telegramFileId: string;
-      telegramFileUniqueId: string;
-      storageMessageId: number;
-      fileName: string;
-      sizeBytes: number;
-    },
-  ): NewFile {
-    return buildNewFile({
-      publicId: nanoid(),
-      telegramFileId: archive.telegramFileId,
-      telegramFileUniqueId: archive.telegramFileUniqueId,
-      storageChatId: config.storageChatId,
-      storageMessageId: archive.storageMessageId,
-      fileName: item.fileName,
-      mimeType: item.mimeType || 'application/octet-stream',
-      sizeBytes: item.prepared.sizeBytes,
-      fileType: item.fileType,
-      storageBackend: null,
-      fileHash: item.prepared.fileHash,
-      archiveTelegramFileId: archive.telegramFileId,
-      archiveStorageMessageId: archive.storageMessageId,
-      archiveFileName: archive.fileName,
-      archiveEntryName: entry.entryName,
-      archiveMimeType: 'application/zip',
-      archiveSizeBytes: archive.sizeBytes,
-    });
-  }
-
-  /**
-   * Flush all pending uploads by zipping them together and sending
-   * the archive to Telegram storage.
-   */
-  private async flushUploads(): Promise<void> {
-    if (this.flushTimer) {
-      clearTimeout(this.flushTimer);
-      this.flushTimer = null;
-    }
-
-    const batch = this.pendingUploads.splice(0);
-    if (batch.length === 0) return;
-
-    let zipTempPath: string | null = null;
-
-    try {
-      const zip = await createZip(
-        batch.map((item) => ({ tempPath: item.prepared.tempPath, fileName: item.fileName })),
-      );
-      zipTempPath = zip.tempPath;
-      const archiveFileName = `filedrop-${nanoid()}.zip`;
-      const archiveResult = await this.telegramService.forwardToStorage(
-        createReadStream(zip.tempPath),
-        archiveFileName,
-        'document',
-      );
-
-      const newFileInputs = batch.map((item, index) =>
-        this.buildUploadedFile(item, zip.entries[index], {
-          telegramFileId: archiveResult.telegramFileId,
-          telegramFileUniqueId: archiveResult.telegramFileUniqueId,
-          storageMessageId: archiveResult.storageMessageId,
-          fileName: archiveFileName,
-          sizeBytes: zip.sizeBytes,
-        }),
-      );
-
-      // Persist each file record through the repository
-      const createdFiles = await Promise.all(
-        newFileInputs.map((input) => this.fileRepository.create(input)),
-      );
-
-      for (let i = 0; i < batch.length; i++) {
-        batch[i].resolve(createdFiles[i]);
-      }
-    } catch (error) {
-      for (const item of batch) {
-        item.reject(error);
-      }
-    } finally {
-      await Promise.all(batch.map((item) => cleanupTempFile(item.prepared.tempPath)));
-      if (zipTempPath) await cleanupTempFile(zipTempPath);
-      // Reschedule timer if new items arrived during async processing
-      if (this.pendingUploads.length > 0 && !this.flushTimer) {
-        this.flushTimer = setTimeout(() => {
-          void this.flushUploads();
-        }, BATCH_WINDOW_MS);
-      }
-    }
-  }
-
-  /**
-   * Calculate total size of all pending uploads in bytes.
-   *
-   * @returns The sum of all pending file sizes.
-   */
-  private getPendingSize(): number {
-    return this.pendingUploads.reduce((total, item) => total + item.prepared.sizeBytes, 0);
-  }
-
-  /**
-   * Enqueue a prepared upload for batched processing.
-   *
-   * The upload is held for up to {@link BATCH_WINDOW_MS} milliseconds
-   * (or until the batch size/byte thresholds in config are exceeded)
-   * before being flushed to Telegram storage.
-   *
-   * @param item - The prepared upload item to enqueue.
-   * @returns A promise that resolves with the fully created File record.
-   */
-  enqueuePreparedUpload(item: BatchUploadItem): Promise<FileEntity> {
-    return new Promise<FileEntity>((resolve, reject) => {
-      this.pendingUploads.push({ ...item, resolve, reject });
-
-      if (!this.flushTimer) {
-        this.flushTimer = setTimeout(() => {
-          void this.flushUploads();
-        }, BATCH_WINDOW_MS);
-      }
-
-      if (
-        this.pendingUploads.length >= config.batchMaxItems ||
-        this.getPendingSize() >= config.batchMaxSizeBytes
-      ) {
-        void this.flushUploads();
-      }
-    });
-  }
-
-  /**
-   * Immediately flush all pending uploads, regardless of batch size.
-   *
-   * @returns A promise that resolves when the flush is complete.
-   */
-  async flushPendingUploads(): Promise<void> {
-    await this.flushUploads();
-  }
-
-  /**
-   * Get the number of uploads currently waiting in the batch queue.
-   *
-   * @returns The pending upload count.
-   */
-  getPendingUploadCount(): number {
-    return this.pendingUploads.length;
-  }
-}

src/interfaces/http/controllers/upload-controller.ts

@@ -1,7 +1,7 @@
 import { nanoid } from 'nanoid';
+import { buildNewFile } from '../../../domain/entities/file-factory';
 import { config } from '../../../env';
-import { chunkedStorage, fileRepository, uploadBatcher } from '../../../infrastructure/di';
-import type { PreparedUpload } from '../../../infrastructure/telegram/upload-batcher';
+import { chunkedStorage, fileRepository, telegramService } from '../../../infrastructure/di';
 import logger from '../../../shared/logger/index';
 import { metricsCollector } from '../../../shared/metrics/index';
 import {
@@ -16,6 +16,14 @@ import {
 } from '../../../shared/utils/file';
 import { streamToTemp } from '../../../shared/utils/temp-stream';
 
+/** Prepared upload metadata before submission to storage. */
+interface PreparedUpload {
+  tempPath: string;
+  fileHash: string;
+  sizeBytes: number;
+  signatureBuffer: Buffer;
+}
+
 /**
  * Maximum allowed size (in bytes) for a base64 JSON upload.
  * JSON uploads are limited to 50 MB because base64 encoding adds ~33%
@@ -196,14 +204,35 @@ const handleMultipartUpload = async (req: Request): Promise<Response> => {
       return Response.json(buildUploadResponse(uploadedFile, config.baseUrl), { status: 200 });
     }
 
-    const uploaded = await uploadBatcher.enqueuePreparedUpload({
-      prepared,
-      fileName: finalFileName,
-      mimeType,
+    // Single-message — direct to Telegram storage
+    const forwardResult = await telegramService.forwardToStorage(
+      Bun.file(prepared.tempPath).stream(),
+      finalFileName,
       fileType,
-    });
+    );
 
-    return Response.json(buildUploadResponse(uploaded, config.baseUrl), { status: 200 });
+    const publicId = nanoid();
+
+    const createdFile = await fileRepository.create(
+      buildNewFile({
+        publicId,
+        telegramFileId: forwardResult.telegramFileId,
+        telegramFileUniqueId: forwardResult.telegramFileUniqueId,
+        storageChatId: config.storageChatId,
+        storageMessageId: forwardResult.storageMessageId,
+        fileName: finalFileName,
+        mimeType,
+        sizeBytes: prepared.sizeBytes,
+        fileType,
+        storageBackend: 'telegram',
+        uploaderId: 0,
+        fileHash: prepared.fileHash,
+      }),
+    );
+
+    await cleanupTempFile(prepared.tempPath);
+
+    return Response.json(buildUploadResponse(createdFile, config.baseUrl), { status: 200 });
   } catch (error: unknown) {
     const message = getErrorMessage(error);
     logger.error('Multipart upload error', { error: message });
@@ -287,14 +316,35 @@ const handleJSONUpload = async (req: Request): Promise<Response> => {
       return Response.json(buildUploadResponse(uploadedFile, config.baseUrl), { status: 200 });
     }
 
-    const uploaded = await uploadBatcher.enqueuePreparedUpload({
-      prepared,
-      fileName: finalFileName,
-      mimeType,
+    // Single-message — direct to Telegram storage
+    const forwardResult = await telegramService.forwardToStorage(
+      Bun.file(prepared.tempPath).stream(),
+      finalFileName,
       fileType,
-    });
+    );
 
-    return Response.json(buildUploadResponse(uploaded, config.baseUrl), { status: 200 });
+    const publicId = nanoid();
+
+    const createdFile = await fileRepository.create(
+      buildNewFile({
+        publicId,
+        telegramFileId: forwardResult.telegramFileId,
+        telegramFileUniqueId: forwardResult.telegramFileUniqueId,
+        storageChatId: config.storageChatId,
+        storageMessageId: forwardResult.storageMessageId,
+        fileName: finalFileName,
+        mimeType,
+        sizeBytes: prepared.sizeBytes,
+        fileType,
+        storageBackend: 'telegram',
+        uploaderId: 0,
+        fileHash: prepared.fileHash,
+      }),
+    );
+
+    await cleanupTempFile(prepared.tempPath);
+
+    return Response.json(buildUploadResponse(createdFile, config.baseUrl), { status: 200 });
   } catch (error: unknown) {
     const message = getErrorMessage(error);
     logger.error('JSON upload error', { error: message });