diff --git a/.env.example b/.env.example index 890eca1..cab25c4 100644 --- a/.env.example +++ b/.env.example @@ -14,4 +14,9 @@ RATE_LIMIT_MAX_REQUESTS=30 # S3_ACCESS_KEY=teleuploader-admin # S3_SECRET_KEY=your-secret-key-here # S3_DEFAULT_REGION=us-east-1 -# S3_VHOST_DOMAINS=upload.asepharyana.my.id,upload.asepharyana.web.id \ No newline at end of file +# S3_VHOST_DOMAINS=upload.asepharyana.my.id,upload.asepharyana.web.id + +# Telegram-safe internal chunking for large stored files +# TELEGRAM_CHUNK_SIZE_BYTES=20971520 +# COMPRESS_CHUNKED_UPLOADS=true +# CHUNK_COMPRESSION_MIN_SIZE_BYTES=4096 \ No newline at end of file diff --git a/.gitignore b/.gitignore index 07eb103..6c7c928 100644 --- a/.gitignore +++ b/.gitignore @@ -23,6 +23,9 @@ report.[0-9]_.[0-9]_.[0-9]_.[0-9]_.json .env.production.local .env.local +# S3 client guide (local reference) +S3_GUIDE.md + # caches .eslintcache .cache diff --git a/docker-compose.yml b/docker-compose.yml index c5c6510..929f44d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -17,6 +17,9 @@ services: - BATCH_MAX_ITEMS=${BATCH_MAX_ITEMS:-20} - BATCH_MAX_SIZE_BYTES=${BATCH_MAX_SIZE_BYTES:-524288000} - MAX_REQUEST_BODY_BYTES=${MAX_REQUEST_BODY_BYTES:-2147483648} + - TELEGRAM_CHUNK_SIZE_BYTES=${TELEGRAM_CHUNK_SIZE_BYTES:-20971520} + - COMPRESS_CHUNKED_UPLOADS=${COMPRESS_CHUNKED_UPLOADS:-true} + - CHUNK_COMPRESSION_MIN_SIZE_BYTES=${CHUNK_COMPRESSION_MIN_SIZE_BYTES:-4096} - RATE_LIMIT_WINDOW_MS=${RATE_LIMIT_WINDOW_MS:-60000} - RATE_LIMIT_MAX_REQUESTS=${RATE_LIMIT_MAX_REQUESTS:-30} - S3_ACCESS_KEY=${S3_ACCESS_KEY:-teleuploader-admin} diff --git a/schema.sql b/schema.sql index 52ba0a3..2b0b0bd 100644 --- a/schema.sql +++ b/schema.sql @@ -42,6 +42,7 @@ ALTER TABLE files ADD COLUMN IF NOT EXISTS s3_key TEXT; ALTER TABLE files ADD COLUMN IF NOT EXISTS storage_backend VARCHAR DEFAULT 'telegram'; ALTER TABLE files ADD COLUMN IF NOT EXISTS is_deleted BOOLEAN DEFAULT false; ALTER TABLE files ADD COLUMN IF NOT EXISTS multipart_upload_id TEXT; +ALTER TABLE files ADD COLUMN IF NOT EXISTS part_count INT; CREATE UNIQUE INDEX IF NOT EXISTS idx_files_bucket_key ON files(bucket_id, s3_key) WHERE is_deleted = false; CREATE INDEX IF NOT EXISTS idx_files_bucket_prefix ON files(bucket_id, s3_key text_pattern_ops); @@ -72,4 +73,24 @@ CREATE TABLE IF NOT EXISTS multipart_parts ( ); CREATE INDEX IF NOT EXISTS idx_multipart_parts_upload ON multipart_parts(upload_id, part_number); -CREATE INDEX IF NOT EXISTS idx_multipart_uploads_status ON multipart_uploads(status); \ No newline at end of file +CREATE INDEX IF NOT EXISTS idx_multipart_uploads_status ON multipart_uploads(status); + +-- Permanent internal chunks for Telegram-safe storage. +-- This is separate from S3 multipart protocol state above. +CREATE TABLE IF NOT EXISTS file_parts ( + id SERIAL PRIMARY KEY, + file_id UUID NOT NULL REFERENCES files(id) ON DELETE CASCADE, + part_number INT NOT NULL, + telegram_file_id VARCHAR NOT NULL, + telegram_file_unique_id VARCHAR NOT NULL, + storage_chat_id BIGINT NOT NULL, + storage_message_id BIGINT NOT NULL, + size_bytes BIGINT NOT NULL, + stored_size_bytes BIGINT NOT NULL, + compression_algorithm VARCHAR, + etag VARCHAR NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + UNIQUE(file_id, part_number) +); + +CREATE INDEX IF NOT EXISTS idx_file_parts_file_id ON file_parts(file_id, part_number); \ No newline at end of file diff --git a/src/db/file-parts.ts b/src/db/file-parts.ts new file mode 100644 index 0000000..45bcc34 --- /dev/null +++ b/src/db/file-parts.ts @@ -0,0 +1,97 @@ +import { sql } from 'drizzle-orm'; +import { db } from './index'; + +export type CompressionAlgorithm = 'gzip' | null; + +export interface FilePart { + id: number; + fileId: string; + partNumber: number; + telegramFileId: string; + telegramFileUniqueId: string; + storageChatId: number; + storageMessageId: number; + sizeBytes: number; + storedSizeBytes: number; + compressionAlgorithm: CompressionAlgorithm; + etag: string; + createdAt: Date; +} + +export type NewFilePartInput = Omit; + +const toNumber = (value: unknown): number => Number(value ?? 0); + +const mapRowToFilePart = (row: Record): FilePart => ({ + id: toNumber(row.id), + fileId: row.file_id as string, + partNumber: toNumber(row.part_number), + telegramFileId: row.telegram_file_id as string, + telegramFileUniqueId: row.telegram_file_unique_id as string, + storageChatId: toNumber(row.storage_chat_id), + storageMessageId: toNumber(row.storage_message_id), + sizeBytes: toNumber(row.size_bytes), + storedSizeBytes: toNumber(row.stored_size_bytes), + compressionAlgorithm: (row.compression_algorithm as CompressionAlgorithm) || null, + etag: row.etag as string, + createdAt: new Date(row.created_at as string), +}); + +export const insertFileParts = async (parts: NewFilePartInput[]): Promise => { + for (const part of parts) { + await db.execute( + sql`INSERT INTO file_parts ( + file_id, + part_number, + telegram_file_id, + telegram_file_unique_id, + storage_chat_id, + storage_message_id, + size_bytes, + stored_size_bytes, + compression_algorithm, + etag + ) VALUES ( + ${part.fileId}::uuid, + ${part.partNumber}, + ${part.telegramFileId}, + ${part.telegramFileUniqueId}, + ${part.storageChatId}, + ${part.storageMessageId}, + ${part.sizeBytes}, + ${part.storedSizeBytes}, + ${part.compressionAlgorithm}, + ${part.etag} + )`, + ); + } +}; + +export const listFileParts = async (fileId: string): Promise => { + const result = (await db.execute( + sql`SELECT id, + file_id, + part_number, + telegram_file_id, + telegram_file_unique_id, + storage_chat_id, + storage_message_id, + size_bytes, + stored_size_bytes, + compression_algorithm, + etag, + created_at + FROM file_parts + WHERE file_id = ${fileId}::uuid + ORDER BY part_number`, + )) as unknown as Record[]; + + return result.map(mapRowToFilePart); +}; + +export const countFileParts = async (fileId: string): Promise => { + const result = (await db.execute( + sql`SELECT COUNT(*) AS count FROM file_parts WHERE file_id = ${fileId}::uuid`, + )) as unknown as Record[]; + return toNumber(result[0]?.count); +}; diff --git a/src/db/files-ext.ts b/src/db/files-ext.ts index d28bf5b..fe4426b 100644 --- a/src/db/files-ext.ts +++ b/src/db/files-ext.ts @@ -51,6 +51,8 @@ const mapDbRowToS3Record = (row: Record): S3FileRecord => { storageBackend: (row.storage_backend as string) || 'telegram', isDeleted: row.is_deleted as boolean, multipartUploadId: row.multipart_upload_id as string | null, + partCount: + row.part_count === null || row.part_count === undefined ? null : toNumber(row.part_count), createdAt: new Date(row.created_at as string), updatedAt: new Date(row.updated_at as string), }; diff --git a/src/db/index.ts b/src/db/index.ts index cb76516..f413737 100644 --- a/src/db/index.ts +++ b/src/db/index.ts @@ -1,6 +1,6 @@ import { drizzle } from 'drizzle-orm/postgres-js'; import postgres from 'postgres'; -import { files } from './schema'; +import { fileParts, files } from './schema'; const client = postgres(process.env.DATABASE_URL!, { max: 10, @@ -8,6 +8,6 @@ const client = postgres(process.env.DATABASE_URL!, { connect_timeout: 10, }); -export const db = drizzle(client, { schema: { files } }); -export { files }; +export const db = drizzle(client, { schema: { fileParts, files } }); +export { fileParts, files }; export default db; diff --git a/src/db/schema.ts b/src/db/schema.ts index fa05941..275e58a 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -1,5 +1,14 @@ import type { InferInsertModel, InferSelectModel } from 'drizzle-orm'; -import { bigint, boolean, pgTable, text, timestamp, uuid } from 'drizzle-orm/pg-core'; +import { + bigint, + boolean, + integer, + pgTable, + serial, + text, + timestamp, + uuid, +} from 'drizzle-orm/pg-core'; export const files = pgTable('files', { id: uuid('id').primaryKey().defaultRandom(), @@ -25,9 +34,27 @@ export const files = pgTable('files', { storageBackend: text('storage_backend').default('telegram'), isDeleted: boolean('is_deleted').default(false), multipartUploadId: text('multipart_upload_id'), + partCount: integer('part_count'), createdAt: timestamp('created_at').defaultNow().notNull(), updatedAt: timestamp('updated_at').defaultNow().notNull(), }); +export const fileParts = pgTable('file_parts', { + id: serial('id').primaryKey(), + fileId: uuid('file_id').notNull(), + partNumber: integer('part_number').notNull(), + telegramFileId: text('telegram_file_id').notNull(), + telegramFileUniqueId: text('telegram_file_unique_id').notNull(), + storageChatId: bigint('storage_chat_id', { mode: 'number' }).notNull(), + storageMessageId: bigint('storage_message_id', { mode: 'number' }).notNull(), + sizeBytes: bigint('size_bytes', { mode: 'number' }).notNull(), + storedSizeBytes: bigint('stored_size_bytes', { mode: 'number' }).notNull(), + compressionAlgorithm: text('compression_algorithm'), + etag: text('etag').notNull(), + createdAt: timestamp('created_at').defaultNow().notNull(), +}); + export type File = InferSelectModel; export type NewFile = InferInsertModel; +export type FilePart = InferSelectModel; +export type NewFilePart = InferInsertModel; diff --git a/src/env.ts b/src/env.ts index f87aac1..43b6400 100644 --- a/src/env.ts +++ b/src/env.ts @@ -16,6 +16,9 @@ interface AppConfig { batchMaxItems: number; batchMaxSizeBytes: number; maxRequestBodyBytes: number; + telegramChunkSizeBytes: number; + compressChunkedUploads: boolean; + chunkCompressionMinSizeBytes: number; s3AccessKey: string; s3SecretKey: string; s3DefaultRegion: string; @@ -85,6 +88,9 @@ export const config: AppConfig = { batchMaxItems: parseNumber(process.env.BATCH_MAX_ITEMS, 20), batchMaxSizeBytes: parseNumber(process.env.BATCH_MAX_SIZE_BYTES, 500 * 1024 * 1024), maxRequestBodyBytes: parseNumber(process.env.MAX_REQUEST_BODY_BYTES, 2 * 1024 * 1024 * 1024), + telegramChunkSizeBytes: parseNumber(process.env.TELEGRAM_CHUNK_SIZE_BYTES, 20 * 1024 * 1024), + compressChunkedUploads: process.env.COMPRESS_CHUNKED_UPLOADS !== 'false', + chunkCompressionMinSizeBytes: parseNumber(process.env.CHUNK_COMPRESSION_MIN_SIZE_BYTES, 4096), s3AccessKey: process.env.S3_ACCESS_KEY || 'teleuploader-admin', s3SecretKey: process.env.S3_SECRET_KEY || '', s3DefaultRegion: process.env.S3_DEFAULT_REGION || 'us-east-1', diff --git a/src/routes/files.ts b/src/routes/files.ts index 5ffbf31..ce41e04 100644 --- a/src/routes/files.ts +++ b/src/routes/files.ts @@ -2,6 +2,7 @@ import { createReadStream } from 'node:fs'; import { nanoid } from 'nanoid'; import { findFileByPublicId } from '../db/files'; import { fileInfoCache } from '../utils/cache'; +import { createChunkedObjectResponse } from '../utils/chunked-storage'; import { cleanupTempFile, formatCreatedAt, getErrorMessage } from '../utils/file'; import logger from '../utils/logger'; import { metricsCollector } from '../utils/metrics'; @@ -52,6 +53,14 @@ export const handleFileRedirect = async (req: RequestWithParams): Promise config.telegramChunkSizeBytes) { + const file = await storeFileInTelegramChunks({ + tempPath, + partFileNamePrefix, + fileName: finalFileName, + mimeType, + sizeBytes: buffer.byteLength, + fileType: 'document', + uploaderId: 0, + bucketId, + s3Key: key, + }); + await cleanupTempFile(tempPath); + return s3Response(null, 200, reqId, { etag: `"${file.fileHash}"` }); + } + const forwardResult = await forwardToStorage( createReadStream(tempPath), - `s3-${bucketRecord.name}-${key.replace(/\//g, '_')}`, + partFileNamePrefix, 'document', ); @@ -566,7 +615,7 @@ const storeFileToTelegram = async ( fileType: 'document', uploaderId: 0, fileHash: hash, - bucketId: bucketRecord.id, + bucketId, s3Key: key, storageBackend: 'telegram', isDeleted: false, @@ -613,6 +662,17 @@ const handleCopyObject = async ( reqId, ); + if (sourceFile.storageBackend === 'chunked') { + // Copying chunked objects is not yet supported. + return s3ErrorResponse( + 'NotImplemented', + 'Copying chunked objects is not yet implemented.', + copySource, + 501, + reqId, + ); + } + // Conditional copy: if-match / if-none-match checks const ifMatch = headers['x-amz-copy-source-if-match']; const ifNoneMatch = headers['x-amz-copy-source-if-none-match']; @@ -883,6 +943,16 @@ const handleUploadPart = async ( const body = await req.arrayBuffer(); const buffer = Buffer.from(body); + if (buffer.byteLength > config.telegramChunkSizeBytes) { + return s3ErrorResponse( + 'EntityTooLarge', + `Your proposed upload part size (${buffer.byteLength} bytes) exceeds the maximum allowed part size (${config.telegramChunkSizeBytes} bytes) for this storage backend. Use smaller part sizes.`, + `/${bucket}/${key}`, + 400, + reqId, + ); + } + const tempPath = `/tmp/teleuploader-mp-${nanoid()}`; await Bun.write(tempPath, buffer); diff --git a/src/routes/upload.ts b/src/routes/upload.ts index 674d26d..bce99fc 100644 --- a/src/routes/upload.ts +++ b/src/routes/upload.ts @@ -2,6 +2,7 @@ import { createWriteStream } from 'node:fs'; import { nanoid } from 'nanoid'; import { findFileByHash } from '../db/files'; import { config } from '../env'; +import { storeFileInTelegramChunks } from '../utils/chunked-storage'; import { buildUploadResponse, checkFileSize, @@ -200,6 +201,20 @@ const handleMultipartUpload = async (req: Request): Promise => { return Response.json({ error: `File size exceeds ${fileType} limit` }, { status: 400 }); } + if (prepared.sizeBytes > config.telegramChunkSizeBytes) { + const file = await storeFileInTelegramChunks({ + tempPath: prepared.tempPath, + partFileNamePrefix: `direct-${prepared.fileHash?.slice(0, 16) || 'upload'}`, + fileName: finalFileName, + mimeType, + sizeBytes: prepared.sizeBytes, + fileType, + uploaderId: 0, + }); + await cleanupTempFile(prepared.tempPath); + return Response.json(buildUploadResponse(file, config.baseUrl), { status: 200 }); + } + const uploaded = await enqueuePreparedUpload({ prepared, fileName: finalFileName, @@ -257,6 +272,21 @@ const handleJSONUpload = async (req: Request): Promise => { } const prepared = await writeBufferToTemp(fileBytes, hash); + + if (prepared.sizeBytes > config.telegramChunkSizeBytes) { + const file = await storeFileInTelegramChunks({ + tempPath: prepared.tempPath, + partFileNamePrefix: `direct-${prepared.fileHash?.slice(0, 16) || 'json'}`, + fileName: finalFileName, + mimeType, + sizeBytes: prepared.sizeBytes, + fileType, + uploaderId: 0, + }); + await cleanupTempFile(prepared.tempPath); + return Response.json(buildUploadResponse(file, config.baseUrl), { status: 200 }); + } + const uploaded = await enqueuePreparedUpload({ prepared, fileName: finalFileName, diff --git a/src/routes/web-api.ts b/src/routes/web-api.ts index c4b049c..0196eba 100644 --- a/src/routes/web-api.ts +++ b/src/routes/web-api.ts @@ -8,6 +8,7 @@ import { softDeleteFile, } from '../db/files-ext'; import { config } from '../env'; +import { createChunkedObjectResponse, storeFileInTelegramChunks } from '../utils/chunked-storage'; import { cleanupTempFile, computeHash, ensureExtension, getErrorMessage } from '../utils/file'; import logger from '../utils/logger'; import { forwardToStorage, getFileInfo } from '../utils/telegram'; @@ -126,9 +127,35 @@ export const handleUploadObjectV1 = async ( file.type || 'application/octet-stream', ); + const partFileNamePrefix = `s3-${bucket.name}-${key.replace(/\//g, '_')}`; + + if (buffer.byteLength > config.telegramChunkSizeBytes) { + const file = await storeFileInTelegramChunks({ + tempPath, + partFileNamePrefix, + fileName: finalFileName, + mimeType, + sizeBytes: buffer.byteLength, + fileType: 'document', + uploaderId: 0, + bucketId: bucket.id, + s3Key: key, + }); + await cleanupTempFile(tempPath); + return json( + { + key, + size: buffer.byteLength, + etag: hash, + downloadUrl: `${config.baseUrl}/f/${file.publicId}`, + }, + 201, + ); + } + const forwardResult = await forwardToStorage( createReadStream(tempPath), - `s3-${bucket.name}-${key.replace(/\//g, '_')}`, + partFileNamePrefix, 'document', ); @@ -183,6 +210,11 @@ export const handleDownloadObjectV1 = async ( const file = await findFileByBucketAndKey(bucket.id, params.key!); if (!file) return jsonError('Object not found', 404); + if (file.storageBackend === 'chunked') { + const range = { type: 'none' as const }; + return createChunkedObjectResponse({ file, range, reqId: '' }); + } + const fileInfo = await getFileInfo(file.telegramFileId); const redirectUrl = `https://api.telegram.org/file/bot${fileInfo.bot_token}/${fileInfo.file_path}`; @@ -209,6 +241,10 @@ export const handleCopyObjectV1 = async (req: Request, params: RouteParams): Pro const sourceFile = await findFileByBucketAndKey(sourceBucket.id, body.sourceKey); if (!sourceFile) return jsonError('Source object not found', 404); + if (sourceFile.storageBackend === 'chunked') { + return json({ error: 'Copying chunked objects is not implemented' }, 501); + } + const publicId = nanoid(); const { db, files: fileSchema } = await import('../db/index'); diff --git a/src/utils/chunked-storage.ts b/src/utils/chunked-storage.ts new file mode 100644 index 0000000..363f03a --- /dev/null +++ b/src/utils/chunked-storage.ts @@ -0,0 +1,226 @@ +import { randomUUID } from 'node:crypto'; +import { createReadStream } from 'node:fs'; +import { gzipSync } from 'node:zlib'; +import { nanoid } from 'nanoid'; +import { db, files as fileSchema } from '../db'; +import { insertFileParts, listFileParts, type NewFilePartInput } from '../db/file-parts'; +import type { File } from '../db/schema'; +import { config } from '../env'; +import { computeHash } from './file'; +import { createGetObjectResponse, type ObjectPartSource } from './s3/object-stream'; +import type { RangeParseResult } from './s3/range'; +import { forwardToStorage, getFileInfo } from './telegram'; + +export type ChunkCompressionAlgorithm = 'gzip' | null; + +export interface ChunkedUploadPart { + partNumber: number; + telegramFileId: string; + telegramFileUniqueId: string; + storageMessageId: number; + sizeBytes: number; + storedSizeBytes: number; + compressionAlgorithm: ChunkCompressionAlgorithm; + etag: string; +} + +export interface ChunkedUploadResult { + parts: ChunkedUploadPart[]; + fileHash: string; + totalSizeBytes: number; +} + +export interface ChunkedFileInput { + tempPath: string; + partFileNamePrefix: string; + fileName: string; + mimeType: string; + sizeBytes: number; + fileType: string; + uploaderId: number; + bucketId?: string | null; + s3Key?: string | null; +} + +const asSafeChunkSize = (chunkSizeBytes: number): number => { + if (!Number.isSafeInteger(chunkSizeBytes) || chunkSizeBytes <= 0) { + throw new Error('Invalid Telegram chunk size'); + } + return chunkSizeBytes; +}; + +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' }; +}; + +export const uploadFileInTelegramChunks = async (input: { + tempPath: string; + partFileNamePrefix: string; + chunkSizeBytes: number; + compress: boolean; + compressionMinSizeBytes: number; +}): Promise => { + 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 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, + }; +}; + +export const storeFileInTelegramChunks = async (input: ChunkedFileInput): Promise => { + const upload = await 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 now = new Date(); + const fileId = randomUUID(); + const publicId = nanoid(); + const file: File = { + id: fileId, + 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, + createdAt: now, + updatedAt: now, + }; + + await db.insert(fileSchema).values(file); + + const fileParts: NewFilePartInput[] = upload.parts.map((part) => ({ + fileId, + 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 insertFileParts(fileParts); + return file; +}; + +export const buildChunkedObjectSources = async (file: File): Promise => { + const parts = await listFileParts(file.id); + const sources: ObjectPartSource[] = []; + + for (const part of parts) { + const fileInfo = await 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; +}; + +export const createChunkedObjectResponse = async (input: { + file: File; + range: RangeParseResult; + reqId: string; +}): Promise => { + const parts = await 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, + }); +}; diff --git a/src/utils/s3/object-stream.ts b/src/utils/s3/object-stream.ts index 2b4fcf8..ca1e211 100644 --- a/src/utils/s3/object-stream.ts +++ b/src/utils/s3/object-stream.ts @@ -1,3 +1,4 @@ +import { gunzipSync } from 'node:zlib'; import { applyS3Headers } from './headers'; import { contentRange, type RangeParseResult } from './range'; @@ -6,6 +7,8 @@ export interface ObjectPartSource { telegramUrl: string; sizeBytes: number; partNumber: number; + storedSizeBytes?: number; + compressionAlgorithm?: 'gzip' | null; } export interface ObjectResponseInput { @@ -54,10 +57,26 @@ const planParts = (parts: ObjectPartSource[], start: number, end: number): Plann return planned; }; +const streamFromBytes = (bytes: Uint8Array): ReadableStream => + new Response(bytes).body!; + +const fetchWholePartBytes = async (telegramUrl: string): Promise => { + const res = await fetch(telegramUrl); + if (!res.ok) throw new Error(`Telegram fetch failed: ${res.status}`); + return new Uint8Array(await res.arrayBuffer()); +}; + const fetchPartBody = async (planned: PlannedPart): Promise> => { - const rangeHeader = `bytes=${planned.relativeStart}-${planned.relativeEnd}`; const wantsWholePart = planned.relativeStart === 0 && planned.relativeEnd === planned.part.sizeBytes - 1; + + if (planned.part.compressionAlgorithm === 'gzip') { + const storedBytes = await fetchWholePartBytes(planned.part.telegramUrl); + const bytes = gunzipSync(storedBytes); + return streamFromBytes(bytes.subarray(planned.relativeStart, planned.relativeEnd + 1)); + } + + const rangeHeader = `bytes=${planned.relativeStart}-${planned.relativeEnd}`; const res = await fetch( planned.part.telegramUrl, wantsWholePart ? undefined : { headers: { range: rangeHeader } }, @@ -66,7 +85,7 @@ const fetchPartBody = async (planned: PlannedPart): Promise => diff --git a/test/chunked-storage.test.ts b/test/chunked-storage.test.ts new file mode 100644 index 0000000..9f7e5ca --- /dev/null +++ b/test/chunked-storage.test.ts @@ -0,0 +1,174 @@ +import { beforeEach, describe, expect, it, mock } from 'bun:test'; + +// Mock DB layer (chunked-storage imports db, insertFileParts, listFileParts) +const mockInsert = mock(() => Promise.resolve()); +const mockPartsInsert = mock(() => Promise.resolve()); +const mockPartsSelect = mock(() => Promise.resolve([])); + +mock.module('../src/db/index', () => ({ + db: { + insert: mockInsert, + select: () => ({ from: () => ({ where: () => ({ limit: () => Promise.resolve([]) }) }) }), + execute: mock(() => Promise.resolve([])), + }, + files: {}, + fileParts: {}, +})); + +mock.module('../src/db/file-parts', () => ({ + insertFileParts: mockPartsInsert, + listFileParts: mockPartsSelect, +})); + +mock.module('../src/utils/telegram', () => ({ + forwardToStorage: async (_bytes: unknown, fileName: string) => ({ + telegramFileId: `tg-${fileName}`, + telegramFileUniqueId: `tg-unique-${fileName}`, + storageMessageId: Math.floor(Math.random() * 100000) + 1, + }), + getFileInfo: async (telegramFileId: string) => ({ + file_size: 0, + mime_type: 'application/octet-stream', + file_path: `documents/${telegramFileId}`, + bot_token: '123456:ABC-DEF', + }), +})); + +const writeTemp = async (path: string, data: Buffer): Promise => { + await Bun.write(path, data); +}; + +const rmTemp = async (path: string): Promise => { + try { + await Bun.$`rm -f ${path}`; + } catch { + /* ignore */ + } +}; + +describe('chunked-storage utility', () => { + beforeEach(() => { + mockInsert.mockClear(); + mockPartsInsert.mockClear(); + mockPartsSelect.mockClear(); + }); + + it('should split a file into correct number of chunks', async () => { + const { uploadFileInTelegramChunks } = await import('../src/utils/chunked-storage'); + + const data = Buffer.from('1234567890ab'); + const path = '/tmp/test-chunk-1'; + await writeTemp(path, data); + + const result = await uploadFileInTelegramChunks({ + tempPath: path, + partFileNamePrefix: 'test-1', + chunkSizeBytes: 4, + compress: false, + compressionMinSizeBytes: 4096, + }); + + await rmTemp(path); + + // 12 bytes at 4 bytes/chunk = 3 chunks + expect(result.parts.length).toBe(3); + expect(result.totalSizeBytes).toBe(12); + expect(result.parts[0].partNumber).toBe(1); + expect(result.parts[1].partNumber).toBe(2); + expect(result.parts[2].partNumber).toBe(3); + expect(result.parts[0].storedSizeBytes).toBe(4); + expect(result.parts[0].compressionAlgorithm).toBeNull(); + }); + + it('should compute correct full-file hash', async () => { + const { uploadFileInTelegramChunks } = await import('../src/utils/chunked-storage'); + const { createHash } = await import('node:crypto'); + + const data = Buffer.from('Hello, chunked storage!'); + const path = '/tmp/test-chunk-hash'; + await writeTemp(path, data); + + // Expected SHA-256 + const expectedHash = createHash('sha256').update(data).digest('hex'); + + const result = await uploadFileInTelegramChunks({ + tempPath: path, + partFileNamePrefix: 'test-hash', + chunkSizeBytes: 10, + compress: false, + compressionMinSizeBytes: 4096, + }); + + await rmTemp(path); + + expect(result.fileHash).toBe(expectedHash); + expect(result.totalSizeBytes).toBe(data.byteLength); + }); + + it('should gzip compressible chunks and skip incompressible ones', async () => { + const { uploadFileInTelegramChunks } = await import('../src/utils/chunked-storage'); + + // Use data large enough to exceed compressionMinSizeBytes + const data = Buffer.from('AAAAAAAAAA'.repeat(100)); // 1000 bytes, very compressible + const path = '/tmp/test-chunk-compress'; + await writeTemp(path, data); + + const result = await uploadFileInTelegramChunks({ + tempPath: path, + partFileNamePrefix: 'test-comp', + chunkSizeBytes: 512, + compress: true, + compressionMinSizeBytes: 10, + }); + + await rmTemp(path); + + expect(result.parts.length).toBe(2); + // At least one chunk was compressed (gzip) + for (const part of result.parts) { + expect(part.storedSizeBytes).toBeLessThanOrEqual(part.sizeBytes); + if (part.storedSizeBytes < part.sizeBytes) { + expect(part.compressionAlgorithm).toBe('gzip'); + } + } + expect(result.totalSizeBytes).toBe(1000); + }); + + it('should not attempt compression for incompressible data', async () => { + const { uploadFileInTelegramChunks } = await import('../src/utils/chunked-storage'); + const { gzipSync } = await import('node:zlib'); + + const original = Buffer.from('AAAA'.repeat(100)); + const compressed = gzipSync(original); + const path = '/tmp/test-chunk-incompress'; + await writeTemp(path, compressed); + + const result = await uploadFileInTelegramChunks({ + tempPath: path, + partFileNamePrefix: 'test-inc', + chunkSizeBytes: 1024, + compress: true, + compressionMinSizeBytes: 10, + }); + + await rmTemp(path); + + // Incompressible data should stay uncompressed + expect(result.parts[0].compressionAlgorithm).toBeNull(); + expect(result.parts[0].sizeBytes).toBe(result.parts[0].storedSizeBytes); + }); + + it('should reject chunk size of zero', async () => { + const { uploadFileInTelegramChunks } = await import('../src/utils/chunked-storage'); + + expect( + uploadFileInTelegramChunks({ + tempPath: '/nonexistent', + partFileNamePrefix: 'err', + chunkSizeBytes: 0, + compress: false, + compressionMinSizeBytes: 4096, + }), + ).rejects.toThrow('Invalid Telegram chunk size'); + }); +}); diff --git a/test/files.test.ts b/test/files.test.ts index ee4cb12..73fca36 100644 --- a/test/files.test.ts +++ b/test/files.test.ts @@ -68,6 +68,11 @@ const mockGetFileInfo = mock(async (_telegramFileId: string) => ({ })); mock.module('../src/utils/telegram', () => ({ + forwardToStorage: async () => ({ + telegramFileId: 'mock-tg-id', + telegramFileUniqueId: 'mock-tg-unique', + storageMessageId: 12345, + }), getFileInfo: mockGetFileInfo, })); diff --git a/test/upload.test.ts b/test/upload.test.ts index b6a27bb..b699301 100644 --- a/test/upload.test.ts +++ b/test/upload.test.ts @@ -79,6 +79,12 @@ const mockForwardToStorage = mock(() => mock.module('../src/utils/telegram', () => ({ forwardToStorage: mockForwardToStorage, + getFileInfo: async (telegramFileId: string) => ({ + file_size: 0, + mime_type: 'application/octet-stream', + file_path: `documents/${telegramFileId}`, + bot_token: '123456:ABC-DEF', + }), getBot: () => ({ telegram: { getFile: mock(() =>