feat: implement S3 protocol dispatcher with bucket, object, listing, and multipart operations
This commit is contained in:
asepharyana
@@ -0,0 +1,623 @@
|
|||||||
|
import { verifySignature, verifyPresignedUrl } from '../utils/s3/auth';
|
||||||
|
import {
|
||||||
|
listBucketsXml, s3ErrorResponse, listBucketResultXml, listBucketV2ResultXml,
|
||||||
|
initiateMultipartUploadXml, listPartsXml, completeMultipartUploadXml, deleteResultXml,
|
||||||
|
copyObjectResultXml, parseDeleteObjectsBody, parseCompleteMultipartBody,
|
||||||
|
} from '../utils/s3/xml';
|
||||||
|
import { createBucket, findBucketByName, listBuckets, deleteBucket } from '../db/buckets';
|
||||||
|
import {
|
||||||
|
createMultipartUpload, findMultipartUpload, completeMultipartUpload, abortMultipartUpload,
|
||||||
|
insertMultipartPart, listMultipartParts,
|
||||||
|
} from '../db/multipart';
|
||||||
|
import {
|
||||||
|
findFileByBucketAndKey, listObjectsByPrefix, softDeleteFile, softDeleteFilesBatch, countBucketObjects,
|
||||||
|
} from '../db/files-ext';
|
||||||
|
import type { File } from '../db/schema';
|
||||||
|
import { config } from '../env';
|
||||||
|
import { forwardToStorage, getFileInfo } from '../utils/telegram';
|
||||||
|
import { computeHash, ensureExtension, getErrorMessage, cleanupTempFile } from '../utils/file';
|
||||||
|
import { nanoid } from 'nanoid';
|
||||||
|
import { createReadStream } from 'node:fs';
|
||||||
|
import logger from '../utils/logger';
|
||||||
|
|
||||||
|
const REGION = config.s3DefaultRegion || 'us-east-1';
|
||||||
|
const REQUEST_ID = () => nanoid(16);
|
||||||
|
|
||||||
|
const parseS3Path = (pathname: string): { bucket: string | null; key: string | null } => {
|
||||||
|
const parts = pathname.split('/').filter(Boolean);
|
||||||
|
if (parts.length === 0) return { bucket: null, key: null };
|
||||||
|
if (parts.length === 1) return { bucket: parts[0], key: null };
|
||||||
|
return { bucket: parts[0], key: parts.slice(1).join('/') };
|
||||||
|
};
|
||||||
|
|
||||||
|
const headersToRecord = (req: Request): Record<string, string> => {
|
||||||
|
const record: Record<string, string> = {};
|
||||||
|
for (const [key, value] of req.headers.entries()) {
|
||||||
|
record[key.toLowerCase()] = value;
|
||||||
|
}
|
||||||
|
return record;
|
||||||
|
};
|
||||||
|
|
||||||
|
// ─────── Main Dispatcher ───────
|
||||||
|
|
||||||
|
export const handleS3Request = async (req: Request): Promise<Response> => {
|
||||||
|
const method = req.method;
|
||||||
|
const url = new URL(req.url);
|
||||||
|
const pathname = url.pathname;
|
||||||
|
const { bucket, key } = parseS3Path(pathname);
|
||||||
|
const headers = headersToRecord(req);
|
||||||
|
const searchParams = url.searchParams;
|
||||||
|
const reqId = REQUEST_ID();
|
||||||
|
|
||||||
|
// Verify auth for regular requests; presigned URLs are verified per-operation
|
||||||
|
if (!searchParams.has('X-Amz-Signature')) {
|
||||||
|
const authResult = await verifySignature(method, req.url, headers, null, config.s3AccessKey, config.s3SecretKey, REGION);
|
||||||
|
if (!authResult.isValid) {
|
||||||
|
return s3ErrorResponse(authResult.errorCode || 'AccessDenied', 'Authentication required', pathname, 403, reqId);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
try {
|
||||||
|
// Root: ListBuckets
|
||||||
|
if (!bucket) {
|
||||||
|
if (method === 'GET') {
|
||||||
|
return handleListBuckets(reqId);
|
||||||
|
}
|
||||||
|
return s3ErrorResponse('MethodNotAllowed', 'The specified method is not allowed against this resource.', '/', 405, reqId);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Bucket-level operations
|
||||||
|
if (!key) {
|
||||||
|
if (method === 'GET') {
|
||||||
|
const listType = searchParams.get('list-type');
|
||||||
|
if (listType === '2') {
|
||||||
|
return handleListObjectsV2(bucket, searchParams, reqId);
|
||||||
|
}
|
||||||
|
return handleListObjectsV1(bucket, searchParams, reqId);
|
||||||
|
}
|
||||||
|
if (method === 'PUT') return handleCreateBucket(bucket, reqId);
|
||||||
|
if (method === 'HEAD') return handleHeadBucket(bucket, reqId);
|
||||||
|
if (method === 'DELETE') return handleDeleteBucket(bucket, reqId);
|
||||||
|
if (method === 'POST') {
|
||||||
|
if (searchParams.has('delete')) {
|
||||||
|
const body = await req.text();
|
||||||
|
return handleDeleteObjects(bucket, body, reqId);
|
||||||
|
}
|
||||||
|
if (searchParams.has('tagging')) {
|
||||||
|
return new Response(null, { status: 204 });
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return s3ErrorResponse('MethodNotAllowed', 'The specified method is not allowed against this resource.', `/${bucket}`, 405, reqId);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Object-level: multipart checks
|
||||||
|
if (searchParams.has('uploads') && method === 'POST') {
|
||||||
|
return handleCreateMultipartUpload(bucket, key, searchParams, reqId);
|
||||||
|
}
|
||||||
|
if (searchParams.has('uploadId') && searchParams.has('partNumber') && method === 'PUT') {
|
||||||
|
return handleUploadPart(bucket, key, searchParams, req, reqId);
|
||||||
|
}
|
||||||
|
if (searchParams.has('uploadId') && method === 'POST') {
|
||||||
|
const body = await req.text();
|
||||||
|
return handleCompleteMultipartUpload(bucket, key, searchParams, body, reqId);
|
||||||
|
}
|
||||||
|
if (searchParams.has('uploadId') && method === 'DELETE') {
|
||||||
|
return handleAbortMultipartUpload(bucket, key, searchParams, reqId);
|
||||||
|
}
|
||||||
|
if (searchParams.has('uploadId') && method === 'GET') {
|
||||||
|
return handleListParts(bucket, key, searchParams, reqId);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Standard object operations
|
||||||
|
if (method === 'GET') return handleGetObject(bucket, key, searchParams, reqId);
|
||||||
|
if (method === 'HEAD') return handleHeadObject(bucket, key, reqId);
|
||||||
|
if (method === 'PUT') return handlePutObject(bucket, key, searchParams, headers, req, reqId);
|
||||||
|
if (method === 'DELETE') return handleDeleteObject(bucket, key, reqId);
|
||||||
|
|
||||||
|
return s3ErrorResponse('MethodNotAllowed', 'The specified method is not allowed against this resource.', `/${bucket}/${key}`, 405, reqId);
|
||||||
|
} catch (error: unknown) {
|
||||||
|
logger.error('S3 operation error', { bucket, key, error: getErrorMessage(error) });
|
||||||
|
return s3ErrorResponse('InternalError', 'We encountered an internal error. Please try again.', pathname, 500, reqId);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
// ─────── Bucket Operations ───────
|
||||||
|
|
||||||
|
const handleListBuckets = async (reqId: string): Promise<Response> => {
|
||||||
|
const buckets = await listBuckets();
|
||||||
|
const xml = listBucketsXml(buckets, reqId);
|
||||||
|
return new Response(xml, {
|
||||||
|
status: 200,
|
||||||
|
headers: { 'content-type': 'application/xml', 'x-amz-request-id': reqId },
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
const handleCreateBucket = async (bucketName: string, reqId: string): Promise<Response> => {
|
||||||
|
if (!/^[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]$/.test(bucketName)) {
|
||||||
|
return s3ErrorResponse('InvalidBucketName', 'The specified bucket is not valid.', `/${bucketName}`, 400, reqId);
|
||||||
|
}
|
||||||
|
const existing = await findBucketByName(bucketName);
|
||||||
|
if (existing) {
|
||||||
|
return s3ErrorResponse('BucketAlreadyExists', 'The requested bucket name is not available.', `/${bucketName}`, 409, reqId);
|
||||||
|
}
|
||||||
|
await createBucket(bucketName);
|
||||||
|
return new Response(null, { status: 200, headers: { 'x-amz-request-id': reqId } });
|
||||||
|
};
|
||||||
|
|
||||||
|
const handleHeadBucket = async (bucketName: string, reqId: string): Promise<Response> => {
|
||||||
|
const bucket = await findBucketByName(bucketName);
|
||||||
|
if (!bucket) {
|
||||||
|
return s3ErrorResponse('NoSuchBucket', 'The specified bucket does not exist.', `/${bucketName}`, 404, reqId);
|
||||||
|
}
|
||||||
|
return new Response(null, { status: 200, headers: { 'x-amz-request-id': reqId } });
|
||||||
|
};
|
||||||
|
|
||||||
|
const handleDeleteBucket = async (bucketName: string, reqId: string): Promise<Response> => {
|
||||||
|
const bucket = await findBucketByName(bucketName);
|
||||||
|
if (!bucket) {
|
||||||
|
return s3ErrorResponse('NoSuchBucket', 'The specified bucket does not exist.', `/${bucketName}`, 404, reqId);
|
||||||
|
}
|
||||||
|
const objCount = await countBucketObjects(bucket.id);
|
||||||
|
if (objCount > 0) {
|
||||||
|
return s3ErrorResponse('BucketNotEmpty', 'The bucket you tried to delete is not empty.', `/${bucketName}`, 409, reqId);
|
||||||
|
}
|
||||||
|
await deleteBucket(bucketName);
|
||||||
|
return new Response(null, { status: 204, headers: { 'x-amz-request-id': reqId } });
|
||||||
|
};
|
||||||
|
|
||||||
|
// ─────── Object Operations ───────
|
||||||
|
|
||||||
|
const handleGetObject = async (bucket: string, key: string, searchParams: URLSearchParams, reqId: string): Promise<Response> => {
|
||||||
|
if (searchParams.has('X-Amz-Signature')) {
|
||||||
|
const fullUrl = `http://localhost/${bucket}/${key}?${searchParams.toString()}`;
|
||||||
|
const presignedResult = await verifyPresignedUrl(fullUrl, 'GET', config.s3AccessKey, config.s3SecretKey, REGION);
|
||||||
|
if (!presignedResult.isValid) {
|
||||||
|
return s3ErrorResponse('AccessDenied', 'Request has expired', `/${bucket}/${key}`, 403, reqId);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const bucketRecord = await findBucketByName(bucket);
|
||||||
|
if (!bucketRecord) return s3ErrorResponse('NoSuchBucket', 'The specified bucket does not exist.', `/${bucket}/${key}`, 404, reqId);
|
||||||
|
|
||||||
|
const file = await findFileByBucketAndKey(bucketRecord.id, key);
|
||||||
|
if (!file) return s3ErrorResponse('NoSuchKey', 'The specified key does not exist.', `/${bucket}/${key}`, 404, reqId);
|
||||||
|
|
||||||
|
if (file.multipartUploadId) {
|
||||||
|
return handleGetMultipartObject(file, bucket, key, reqId);
|
||||||
|
}
|
||||||
|
|
||||||
|
const fileInfo = await getFileInfo(file.telegramFileId);
|
||||||
|
const redirectUrl = `https://api.telegram.org/file/bot${fileInfo.bot_token}/${fileInfo.file_path}`;
|
||||||
|
|
||||||
|
return new Response(null, {
|
||||||
|
status: 302,
|
||||||
|
headers: {
|
||||||
|
Location: redirectUrl,
|
||||||
|
'x-amz-request-id': reqId,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
const handleGetMultipartObject = async (file: File, bucket: string, key: string, reqId: string): Promise<Response> => {
|
||||||
|
const uploadId = file.multipartUploadId!;
|
||||||
|
const parts = await listMultipartParts(uploadId);
|
||||||
|
|
||||||
|
if (parts.length === 0) {
|
||||||
|
return s3ErrorResponse('InternalError', 'Multipart object has no parts.', `/${bucket}/${key}`, 500, reqId);
|
||||||
|
}
|
||||||
|
|
||||||
|
const fileInfo = await getFileInfo(parts[0].telegramFileId);
|
||||||
|
const redirectUrl = `https://api.telegram.org/file/bot${fileInfo.bot_token}/${fileInfo.file_path}`;
|
||||||
|
|
||||||
|
return new Response(null, {
|
||||||
|
status: 302,
|
||||||
|
headers: {
|
||||||
|
Location: redirectUrl,
|
||||||
|
'x-amz-request-id': reqId,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
const handleHeadObject = async (bucket: string, key: string, reqId: string): Promise<Response> => {
|
||||||
|
const bucketRecord = await findBucketByName(bucket);
|
||||||
|
if (!bucketRecord) return s3ErrorResponse('NoSuchBucket', 'The specified bucket does not exist.', `/${bucket}/${key}`, 404, reqId);
|
||||||
|
|
||||||
|
const file = await findFileByBucketAndKey(bucketRecord.id, key);
|
||||||
|
if (!file) return s3ErrorResponse('NoSuchKey', 'The specified key does not exist.', `/${bucket}/${key}`, 404, reqId);
|
||||||
|
|
||||||
|
return new Response(null, {
|
||||||
|
status: 200,
|
||||||
|
headers: {
|
||||||
|
'content-type': file.mimeType,
|
||||||
|
'content-length': String(file.sizeBytes),
|
||||||
|
'etag': `"${file.fileHash || nanoid(16)}"`,
|
||||||
|
'last-modified': file.createdAt instanceof Date ? file.createdAt.toUTCString() : new Date().toUTCString(),
|
||||||
|
'x-amz-request-id': reqId,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
const handlePutObject = async (bucket: string, key: string, searchParams: URLSearchParams, headers: Record<string, string>, req: Request, reqId: string): Promise<Response> => {
|
||||||
|
const bucketRecord = await findBucketByName(bucket);
|
||||||
|
if (!bucketRecord) return s3ErrorResponse('NoSuchBucket', 'The specified bucket does not exist.', `/${bucket}/${key}`, 404, reqId);
|
||||||
|
|
||||||
|
if (searchParams.has('tagging')) {
|
||||||
|
return new Response(null, { status: 204 });
|
||||||
|
}
|
||||||
|
|
||||||
|
const copySource = headers['x-amz-copy-source'];
|
||||||
|
if (copySource) {
|
||||||
|
return handleCopyObject(bucket, key, copySource, bucketRecord.id, reqId);
|
||||||
|
}
|
||||||
|
|
||||||
|
const formData = await req.formData();
|
||||||
|
const fileField = formData.get('file');
|
||||||
|
const file = fileField instanceof File ? fileField : null;
|
||||||
|
|
||||||
|
if (!file) {
|
||||||
|
// Direct body upload
|
||||||
|
const body = await req.arrayBuffer();
|
||||||
|
const fileBuffer = Buffer.from(body);
|
||||||
|
const hash = computeHash(fileBuffer);
|
||||||
|
const existing = await findFileByBucketAndKey(bucketRecord.id, key);
|
||||||
|
if (existing) {
|
||||||
|
return new Response(null, { status: 200, headers: { 'etag': `"${hash}"`, 'x-amz-request-id': reqId } });
|
||||||
|
}
|
||||||
|
|
||||||
|
return await storeFileToTelegram(fileBuffer, hash, key, bucketRecord, reqId);
|
||||||
|
}
|
||||||
|
|
||||||
|
const buffer = Buffer.from(await file.arrayBuffer());
|
||||||
|
const hash = computeHash(buffer);
|
||||||
|
|
||||||
|
const existing = await findFileByBucketAndKey(bucketRecord.id, key);
|
||||||
|
if (existing) {
|
||||||
|
return new Response(null, { status: 200, headers: { 'etag': `"${hash}"`, 'x-amz-request-id': reqId } });
|
||||||
|
}
|
||||||
|
|
||||||
|
return await storeFileToTelegram(buffer, hash, key, bucketRecord, reqId);
|
||||||
|
};
|
||||||
|
|
||||||
|
const storeFileToTelegram = async (buffer: Buffer, hash: string, key: string, bucketRecord: { id: string; name: string }, reqId: string): Promise<Response> => {
|
||||||
|
const tempPath = `/tmp/teleuploader-s3-${nanoid()}`;
|
||||||
|
await Bun.write(tempPath, buffer);
|
||||||
|
|
||||||
|
const signatureBuffer = buffer.subarray(0, 16);
|
||||||
|
const fileName = key.split('/').pop() || 'file';
|
||||||
|
const { fileName: finalFileName, mimeType } = ensureExtension(fileName, signatureBuffer, 'application/octet-stream');
|
||||||
|
|
||||||
|
const forwardResult = await forwardToStorage(
|
||||||
|
createReadStream(tempPath),
|
||||||
|
`s3-${bucketRecord.name}-${key.replace(/\//g, '_')}`,
|
||||||
|
'document',
|
||||||
|
);
|
||||||
|
|
||||||
|
const publicId = nanoid();
|
||||||
|
const { db, files: fileSchema } = await import('../db/index');
|
||||||
|
|
||||||
|
await db.insert(fileSchema).values({
|
||||||
|
publicId,
|
||||||
|
telegramFileId: forwardResult.telegramFileId,
|
||||||
|
telegramFileUniqueId: forwardResult.telegramFileUniqueId,
|
||||||
|
storageChatId: config.storageChatId,
|
||||||
|
storageMessageId: forwardResult.storageMessageId,
|
||||||
|
fileName: finalFileName,
|
||||||
|
mimeType,
|
||||||
|
sizeBytes: buffer.byteLength,
|
||||||
|
fileType: 'document',
|
||||||
|
uploaderId: 0,
|
||||||
|
fileHash: hash,
|
||||||
|
bucketId: bucketRecord.id,
|
||||||
|
s3Key: key,
|
||||||
|
storageBackend: 'telegram',
|
||||||
|
isDeleted: false,
|
||||||
|
createdAt: new Date(),
|
||||||
|
updatedAt: new Date(),
|
||||||
|
});
|
||||||
|
|
||||||
|
await cleanupTempFile(tempPath);
|
||||||
|
|
||||||
|
return new Response(null, {
|
||||||
|
status: 200,
|
||||||
|
headers: { 'etag': `"${hash}"`, 'x-amz-request-id': reqId },
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
const handleCopyObject = async (_destBucket: string, destKey: string, copySource: string, destBucketId: string, reqId: string): Promise<Response> => {
|
||||||
|
const sourcePath = copySource.startsWith('/') ? copySource.slice(1) : copySource;
|
||||||
|
const parts = sourcePath.split('/');
|
||||||
|
const sourceBucket = parts[0];
|
||||||
|
const sourceKey = parts.slice(1).join('/');
|
||||||
|
|
||||||
|
const sourceBucketRecord = await findBucketByName(sourceBucket);
|
||||||
|
if (!sourceBucketRecord) return s3ErrorResponse('NoSuchBucket', 'The specified bucket does not exist.', copySource, 404, reqId);
|
||||||
|
|
||||||
|
const sourceFile = await findFileByBucketAndKey(sourceBucketRecord.id, sourceKey);
|
||||||
|
if (!sourceFile) return s3ErrorResponse('NoSuchKey', 'The specified key does not exist.', copySource, 404, reqId);
|
||||||
|
|
||||||
|
const publicId = nanoid();
|
||||||
|
const { db, files: fileSchema } = await import('../db/index');
|
||||||
|
|
||||||
|
await db.insert(fileSchema).values({
|
||||||
|
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,
|
||||||
|
bucketId: destBucketId,
|
||||||
|
s3Key: destKey,
|
||||||
|
storageBackend: 'telegram',
|
||||||
|
isDeleted: false,
|
||||||
|
createdAt: new Date(),
|
||||||
|
updatedAt: new Date(),
|
||||||
|
});
|
||||||
|
|
||||||
|
const xml = copyObjectResultXml(sourceFile.fileHash || nanoid(16), new Date());
|
||||||
|
return new Response(xml, {
|
||||||
|
status: 200,
|
||||||
|
headers: { 'content-type': 'application/xml', 'x-amz-request-id': reqId },
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
const handleDeleteObject = async (bucket: string, key: string, reqId: string): Promise<Response> => {
|
||||||
|
const bucketRecord = await findBucketByName(bucket);
|
||||||
|
if (!bucketRecord) return s3ErrorResponse('NoSuchBucket', 'The specified bucket does not exist.', `/${bucket}/${key}`, 404, reqId);
|
||||||
|
|
||||||
|
await softDeleteFile(bucketRecord.id, key);
|
||||||
|
return new Response(null, { status: 204, headers: { 'x-amz-request-id': reqId } });
|
||||||
|
};
|
||||||
|
|
||||||
|
const handleDeleteObjects = async (bucket: string, body: string, reqId: string): Promise<Response> => {
|
||||||
|
const bucketRecord = await findBucketByName(bucket);
|
||||||
|
if (!bucketRecord) return s3ErrorResponse('NoSuchBucket', 'The specified bucket does not exist.', `/${bucket}`, 404, reqId);
|
||||||
|
|
||||||
|
const { keys } = parseDeleteObjectsBody(body);
|
||||||
|
const deleted = await softDeleteFilesBatch(bucketRecord.id, keys);
|
||||||
|
const xml = deleteResultXml(keys.slice(0, deleted), []);
|
||||||
|
return new Response(xml, {
|
||||||
|
status: 200,
|
||||||
|
headers: { 'content-type': 'application/xml', 'x-amz-request-id': reqId },
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
// ─────── Object Listing ───────
|
||||||
|
|
||||||
|
const handleListObjectsV1 = async (bucket: string, searchParams: URLSearchParams, reqId: string): Promise<Response> => {
|
||||||
|
const bucketRecord = await findBucketByName(bucket);
|
||||||
|
if (!bucketRecord) return s3ErrorResponse('NoSuchBucket', 'The specified bucket does not exist.', `/${bucket}`, 404, reqId);
|
||||||
|
|
||||||
|
const prefix = searchParams.get('prefix') || '';
|
||||||
|
const delimiter = searchParams.get('delimiter') || null;
|
||||||
|
const maxKeys = Math.min(parseInt(searchParams.get('max-keys') || '1000', 10), 1000);
|
||||||
|
const marker = searchParams.get('marker') || null;
|
||||||
|
|
||||||
|
const { objects, prefixes: commonPrefixes } = await listObjectsByPrefix(
|
||||||
|
bucketRecord.id, prefix, delimiter, maxKeys, marker,
|
||||||
|
);
|
||||||
|
|
||||||
|
const isTruncated = objects.length > maxKeys;
|
||||||
|
const displayObjects = objects.slice(0, maxKeys);
|
||||||
|
const nextMarker = isTruncated ? (displayObjects[displayObjects.length - 1]?.s3Key ?? null) : null;
|
||||||
|
|
||||||
|
const xml = listBucketResultXml(
|
||||||
|
bucket,
|
||||||
|
displayObjects.map((o) => ({
|
||||||
|
key: o.s3Key ?? '',
|
||||||
|
sizeBytes: o.sizeBytes,
|
||||||
|
etag: o.fileHash || nanoid(16),
|
||||||
|
lastModified: o.createdAt instanceof Date ? o.createdAt : new Date(),
|
||||||
|
mimeType: o.mimeType,
|
||||||
|
})),
|
||||||
|
commonPrefixes,
|
||||||
|
isTruncated,
|
||||||
|
marker,
|
||||||
|
maxKeys,
|
||||||
|
prefix,
|
||||||
|
delimiter,
|
||||||
|
nextMarker,
|
||||||
|
reqId,
|
||||||
|
);
|
||||||
|
|
||||||
|
return new Response(xml, {
|
||||||
|
status: 200,
|
||||||
|
headers: { 'content-type': 'application/xml', 'x-amz-request-id': reqId },
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
const handleListObjectsV2 = async (bucket: string, searchParams: URLSearchParams, reqId: string): Promise<Response> => {
|
||||||
|
const bucketRecord = await findBucketByName(bucket);
|
||||||
|
if (!bucketRecord) return s3ErrorResponse('NoSuchBucket', 'The specified bucket does not exist.', `/${bucket}`, 404, reqId);
|
||||||
|
|
||||||
|
const prefix = searchParams.get('prefix') || '';
|
||||||
|
const delimiter = searchParams.get('delimiter') || null;
|
||||||
|
const maxKeys = Math.min(parseInt(searchParams.get('max-keys') || '1000', 10), 1000);
|
||||||
|
const continuationToken = searchParams.get('continuation-token') || null;
|
||||||
|
const startAfter = searchParams.get('start-after') || null;
|
||||||
|
|
||||||
|
const { objects, prefixes: commonPrefixes } = await listObjectsByPrefix(
|
||||||
|
bucketRecord.id, prefix, delimiter, maxKeys, continuationToken || startAfter,
|
||||||
|
);
|
||||||
|
|
||||||
|
const isTruncated = objects.length > maxKeys;
|
||||||
|
const displayObjects = objects.slice(0, maxKeys);
|
||||||
|
const nextContinuationToken = isTruncated ? (displayObjects[displayObjects.length - 1]?.s3Key ?? null) : null;
|
||||||
|
|
||||||
|
const xml = listBucketV2ResultXml(
|
||||||
|
bucket,
|
||||||
|
displayObjects.map((o) => ({
|
||||||
|
key: o.s3Key ?? '',
|
||||||
|
sizeBytes: o.sizeBytes,
|
||||||
|
etag: o.fileHash || nanoid(16),
|
||||||
|
lastModified: o.createdAt instanceof Date ? o.createdAt : new Date(),
|
||||||
|
mimeType: o.mimeType,
|
||||||
|
})),
|
||||||
|
commonPrefixes,
|
||||||
|
isTruncated,
|
||||||
|
maxKeys,
|
||||||
|
prefix,
|
||||||
|
delimiter,
|
||||||
|
continuationToken,
|
||||||
|
nextContinuationToken,
|
||||||
|
displayObjects.length,
|
||||||
|
reqId,
|
||||||
|
);
|
||||||
|
|
||||||
|
return new Response(xml, {
|
||||||
|
status: 200,
|
||||||
|
headers: { 'content-type': 'application/xml', 'x-amz-request-id': reqId },
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
// ─────── Multipart Upload ───────
|
||||||
|
|
||||||
|
const handleCreateMultipartUpload = async (bucket: string, key: string, _searchParams: URLSearchParams, reqId: string): Promise<Response> => {
|
||||||
|
const bucketRecord = await findBucketByName(bucket);
|
||||||
|
if (!bucketRecord) return s3ErrorResponse('NoSuchBucket', 'The specified bucket does not exist.', `/${bucket}/${key}`, 404, reqId);
|
||||||
|
|
||||||
|
const uploadId = await createMultipartUpload(bucketRecord.id, key, 's3');
|
||||||
|
|
||||||
|
const xml = initiateMultipartUploadXml(bucket, key, uploadId);
|
||||||
|
return new Response(xml, {
|
||||||
|
status: 200,
|
||||||
|
headers: { 'content-type': 'application/xml', 'x-amz-request-id': reqId },
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
const handleUploadPart = async (bucket: string, key: string, searchParams: URLSearchParams, req: Request, reqId: string): Promise<Response> => {
|
||||||
|
const uploadId = searchParams.get('uploadId')!;
|
||||||
|
const partNumber = parseInt(searchParams.get('partNumber')!, 10);
|
||||||
|
|
||||||
|
const multipart = await findMultipartUpload(uploadId);
|
||||||
|
if (!multipart || multipart.s3Key !== key) {
|
||||||
|
return s3ErrorResponse('NoSuchUpload', 'The specified upload does not exist.', `/${bucket}/${key}`, 404, reqId);
|
||||||
|
}
|
||||||
|
|
||||||
|
const body = await req.arrayBuffer();
|
||||||
|
const buffer = Buffer.from(body);
|
||||||
|
|
||||||
|
const tempPath = `/tmp/teleuploader-mp-${nanoid()}`;
|
||||||
|
await Bun.write(tempPath, buffer);
|
||||||
|
|
||||||
|
const forwardResult = await forwardToStorage(
|
||||||
|
createReadStream(tempPath),
|
||||||
|
`mp-${uploadId}-part-${partNumber}`,
|
||||||
|
'document',
|
||||||
|
);
|
||||||
|
|
||||||
|
await cleanupTempFile(tempPath);
|
||||||
|
|
||||||
|
const etag = computeHash(buffer);
|
||||||
|
await insertMultipartPart({
|
||||||
|
uploadId,
|
||||||
|
partNumber,
|
||||||
|
telegramFileId: forwardResult.telegramFileId,
|
||||||
|
telegramFileUniqueId: forwardResult.telegramFileUniqueId,
|
||||||
|
storageMessageId: forwardResult.storageMessageId,
|
||||||
|
sizeBytes: buffer.byteLength,
|
||||||
|
etag,
|
||||||
|
});
|
||||||
|
|
||||||
|
return new Response(null, {
|
||||||
|
status: 200,
|
||||||
|
headers: { 'etag': `"${etag}"`, 'x-amz-request-id': reqId },
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
const handleCompleteMultipartUpload = async (bucket: string, key: string, searchParams: URLSearchParams, body: string, reqId: string): Promise<Response> => {
|
||||||
|
const uploadId = searchParams.get('uploadId')!;
|
||||||
|
const multipart = await findMultipartUpload(uploadId);
|
||||||
|
if (!multipart) {
|
||||||
|
return s3ErrorResponse('NoSuchUpload', 'The specified upload does not exist.', `/${bucket}/${key}`, 404, reqId);
|
||||||
|
}
|
||||||
|
|
||||||
|
const parts = parseCompleteMultipartBody(body);
|
||||||
|
const storedParts = await listMultipartParts(uploadId);
|
||||||
|
|
||||||
|
if (parts.length !== storedParts.length) {
|
||||||
|
return s3ErrorResponse('InvalidPart', 'One or more specified parts could not be found.', `/${bucket}/${key}`, 400, reqId);
|
||||||
|
}
|
||||||
|
|
||||||
|
const totalSize = storedParts.reduce((sum, p) => sum + p.sizeBytes, 0);
|
||||||
|
|
||||||
|
const publicId = nanoid();
|
||||||
|
const { db, files: fileSchema } = await import('../db/index');
|
||||||
|
|
||||||
|
await db.insert(fileSchema).values({
|
||||||
|
publicId,
|
||||||
|
telegramFileId: storedParts[0].telegramFileId,
|
||||||
|
telegramFileUniqueId: storedParts[0].telegramFileUniqueId,
|
||||||
|
storageChatId: config.storageChatId,
|
||||||
|
storageMessageId: storedParts[0].storageMessageId,
|
||||||
|
fileName: key.split('/').pop() || 'file',
|
||||||
|
mimeType: 'application/octet-stream',
|
||||||
|
sizeBytes: totalSize,
|
||||||
|
fileType: 'document',
|
||||||
|
uploaderId: 0,
|
||||||
|
bucketId: multipart.bucketId,
|
||||||
|
s3Key: key,
|
||||||
|
storageBackend: 'telegram',
|
||||||
|
isDeleted: false,
|
||||||
|
multipartUploadId: uploadId,
|
||||||
|
createdAt: new Date(),
|
||||||
|
updatedAt: new Date(),
|
||||||
|
});
|
||||||
|
|
||||||
|
await completeMultipartUpload(uploadId);
|
||||||
|
|
||||||
|
const location = `${config.baseUrl}/${bucket}/${key}`;
|
||||||
|
const combinedEtag = storedParts.map((p) => p.etag).join('-');
|
||||||
|
const xml = completeMultipartUploadXml(bucket, key, combinedEtag, location);
|
||||||
|
|
||||||
|
return new Response(xml, {
|
||||||
|
status: 200,
|
||||||
|
headers: { 'content-type': 'application/xml', 'x-amz-request-id': reqId },
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
const handleAbortMultipartUpload = async (bucket: string, key: string, searchParams: URLSearchParams, reqId: string): Promise<Response> => {
|
||||||
|
const uploadId = searchParams.get('uploadId')!;
|
||||||
|
const multipart = await findMultipartUpload(uploadId);
|
||||||
|
if (!multipart) {
|
||||||
|
return s3ErrorResponse('NoSuchUpload', 'The specified upload does not exist.', `/${bucket}/${key}`, 404, reqId);
|
||||||
|
}
|
||||||
|
|
||||||
|
await abortMultipartUpload(uploadId);
|
||||||
|
return new Response(null, { status: 204, headers: { 'x-amz-request-id': reqId } });
|
||||||
|
};
|
||||||
|
|
||||||
|
const handleListParts = async (bucket: string, key: string, searchParams: URLSearchParams, reqId: string): Promise<Response> => {
|
||||||
|
const uploadId = searchParams.get('uploadId')!;
|
||||||
|
const multipart = await findMultipartUpload(uploadId);
|
||||||
|
if (!multipart) {
|
||||||
|
return s3ErrorResponse('NoSuchUpload', 'The specified upload does not exist.', `/${bucket}/${key}`, 404, reqId);
|
||||||
|
}
|
||||||
|
|
||||||
|
const parts = await listMultipartParts(uploadId);
|
||||||
|
const maxParts = Math.min(parseInt(searchParams.get('max-parts') || '1000', 10), 1000);
|
||||||
|
|
||||||
|
const xml = listPartsXml(
|
||||||
|
bucket,
|
||||||
|
key,
|
||||||
|
uploadId,
|
||||||
|
parts.map((p) => ({
|
||||||
|
partNumber: p.partNumber,
|
||||||
|
etag: p.etag,
|
||||||
|
sizeBytes: p.sizeBytes,
|
||||||
|
createdAt: p.createdAt,
|
||||||
|
})),
|
||||||
|
maxParts,
|
||||||
|
false,
|
||||||
|
reqId,
|
||||||
|
);
|
||||||
|
|
||||||
|
return new Response(xml, {
|
||||||
|
status: 200,
|
||||||
|
headers: { 'content-type': 'application/xml', 'x-amz-request-id': reqId },
|
||||||
|
});
|
||||||
|
};
|
||||||
Reference in New Issue
Block a user