feat: dedup wave 3 — constant + streamToTemp utility
Deploy FileDrop / deploy (push) Successful in 44s

- ♻️ 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 
This commit is contained in:
Claude
2026-07-29 16:10:49 +07:00
parent 3501d547c0
commit fab91ad69c
7 changed files with 190 additions and 182 deletions
@@ -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,
+9 -5
View File
@@ -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,
@@ -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<Uint8Array>;
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,
@@ -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<PreparedUpload> => {
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<void> => {
if (!writer.write(chunk)) {
await new Promise<void>((resolve, reject) => {
writer.once('drain', resolve);
writer.once('error', reject);
});
}
};
const finishWriter = async (): Promise<void> => {
await new Promise<void>((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;
};
/**
@@ -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,
);
};
+3
View File
@@ -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<string, number> = {
document: 2 * 1024 * 1024 * 1024, // 2GB
+132
View File
@@ -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<Uint8Array>,
options?: StreamToTempOptions,
): Promise<StreamToTempResult> => {
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
}
}
};