From db5de1367d10b0cfe302307583f978f4e7f26004 Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Fri, 22 May 2026 00:51:49 +0700 Subject: [PATCH] feat: extend file schema with archive metadata, implement batch upload processing, and add zip utilities for file handling --- schema.sql | 13 ++ src/db/schema.ts | 6 + src/routes/files.ts | 70 +++++++--- src/routes/upload.ts | 74 ++--------- src/utils/uploadBatcher.ts | 155 +++++++++++++++++++++++ src/utils/zip.ts | 253 +++++++++++++++++++++++++++++++++++++ test/zip.test.ts | 45 +++++++ 7 files changed, 538 insertions(+), 78 deletions(-) create mode 100644 src/utils/uploadBatcher.ts create mode 100644 src/utils/zip.ts create mode 100644 test/zip.test.ts diff --git a/schema.sql b/schema.sql index fe17f61..bcaf1c5 100644 --- a/schema.sql +++ b/schema.sql @@ -11,14 +11,27 @@ CREATE TABLE IF NOT EXISTS files ( file_type VARCHAR NOT NULL, uploader_id BIGINT NOT NULL, file_hash VARCHAR, + archive_telegram_file_id VARCHAR, + archive_storage_message_id BIGINT, + archive_file_name VARCHAR, + archive_entry_name VARCHAR, + archive_mime_type VARCHAR, + archive_size_bytes BIGINT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); ALTER TABLE files ADD COLUMN IF NOT EXISTS file_hash VARCHAR; +ALTER TABLE files ADD COLUMN IF NOT EXISTS archive_telegram_file_id VARCHAR; +ALTER TABLE files ADD COLUMN IF NOT EXISTS archive_storage_message_id BIGINT; +ALTER TABLE files ADD COLUMN IF NOT EXISTS archive_file_name VARCHAR; +ALTER TABLE files ADD COLUMN IF NOT EXISTS archive_entry_name VARCHAR; +ALTER TABLE files ADD COLUMN IF NOT EXISTS archive_mime_type VARCHAR; +ALTER TABLE files ADD COLUMN IF NOT EXISTS archive_size_bytes BIGINT; CREATE INDEX IF NOT EXISTS idx_files_public_id ON files(public_id); CREATE INDEX IF NOT EXISTS idx_files_telegram_file_id ON files(telegram_file_id); CREATE INDEX IF NOT EXISTS idx_files_file_hash ON files(file_hash); +CREATE INDEX IF NOT EXISTS idx_files_archive_telegram_file_id ON files(archive_telegram_file_id); CREATE INDEX IF NOT EXISTS idx_files_uploader_id ON files(uploader_id); CREATE INDEX IF NOT EXISTS idx_files_created_at ON files(created_at DESC); \ No newline at end of file diff --git a/src/db/schema.ts b/src/db/schema.ts index 03265be..1a1d1a5 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -14,6 +14,12 @@ export const files = pgTable('files', { fileType: text('file_type').notNull(), uploaderId: bigint('uploader_id', { mode: 'number' }).notNull(), fileHash: text('file_hash'), + archiveTelegramFileId: text('archive_telegram_file_id'), + archiveStorageMessageId: bigint('archive_storage_message_id', { mode: 'number' }), + archiveFileName: text('archive_file_name'), + archiveEntryName: text('archive_entry_name'), + archiveMimeType: text('archive_mime_type'), + archiveSizeBytes: bigint('archive_size_bytes', { mode: 'number' }), createdAt: timestamp('created_at').defaultNow().notNull(), updatedAt: timestamp('updated_at').defaultNow().notNull(), }); diff --git a/src/routes/files.ts b/src/routes/files.ts index 79fe3d1..829b5dc 100644 --- a/src/routes/files.ts +++ b/src/routes/files.ts @@ -4,6 +4,7 @@ import { formatCreatedAt, getErrorMessage } from '../utils/file'; import logger from '../utils/logger'; import { checkRateLimit } from '../utils/rateLimit'; import { getBot } from '../utils/telegram'; +import { extractZipEntry } from '../utils/zip'; type RequestWithParams = Request & { params?: { @@ -11,6 +12,30 @@ type RequestWithParams = Request & { }; }; +const getTelegramFileInfo = async (telegramFileId: string, public_id: string) => { + const cacheKey = `file_info_${telegramFileId}`; + let fileInfo = fileInfoCache.get(cacheKey); + + if (!fileInfo) { + const bot = getBot(); + const apiFileInfo = await bot.telegram.getFile(telegramFileId); + fileInfo = { + file_size: (apiFileInfo as any).file_size || 0, + mime_type: (apiFileInfo as any).mime_type || 'application/octet-stream', + file_path: (apiFileInfo as any).file_path || '', + }; + fileInfoCache.set(cacheKey, fileInfo); + logger.debug('File info cached', { public_id, cacheKey }); + } else { + logger.debug('File info from cache', { public_id, cacheKey }); + } + + return fileInfo; +}; + +const buildTelegramFileUrl = (filePath: string): string => + `https://api.telegram.org/file/bot${process.env.BOT_TOKEN}/${filePath}`; + export const handleFileRedirect = async (req: RequestWithParams): Promise => { const public_id = req.params?.public_id; try { @@ -27,27 +52,36 @@ export const handleFileRedirect = async (req: RequestWithParams): Promise { const JSON_UPLOAD_LIMIT_BYTES = 50 * 1024 * 1024; const SIGNATURE_BYTES = 16; -type PreparedUpload = { - tempPath: string; - fileHash: string; - sizeBytes: number; - signatureBuffer: Buffer; -}; - const cleanupTempFile = async (tempPath: string): Promise => { try { await unlink(tempPath); @@ -137,46 +123,6 @@ const writeBufferToTemp = async (fileBuffer: Buffer, fileHash: string): Promise< } }; -const closeFileStream = async (fileStream: ReturnType): Promise => { - if (fileStream.closed) return; - - await new Promise((resolve) => { - fileStream.once('close', resolve); - fileStream.destroy(); - }); -}; - -const performUpload = async ( - prepared: PreparedUpload, - fileName: string, - mimeType: string, - fileType: string, -): Promise => { - const fileStream = createReadStream(prepared.tempPath); - try { - const result = await forwardToStorage(fileStream, fileName, fileType); - - return { - publicId: nanoid(), - telegramFileId: result.telegramFileId, - telegramFileUniqueId: result.telegramFileUniqueId, - storageChatId: config.storageChatId, - storageMessageId: result.storageMessageId, - fileName, - mimeType: mimeType || 'application/octet-stream', - sizeBytes: prepared.sizeBytes, - fileType, - uploaderId: 0, - fileHash: prepared.fileHash, - createdAt: new Date(), - updatedAt: new Date(), - }; - } finally { - await closeFileStream(fileStream); - await cleanupTempFile(prepared.tempPath); - } -}; - export const handleUpload = async (req: Request): Promise => { try { const contentType = req.headers.get('content-type') || ''; @@ -230,8 +176,12 @@ const handleMultipartUpload = async (req: Request): Promise => { return Response.json({ error: `File size exceeds ${fileType} limit` }, { status: 400 }); } - const uploaded = await performUpload(prepared, finalFileName, mimeType, fileType); - await db.insert(fileSchema).values(uploaded); + const uploaded = await enqueuePreparedUpload({ + prepared, + fileName: finalFileName, + mimeType, + fileType, + }); return Response.json(buildUploadResponse(uploaded, config.baseUrl), { status: 200 }); } catch (error: unknown) { @@ -280,8 +230,12 @@ const handleJSONUpload = async (req: Request): Promise => { } const prepared = await writeBufferToTemp(fileBytes, hash); - const uploaded = await performUpload(prepared, finalFileName, mimeType, fileType); - await db.insert(fileSchema).values(uploaded); + const uploaded = await enqueuePreparedUpload({ + prepared, + fileName: finalFileName, + mimeType, + fileType, + }); return Response.json(buildUploadResponse(uploaded, config.baseUrl), { status: 200 }); } catch (error: unknown) { diff --git a/src/utils/uploadBatcher.ts b/src/utils/uploadBatcher.ts new file mode 100644 index 0000000..b967cad --- /dev/null +++ b/src/utils/uploadBatcher.ts @@ -0,0 +1,155 @@ +import { createReadStream } from 'node:fs'; +import { unlink } from 'node:fs/promises'; +import { nanoid } from 'nanoid'; +import { db, files as fileSchema } from '../db'; +import type { NewFile } from '../db/schema'; +import { config } from '../env'; +import { getErrorMessage } from './file'; +import logger from './logger'; +import { forwardToStorage } from './telegram'; +import { createZip, type ZipEntry } from './zip'; + +export type PreparedUpload = { + tempPath: string; + fileHash: string; + sizeBytes: number; + signatureBuffer: Buffer; +}; + +export type UploadedFile = NewFile & { + createdAt: Date; + updatedAt: Date; +}; + +export type BatchUploadItem = { + prepared: PreparedUpload; + fileName: string; + mimeType: string; + fileType: string; +}; + +type PendingUpload = BatchUploadItem & { + resolve: (file: UploadedFile) => void; + reject: (error: unknown) => void; +}; + +const BATCH_WINDOW_MS = 2000; +const MAX_BATCH_ITEMS = 100; +const MAX_BATCH_SIZE_BYTES = 2 * 1024 * 1024 * 1024; + +let pendingUploads: PendingUpload[] = []; +let flushTimer: ReturnType | null = null; + +const cleanupTempFile = async (tempPath: string): Promise => { + try { + await unlink(tempPath); + } catch (error) { + logger.warn('Failed to cleanup temp file', { tempPath, error: getErrorMessage(error) }); + } +}; + +const buildUploadedFile = ( + item: BatchUploadItem, + entry: ZipEntry, + archive: { + telegramFileId: string; + telegramFileUniqueId: string; + storageMessageId: number; + fileName: string; + sizeBytes: number; + }, +): UploadedFile => ({ + 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, + createdAt: new Date(), + updatedAt: new Date(), +}); + +const flushUploads = async (): Promise => { + if (flushTimer) { + clearTimeout(flushTimer); + flushTimer = null; + } + + const batch = pendingUploads; + pendingUploads = []; + 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 = `teleuploader-${nanoid()}.zip`; + const archiveResult = await forwardToStorage( + createReadStream(zip.tempPath), + archiveFileName, + 'document', + ); + + const uploadedFiles = batch.map((item, index) => + buildUploadedFile(item, zip.entries[index], { + telegramFileId: archiveResult.telegramFileId, + telegramFileUniqueId: archiveResult.telegramFileUniqueId, + storageMessageId: archiveResult.storageMessageId, + fileName: archiveFileName, + sizeBytes: zip.sizeBytes, + }), + ); + + await db.insert(fileSchema).values(uploadedFiles); + + for (let i = 0; i < batch.length; i++) { + batch[i].resolve(uploadedFiles[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); + } +}; + +const getPendingSize = (): number => + pendingUploads.reduce((total, item) => total + item.prepared.sizeBytes, 0); + +export const enqueuePreparedUpload = (item: BatchUploadItem): Promise => { + return new Promise((resolve, reject) => { + pendingUploads.push({ ...item, resolve, reject }); + + if (!flushTimer) { + flushTimer = setTimeout(() => { + void flushUploads(); + }, BATCH_WINDOW_MS); + } + + if (pendingUploads.length >= MAX_BATCH_ITEMS || getPendingSize() >= MAX_BATCH_SIZE_BYTES) { + void flushUploads(); + } + }); +}; + +export const flushPendingUploads = async (): Promise => { + await flushUploads(); +}; + +export const getPendingUploadCount = (): number => pendingUploads.length; diff --git a/src/utils/zip.ts b/src/utils/zip.ts new file mode 100644 index 0000000..30c014b --- /dev/null +++ b/src/utils/zip.ts @@ -0,0 +1,253 @@ +import { createReadStream, createWriteStream } from 'node:fs'; +import { stat } from 'node:fs/promises'; +import { basename } from 'node:path'; +import { nanoid } from 'nanoid'; + +export type ZipInputFile = { + tempPath: string; + fileName: string; +}; + +export type ZipEntry = { + fileName: string; + entryName: string; + crc32: number; + compressedSize: number; + uncompressedSize: number; + localHeaderOffset: number; +}; + +export type CreatedZip = { + tempPath: string; + sizeBytes: number; + fileHash: string; + entries: ZipEntry[]; +}; + +const CRC32_TABLE = new Uint32Array(256).map((_, index) => { + let value = index; + for (let bit = 0; bit < 8; bit++) { + value = value & 1 ? 0xedb88320 ^ (value >>> 1) : value >>> 1; + } + return value >>> 0; +}); + +const updateCrc32 = (crc: number, chunk: Buffer): number => { + let value = crc; + for (const byte of chunk) { + value = CRC32_TABLE[(value ^ byte) & 0xff] ^ (value >>> 8); + } + return value >>> 0; +}; + +const dosDateTime = (date = new Date()): { date: number; time: number } => { + const year = Math.max(date.getFullYear(), 1980); + return { + time: (date.getHours() << 11) | (date.getMinutes() << 5) | Math.floor(date.getSeconds() / 2), + date: ((year - 1980) << 9) | ((date.getMonth() + 1) << 5) | date.getDate(), + }; +}; + +const writeUInt16 = (value: number): Buffer => { + const buffer = Buffer.allocUnsafe(2); + buffer.writeUInt16LE(value & 0xffff, 0); + return buffer; +}; + +const writeUInt32 = (value: number): Buffer => { + const buffer = Buffer.allocUnsafe(4); + buffer.writeUInt32LE(value >>> 0, 0); + return buffer; +}; + +const writeChunk = async ( + writer: ReturnType, + chunk: Buffer, +): Promise => { + if (!writer.write(chunk)) { + await new Promise((resolve, reject) => { + writer.once('drain', resolve); + writer.once('error', reject); + }); + } +}; + +const finishWriter = async (writer: ReturnType): Promise => { + await new Promise((resolve, reject) => { + writer.end(() => resolve()); + writer.once('error', reject); + }); +}; + +export const sanitizeZipEntryName = (fileName: string, usedNames = new Set()): string => { + const cleaned = basename(fileName) + .replace(/[\\/]+/g, '_') + .replace(/\.\.+/g, '.') + .trim(); + const fallback = cleaned && cleaned !== '.' && cleaned !== '..' ? cleaned : 'file'; + const dotIndex = fallback.lastIndexOf('.'); + const baseName = dotIndex > 0 ? fallback.slice(0, dotIndex) : fallback; + const extension = dotIndex > 0 ? fallback.slice(dotIndex) : ''; + let candidate = fallback; + let counter = 1; + + while (usedNames.has(candidate)) { + candidate = `${baseName}-${counter}${extension}`; + counter++; + } + + usedNames.add(candidate); + return candidate; +}; + +export const createZip = async (files: ZipInputFile[]): Promise => { + const tempPath = `/tmp/teleuploader-${nanoid()}.zip`; + const writer = createWriteStream(tempPath); + const hasher = new Bun.CryptoHasher('sha256'); + const entries: ZipEntry[] = []; + const usedNames = new Set(); + let offset = 0; + + const writeHashed = async (chunk: Buffer): Promise => { + hasher.update(chunk); + await writeChunk(writer, chunk); + offset += chunk.byteLength; + }; + + try { + for (const file of files) { + const entryName = sanitizeZipEntryName(file.fileName, usedNames); + const nameBuffer = Buffer.from(entryName); + const fileStats = await stat(file.tempPath); + const { date, time } = dosDateTime(); + const localHeaderOffset = offset; + let crc = 0xffffffff; + + const chunks: Buffer[] = []; + await new Promise((resolve, reject) => { + const reader = createReadStream(file.tempPath); + reader.on('data', (chunk: Buffer) => { + crc = updateCrc32(crc, chunk); + chunks.push(chunk); + }); + reader.once('end', resolve); + reader.once('error', reject); + }); + + const crc32 = (crc ^ 0xffffffff) >>> 0; + const localHeader = Buffer.concat([ + writeUInt32(0x04034b50), + writeUInt16(20), + writeUInt16(0), + writeUInt16(0), + writeUInt16(time), + writeUInt16(date), + writeUInt32(crc32), + writeUInt32(fileStats.size), + writeUInt32(fileStats.size), + writeUInt16(nameBuffer.byteLength), + writeUInt16(0), + nameBuffer, + ]); + + await writeHashed(localHeader); + for (const chunk of chunks) { + await writeHashed(chunk); + } + + entries.push({ + fileName: file.fileName, + entryName, + crc32, + compressedSize: fileStats.size, + uncompressedSize: fileStats.size, + localHeaderOffset, + }); + } + + const centralDirectoryOffset = offset; + for (const entry of entries) { + const nameBuffer = Buffer.from(entry.entryName); + const { date, time } = dosDateTime(); + await writeHashed( + Buffer.concat([ + writeUInt32(0x02014b50), + writeUInt16(20), + writeUInt16(20), + writeUInt16(0), + writeUInt16(0), + writeUInt16(time), + writeUInt16(date), + writeUInt32(entry.crc32), + writeUInt32(entry.compressedSize), + writeUInt32(entry.uncompressedSize), + writeUInt16(nameBuffer.byteLength), + writeUInt16(0), + writeUInt16(0), + writeUInt16(0), + writeUInt16(0), + writeUInt32(0), + writeUInt32(entry.localHeaderOffset), + nameBuffer, + ]), + ); + } + + const centralDirectorySize = offset - centralDirectoryOffset; + await writeHashed( + Buffer.concat([ + writeUInt32(0x06054b50), + writeUInt16(0), + writeUInt16(0), + writeUInt16(entries.length), + writeUInt16(entries.length), + writeUInt32(centralDirectorySize), + writeUInt32(centralDirectoryOffset), + writeUInt16(0), + ]), + ); + + await finishWriter(writer); + + return { + tempPath, + sizeBytes: offset, + fileHash: hasher.digest('hex'), + entries, + }; + } catch (error) { + writer.destroy(); + throw error; + } +}; + +export const extractZipEntry = async ( + zipBuffer: Buffer, + entryName: string, +): Promise => { + let offset = 0; + + while (offset + 30 <= zipBuffer.byteLength) { + const signature = zipBuffer.readUInt32LE(offset); + if (signature !== 0x04034b50) break; + + const compressionMethod = zipBuffer.readUInt16LE(offset + 8); + const compressedSize = zipBuffer.readUInt32LE(offset + 18); + const fileNameLength = zipBuffer.readUInt16LE(offset + 26); + const extraLength = zipBuffer.readUInt16LE(offset + 28); + const nameStart = offset + 30; + const nameEnd = nameStart + fileNameLength; + const dataStart = nameEnd + extraLength; + const dataEnd = dataStart + compressedSize; + const currentName = zipBuffer.subarray(nameStart, nameEnd).toString(); + + if (currentName === entryName) { + if (compressionMethod !== 0) return null; + return zipBuffer.subarray(dataStart, dataEnd); + } + + offset = dataEnd; + } + + return null; +}; diff --git a/test/zip.test.ts b/test/zip.test.ts new file mode 100644 index 0000000..fbed833 --- /dev/null +++ b/test/zip.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, it } from 'bun:test'; +import { unlink, writeFile } from 'node:fs/promises'; +import { createZip, extractZipEntry, sanitizeZipEntryName } from '../src/utils/zip'; + +const cleanup = async (...paths: string[]) => { + await Promise.all( + paths.map(async (path) => { + try { + await unlink(path); + } catch {} + }), + ); +}; + +describe('ZIP utilities', () => { + it('should create a zip and extract entries by name', async () => { + const firstPath = `/tmp/teleuploader-test-${crypto.randomUUID()}-1.txt`; + const secondPath = `/tmp/teleuploader-test-${crypto.randomUUID()}-2.txt`; + await writeFile(firstPath, 'hello'); + await writeFile(secondPath, 'world'); + + const zip = await createZip([ + { tempPath: firstPath, fileName: 'greeting.txt' }, + { tempPath: secondPath, fileName: 'greeting.txt' }, + ]); + + try { + const zipBuffer = Buffer.from(await Bun.file(zip.tempPath).arrayBuffer()); + expect(zipBuffer.subarray(0, 2).toString()).toBe('PK'); + expect(zip.entries.map((entry) => entry.entryName)).toEqual([ + 'greeting.txt', + 'greeting-1.txt', + ]); + expect((await extractZipEntry(zipBuffer, 'greeting.txt'))?.toString()).toBe('hello'); + expect((await extractZipEntry(zipBuffer, 'greeting-1.txt'))?.toString()).toBe('world'); + } finally { + await cleanup(firstPath, secondPath, zip.tempPath); + } + }); + + it('should sanitize unsafe entry names', () => { + expect(sanitizeZipEntryName('../secret.txt')).toBe('secret.txt'); + expect(sanitizeZipEntryName('nested/path/file.txt')).toBe('file.txt'); + }); +});