fix: correct import paths and add missing protocol files for DDD structure
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude
@@ -0,0 +1,213 @@
|
||||
import { Telegraf } from 'telegraf';
|
||||
import { config } from '../../env';
|
||||
import logger from '../../shared/logger/index';
|
||||
import type { ITelegramService, ForwardResult, TelegramFileInfo } from '../../domain/ports/telegram-service';
|
||||
import { enqueueUpload } from './upload-queue';
|
||||
import {
|
||||
sendMethodMap,
|
||||
extractUploadedFile,
|
||||
buildSendPayload,
|
||||
type TelegramMessageResult,
|
||||
type SendMethod,
|
||||
} from './types';
|
||||
|
||||
/**
|
||||
* Sleep for a given number of seconds.
|
||||
*
|
||||
* Used as a backoff mechanism when all bots in the pool are rate-limited.
|
||||
*
|
||||
* @param seconds - Number of seconds to sleep.
|
||||
* @returns A promise that resolves after the specified delay.
|
||||
*/
|
||||
const sleep = (seconds: number): Promise<void> => {
|
||||
return new Promise((resolve) => setTimeout(resolve, seconds * 1000));
|
||||
};
|
||||
|
||||
/**
|
||||
* Manages a pool of Telegram bots with automatic rotation and rate-limit handling.
|
||||
*
|
||||
* Distributes uploads across multiple bot tokens to maximise throughput.
|
||||
* When a bot receives a 429 (rate-limit) error, the pool instantly rotates
|
||||
* to the next available bot. If all bots are rate-limited, a coordinated
|
||||
* sleep is performed before retrying.
|
||||
*
|
||||
* Implements the {@link ITelegramService} contract.
|
||||
*/
|
||||
export class BotPool implements ITelegramService {
|
||||
private readonly bots: Telegraf[];
|
||||
private readonly botTokens: string[];
|
||||
private nextBotIndex = 0;
|
||||
|
||||
/** Create a new BotPool from the application configuration. */
|
||||
constructor() {
|
||||
this.botTokens = Array.from(new Set([config.botToken, ...config.additionalBotTokens]));
|
||||
this.bots = this.botTokens.map((token) => new Telegraf(token));
|
||||
}
|
||||
|
||||
/**
|
||||
* Claim the next bot index using round-robin rotation.
|
||||
*
|
||||
* @returns The index of the selected bot.
|
||||
*/
|
||||
private claimBotIndex(): number {
|
||||
const botIndex = this.nextBotIndex;
|
||||
this.nextBotIndex = (this.nextBotIndex + 1) % this.bots.length;
|
||||
return botIndex;
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a Telegram API action with automatic retry and bot rotation.
|
||||
*
|
||||
* On 429 errors the pool either:
|
||||
* 1. Rotates to the next bot immediately (if another bot is available), or
|
||||
* 2. Sleeps for the required duration after all bots are exhausted, then retries.
|
||||
*
|
||||
* @param action - The action to execute on a bot instance.
|
||||
* @param retries - Number of full-pool retry cycles remaining.
|
||||
* @param attemptedBots - Number of bots attempted in the current cycle.
|
||||
* @returns The result of the action.
|
||||
*/
|
||||
private async executeWithBotRetry<T>(
|
||||
action: (botInstance: Telegraf, botToken: string) => Promise<T>,
|
||||
retries = 5,
|
||||
attemptedBots = 0,
|
||||
): Promise<T> {
|
||||
const botIndex = this.claimBotIndex();
|
||||
const currentBot = this.bots[botIndex];
|
||||
const currentToken = this.botTokens[botIndex];
|
||||
try {
|
||||
return await action(currentBot, currentToken);
|
||||
} catch (error: unknown) {
|
||||
const errorStr = error instanceof Error ? error.message : String(error);
|
||||
const match = errorStr.match(/retry after (\d+)/i);
|
||||
|
||||
if (match) {
|
||||
const nextIndex = this.nextBotIndex;
|
||||
const nextAttemptedBots = attemptedBots + 1;
|
||||
|
||||
if (nextAttemptedBots < this.bots.length) {
|
||||
logger.info(
|
||||
`Bot Index ${botIndex} hit 429. Instantly rotating to Bot Index ${nextIndex}...`,
|
||||
);
|
||||
return this.executeWithBotRetry(action, retries, nextAttemptedBots);
|
||||
}
|
||||
|
||||
if (retries > 0) {
|
||||
const seconds = parseInt(match[1], 10);
|
||||
logger.warn(
|
||||
`All bots in the pool are rate-limited. Sleeping for ${seconds} seconds...`,
|
||||
{ error: errorStr },
|
||||
);
|
||||
await sleep(seconds);
|
||||
return this.executeWithBotRetry(action, retries - 1, 0);
|
||||
}
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Forward a file chunk to the configured Telegram storage chat.
|
||||
*
|
||||
* The upload is queued (via {@link enqueueUpload}) and executed with
|
||||
* automatic bot rotation on rate-limit errors.
|
||||
*
|
||||
* @param fileChunk - The file data (ReadStream, Buffer, or file path).
|
||||
* @param fileName - The original file name.
|
||||
* @param fileType - The file type classification (e.g. "document", "photo").
|
||||
* @returns The Telegram identifiers of the stored file.
|
||||
*/
|
||||
async forwardToStorage(
|
||||
fileChunk: unknown,
|
||||
fileName: string,
|
||||
fileType: string,
|
||||
): Promise<ForwardResult> {
|
||||
try {
|
||||
const result = await this.enqueueUpload<TelegramMessageResult>(async () => {
|
||||
const filePayload = { source: fileChunk, filename: fileName };
|
||||
const sendMethodName = sendMethodMap[fileType] || 'sendDocument';
|
||||
const payload = buildSendPayload(fileType, fileName);
|
||||
|
||||
return this.executeWithBotRetry<TelegramMessageResult>((activeBot) => {
|
||||
const telegram = activeBot.telegram as unknown as Record<string, SendMethod>;
|
||||
return telegram[sendMethodName](config.storageChatId, filePayload, payload);
|
||||
});
|
||||
});
|
||||
|
||||
const uploadedFile = extractUploadedFile(result, fileType);
|
||||
logger.info('File forwarded to storage', { fileName, message: result.message_id });
|
||||
|
||||
return {
|
||||
telegramFileId: uploadedFile?.file_id || '',
|
||||
telegramFileUniqueId: uploadedFile?.file_unique_id || '',
|
||||
storageMessageId: result.message_id,
|
||||
};
|
||||
} catch (error: unknown) {
|
||||
logger.error('Failed to forward file to storage', {
|
||||
fileName,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Retrieve file metadata from Telegram by file ID.
|
||||
*
|
||||
* Tries all configured bots sequentially; returns info from the first
|
||||
* bot that can retrieve the file. Errors indicating the file belongs
|
||||
* to a different bot are silently skipped.
|
||||
*
|
||||
* @param telegramFileId - The Telegram file_id to look up.
|
||||
* @returns Metadata including size, MIME type, download path, and bot token.
|
||||
*/
|
||||
async getFileInfo(telegramFileId: string): Promise<TelegramFileInfo> {
|
||||
let lastError: unknown;
|
||||
for (const activeBot of this.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;
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue a task for sequential upload execution.
|
||||
*
|
||||
* Delegates to the shared upload queue to ensure only a limited number
|
||||
* of Telegram uploads run concurrently.
|
||||
*
|
||||
* @param task - An async function performing the upload.
|
||||
* @returns The result of the task.
|
||||
*/
|
||||
enqueueUpload<T>(task: () => Promise<T>): Promise<T> {
|
||||
return enqueueUpload(task);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Singleton BotPool instance initialised from application configuration.
|
||||
*/
|
||||
export const botPool = new BotPool();
|
||||
@@ -0,0 +1,336 @@
|
||||
import { createReadStream } from 'node:fs';
|
||||
import { gzipSync } from 'node:zlib';
|
||||
import { nanoid } from 'nanoid';
|
||||
import { config } from '../../env';
|
||||
import { computeHash } from '../../utils/file';
|
||||
import { createGetObjectResponse, type ObjectPartSource } from '../../utils/s3/object-stream';
|
||||
import type { RangeParseResult } from '../../utils/s3/range';
|
||||
import type { IFileRepository } from '../../domain/ports/file-repository';
|
||||
import type { IFilePartRepository } from '../../domain/ports/file-part-repository';
|
||||
import type { ITelegramService } from '../../domain/ports/telegram-service';
|
||||
import type { File as FileEntity } from '../../domain/entities/file';
|
||||
import type { NewFilePart, CompressionAlgorithm } from '../../domain/entities/file-part';
|
||||
|
||||
/**
|
||||
* Chunk compression algorithm identifier.
|
||||
* `"gzip"` if gzip compression was applied, `null` for uncompressed.
|
||||
*/
|
||||
export type ChunkCompressionAlgorithm = CompressionAlgorithm;
|
||||
|
||||
/**
|
||||
* Metadata about a single uploaded chunk (part) stored in Telegram.
|
||||
*/
|
||||
export interface ChunkedUploadPart {
|
||||
/** 1-based part number within the file */
|
||||
partNumber: number;
|
||||
/** Telegram file_id for retrieving this part */
|
||||
telegramFileId: string;
|
||||
/** Telegram unique file_id (stable across bot tokens) */
|
||||
telegramFileUniqueId: string;
|
||||
/** Message ID within the storage chat */
|
||||
storageMessageId: number;
|
||||
/** Original (pre-compression) size in bytes */
|
||||
sizeBytes: number;
|
||||
/** Stored (post-compression) size in bytes */
|
||||
storedSizeBytes: number;
|
||||
/** Compression algorithm applied, or null */
|
||||
compressionAlgorithm: ChunkCompressionAlgorithm;
|
||||
/** ETag (SHA-256 hash) of the original chunk */
|
||||
etag: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Result of uploading a file in Telegram chunks.
|
||||
*/
|
||||
export interface ChunkedUploadResult {
|
||||
/** Ordered list of uploaded parts */
|
||||
parts: ChunkedUploadPart[];
|
||||
/** SHA-256 hash of the complete file content */
|
||||
fileHash: string;
|
||||
/** Total file size in bytes */
|
||||
totalSizeBytes: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Input parameters for storing a file via chunked Telegram uploads.
|
||||
*/
|
||||
export interface ChunkedFileInput {
|
||||
/** Path to the temporary file on disk */
|
||||
tempPath: string;
|
||||
/** Prefix for generated part file names */
|
||||
partFileNamePrefix: string;
|
||||
/** Original file name */
|
||||
fileName: string;
|
||||
/** MIME type of the file */
|
||||
mimeType: string;
|
||||
/** File size in bytes */
|
||||
sizeBytes: number;
|
||||
/** File type classification (e.g. "document", "video") */
|
||||
fileType: string;
|
||||
/** Telegram user ID of the uploader */
|
||||
uploaderId: number;
|
||||
/** S3 bucket ID if the file is also tracked in S3, or null */
|
||||
bucketId?: string | null;
|
||||
/** S3 object key if the file is also tracked in S3, or null */
|
||||
s3Key?: string | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate and sanitise the Telegram chunk size.
|
||||
*
|
||||
* @param chunkSizeBytes - The desired chunk size in bytes.
|
||||
* @returns The validated chunk size.
|
||||
* @throws {Error} If the chunk size is not a safe positive integer.
|
||||
*/
|
||||
const asSafeChunkSize = (chunkSizeBytes: number): number => {
|
||||
if (!Number.isSafeInteger(chunkSizeBytes) || chunkSizeBytes <= 0) {
|
||||
throw new Error('Invalid Telegram chunk size');
|
||||
}
|
||||
return chunkSizeBytes;
|
||||
};
|
||||
|
||||
/**
|
||||
* Optionally compress a chunk with gzip.
|
||||
*
|
||||
* Compression is skipped if:
|
||||
* - The `compress` flag is false.
|
||||
* - The chunk is smaller than `compressionMinSizeBytes`.
|
||||
* - The compressed result is larger than the original.
|
||||
*
|
||||
* @param chunk - The raw chunk buffer.
|
||||
* @param compress - Whether compression is enabled.
|
||||
* @param compressionMinSizeBytes - Minimum chunk size to attempt compression.
|
||||
* @returns The (possibly compressed) bytes and the algorithm used.
|
||||
*/
|
||||
const maybeCompressChunk = (
|
||||
chunk: Buffer,
|
||||
compress: boolean,
|
||||
compressionMinSizeBytes: number,
|
||||
): { bytes: Buffer; compressionAlgorithm: ChunkCompressionAlgorithm } => {
|
||||
if (!compress || chunk.byteLength < compressionMinSizeBytes) {
|
||||
return { bytes: chunk, compressionAlgorithm: null };
|
||||
}
|
||||
|
||||
const gzipped = gzipSync(chunk);
|
||||
if (gzipped.byteLength >= chunk.byteLength) {
|
||||
return { bytes: chunk, compressionAlgorithm: null };
|
||||
}
|
||||
|
||||
return { bytes: gzipped, compressionAlgorithm: 'gzip' };
|
||||
};
|
||||
|
||||
/**
|
||||
* Manages chunked storage of large files in Telegram.
|
||||
*
|
||||
* Large files are split into smaller chunks, each uploaded as a separate
|
||||
* Telegram document. File and part metadata is persisted through the
|
||||
* provided repository interfaces.
|
||||
*
|
||||
* Injects dependencies via constructor — can be used with any
|
||||
* {@link IFileRepository}, {@link IFilePartRepository}, and
|
||||
* {@link ITelegramService} implementation.
|
||||
*/
|
||||
export class ChunkedStorage {
|
||||
/**
|
||||
* @param fileRepository - Repository for File entity persistence.
|
||||
* @param filePartRepository - Repository for FilePart entity persistence.
|
||||
* @param telegramService - Service for Telegram API interactions.
|
||||
*/
|
||||
constructor(
|
||||
private readonly fileRepository: IFileRepository,
|
||||
private readonly filePartRepository: IFilePartRepository,
|
||||
private readonly telegramService: ITelegramService,
|
||||
) {}
|
||||
|
||||
/**
|
||||
* Upload a file to Telegram in chunks and return chunk metadata.
|
||||
*
|
||||
* Reads the file from disk in fixed-size chunks, compresses each chunk
|
||||
* if beneficial, and forwards each chunk to Telegram storage.
|
||||
*
|
||||
* @param input - Upload parameters including temp path, chunk size, and compression settings.
|
||||
* @returns Metadata about all uploaded chunks and the file hash.
|
||||
*/
|
||||
async uploadFileInTelegramChunks(input: {
|
||||
tempPath: string;
|
||||
partFileNamePrefix: string;
|
||||
chunkSizeBytes: number;
|
||||
compress: boolean;
|
||||
compressionMinSizeBytes: number;
|
||||
}): Promise<ChunkedUploadResult> {
|
||||
const chunkSizeBytes = asSafeChunkSize(input.chunkSizeBytes);
|
||||
const hasher = new Bun.CryptoHasher('sha256');
|
||||
const parts: ChunkedUploadPart[] = [];
|
||||
let totalSizeBytes = 0;
|
||||
let partNumber = 0;
|
||||
|
||||
const stream = createReadStream(input.tempPath, { highWaterMark: chunkSizeBytes });
|
||||
|
||||
for await (const data of stream) {
|
||||
const chunk = Buffer.isBuffer(data) ? data : Buffer.from(data as Uint8Array);
|
||||
if (chunk.byteLength === 0) continue;
|
||||
|
||||
partNumber += 1;
|
||||
totalSizeBytes += chunk.byteLength;
|
||||
hasher.update(chunk);
|
||||
|
||||
const { bytes, compressionAlgorithm } = maybeCompressChunk(
|
||||
chunk,
|
||||
input.compress,
|
||||
input.compressionMinSizeBytes,
|
||||
);
|
||||
const forwardResult = await this.telegramService.forwardToStorage(
|
||||
bytes,
|
||||
`${input.partFileNamePrefix}.part-${partNumber}`,
|
||||
'document',
|
||||
);
|
||||
|
||||
parts.push({
|
||||
partNumber,
|
||||
telegramFileId: forwardResult.telegramFileId,
|
||||
telegramFileUniqueId: forwardResult.telegramFileUniqueId,
|
||||
storageMessageId: forwardResult.storageMessageId,
|
||||
sizeBytes: chunk.byteLength,
|
||||
storedSizeBytes: bytes.byteLength,
|
||||
compressionAlgorithm,
|
||||
etag: computeHash(chunk),
|
||||
});
|
||||
}
|
||||
|
||||
return {
|
||||
parts,
|
||||
fileHash: hasher.digest('hex'),
|
||||
totalSizeBytes,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Upload a file to Telegram in chunks and persist file + part records.
|
||||
*
|
||||
* Combines chunk upload ({@link uploadFileInTelegramChunks}) with
|
||||
* repository persistence for both the File and FilePart entities.
|
||||
*
|
||||
* @param input - The file metadata and upload parameters.
|
||||
* @returns The persisted File entity.
|
||||
*/
|
||||
async storeFileInTelegramChunks(input: ChunkedFileInput): Promise<FileEntity> {
|
||||
const upload = await this.uploadFileInTelegramChunks({
|
||||
tempPath: input.tempPath,
|
||||
partFileNamePrefix: input.partFileNamePrefix,
|
||||
chunkSizeBytes: config.telegramChunkSizeBytes,
|
||||
compress: config.compressChunkedUploads,
|
||||
compressionMinSizeBytes: config.chunkCompressionMinSizeBytes,
|
||||
});
|
||||
|
||||
const firstPart = upload.parts[0];
|
||||
if (!firstPart) {
|
||||
throw new Error('Chunked upload produced no parts');
|
||||
}
|
||||
|
||||
const publicId = nanoid();
|
||||
|
||||
const file = await this.fileRepository.create({
|
||||
publicId,
|
||||
telegramFileId: firstPart.telegramFileId,
|
||||
telegramFileUniqueId: firstPart.telegramFileUniqueId,
|
||||
storageChatId: config.storageChatId,
|
||||
storageMessageId: firstPart.storageMessageId,
|
||||
fileName: input.fileName,
|
||||
mimeType: input.mimeType,
|
||||
sizeBytes: upload.totalSizeBytes,
|
||||
fileType: input.fileType,
|
||||
uploaderId: input.uploaderId,
|
||||
fileHash: upload.fileHash,
|
||||
archiveTelegramFileId: null,
|
||||
archiveStorageMessageId: null,
|
||||
archiveFileName: null,
|
||||
archiveEntryName: null,
|
||||
archiveMimeType: null,
|
||||
archiveSizeBytes: null,
|
||||
bucketId: input.bucketId ?? null,
|
||||
s3Key: input.s3Key ?? null,
|
||||
storageBackend: 'chunked',
|
||||
isDeleted: false,
|
||||
multipartUploadId: null,
|
||||
partCount: upload.parts.length,
|
||||
});
|
||||
|
||||
const fileParts: NewFilePart[] = upload.parts.map((part) => ({
|
||||
fileId: file.id,
|
||||
partNumber: part.partNumber,
|
||||
telegramFileId: part.telegramFileId,
|
||||
telegramFileUniqueId: part.telegramFileUniqueId,
|
||||
storageChatId: config.storageChatId,
|
||||
storageMessageId: part.storageMessageId,
|
||||
sizeBytes: part.sizeBytes,
|
||||
storedSizeBytes: part.storedSizeBytes,
|
||||
compressionAlgorithm: part.compressionAlgorithm,
|
||||
etag: part.etag,
|
||||
}));
|
||||
|
||||
await this.filePartRepository.insert(fileParts);
|
||||
return file;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build a list of object-part sources for reconstructing a chunked file.
|
||||
*
|
||||
* Queries the file-part repository and enriches each part with
|
||||
* the Telegram download URL by calling {@link ITelegramService.getFileInfo}.
|
||||
*
|
||||
* @param file - The File entity whose parts should be resolved.
|
||||
* @returns An ordered list of object part sources ready for streaming.
|
||||
*/
|
||||
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,
|
||||
});
|
||||
}
|
||||
|
||||
return sources;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create an HTTP Response that streams a chunked file's content.
|
||||
*
|
||||
* Supports HTTP range requests for partial content delivery.
|
||||
* The response is constructed by reassembling parts in order and
|
||||
* optionally decompressing gzip-compressed parts.
|
||||
*
|
||||
* @param input - Parameters including the file entity, range, and request ID.
|
||||
* @returns A Response object streaming the requested byte range.
|
||||
*/
|
||||
async createChunkedObjectResponse(input: {
|
||||
file: FileEntity;
|
||||
range: RangeParseResult;
|
||||
reqId: string;
|
||||
}): Promise<Response> {
|
||||
const parts = await this.buildChunkedObjectSources(input.file);
|
||||
if (parts.length === 0) {
|
||||
throw new Error('Chunked object has no parts');
|
||||
}
|
||||
|
||||
return createGetObjectResponse({
|
||||
reqId: input.reqId,
|
||||
contentType: input.file.mimeType,
|
||||
etag: input.file.fileHash || parts.map((p) => p.telegramFileId).join('-'),
|
||||
lastModified:
|
||||
input.file.createdAt instanceof Date
|
||||
? input.file.createdAt
|
||||
: new Date(input.file.createdAt),
|
||||
totalSize: Number(input.file.sizeBytes),
|
||||
parts,
|
||||
range: input.range,
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,130 @@
|
||||
/**
|
||||
* Telegram-specific types used internally by the infrastructure layer.
|
||||
*
|
||||
* These types represent the raw Telegram Bot API response shapes and
|
||||
* the internal abstractions built on top of them. The higher-level domain
|
||||
* types (ForwardResult, TelegramFileInfo) are defined in
|
||||
* src/domain/ports/telegram-service.ts.
|
||||
*/
|
||||
|
||||
/**
|
||||
* File reference within a Telegram message result.
|
||||
* Contains identifiers returned by the Telegram API for uploaded media.
|
||||
*/
|
||||
export interface UploadedTelegramFile {
|
||||
/** Telegram file_id for retrieving the file */
|
||||
file_id?: string;
|
||||
/** Telegram unique file_id (stable across bot tokens) */
|
||||
file_unique_id?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Result structure returned by Telegram send* API methods.
|
||||
* Covers all media types a Telegram message can carry.
|
||||
*/
|
||||
export interface TelegramMessageResult {
|
||||
/** Unique message identifier inside the chat */
|
||||
message_id: number;
|
||||
/** Sent document, if applicable */
|
||||
document?: UploadedTelegramFile;
|
||||
/** Sent photo (array of sizes, last element is largest), if applicable */
|
||||
photo?: UploadedTelegramFile[];
|
||||
/** Sent video, if applicable */
|
||||
video?: UploadedTelegramFile;
|
||||
/** Sent audio, if applicable */
|
||||
audio?: UploadedTelegramFile;
|
||||
/** Sent voice message, if applicable */
|
||||
voice?: UploadedTelegramFile;
|
||||
/** Sent animation (GIF), if applicable */
|
||||
animation?: UploadedTelegramFile;
|
||||
/** Sent sticker, if applicable */
|
||||
sticker?: UploadedTelegramFile;
|
||||
/** Sent video note, if applicable */
|
||||
video_note?: UploadedTelegramFile;
|
||||
/** Catch-all for any additional Telegram response fields */
|
||||
[key: string]: unknown;
|
||||
}
|
||||
|
||||
/**
|
||||
* Payload structure for sending a file via the Telegram Bot API.
|
||||
*
|
||||
* @internal
|
||||
*/
|
||||
export type FilePayload = { source: unknown; filename: string };
|
||||
|
||||
/**
|
||||
* Additional optional payload for Telegram send method calls.
|
||||
*
|
||||
* @internal
|
||||
*/
|
||||
export type SendPayload = { caption?: string };
|
||||
|
||||
/**
|
||||
* Function signature for Telegram send* method calls on a bot instance.
|
||||
*
|
||||
* @internal
|
||||
*/
|
||||
export type SendMethod = (
|
||||
chatId: number,
|
||||
filePayload: FilePayload,
|
||||
payload?: SendPayload,
|
||||
) => Promise<TelegramMessageResult>;
|
||||
|
||||
/**
|
||||
* Mapping from file type identifier to Telegram Bot API method name.
|
||||
*
|
||||
* Each key corresponds to a Telegram media type; the value is the
|
||||
* method name to call on `bot.telegram`.
|
||||
*/
|
||||
export const sendMethodMap: Record<string, string> = {
|
||||
photo: 'sendPhoto',
|
||||
audio: 'sendAudio',
|
||||
video: 'sendVideo',
|
||||
voice: 'sendVoice',
|
||||
animation: 'sendAnimation',
|
||||
sticker: 'sendSticker',
|
||||
document: 'sendDocument',
|
||||
video_note: 'sendDocument',
|
||||
};
|
||||
|
||||
/**
|
||||
* Extract the uploaded file reference from a Telegram message result
|
||||
* based on the media type present in the result.
|
||||
*
|
||||
* Falls back to looking up the file type key directly on the result object.
|
||||
*
|
||||
* @param result - The message result from a Telegram send* call.
|
||||
* @param fileType - The file type classification (e.g. "document", "photo").
|
||||
* @returns The uploaded file reference, or `undefined` if none was found.
|
||||
*/
|
||||
export const extractUploadedFile = (
|
||||
result: TelegramMessageResult,
|
||||
fileType: string,
|
||||
): UploadedTelegramFile | undefined => {
|
||||
if (result.document) return result.document;
|
||||
if (result.photo) return result.photo?.slice(-1)[0];
|
||||
if (result.video) return result.video;
|
||||
if (result.audio) return result.audio;
|
||||
if (result.voice) return result.voice;
|
||||
if (result.animation) return result.animation;
|
||||
if (result.sticker) return result.sticker;
|
||||
if (result.video_note) return result.video_note;
|
||||
return result[fileType] as UploadedTelegramFile | undefined;
|
||||
};
|
||||
|
||||
/**
|
||||
* Build the send payload (caption, etc.) for a Telegram send* method call.
|
||||
*
|
||||
* Stickers do not support captions. Documents get a labelled caption
|
||||
* with the file name. All other types use the plain file name as caption.
|
||||
*
|
||||
* @param fileType - The file type (e.g. "document", "photo", "sticker").
|
||||
* @param fileName - The file name to use in the caption.
|
||||
* @returns The payload object with caption (or empty for sticker).
|
||||
*/
|
||||
export const buildSendPayload = (fileType: string, fileName: string): SendPayload => {
|
||||
const basePayload: SendPayload = { caption: fileName };
|
||||
if (fileType === 'sticker') return {};
|
||||
if (fileType === 'document') return { caption: `📁 ${fileName}` };
|
||||
return basePayload;
|
||||
};
|
||||
@@ -0,0 +1,238 @@
|
||||
import { createReadStream } from 'node:fs';
|
||||
import { nanoid } from 'nanoid';
|
||||
import { config } from '../../env';
|
||||
import { cleanupTempFile } from '../../shared/utils/file';
|
||||
import { createZip, type ZipEntry } from '../../shared/utils/zip';
|
||||
import type { IFileRepository } from '../../domain/ports/file-repository';
|
||||
import type { ITelegramService } from '../../domain/ports/telegram-service';
|
||||
import type { File as FileEntity, NewFile } from '../../domain/entities/file';
|
||||
|
||||
/**
|
||||
* 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 {
|
||||
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,
|
||||
uploaderId: 0,
|
||||
fileHash: item.prepared.fileHash,
|
||||
archiveTelegramFileId: archive.telegramFileId,
|
||||
archiveStorageMessageId: archive.storageMessageId,
|
||||
archiveFileName: archive.fileName,
|
||||
archiveEntryName: entry.entryName,
|
||||
archiveMimeType: 'application/zip',
|
||||
archiveSizeBytes: archive.sizeBytes,
|
||||
bucketId: null,
|
||||
s3Key: null,
|
||||
storageBackend: null,
|
||||
isDeleted: null,
|
||||
multipartUploadId: null,
|
||||
partCount: null,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* 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;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
import PQueue from 'p-queue';
|
||||
import { config } from '../../env';
|
||||
import logger from '../../shared/logger/index';
|
||||
|
||||
/**
|
||||
* P-queue instance for serialising Telegram upload tasks.
|
||||
*
|
||||
* Concurrency is governed by {@link config.uploadConcurrency}.
|
||||
* Built-in logging emits warnings when the queue grows beyond 5 pending items.
|
||||
*/
|
||||
const uploadQueue = new PQueue({
|
||||
concurrency: config.uploadConcurrency,
|
||||
});
|
||||
|
||||
/* Monitor queue growth and emit warnings for large backlogs */
|
||||
uploadQueue.on('add', () => {
|
||||
const stats = getQueueStats();
|
||||
if (stats.size > 5) {
|
||||
logger.warn('Upload queue building up', { pending: stats.pending, size: stats.size });
|
||||
}
|
||||
});
|
||||
|
||||
uploadQueue.on('next', () => {
|
||||
const stats = getQueueStats();
|
||||
logger.debug('Processing next upload', { pending: stats.pending, size: stats.size });
|
||||
});
|
||||
|
||||
/**
|
||||
* Enqueue an upload task to be executed by the queue.
|
||||
*
|
||||
* Tasks are executed in FIFO order, subject to the concurrency limit.
|
||||
*
|
||||
* @param task - An async function representing the upload operation.
|
||||
* @returns A promise that resolves with the task's result.
|
||||
*/
|
||||
export const enqueueUpload = <T>(task: () => Promise<T>): Promise<T> => {
|
||||
return uploadQueue.add(task);
|
||||
};
|
||||
|
||||
/**
|
||||
* Get current queue statistics.
|
||||
*
|
||||
* @returns An object with `pending` (actively executing) and `size` (waiting) counts.
|
||||
*/
|
||||
export const getQueueStats = (): { pending: number; size: number } => ({
|
||||
pending: uploadQueue.pending,
|
||||
size: uploadQueue.size,
|
||||
});
|
||||
|
||||
/**
|
||||
* Get the number of items waiting in the queue (not yet started).
|
||||
*
|
||||
* @returns The number of queued items.
|
||||
*/
|
||||
export const getQueueSize = (): number => uploadQueue.size;
|
||||
|
||||
/**
|
||||
* Get the number of items currently being processed.
|
||||
*
|
||||
* @returns The number of pending (in-flight) items.
|
||||
*/
|
||||
export const getPendingCount = (): number => uploadQueue.pending;
|
||||
|
||||
/**
|
||||
* Clear all pending items and wait for in-flight ones to finish.
|
||||
*
|
||||
* @returns A promise that resolves when the queue is idle after clearing.
|
||||
*/
|
||||
export const clearQueue = async (): Promise<void> => {
|
||||
uploadQueue.clear();
|
||||
await uploadQueue.onIdle();
|
||||
};
|
||||
|
||||
/**
|
||||
* Wait for the queue to become idle (all tasks finished).
|
||||
*
|
||||
* @returns A promise that resolves when no tasks are pending or in-flight.
|
||||
*/
|
||||
export const waitForQueue = async (): Promise<void> => {
|
||||
await uploadQueue.onIdle();
|
||||
};
|
||||
Reference in New Issue
Block a user