diff --git a/src/application/use-cases/manage-bucket.ts b/src/application/use-cases/manage-bucket.ts new file mode 100644 index 0000000..a51e0b9 --- /dev/null +++ b/src/application/use-cases/manage-bucket.ts @@ -0,0 +1,168 @@ +import type { Bucket } from '../../domain/entities/bucket'; +import type { IBucketRepository } from '../../domain/ports/bucket-repository'; +import type { IFileRepository } from '../../domain/ports/file-repository'; + +/** + * S3 bucket name validation regex. + * + * Bucket names must be 3-63 characters, start/end with a lowercase letter or + * digit, and contain only lowercase letters, digits, dots, and hyphens. + */ +const BUCKET_NAME_REGEX = /^[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]$/; + +/** + * Error type for bucket-level application errors that carry an S3-compatible + * error code and an HTTP status suggestion. + */ +export class BucketError extends Error { + /** S3-compatible error code (e.g. "NoSuchBucket", "BucketAlreadyExists"). */ + readonly code: string; + /** Suggested HTTP status code. */ + readonly status: number; + + /** + * @param code - The S3 error code. + * @param message - Human-readable error description. + * @param status - Suggested HTTP status. + */ + constructor(code: string, message: string, status: number) { + super(message); + this.name = 'BucketError'; + this.code = code; + this.status = status; + } +} + +/** + * Describes a bucket entry returned by the list-buckets use case, + * enriched with the current count of non-deleted objects. + */ +export interface BucketWithCount { + /** The bucket domain entity. */ + bucket: Bucket; + /** Number of non-deleted objects in the bucket. */ + objectCount: number; +} + +/** Dependencies required by the manage-bucket use case factories. */ +export interface ManageBucketDeps { + /** Bucket repository for CRUD operations. */ + bucketRepo: IBucketRepository; + /** File repository for counting and checking objects within buckets. */ + fileRepo: IFileRepository; +} + +/** + * Creates a use case that lists all buckets together with their object counts. + * + * @param deps - The injected dependencies. + * @returns An async function that returns a list of buckets with counts. + */ +export function createListBucketsUseCase(deps: ManageBucketDeps) { + return async (): Promise => { + const buckets = await deps.bucketRepo.list(); + const results = await Promise.all( + buckets.map(async (bucket) => ({ + bucket, + objectCount: await deps.fileRepo.countByBucket(bucket.id), + })), + ); + return results; + }; +} + +/** + * Creates a use case that retrieves a single bucket by its name. + * + * @param deps - The injected dependencies. + * @returns An async function accepting a bucket name and returning the + * bucket, or `null` when not found. + */ +export function createGetBucketUseCase(deps: ManageBucketDeps) { + return async (name: string): Promise => { + return deps.bucketRepo.findByName(name); + }; +} + +/** + * Creates a use case that creates a new bucket. + * + * Validates the bucket name format and checks for duplicates before + * persisting. + * + * @param deps - The injected dependencies. + * @returns An async function accepting a bucket name and returning the + * newly created bucket. + * @throws {BucketError} When the name is invalid or the bucket already exists. + */ +export function createCreateBucketUseCase(deps: ManageBucketDeps) { + return async (name: string): Promise => { + if (!BUCKET_NAME_REGEX.test(name)) { + throw new BucketError( + 'InvalidBucketName', + 'The specified bucket is not valid.', + 400, + ); + } + + const existing = await deps.bucketRepo.findByName(name); + if (existing) { + throw new BucketError( + 'BucketAlreadyExists', + 'The requested bucket name is not available.', + 409, + ); + } + + return deps.bucketRepo.create(name); + }; +} + +/** + * Creates a use case that deletes a bucket. + * + * Ensures the bucket exists and is empty (no non-deleted objects) before + * proceeding with deletion. + * + * @param deps - The injected dependencies. + * @returns An async function accepting a bucket name. Returns `true` when + * the bucket was deleted, throws when the bucket is missing or + * not empty. + * @throws {BucketError} When the bucket does not exist or is not empty. + */ +export function createDeleteBucketUseCase(deps: ManageBucketDeps) { + return async (name: string): Promise => { + const bucket = await deps.bucketRepo.findByName(name); + if (!bucket) { + throw new BucketError( + 'NoSuchBucket', + 'The specified bucket does not exist.', + 404, + ); + } + + const objectCount = await deps.fileRepo.countByBucket(bucket.id); + if (objectCount > 0) { + throw new BucketError( + 'BucketNotEmpty', + 'The bucket you tried to delete is not empty.', + 409, + ); + } + + return deps.bucketRepo.delete(name); + }; +} + +/** + * Creates a use case that checks whether a bucket exists. + * + * @param deps - The injected dependencies. + * @returns An async function accepting a bucket name and returning `true` + * when the bucket exists. + */ +export function createBucketExistsUseCase(deps: ManageBucketDeps) { + return async (name: string): Promise => { + return deps.bucketRepo.exists(name); + }; +} \ No newline at end of file diff --git a/src/application/use-cases/multipart-upload.ts b/src/application/use-cases/multipart-upload.ts new file mode 100644 index 0000000..4bf8148 --- /dev/null +++ b/src/application/use-cases/multipart-upload.ts @@ -0,0 +1,378 @@ +import { nanoid } from 'nanoid'; +import type { MultipartUpload } from '../../domain/entities/multipart'; +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'; + +// ─── Types ────────────────────────────────────────────────────────── + +/** + * A single part reference as submitted in a complete-multipart-upload request. + */ +export interface CompletePartInput { + /** 1-based part number. */ + partNumber: number; + /** ETag returned when the part was uploaded. */ + etag: string; +} + +/** + * Result of initiating a multipart upload. + */ +export interface InitiateMultipartResult { + /** The generated upload identifier (nanoid). */ + uploadId: string; + /** The bucket name. */ + bucket: string; + /** The S3 object key. */ + key: string; +} + +/** + * Result of uploading a single part. + */ +export interface UploadPartResult { + /** ETag of the uploaded part (SHA-256 hex digest). */ + etag: string; +} + +/** + * Result of completing a multipart upload. + */ +export interface CompleteMultipartResult { + /** Public-facing unique identifier of the created file record. */ + publicId: string; + /** The S3 location URL of the completed object. */ + location: string; + /** Combined ETag (all part etags joined by hyphens). */ + etag: string; + /** Total object size in bytes. */ + sizeBytes: number; +} + +/** + * Summary of a single part within a multipart upload used in listing results. + */ +export interface PartSummary { + /** 1-based part number. */ + partNumber: number; + /** ETag of the part content. */ + etag: string; + /** Part size in bytes. */ + sizeBytes: number; + /** ISO-8601 timestamp when the part was stored. */ + createdAt: Date; +} + +/** + * Result of listing multipart uploads within a bucket. + */ +export interface ListMultipartUploadsResult { + /** Array of in-progress upload summaries. */ + uploads: MultipartUpload[]; + /** Whether more results are available. */ + isTruncated: boolean; + /** Marker for the next page, or null when not truncated. */ + nextKeyMarker: string | null; +} + +// ─── Config ───────────────────────────────────────────────────────── + +/** Subset of application configuration consumed by the multipart use cases. */ +export interface MultipartConfig { + /** Maximum chunk size in bytes for Telegram uploads (part size limit). */ + telegramChunkSizeBytes: number; + /** Telegram chat ID where part data is stored. */ + storageChatId: number; + /** Server base URL for constructing location URLs. */ + baseUrl: string; +} + +/** Dependencies required by the multipart upload use case factories. */ +export interface MultipartDeps { + /** Bucket repository for bucket lookups. */ + bucketRepo: IBucketRepository; + /** File repository for creating the final file record on completion. */ + fileRepo: IFileRepository; + /** Multipart repository for managing upload sessions and parts. */ + multipartRepo: IMultipartRepository; + /** Telegram service for forwarding part data to storage. */ + telegramService: ITelegramService; + /** Application configuration subset. */ + config: MultipartConfig; +} + +// ─── Use Case Factories ───────────────────────────────────────────── + +/** + * Creates a use case that initiates an S3 multipart upload. + * + * Validates the bucket exists and creates a new multipart upload session. + * + * @param deps - The injected dependencies. + * @returns An async function accepting bucket name and object key, returning + * the upload initiation result, or `null` when the bucket is not found. + */ +export function createInitiateMultipartUploadUseCase(deps: MultipartDeps) { + return async ( + bucketName: string, + key: string, + ): Promise => { + const bucket = await deps.bucketRepo.findByName(bucketName); + if (!bucket) return null; + + const uploadId = await deps.multipartRepo.create(bucket.id, key, 's3'); + + return { uploadId, bucket: bucketName, key }; + }; +} + +/** + * Creates a use case that uploads a single part of a multipart upload. + * + * Validates the part number range (1-10000), checks the upload session exists + * and matches the expected key, checks part size against the configured limit, + * forwards the part data to Telegram storage, and persists the part record. + * + * @param deps - The injected dependencies. + * @returns An async function accepting upload details and part data, returning + * the part ETag, or `null` when the upload session is not found. + */ +export function createUploadPartUseCase(deps: MultipartDeps) { + return async (input: { + /** Bucket name for the multipart upload. */ + bucketName: string; + /** S3 object key for the multipart upload. */ + key: string; + /** Upload identifier returned by initiate. */ + uploadId: string; + /** 1-based part number (1-10000). */ + partNumber: number; + /** Raw part data. */ + body: Buffer; + }): Promise => { + if (input.partNumber < 1 || input.partNumber > 10000) { + throw new MultipartError( + 'InvalidArgument', + 'Part number must be an integer between 1 and 10000', + 400, + ); + } + + const multipart = await deps.multipartRepo.findById(input.uploadId); + if (!multipart || multipart.s3Key !== input.key) { + return null; + } + + if (input.body.byteLength > deps.config.telegramChunkSizeBytes) { + throw new MultipartError( + 'EntityTooLarge', + `Your proposed upload part size (${input.body.byteLength} bytes) exceeds the maximum allowed part size (${deps.config.telegramChunkSizeBytes} bytes) for this storage backend. Use smaller part sizes.`, + 400, + ); + } + + const forwardResult = await deps.telegramService.forwardToStorage( + input.body, + `mp-${input.uploadId}-part-${input.partNumber}`, + 'document', + ); + + const etag = computeHash(input.body); + + await deps.multipartRepo.insertPart({ + uploadId: input.uploadId, + partNumber: input.partNumber, + telegramFileId: forwardResult.telegramFileId, + telegramFileUniqueId: forwardResult.telegramFileUniqueId, + storageMessageId: forwardResult.storageMessageId, + sizeBytes: input.body.byteLength, + etag, + }); + + return { etag: `"${etag}"` }; + }; +} + +/** + * Error type for multipart-level application errors. + */ +export class MultipartError extends Error { + /** S3-compatible error code. */ + readonly code: string; + /** Suggested HTTP status code. */ + readonly status: number; + + /** + * @param code - The S3 error code. + * @param message - Human-readable error description. + * @param status - Suggested HTTP status. + */ + constructor(code: string, message: string, status: number) { + super(message); + this.name = 'MultipartError'; + this.code = code; + this.status = status; + } +} + +/** + * Creates a use case that completes an S3 multipart upload. + * + * Validates the submitted part list (all parts must be present and in ascending + * order), creates the final file record referencing the first part's Telegram + * data, marks the upload session as completed, and returns the combined result. + * + * @param deps - The injected dependencies. + * @returns An async function accepting upload details and submitted parts, + * returning the completion result, or `null` when the upload session + * is not found. + */ +export function createCompleteMultipartUploadUseCase(deps: MultipartDeps) { + return async (input: { + /** Bucket name for the multipart upload. */ + bucketName: string; + /** S3 object key for the multipart upload. */ + key: string; + /** Upload identifier. */ + uploadId: string; + /** Parts submitted by the client (in ascending part number order). */ + parts: CompletePartInput[]; + }): Promise => { + const multipart = await deps.multipartRepo.findById(input.uploadId); + if (!multipart) return null; + + const storedParts = await deps.multipartRepo.listParts(input.uploadId); + + // Validate ascending part order + const partNumbers = input.parts.map((p) => p.partNumber); + if (partNumbers.length > 1 && partNumbers.some((n, i) => i > 0 && n <= partNumbers[i - 1])) { + throw new MultipartError( + 'InvalidPartOrder', + 'The list of parts was not in ascending order.', + 400, + ); + } + + // Validate part count matches + if (input.parts.length !== storedParts.length) { + throw new MultipartError( + 'InvalidPart', + 'One or more specified parts could not be found.', + 400, + ); + } + + const totalSize = storedParts.reduce((sum, p) => sum + p.sizeBytes, 0); + const firstPart = storedParts[0]; + if (!firstPart) { + throw new MultipartError('InternalError', 'Multipart object has no parts.', 500); + } + + const publicId = nanoid(); + + await deps.fileRepo.create({ + publicId, + telegramFileId: firstPart.telegramFileId, + telegramFileUniqueId: firstPart.telegramFileUniqueId, + storageChatId: deps.config.storageChatId, + storageMessageId: firstPart.storageMessageId, + fileName: input.key.split('/').pop() || 'file', + mimeType: 'application/octet-stream', + sizeBytes: totalSize, + fileType: 'document', + uploaderId: 0, + fileHash: null, + archiveTelegramFileId: null, + archiveStorageMessageId: null, + archiveFileName: null, + archiveEntryName: null, + archiveMimeType: null, + archiveSizeBytes: null, + bucketId: multipart.bucketId, + s3Key: input.key, + storageBackend: 'telegram', + isDeleted: false, + multipartUploadId: input.uploadId, + partCount: null, + }); + + await deps.multipartRepo.complete(input.uploadId); + + const location = `${deps.config.baseUrl}/${input.bucketName}/${input.key}`; + const combinedEtag = storedParts.map((p) => p.etag).join('-'); + + return { + publicId, + location, + etag: combinedEtag, + sizeBytes: totalSize, + }; + }; +} + +/** + * Creates a use case that aborts an S3 multipart upload. + * + * @param deps - The injected dependencies. + * @returns An async function accepting an upload identifier, returning `true` + * when the upload was aborted, or `null` when the upload session + * is not found. + */ +export function createAbortMultipartUploadUseCase(deps: MultipartDeps) { + return async (uploadId: string): Promise => { + const multipart = await deps.multipartRepo.findById(uploadId); + if (!multipart) return null; + + await deps.multipartRepo.abort(uploadId); + return true; + }; +} + +/** + * Creates a use case that lists in-progress multipart uploads within a bucket. + * + * @param deps - The injected dependencies. + * @returns An async function accepting query parameters and returning the + * listing result, or `null` when the bucket is not found. + */ +export function createListMultipartUploadsUseCase(deps: MultipartDeps) { + return async (input: { + /** Bucket name to list uploads from. */ + bucketName: string; + /** Maximum number of uploads to return (clamped 1-1000). */ + maxUploads: number; + /** Return only uploads whose S3 key is strictly greater than this, or null. */ + keyMarker: string | null; + }): Promise => { + const bucket = await deps.bucketRepo.findByName(input.bucketName); + if (!bucket) return null; + + return deps.multipartRepo.listByBucket(bucket.id, input.maxUploads, input.keyMarker); + }; +} + +/** + * Creates a use case that lists parts of a specific multipart upload. + * + * @param deps - The injected dependencies. + * @returns An async function accepting an upload identifier and returning the + * list of parts, or `null` when the upload session is not found. + */ +export function createListPartsUseCase(deps: MultipartDeps) { + return async (uploadId: string): Promise => { + const multipart = await deps.multipartRepo.findById(uploadId); + if (!multipart) return null; + + const parts = await deps.multipartRepo.listParts(uploadId); + + return parts.map((p) => ({ + partNumber: p.partNumber, + etag: p.etag, + sizeBytes: p.sizeBytes, + createdAt: p.createdAt, + })); + }; +} \ No newline at end of file diff --git a/src/application/use-cases/s3-object.ts b/src/application/use-cases/s3-object.ts new file mode 100644 index 0000000..d125404 --- /dev/null +++ b/src/application/use-cases/s3-object.ts @@ -0,0 +1,774 @@ +import { randomUUID } from 'node:crypto'; +import { gzipSync } from 'node:zlib'; +import { nanoid } from 'nanoid'; +import type { File, NewFile } from '../../domain/entities/file'; +import type { NewFilePart } from '../../domain/entities/file-part'; +import type { MultipartPart } from '../../domain/entities/multipart'; +import type { IBucketRepository } from '../../domain/ports/bucket-repository'; +import type { IFilePartRepository } from '../../domain/ports/file-part-repository'; +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 { ensureExtension, computeHash, formatCreatedAt } from '../../shared/utils/file'; + +// ─── Types ────────────────────────────────────────────────────────── + +/** + * Compression algorithm used for chunked object storage. + */ +type CompressionAlgorithm = 'gzip' | null; + +/** + * A single part source for building a multi-part streaming response. + * Each part corresponds to a Telegram-stored file chunk. + */ +export interface ObjectPartSource { + /** Telegram file identifier for retrieving this part. */ + telegramFileId: string; + /** Telegram CDN URL for downloading this part. */ + telegramUrl: string; + /** Original size of this part in bytes. */ + sizeBytes: number; + /** 1-based part number within the object. */ + partNumber: number; + /** Stored (post-compression) size in bytes, when applicable. */ + storedSizeBytes?: number; + /** Compression algorithm applied, or null if uncompressed. */ + compressionAlgorithm?: CompressionAlgorithm; +} + +/** + * A regular (direct) S3 object resolved to a Telegram CDN URL. + */ +export interface DirectObjectResult { + /** Discriminant. */ + type: 'direct'; + /** The resolved file entity. */ + file: File; + /** Full Telegram CDN URL for downloading the object. */ + telegramUrl: string; + /** Telegram file metadata. */ + fileInfo: TelegramFileInfo; +} + +/** + * A chunked S3 object stored across multiple Telegram file parts. + */ +export interface ChunkedObjectResult { + /** Discriminant. */ + type: 'chunked'; + /** The resolved file entity. */ + file: File; +} + +/** + * An S3 object assembled from a completed multipart upload. + */ +export interface MultipartObjectResult { + /** Discriminant. */ + type: 'multipart'; + /** The resolved file entity. */ + file: File; + /** Resolved part sources with Telegram CDN URLs. */ + parts: ObjectPartSource[]; +} + +/** + * Discriminated union of all possible S3 get-object outcomes. + */ +export type GetObjectResult = DirectObjectResult | ChunkedObjectResult | MultipartObjectResult; + +/** + * Result of an S3 put-object operation. + */ +export interface PutObjectResult { + /** SHA-256 hex digest of the object content. */ + etag: string; +} + +/** + * Result of an S3 copy-object operation. + */ +export interface CopyObjectResult { + /** SHA-256 hex digest of the source object content. */ + etag: string; + /** ISO-8601 timestamp of the copy operation. */ + lastModified: string; +} + +/** + * A single S3 object as returned in listing results. + */ +export interface ListObjectEntry { + /** The object key (full path within the bucket). */ + key: string; + /** Object size in bytes. */ + sizeBytes: number; + /** SHA-256 hex digest or fallback identifier. */ + etag: string; + /** ISO-8601 timestamp of last modification. */ + lastModified: string; + /** MIME type of the stored object. */ + mimeType: string; +} + +/** + * Result of an S3 list-objects operation (both V1 and V2). + */ +export interface ListObjectsResult { + /** Array of object summaries. */ + objects: ListObjectEntry[]; + /** Common prefixes when a delimiter was used. */ + prefixes: string[]; + /** Whether more results are available. */ + isTruncated: boolean; + /** The last key in the returned page, for use as the next marker. */ + nextMarker: string | null; +} + +/** + * Head-object metadata. + */ +export interface HeadObjectMetadata { + /** MIME type of the object. */ + contentType: string; + /** Object size in bytes. */ + contentLength: number; + /** Entity tag (SHA-256 hex digest or fallback). */ + etag: string; + /** ISO-8601 timestamp of last modification. */ + lastModified: string; +} + +// ─── Config ───────────────────────────────────────────────────────── + +/** Subset of application configuration consumed by the s3-object use cases. */ +export interface S3ObjectConfig { + /** Maximum chunk size in bytes for Telegram chunked uploads. */ + telegramChunkSizeBytes: number; + /** Whether gzip compression is enabled for chunked uploads. */ + compressChunkedUploads: boolean; + /** Minimum chunk size in bytes below which compression is skipped. */ + chunkCompressionMinSizeBytes: number; + /** Telegram chat ID where file parts are stored. */ + storageChatId: number; + /** Server base URL for constructing download links. */ + baseUrl: string; + /** Whether to proxy S3 GET requests through the server. */ + proxyS3Get: boolean; +} + +/** Dependencies required by the s3-object use case factories. */ +export interface S3ObjectDeps { + /** Bucket repository for bucket lookups. */ + bucketRepo: IBucketRepository; + /** File repository for object CRUD operations. */ + fileRepo: IFileRepository; + /** File-part repository for chunked upload part records. */ + filePartRepo: IFilePartRepository; + /** Multipart repository for resolving multipart-upload objects. */ + multipartRepo: IMultipartRepository; + /** Telegram service for uploading and resolving file metadata. */ + telegramService: ITelegramService; + /** Application configuration subset. */ + config: S3ObjectConfig; +} + +// ─── Helpers ──────────────────────────────────────────────────────── + +/** + * Validates the configured chunk size and returns it as a safe integer. + * + * @param chunkSizeBytes - The configured chunk size in bytes. + * @returns The same value if it is a positive safe integer. + */ +const asSafeChunkSize = (chunkSizeBytes: number): number => { + if (!Number.isSafeInteger(chunkSizeBytes) || chunkSizeBytes <= 0) { + throw new Error('Invalid Telegram chunk size'); + } + return chunkSizeBytes; +}; + +/** + * Optionally gzip-compresses a chunk if compression is enabled and the chunk + * is large enough to benefit from it. + * + * @param chunk - The raw chunk buffer. + * @param compress - Whether compression is enabled. + * @param compressionMinSizeBytes - Minimum chunk size to attempt compression. + * @returns The (possibly compressed) buffer and the algorithm used. + */ +const maybeCompressChunk = ( + chunk: Buffer, + compress: boolean, + compressionMinSizeBytes: number, +): { bytes: Buffer; compressionAlgorithm: CompressionAlgorithm } => { + 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' }; +}; + +/** + * Metadata for a single uploaded chunk/part during S3 put-object. + */ +interface UploadedPart { + /** 1-based part number. */ + partNumber: number; + /** Telegram file identifier for this part. */ + telegramFileId: string; + /** Telegram unique file identifier (stable across bot tokens). */ + telegramFileUniqueId: string; + /** Message ID within the storage chat. */ + storageMessageId: number; + /** Original size of the chunk in bytes before compression. */ + sizeBytes: number; + /** Stored (post-compression) size in bytes. */ + storedSizeBytes: number; + /** Compression algorithm applied, or null if uncompressed. */ + compressionAlgorithm: CompressionAlgorithm; + /** SHA-256 hash of the original chunk content. */ + etag: string; +} + +/** Result of uploading an object in multiple Telegram chunks. */ +interface ChunkedUploadResult { + /** Metadata for each uploaded part. */ + parts: UploadedPart[]; + /** SHA-256 hex digest of the complete object content. */ + fileHash: string; + /** Total object size in bytes (sum of all original chunks). */ + totalSizeBytes: number; +} + +/** + * Uploads a buffer to Telegram in chunks, returning metadata for all parts. + * + * @param buffer - The full object buffer. + * @param partFileNamePrefix - Prefix used for each chunk's file name in Telegram. + * @param chunkSizeBytes - Maximum size of each chunk in bytes. + * @param compress - Whether gzip compression is enabled. + * @param compressionMinSizeBytes - Minimum chunk size to attempt compression. + * @param telegramService - The Telegram service to forward each chunk. + * @returns The aggregated chunked upload result. + */ +const uploadInChunks = async ( + buffer: Buffer, + partFileNamePrefix: string, + chunkSizeBytes: number, + compress: boolean, + compressionMinSizeBytes: number, + telegramService: ITelegramService, +): Promise => { + const safeChunkSize = asSafeChunkSize(chunkSizeBytes); + const hasher = new Bun.CryptoHasher('sha256'); + const parts: UploadedPart[] = []; + let totalSizeBytes = 0; + let partNumber = 0; + let offset = 0; + + while (offset < buffer.byteLength) { + const chunk = buffer.subarray(offset, offset + safeChunkSize); + if (chunk.byteLength === 0) break; + + partNumber += 1; + totalSizeBytes += chunk.byteLength; + hasher.update(chunk); + + const { bytes, compressionAlgorithm } = maybeCompressChunk( + chunk, + compress, + compressionMinSizeBytes, + ); + + const forwardResult = await telegramService.forwardToStorage( + bytes, + `${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), + }); + + offset += safeChunkSize; + } + + return { + parts, + fileHash: hasher.digest('hex'), + totalSizeBytes, + }; +}; + +/** + * Resolves a list of multipart parts to their Telegram CDN URLs. + * + * @param parts - The stored multipart parts. + * @param telegramService - The Telegram service for resolving file metadata. + * @returns An array of resolved part sources. + */ +const resolveMultipartParts = async ( + parts: MultipartPart[], + telegramService: ITelegramService, +): Promise => { + const sources: ObjectPartSource[] = []; + for (const part of parts) { + const fileInfo = await 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, + partNumber: part.partNumber, + }); + } + return sources; +}; + +/** + * Formats a `createdAt` value into an HTTP Last-Modified header value. + * + * @param date - The date to format. + * @returns The UTC string representation. + */ +const formatLastModified = (date: Date | string | number): string => { + return date instanceof Date ? date.toUTCString() : new Date(date).toUTCString(); +}; + +// ─── Use Case Factories ───────────────────────────────────────────── + +/** + * Creates a use case that resolves an S3 object for retrieval (GET). + * + * Looks up the bucket and file by key, then determines the storage type: + * - **direct**: regular Telegram-stored object — resolves the Telegram CDN URL. + * - **chunked**: object stored across multiple Telegram file parts. + * - **multipart**: object assembled from a completed multipart upload — resolves + * the Telegram CDN URLs for each part. + * + * @param deps - The injected dependencies. + * @returns An async function accepting bucket name and object key, returning + * a discriminated union of possible results, or `null` when the + * bucket or file is not found. + */ +export function createGetObjectUseCase(deps: S3ObjectDeps) { + return async (bucketName: string, key: string): Promise => { + const bucket = await deps.bucketRepo.findByName(bucketName); + if (!bucket) return null; + + const file = await deps.fileRepo.findByBucketAndKey(bucket.id, key); + if (!file) return null; + + // Chunked storage — return the entity; the caller resolves parts via + // chunked-storage helpers. + if (file.storageBackend === 'chunked') { + return { type: 'chunked', file }; + } + + // Multipart upload object — resolve part Telegram URLs + if (file.multipartUploadId) { + const parts = await deps.multipartRepo.listParts(file.multipartUploadId); + const resolvedParts = await resolveMultipartParts(parts, deps.telegramService); + return { type: 'multipart', file, parts: resolvedParts }; + } + + // Regular direct object — resolve Telegram CDN URL + const fileInfo = await deps.telegramService.getFileInfo(file.telegramFileId); + const telegramUrl = `https://api.telegram.org/file/bot${fileInfo.bot_token}/${fileInfo.file_path}`; + + return { type: 'direct', file, telegramUrl, fileInfo }; + }; +} + +/** + * Creates a use case that retrieves S3 object metadata (HEAD). + * + * @param deps - The injected dependencies. + * @returns An async function accepting bucket name and object key, returning + * metadata or `null` when the bucket or file is not found. + */ +export function createHeadObjectUseCase(deps: S3ObjectDeps) { + return async (bucketName: string, key: string): Promise => { + const bucket = await deps.bucketRepo.findByName(bucketName); + if (!bucket) return null; + + const file = await deps.fileRepo.findByBucketAndKey(bucket.id, key); + if (!file) return null; + + return { + contentType: file.mimeType, + contentLength: file.sizeBytes, + etag: file.fileHash || nanoid(16), + lastModified: formatLastModified(file.createdAt), + }; + }; +} + +/** + * Creates a use case that stores an S3 object (PUT). + * + * Handles both chunked (large) and single-message (small) upload paths, + * deduplicates by bucket+key (idempotent PUT), and persists the file + * record and (for chunked storage) part records. + * + * @param deps - The injected dependencies. + * @returns An async function accepting bucket name, key, body buffer, and + * content type, returning the etag of the stored object. Returns + * `null` when the bucket is not found. + */ +export function createPutObjectUseCase(deps: S3ObjectDeps) { + return async ( + bucketName: string, + key: string, + body: Buffer, + contentType: string, + ): Promise => { + const bucket = await deps.bucketRepo.findByName(bucketName); + if (!bucket) return null; + + const hash = computeHash(body); + + // Idempotent PUT: if the object already exists, skip upload + const existing = await deps.fileRepo.findByBucketAndKey(bucket.id, key); + if (existing) { + return { etag: `"${hash}"` }; + } + + const fileName = key.split('/').pop() || 'file'; + const signatureBuffer = body.subarray(0, 16); + const { fileName: finalFileName, mimeType } = ensureExtension( + fileName, + signatureBuffer, + contentType, + ); + + const partFileNamePrefix = `s3-${bucket.name}-${key.replace(/\//g, '_')}`; + const { telegramChunkSizeBytes, compressChunkedUploads, chunkCompressionMinSizeBytes, storageChatId } = deps.config; + + if (body.byteLength > telegramChunkSizeBytes) { + // Chunked upload path + const chunkResult = await uploadInChunks( + body, + partFileNamePrefix, + telegramChunkSizeBytes, + compressChunkedUploads, + chunkCompressionMinSizeBytes, + deps.telegramService, + ); + + const firstPart = chunkResult.parts[0]; + if (!firstPart) { + throw new Error('Chunked upload produced no parts'); + } + + const fileId = randomUUID(); + const publicId = nanoid(); + + await deps.fileRepo.create({ + publicId, + telegramFileId: firstPart.telegramFileId, + telegramFileUniqueId: firstPart.telegramFileUniqueId, + storageChatId, + storageMessageId: firstPart.storageMessageId, + fileName: finalFileName, + mimeType, + sizeBytes: chunkResult.totalSizeBytes, + fileType: 'document', + uploaderId: 0, + fileHash: chunkResult.fileHash, + archiveTelegramFileId: null, + archiveStorageMessageId: null, + archiveFileName: null, + archiveEntryName: null, + archiveMimeType: null, + archiveSizeBytes: null, + bucketId: bucket.id, + s3Key: key, + storageBackend: 'chunked', + isDeleted: false, + multipartUploadId: null, + partCount: chunkResult.parts.length, + }); + + const fileParts: NewFilePart[] = chunkResult.parts.map((part) => ({ + fileId, + partNumber: part.partNumber, + telegramFileId: part.telegramFileId, + telegramFileUniqueId: part.telegramFileUniqueId, + storageChatId, + storageMessageId: part.storageMessageId, + sizeBytes: part.sizeBytes, + storedSizeBytes: part.storedSizeBytes, + compressionAlgorithm: part.compressionAlgorithm, + etag: part.etag, + })); + + await deps.filePartRepo.insert(fileParts); + + return { etag: `"${chunkResult.fileHash}"` }; + } + + // Single-message upload path + const forwardResult = await deps.telegramService.forwardToStorage( + body, + partFileNamePrefix, + 'document', + ); + + const publicId = nanoid(); + + await deps.fileRepo.create({ + publicId, + telegramFileId: forwardResult.telegramFileId, + telegramFileUniqueId: forwardResult.telegramFileUniqueId, + storageChatId, + storageMessageId: forwardResult.storageMessageId, + fileName: finalFileName, + mimeType, + sizeBytes: body.byteLength, + fileType: 'document', + uploaderId: 0, + fileHash: hash, + archiveTelegramFileId: null, + archiveStorageMessageId: null, + archiveFileName: null, + archiveEntryName: null, + archiveMimeType: null, + archiveSizeBytes: null, + bucketId: bucket.id, + s3Key: key, + storageBackend: 'telegram', + isDeleted: false, + multipartUploadId: null, + partCount: null, + }); + + return { etag: `"${hash}"` }; + }; +} + +/** + * Creates a use case that copies an S3 object to a new key (PUT with + * x-amz-copy-source). + * + * Creates a new file record referencing the same Telegram-stored data. + * Chunked source objects are not supported for copy. + * + * @param deps - The injected dependencies. + * @returns An async function accepting source + destination identifiers and + * optional precondition headers, returning the copy result or + * `null` when a required bucket or file is not found. + */ +export function createCopyObjectUseCase(deps: S3ObjectDeps) { + return async (input: { + /** Source bucket name. */ + sourceBucket: string; + /** Source object key. */ + sourceKey: string; + /** Destination bucket UUID (must already exist). */ + destBucketId: string; + /** Destination object key. */ + destKey: string; + /** Optional if-match precondition (raw etag value, without surrounding quotes). */ + ifMatch?: string | null; + /** Optional if-none-match precondition (raw etag value, without surrounding quotes). */ + ifNoneMatch?: string | null; + }): Promise => { + const sourceBucket = await deps.bucketRepo.findByName(input.sourceBucket); + if (!sourceBucket) return null; + + const sourceFile = await deps.fileRepo.findByBucketAndKey(sourceBucket.id, input.sourceKey); + if (!sourceFile) return null; + + if (sourceFile.storageBackend === 'chunked') { + throw new ObjectError('NotImplemented', 'Copying chunked objects is not yet implemented.', 501); + } + + // Conditional copy: if-match / if-none-match checks + const sourceEtag = sourceFile.fileHash; + if (input.ifMatch && sourceEtag && input.ifMatch !== sourceEtag) { + throw new ObjectError('PreconditionFailed', 'The preconditions you specified did not hold.', 412); + } + if (input.ifNoneMatch && sourceEtag && input.ifNoneMatch === sourceEtag) { + throw new ObjectError('PreconditionFailed', 'The preconditions you specified did not hold.', 412); + } + + const publicId = nanoid(); + + await deps.fileRepo.create({ + publicId, + telegramFileId: sourceFile.telegramFileId, + telegramFileUniqueId: sourceFile.telegramFileUniqueId, + storageChatId: sourceFile.storageChatId, + storageMessageId: sourceFile.storageMessageId, + fileName: sourceFile.fileName, + mimeType: sourceFile.mimeType, + sizeBytes: sourceFile.sizeBytes, + fileType: sourceFile.fileType, + uploaderId: 0, + fileHash: sourceFile.fileHash, + archiveTelegramFileId: sourceFile.archiveTelegramFileId, + archiveStorageMessageId: sourceFile.archiveStorageMessageId, + archiveFileName: sourceFile.archiveFileName, + archiveEntryName: sourceFile.archiveEntryName, + archiveMimeType: sourceFile.archiveMimeType, + archiveSizeBytes: sourceFile.archiveSizeBytes, + bucketId: input.destBucketId, + s3Key: input.destKey, + storageBackend: 'telegram', + isDeleted: false, + multipartUploadId: null, + partCount: null, + }); + + return { + etag: sourceEtag || nanoid(16), + lastModified: new Date().toISOString(), + }; + }; +} + +/** + * Error type for S3 object-level application errors. + */ +export class ObjectError extends Error { + /** S3-compatible error code. */ + readonly code: string; + /** Suggested HTTP status code. */ + readonly status: number; + + /** + * @param code - The S3 error code. + * @param message - Human-readable error description. + * @param status - Suggested HTTP status. + */ + constructor(code: string, message: string, status: number) { + super(message); + this.name = 'ObjectError'; + this.code = code; + this.status = status; + } +} + +/** + * Creates a use case that soft-deletes an S3 object (DELETE). + * + * @param deps - The injected dependencies. + * @returns An async function accepting bucket name and object key, returning + * `true` if a row was soft-deleted. Returns `null` when the bucket + * is not found. + */ +export function createDeleteObjectUseCase(deps: S3ObjectDeps) { + return async (bucketName: string, key: string): Promise => { + const bucket = await deps.bucketRepo.findByName(bucketName); + if (!bucket) return null; + return deps.fileRepo.softDelete(bucket.id, key); + }; +} + +/** + * Creates a use case that batch-deletes multiple S3 objects (POST with + * ?delete). + * + * @param deps - The injected dependencies. + * @returns An async function accepting bucket name and an array of keys, + * returning the array of keys that were actually deleted. Returns + * `null` when the bucket is not found. + */ +export function createDeleteObjectsUseCase(deps: S3ObjectDeps) { + return async (bucketName: string, keys: string[]): Promise => { + const bucket = await deps.bucketRepo.findByName(bucketName); + if (!bucket) return null; + + const deletedKeys: string[] = []; + for (const key of keys) { + const ok = await deps.fileRepo.softDelete(bucket.id, key); + if (ok) deletedKeys.push(key); + } + return deletedKeys; + }; +} + +/** + * Creates a use case that lists objects within a bucket (ListObjectsV1/V2). + * + * Supports prefix filtering, delimiter-based pseudo-directory grouping, and + * pagination via marker/startAfter. + * + * @param deps - The injected dependencies. + * @returns An async function accepting query parameters and returning the + * listing result, or `null` when the bucket is not found. + */ +export function createListObjectsUseCase(deps: S3ObjectDeps) { + return async (input: { + /** Bucket name to list from. */ + bucketName: string; + /** Key prefix to filter by (empty string for no filter). */ + prefix: string; + /** Delimiter character (e.g. "/") or null for flat listing. */ + delimiter: string | null; + /** Maximum number of object records to return (clamped to 1000). */ + maxKeys: number; + /** Return only keys strictly greater than this value, or null. */ + startAfter: string | null; + }): Promise => { + const bucket = await deps.bucketRepo.findByName(input.bucketName); + if (!bucket) return null; + + const clampedMaxKeys = Math.min(input.maxKeys, 1000); + + const { objects, prefixes } = await deps.fileRepo.listByPrefix( + bucket.id, + input.prefix, + input.delimiter, + clampedMaxKeys, + input.startAfter, + ); + + const isTruncated = objects.length > clampedMaxKeys; + const displayObjects = objects.slice(0, clampedMaxKeys); + const nextMarker = isTruncated + ? (displayObjects[displayObjects.length - 1]?.s3Key ?? null) + : null; + + return { + objects: displayObjects.map((o: S3FileRecord) => ({ + key: o.s3Key, + sizeBytes: o.sizeBytes, + etag: o.fileHash || nanoid(16), + lastModified: formatCreatedAt(o.createdAt), + mimeType: o.mimeType, + })), + prefixes, + isTruncated, + nextMarker, + }; + }; +} + +/** + * Creates a use case that checks whether an object exists and is accessible + * within a bucket. + * + * @param deps - The injected dependencies. + * @returns An async function accepting a bucket ID and object key, + * returning the file entity or null. + */ +export function createFindObjectUseCase(deps: Pick) { + return async (bucketId: string, key: string): Promise => { + return deps.fileRepo.findByBucketAndKey(bucketId, key); + }; +} \ No newline at end of file