perf: parallelize S3 object part fetch + cache Telegram file info

Optimize the S3 GET path for chunked/multipart objects and reduce Telegram
API round-trips:

- object-stream: fetch object parts concurrently (bounded, in-order fan-in)
  instead of serializing N sequential Telegram CDN fetches. Response latency
  is now ~the slowest part fetch, not the sum of all part fetches.
- bot-pool.getFileInfo: cache file_id -> file_path in the existing in-memory
  cache so repeated S3 GET/HEAD of the same object skip the Telegram API call
  (file-controller had its own cache wrapper; the S3 path did not).
- chunked-storage + s3-controller: resolve multipart/chunked part CDN URLs
  concurrently via Promise.all instead of sequentially.
- s3-controller multipart: await writer.end() before re-reading the temp part
  file to avoid a flush race.

Adds object-stream-parallel.test.ts covering in-order fan-in, byte ranges,
and single-part responses even when the slowest part resolves out of order.

src/infrastructure/telegram/bot-pool.ts

@@ -7,6 +7,7 @@ import type {
 } from '../../domain/ports/telegram-service';
 import { config } from '../../env';
 import logger from '../../shared/logger/index';
+import { fileInfoCache } from '../cache/index';
 import {
   buildSendPayload,
   extractUploadedFile,
@@ -259,18 +260,31 @@ export class BotPool implements ITelegramService {
   }
 
   async getFileInfo(telegramFileId: string): Promise<TelegramFileInfo> {
+    // Telegram file_id → file_path mapping is stable for the lifetime of the
+    // file. Cache it to avoid a Telegram API round-trip on every S3 GET / HEAD
+    // of the same object. TTL 1h; a cached (possibly stale) file_path would
+    // only surface if Telegram recycles a file_id, which it does not for
+    // documents we own.
+    const cacheKey = `file_info_${telegramFileId}`;
+    const cached = fileInfoCache.get(cacheKey) as TelegramFileInfo | null;
+    if (cached) {
+      return cached;
+    }
+
     let lastError: unknown;
     for (const bot of this.bots) {
       for (let retry = 0; retry <= MAX_TRANSIENT_RETRIES; retry++) {
         try {
           const result = await bot.instance.telegram.getFile(telegramFileId);
           const fileData = result as unknown as Omit<TelegramFileInfo, 'bot_token'>;
-          return {
+          const fileInfo: TelegramFileInfo = {
             file_size: fileData.file_size || 0,
             mime_type: fileData.mime_type || 'application/octet-stream',
             file_path: fileData.file_path || '',
             bot_token: bot.token,
           };
+          fileInfoCache.set(cacheKey, fileInfo);
+          return fileInfo;
         } catch (error: unknown) {
           lastError = error;
           const errorStr = error instanceof Error ? error.message : String(error);

src/infrastructure/telegram/chunked-storage.ts

@@ -229,21 +229,27 @@ export class ChunkedStorage {
    */
   async buildChunkedObjectSources(file: FileEntity): Promise<ObjectPartSource[]> {
     const parts = await this.filePartRepository.listByFileId(file.id);
-    const sources: ObjectPartSource[] = [];
 
-    for (const part of parts) {
-      const fileInfo = await this.telegramService.getFileInfo(part.telegramFileId);
-      sources.push({
-        telegramFileId: part.telegramFileId,
-        telegramUrl: `https://api.telegram.org/file/bot${fileInfo.bot_token}/${fileInfo.file_path}`,
-        sizeBytes: part.sizeBytes,
-        storedSizeBytes: part.storedSizeBytes,
-        compressionAlgorithm: part.compressionAlgorithm,
-        partNumber: part.partNumber,
-      });
-    }
+    // Resolve every part's Telegram CDN URL concurrently (each is an independent
+    // getFile call) so total resolution time is ~1 round-trip, not N.
+    const partInfos = await Promise.all(
+      parts.map(async (part) => {
+        const fileInfo = await this.telegramService.getFileInfo(part.telegramFileId);
+        return {
+          part,
+          url: `https://api.telegram.org/file/bot${fileInfo.bot_token}/${fileInfo.file_path}`,
+        };
+      }),
+    );
 
-    return sources;
+    return partInfos.map(({ part, url }) => ({
+      telegramFileId: part.telegramFileId,
+      telegramUrl: url,
+      sizeBytes: part.sizeBytes,
+      storedSizeBytes: part.storedSizeBytes,
+      compressionAlgorithm: part.compressionAlgorithm,
+      partNumber: part.partNumber,
+    }));
   }
 
   /**

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

@@ -702,15 +702,21 @@ const handleGetMultipartObject = async (
   }
 
   const sources: ObjectPartSource[] = [];
-  for (const part of parts) {
-    const fileInfo = await botPool.getFileInfo(part.telegramFileId);
-    sources.push({
-      telegramFileId: part.telegramFileId,
-      telegramUrl: `https://api.telegram.org/file/bot${fileInfo.bot_token}/${fileInfo.file_path}`,
-      sizeBytes: part.sizeBytes,
-      partNumber: part.partNumber,
-    });
-  }
+  // Resolve all part CDN URLs concurrently (independent getFile calls) so
+  // assembly latency is ~1 round-trip instead of N.
+  sources.push(
+    ...(await Promise.all(
+      parts.map(async (part) => {
+        const fileInfo = await botPool.getFileInfo(part.telegramFileId);
+        return {
+          telegramFileId: part.telegramFileId,
+          telegramUrl: `https://api.telegram.org/file/bot${fileInfo.bot_token}/${fileInfo.file_path}`,
+          sizeBytes: part.sizeBytes,
+          partNumber: part.partNumber,
+        };
+      }),
+    )),
+  );
 
   // H1: Always proxy — never expose bot token in redirect URL
 
@@ -1479,9 +1485,13 @@ const handleUploadPart = async (
       hasher.update(chunk);
       writer.write(chunk);
     }
-    writer.end();
+    await writer.end();
   } catch (error) {
-    writer.end();
+    try {
+      writer.end();
+    } catch {
+      // ignore during error path
+    }
     await cleanupTempFile(tempPath);
     throw error;
   } finally {

src/interfaces/s3/object-stream.ts

@@ -97,12 +97,53 @@ const fetchPartBody = async (planned: PlannedPart): Promise<ReadableStream<Uint8
   return streamFromBytes(bytes.slice(planned.relativeStart, planned.relativeEnd + 1));
 };
 
+/**
+ * Maximum number of Telegram CDN part fetches run concurrently while
+ * assembling a chunked/multipart object response.
+ *
+ * Part fetches are initiated in parallel (bounded by this constant) to avoid
+ * serializing N sequential network round-trips on the Telegram CDN, then the
+ * results are fanned-in to the response stream in part-number order so the
+ * object bytes remain correctly ordered.
+ */
+const PART_FETCH_CONCURRENCY = 6;
+
+/**
+ * Runs an async mapper over the parts with bounded concurrency, returning the
+ * results in the same order as the input. Each worker claims the next not-yet-
+ * claimed index, so array slots are filled by exactly one worker each.
+ */
+const mapBounded = async <T, R>(
+  items: T[],
+  limit: number,
+  fn: (item: T, index: number) => Promise<R>,
+): Promise<R[]> => {
+  const results: R[] = new Array(items.length);
+  let next = 0;
+
+  const worker = async (): Promise<void> => {
+    while (true) {
+      const idx = next++;
+      if (idx >= items.length) return;
+      results[idx] = await fn(items[idx], idx);
+    }
+  };
+
+  const workers = Array.from({ length: Math.max(1, Math.min(limit, items.length)) }, worker);
+  await Promise.all(workers);
+  return results;
+};
+
 const concatPartStreams = (plannedParts: PlannedPart[]): ReadableStream<Uint8Array> =>
   new ReadableStream<Uint8Array>({
     async start(controller) {
       try {
-        for (const planned of plannedParts) {
-          const stream = await fetchPartBody(planned);
+        // Fetch every part body concurrently (bounded) so the slowest Telegram
+        // CDN fetch dictates latency instead of the sum of all fetches.
+        const partStreams = await mapBounded(plannedParts, PART_FETCH_CONCURRENCY, fetchPartBody);
+
+        // Fan-in in part order — object bytes stay correctly ordered.
+        for (const stream of partStreams) {
           const reader = stream.getReader();
           while (true) {
             const { value, done } = await reader.read();

test/object-stream-parallel.test.ts

@@ -0,0 +1,130 @@
+import { afterEach, beforeEach, describe, expect, it, mock } from 'bun:test';
+import { createGetObjectResponse, type ObjectPartSource } from '../src/interfaces/s3/object-stream';
+import type { RangeParseResult } from '../src/interfaces/s3/range';
+
+/**
+ * Stubs `globalThis.fetch` so every Telegram part URL returns a body whose
+ * bytes encode the part number. The first (and only the first) fetch is
+ * delayed heavily, so it resolves LAST — proving the response still streams
+ * parts back in correct order despite out-of-order completion.
+ */
+const installFetchMock = () => {
+  const original = globalThis.fetch;
+
+  const fakeFetch = mock((url: string | URL | Request) => {
+    const u = url.toString();
+    const match = u.match(/part(\d+)\.bin/);
+    const part = match ? Number.parseInt(match[1], 10) : 0;
+    const bytes = new Uint8Array(8).fill(0);
+    // Encode part number in the last byte so we can assert ordering.
+    bytes[7] = part;
+
+    const delay = part === 1 ? 60 : 0;
+    return new Promise<Response>((resolve) => {
+      setTimeout(() => {
+        resolve(
+          new Response(bytes, {
+            status: 200,
+            headers: { 'content-length': String(bytes.byteLength) },
+          }),
+        );
+      }, delay);
+    });
+  }) as unknown as typeof fetch;
+
+  globalThis.fetch = fakeFetch;
+  return { original, fakeFetch };
+};
+
+const makePart = (partNumber: number, sizeBytes: number): ObjectPartSource => ({
+  telegramFileId: `id-${partNumber}`,
+  telegramUrl: `https://api.telegram.org/file/botTOKEN/part${partNumber}.bin`,
+  sizeBytes,
+  partNumber,
+});
+
+const fullRange = (): RangeParseResult => ({
+  type: 'none',
+});
+
+const collect = async (res: Response): Promise<Uint8Array> => {
+  const buf = await res.arrayBuffer();
+  return new Uint8Array(buf);
+};
+
+describe('object-stream parallel fan-in', () => {
+  let restore: { original: typeof fetch; fakeFetch: unknown } | null = null;
+
+  beforeEach(() => {
+    restore = installFetchMock();
+  });
+
+  afterEach(() => {
+    if (restore) {
+      globalThis.fetch = restore.original;
+      restore = null;
+    }
+  });
+
+  it('streams multiple parts in order even when part 1 resolves last', async () => {
+    const parts = [makePart(1, 8), makePart(2, 8), makePart(3, 8)];
+    const res = await createGetObjectResponse({
+      reqId: 'req-1',
+      contentType: 'application/octet-stream',
+      etag: 'etag',
+      lastModified: new Date('2026-01-01T00:00:00Z'),
+      totalSize: 24,
+      parts,
+      range: fullRange(),
+    });
+
+    expect(res.status).toBe(200);
+    const body = await collect(res);
+    expect(body.byteLength).toBe(24);
+
+    // Each 8-byte part ends with its part number; order must be 1,2,3.
+    expect(body[7]).toBe(1);
+    expect(body[15]).toBe(2);
+    expect(body[23]).toBe(3);
+  });
+
+  it('respects byte range across parts', async () => {
+    const parts = [makePart(1, 8), makePart(2, 8), makePart(3, 8)];
+    const range: RangeParseResult = { type: 'valid', start: 4, end: 19 };
+    const res = await createGetObjectResponse({
+      reqId: 'req-2',
+      contentType: 'application/octet-stream',
+      etag: 'etag',
+      lastModified: new Date('2026-01-01T00:00:00Z'),
+      totalSize: 24,
+      parts,
+      range,
+    });
+
+    expect(res.status).toBe(206);
+    const body = await collect(res);
+    // 16 bytes: tail of part1 (byte 4-7 => part1 marker), full part2, head of part3 (0-3)
+    expect(body.byteLength).toBe(16);
+    expect(body[3]).toBe(1); // last byte of part 1 (marker)
+    expect(body[11]).toBe(2); // last byte of part 2 (marker)
+    // Part 3 contributes only its first 4 bytes (0-3); its marker lives at
+    // relative byte 7, which is outside the requested range — so no marker here.
+    expect(body[15]).toBe(0);
+  });
+
+  it('handles a single part without error', async () => {
+    const parts = [makePart(1, 8)];
+    const res = await createGetObjectResponse({
+      reqId: 'req-3',
+      contentType: 'application/octet-stream',
+      etag: 'etag',
+      lastModified: new Date('2026-01-01T00:00:00Z'),
+      totalSize: 8,
+      parts,
+      range: fullRange(),
+    });
+    const body = await collect(res);
+    expect(body.byteLength).toBe(8);
+    expect(body[7]).toBe(1);
+  });
+});