From fab91ad69c23ad03035810c908cfad5b2fdf73f7 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 29 Jul 2026 16:10:49 +0700 Subject: [PATCH] =?UTF-8?q?feat:=20dedup=20wave=203=20=E2=80=94=20constant?= =?UTF-8?q?=20+=20streamToTemp=20utility?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ♻️ randomUUID() → nanoid() di s3-object.ts (eliminasi node:crypto) - ✨ DEFAULT_FILE_TYPE constant, ganti 8× hardcoded 'document' - ✨ streamToTemp() shared utility (src/shared/utils/temp-stream.ts) - ♻️ 3× streaming-to-temp pattern di s3-controller, upload-controller, web-api-controller → pake streamToTemp() - Lint ✅ Build ✅ --- src/application/use-cases/multipart-upload.ts | 4 +- src/application/use-cases/s3-object.ts | 14 +- .../http/controllers/s3-controller.ts | 75 ++-------- .../http/controllers/upload-controller.ts | 67 +-------- .../http/controllers/web-api-controller.ts | 77 ++++------ src/shared/utils/file.ts | 3 + src/shared/utils/temp-stream.ts | 132 ++++++++++++++++++ 7 files changed, 190 insertions(+), 182 deletions(-) create mode 100644 src/shared/utils/temp-stream.ts diff --git a/src/application/use-cases/multipart-upload.ts b/src/application/use-cases/multipart-upload.ts index 4db78ea..92710bd 100644 --- a/src/application/use-cases/multipart-upload.ts +++ b/src/application/use-cases/multipart-upload.ts @@ -5,7 +5,7 @@ import type { IBucketRepository } from '../../domain/ports/bucket-repository'; import type { IFileRepository } from '../../domain/ports/file-repository'; import type { IMultipartRepository } from '../../domain/ports/multipart-repository'; import type { ITelegramService } from '../../domain/ports/telegram-service'; -import { computeHash } from '../../shared/utils/file'; +import { computeHash, DEFAULT_FILE_TYPE } from '../../shared/utils/file'; // ─── Types ────────────────────────────────────────────────────────── @@ -281,7 +281,7 @@ export function createCompleteMultipartUploadUseCase(deps: MultipartDeps) { fileName: input.key.split('/').pop() || 'file', mimeType: 'application/octet-stream', sizeBytes: totalSize, - fileType: 'document', + fileType: DEFAULT_FILE_TYPE, storageBackend: 'telegram', bucketId: multipart.bucketId, s3Key: input.key, diff --git a/src/application/use-cases/s3-object.ts b/src/application/use-cases/s3-object.ts index 7cc411b..1421ada 100644 --- a/src/application/use-cases/s3-object.ts +++ b/src/application/use-cases/s3-object.ts @@ -1,4 +1,3 @@ -import { randomUUID } from 'node:crypto'; import { gzipSync } from 'node:zlib'; import { nanoid } from 'nanoid'; import type { File } from '../../domain/entities/file'; @@ -10,7 +9,12 @@ import type { IFilePartRepository } from '../../domain/ports/file-part-repositor import type { IFileRepository, S3FileRecord } from '../../domain/ports/file-repository'; import type { IMultipartRepository } from '../../domain/ports/multipart-repository'; import type { ITelegramService, TelegramFileInfo } from '../../domain/ports/telegram-service'; -import { computeHash, ensureExtension, formatCreatedAt } from '../../shared/utils/file'; +import { + computeHash, + DEFAULT_FILE_TYPE, + ensureExtension, + formatCreatedAt, +} from '../../shared/utils/file'; // ─── Types ────────────────────────────────────────────────────────── @@ -480,7 +484,7 @@ export function createPutObjectUseCase(deps: S3ObjectDeps) { throw new Error('Chunked upload produced no parts'); } - const fileId = randomUUID(); + const fileId = nanoid(); const publicId = nanoid(); await deps.fileRepo.create( @@ -493,7 +497,7 @@ export function createPutObjectUseCase(deps: S3ObjectDeps) { fileName: finalFileName, mimeType, sizeBytes: chunkResult.totalSizeBytes, - fileType: 'document', + fileType: DEFAULT_FILE_TYPE, fileHash: chunkResult.fileHash, bucketId: bucket.id, s3Key: key, @@ -539,7 +543,7 @@ export function createPutObjectUseCase(deps: S3ObjectDeps) { fileName: finalFileName, mimeType, sizeBytes: body.byteLength, - fileType: 'document', + fileType: DEFAULT_FILE_TYPE, fileHash: hash, bucketId: bucket.id, s3Key: key, diff --git a/src/interfaces/http/controllers/s3-controller.ts b/src/interfaces/http/controllers/s3-controller.ts index 903c30a..8cf028b 100644 --- a/src/interfaces/http/controllers/s3-controller.ts +++ b/src/interfaces/http/controllers/s3-controller.ts @@ -12,7 +12,13 @@ import { } from '../../../infrastructure/di'; import { botPool } from '../../../infrastructure/telegram/bot-pool'; import logger from '../../../shared/logger/index'; -import { cleanupTempFile, ensureExtension, getErrorMessage } from '../../../shared/utils/file'; +import { + cleanupTempFile, + DEFAULT_FILE_TYPE, + ensureExtension, + getErrorMessage, +} from '../../../shared/utils/file'; +import { streamToTemp } from '../../../shared/utils/temp-stream'; import { verifyBodyHash, verifyPresignedUrl, verifySignature } from '../../s3/auth'; import { S3_CORS_HEADERS, s3Headers } from '../../s3/headers'; import { createGetObjectResponse, type ObjectPartSource } from '../../s3/object-stream'; @@ -818,17 +824,10 @@ const streamBodyToTemp = async ( ): Promise<{ tempPath: string; fileHash: string; - md5Hash: string; + md5Hash?: string; sizeBytes: number; signatureBuffer: Buffer; }> => { - const tempPath = `/tmp/filedrop-s3-${nanoid()}`; - const writer = Bun.file(tempPath).writer(); - const sha256 = new Bun.CryptoHasher('sha256'); - const md5 = new Bun.CryptoHasher('md5'); - let writerFailed = false; - - // Handle body being null (GET/HEAD/DELETE or empty PUT) const reader = ( body ?? new ReadableStream({ @@ -836,56 +835,8 @@ const streamBodyToTemp = async ( c.close(); }, }) - ).getReader(); - const SIGNATURE_BYTES = 16; - const signatureChunks: Buffer[] = []; - let signatureBytes = 0; - let sizeBytes = 0; - - try { - while (true) { - const { done, value } = await reader.read(); - if (done) break; - const chunk = Buffer.from(value); - sizeBytes += chunk.byteLength; - sha256.update(chunk); - md5.update(chunk); - writer.write(chunk); - - if (signatureBytes < SIGNATURE_BYTES) { - const remaining = SIGNATURE_BYTES - signatureBytes; - const sigChunk = chunk.subarray(0, remaining); - signatureChunks.push(sigChunk); - signatureBytes += sigChunk.byteLength; - } - } - - try { - writer.end(); - } catch { - writerFailed = true; - } - - return { - tempPath, - fileHash: sha256.digest('hex'), - md5Hash: md5.digest('base64'), - sizeBytes, - signatureBuffer: Buffer.concat(signatureChunks, signatureBytes), - }; - } catch (error) { - if (!writerFailed) { - try { - writer.end(); - } catch { - /* writer may already be errored */ - } - } - await cleanupTempFile(tempPath); - throw error; - } finally { - reader.releaseLock(); - } + ).getReader() as ReadableStreamDefaultReader; + return streamToTemp(reader, { computeMd5: true, prefix: '/tmp/filedrop-s3-' }); }; /** @@ -1049,7 +1000,7 @@ const storeFileFromTemp = async ( fileName: finalFileName, mimeType, sizeBytes: streamed.sizeBytes, - fileType: 'document', + fileType: DEFAULT_FILE_TYPE, uploaderId: 0, bucketId, s3Key: key, @@ -1080,7 +1031,7 @@ const storeFileFromTemp = async ( fileName: finalFileName, mimeType, sizeBytes: streamed.sizeBytes, - fileType: 'document', + fileType: DEFAULT_FILE_TYPE, uploaderId: 0, fileHash: streamed.fileHash, bucketId, @@ -1687,7 +1638,7 @@ const handleCompleteMultipartUpload = async ( fileName: key.split('/').pop() || 'file', mimeType, sizeBytes: totalSize, - fileType: 'document', + fileType: DEFAULT_FILE_TYPE, uploaderId: 0, bucketId: multipart.bucketId, s3Key: key, diff --git a/src/interfaces/http/controllers/upload-controller.ts b/src/interfaces/http/controllers/upload-controller.ts index 2c1ef10..22cc846 100644 --- a/src/interfaces/http/controllers/upload-controller.ts +++ b/src/interfaces/http/controllers/upload-controller.ts @@ -1,4 +1,3 @@ -import { createWriteStream } from 'node:fs'; import { nanoid } from 'nanoid'; import { config } from '../../../env'; import { chunkedStorage, fileRepository, uploadBatcher } from '../../../infrastructure/di'; @@ -15,6 +14,7 @@ import { getErrorMessage, getFileType, } from '../../../shared/utils/file'; +import { streamToTemp } from '../../../shared/utils/temp-stream'; /** * Maximum allowed size (in bytes) for a base64 JSON upload. @@ -93,7 +93,7 @@ const rejectOversizedRequest = (req: Request): Response | null => { * Streams a multipart `File` to a temporary file on disk while computing * its SHA-256 hash and extracting the signature (first 16 bytes). * - * Backpressure from the write stream is respected via the drain event. + * Delegates to the shared {@link streamToTemp} utility. * * @param file - The multipart `File` object. * @param maxSizeBytes - Maximum allowed file size; an error is thrown if @@ -102,67 +102,8 @@ const rejectOversizedRequest = (req: Request): Response | null => { * @throws {Error} When the file size exceeds `maxSizeBytes`. */ const streamFileToTemp = async (file: File, maxSizeBytes: number): Promise => { - const tempPath = `/tmp/filedrop-${nanoid()}`; - const writer = createWriteStream(tempPath); - const hasher = new Bun.CryptoHasher('sha256'); - const reader = file.stream().getReader(); - const signatureChunks: Buffer[] = []; - let signatureBytes = 0; - let sizeBytes = 0; - - const writeChunk = async (chunk: Buffer): Promise => { - if (!writer.write(chunk)) { - await new Promise((resolve, reject) => { - writer.once('drain', resolve); - writer.once('error', reject); - }); - } - }; - - const finishWriter = async (): Promise => { - await new Promise((resolve, reject) => { - writer.end(() => resolve()); - writer.once('error', reject); - }); - }; - - try { - while (true) { - const { done, value } = await reader.read(); - if (done) break; - - const chunk = Buffer.from(value); - sizeBytes += chunk.byteLength; - if (sizeBytes > maxSizeBytes) { - throw new Error('File size exceeds upload limit'); - } - - hasher.update(chunk); - await writeChunk(chunk); - - if (signatureBytes < SIGNATURE_BYTES) { - const remaining = SIGNATURE_BYTES - signatureBytes; - const signatureChunk = chunk.subarray(0, remaining); - signatureChunks.push(signatureChunk); - signatureBytes += signatureChunk.byteLength; - } - } - - await finishWriter(); - - return { - tempPath, - fileHash: hasher.digest('hex'), - sizeBytes, - signatureBuffer: Buffer.concat(signatureChunks, signatureBytes), - }; - } catch (error) { - writer.destroy(); - await cleanupTempFile(tempPath); - throw error; - } finally { - reader.releaseLock(); - } + const result = await streamToTemp(file.stream().getReader(), { maxSizeBytes }); + return result; }; /** diff --git a/src/interfaces/http/controllers/web-api-controller.ts b/src/interfaces/http/controllers/web-api-controller.ts index d928e18..7af744a 100644 --- a/src/interfaces/http/controllers/web-api-controller.ts +++ b/src/interfaces/http/controllers/web-api-controller.ts @@ -5,7 +5,13 @@ import { config } from '../../../env'; import { bucketRepository, chunkedStorage, fileRepository } from '../../../infrastructure/di'; import { botPool } from '../../../infrastructure/telegram/bot-pool'; import logger from '../../../shared/logger/index'; -import { cleanupTempFile, ensureExtension, getErrorMessage } from '../../../shared/utils/file'; +import { + cleanupTempFile, + DEFAULT_FILE_TYPE, + ensureExtension, + getErrorMessage, +} from '../../../shared/utils/file'; +import { streamToTemp } from '../../../shared/utils/temp-stream'; /** * Route parameters extracted from the URL path. @@ -163,67 +169,33 @@ export const handleUploadObjectV1 = async ( } const key = (formData.get('key') as string) || file.name; - const tempPath = `/tmp/filedrop-web-${nanoid()}`; - const writer = Bun.file(tempPath).writer(); - const reader = file.stream().getReader(); - const hasher = new Bun.CryptoHasher('sha256'); - const SIGNATURE_BYTES = 16; - const signatureChunks: Buffer[] = []; - let signatureBytes = 0; - let sizeBytes = 0; - - try { - while (true) { - const { done, value } = await reader.read(); - if (done) break; - const chunk = Buffer.from(value); - sizeBytes += chunk.byteLength; - hasher.update(chunk); - writer.write(chunk); - if (signatureBytes < SIGNATURE_BYTES) { - const remaining = SIGNATURE_BYTES - signatureBytes; - const sigChunk = chunk.subarray(0, remaining); - signatureChunks.push(sigChunk); - signatureBytes += sigChunk.byteLength; - } - } - writer.end(); - } catch (error) { - writer.end(); - await cleanupTempFile(tempPath); - throw error; - } finally { - reader.releaseLock(); - } - - const hash = hasher.digest('hex'); - const signatureBuffer = Buffer.concat(signatureChunks, signatureBytes); + const streamed = await streamToTemp(file.stream().getReader(), { prefix: '/tmp/filedrop-web-' }); const { fileName: finalFileName, mimeType } = ensureExtension( key.split('/').pop() || 'file', - signatureBuffer, + streamed.signatureBuffer, file.type || 'application/octet-stream', ); const partFileNamePrefix = `s3-${bucket.name}-${key.replace(/\//g, '_')}`; - if (sizeBytes > config.telegramChunkSizeBytes) { + if (streamed.sizeBytes > config.telegramChunkSizeBytes) { const uploadedFile = await chunkedStorage.storeFileInTelegramChunks({ - tempPath, + tempPath: streamed.tempPath, partFileNamePrefix, fileName: finalFileName, mimeType, - sizeBytes, - fileType: 'document', + sizeBytes: streamed.sizeBytes, + fileType: DEFAULT_FILE_TYPE, uploaderId: 0, bucketId: bucket.id, s3Key: key, }); - await cleanupTempFile(tempPath); + await cleanupTempFile(streamed.tempPath); return json( { key, - size: sizeBytes, - etag: hash, + size: streamed.sizeBytes, + etag: streamed.fileHash, downloadUrl: `${config.baseUrl}/f/${uploadedFile.publicId}`, }, 201, @@ -231,7 +203,7 @@ export const handleUploadObjectV1 = async ( } const forwardResult = await botPool.forwardToStorage( - createReadStream(tempPath), + createReadStream(streamed.tempPath), partFileNamePrefix, 'document', ); @@ -247,20 +219,25 @@ export const handleUploadObjectV1 = async ( storageMessageId: forwardResult.storageMessageId, fileName: finalFileName, mimeType, - sizeBytes, - fileType: 'document', + sizeBytes: streamed.sizeBytes, + fileType: DEFAULT_FILE_TYPE, uploaderId: 0, - fileHash: hash, + fileHash: streamed.fileHash, bucketId: bucket.id, s3Key: key, storageBackend: 'telegram', }), ); - await cleanupTempFile(tempPath); + await cleanupTempFile(streamed.tempPath); return json( - { key, size: sizeBytes, etag: hash, downloadUrl: `${config.baseUrl}/f/${publicId}` }, + { + key, + size: streamed.sizeBytes, + etag: streamed.fileHash, + downloadUrl: `${config.baseUrl}/f/${publicId}`, + }, 201, ); }; diff --git a/src/shared/utils/file.ts b/src/shared/utils/file.ts index 60f72da..636f6a4 100644 --- a/src/shared/utils/file.ts +++ b/src/shared/utils/file.ts @@ -51,6 +51,9 @@ interface FileMetadata { createdAt: Date | string | number; } +/** Default file type used when no specific type can be determined. */ +export const DEFAULT_FILE_TYPE = 'document'; + /** Per-file-type size limits in bytes. */ const FILE_TYPES: Record = { document: 2 * 1024 * 1024 * 1024, // 2GB diff --git a/src/shared/utils/temp-stream.ts b/src/shared/utils/temp-stream.ts new file mode 100644 index 0000000..2478af8 --- /dev/null +++ b/src/shared/utils/temp-stream.ts @@ -0,0 +1,132 @@ +/** + * Shared utility for streaming data into a temporary file on disk while + * computing its SHA-256 hash (and optionally MD5) and extracting the + * signature (first 16 bytes) for magic-byte detection. + * + * Consolidates the duplicated streaming-to-temp pattern found across + * multiple HTTP controllers (s3-controller, upload-controller, + * web-api-controller) into a single, reusable function. + */ + +import { unlink } from 'node:fs/promises'; +import { nanoid } from 'nanoid'; + +/** Options for the {@link streamToTemp} function. */ +export interface StreamToTempOptions { + /** Temporary file path prefix (default: `'/tmp/filedrop-'`). */ + prefix?: string; + /** When true, also compute the MD5 hash (default: false). */ + computeMd5?: boolean; + /** Maximum allowed bytes; throws if the stream exceeds this size. */ + maxSizeBytes?: number; +} + +/** Result of a successful {@link streamToTemp} call. */ +export interface StreamToTempResult { + /** Absolute path to the written temp file. */ + tempPath: string; + /** SHA-256 hex digest of the entire stream. */ + fileHash: string; + /** MD5 base-64 digest — only present when `computeMd5` was true. */ + md5Hash?: string; + /** Total number of bytes written. */ + sizeBytes: number; + /** First 16 bytes of the stream (padded with zeros if shorter). */ + signatureBuffer: Buffer; +} + +/** + * Streams data from a `ReadableStreamDefaultReader` into a temporary file + * while computing hashes and extracting the first 16 bytes as a signature + * buffer. + * + * The temp file is cleaned up automatically on error. + * + * @param reader - A reader obtained from a `ReadableStream`. + * @param options - Optional behaviour flags. + * @returns A promise resolving with the temp-file metadata. + * @throws {Error} If `maxSizeBytes` is exceeded. + */ +export const streamToTemp = async ( + reader: ReadableStreamDefaultReader, + options?: StreamToTempOptions, +): Promise => { + const prefix = options?.prefix ?? '/tmp/filedrop-'; + const computeMd5 = options?.computeMd5 ?? false; + const maxSizeBytes = options?.maxSizeBytes; + + const tempPath = `${prefix}${nanoid()}`; + const writer = Bun.file(tempPath).writer(); + const sha256 = new Bun.CryptoHasher('sha256'); + const md5 = computeMd5 ? new Bun.CryptoHasher('md5') : null; + + const SIGNATURE_BYTES = 16; + const signatureChunks: Buffer[] = []; + let signatureBytes = 0; + let sizeBytes = 0; + + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + const chunk = Buffer.from(value); + sizeBytes += chunk.byteLength; + + if (maxSizeBytes !== undefined && sizeBytes > maxSizeBytes) { + reader.cancel(); + throw new Error('File size exceeds upload limit'); + } + + sha256.update(chunk); + md5?.update(chunk); + writer.write(chunk); + + if (signatureBytes < SIGNATURE_BYTES) { + const remaining = SIGNATURE_BYTES - signatureBytes; + const sigChunk = chunk.subarray(0, remaining); + signatureChunks.push(sigChunk); + signatureBytes += sigChunk.byteLength; + } + } + + try { + writer.end(); + } catch { + // Writer may have already errored — ignore on success path + } + + const result: StreamToTempResult = { + tempPath, + fileHash: sha256.digest('hex'), + sizeBytes, + signatureBuffer: Buffer.concat(signatureChunks, signatureBytes), + }; + + if (md5) { + result.md5Hash = md5.digest('base64'); + } + + return result; + } catch (error) { + try { + writer.end(); + } catch { + // ignore writer end failure during error path + } + + try { + await unlink(tempPath); + } catch { + // ignore unlink failure + } + + throw error; + } finally { + try { + reader.releaseLock(); + } catch { + // ignore release lock failure + } + } +};