fix: stream complete multipart S3 objects
This commit is contained in:
+40
-24
@@ -312,7 +312,7 @@ const handleGetObject = async (
|
||||
);
|
||||
|
||||
if (file.multipartUploadId) {
|
||||
return handleGetMultipartObject(file, bucket, key, reqId);
|
||||
return handleGetMultipartObject(file, bucket, key, headers, reqId);
|
||||
}
|
||||
|
||||
const fileInfo = await getFileInfo(file.telegramFileId);
|
||||
@@ -377,6 +377,7 @@ const handleGetMultipartObject = async (
|
||||
file: File,
|
||||
bucket: string,
|
||||
key: string,
|
||||
headers: Record<string, string>,
|
||||
reqId: string,
|
||||
): Promise<Response> => {
|
||||
const uploadId = file.multipartUploadId!;
|
||||
@@ -392,21 +393,53 @@ const handleGetMultipartObject = async (
|
||||
);
|
||||
}
|
||||
|
||||
const fileInfo = await getFileInfo(parts[0].telegramFileId);
|
||||
const redirectUrl = `https://api.telegram.org/file/bot${fileInfo.bot_token}/${fileInfo.file_path}`;
|
||||
const totalSize = parts.reduce((sum, p) => sum + p.sizeBytes, 0);
|
||||
const range = parseRangeHeader(headers.range || null, totalSize);
|
||||
if (range.type === 'invalid') {
|
||||
return s3ErrorResponse(
|
||||
'InvalidRange',
|
||||
'The requested range is not satisfiable.',
|
||||
`/${bucket}/${key}`,
|
||||
416,
|
||||
reqId,
|
||||
{
|
||||
'content-range': unsatisfiedContentRange(totalSize),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
const sources: ObjectPartSource[] = [];
|
||||
for (const part of parts) {
|
||||
const fileInfo = await 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,
|
||||
});
|
||||
}
|
||||
|
||||
if (!config.proxyS3Get) {
|
||||
return new Response(null, {
|
||||
status: 302,
|
||||
headers: { Location: redirectUrl, 'x-amz-request-id': reqId },
|
||||
headers: { Location: sources[0].telegramUrl, 'x-amz-request-id': reqId },
|
||||
});
|
||||
}
|
||||
|
||||
const tgResponse = await fetch(redirectUrl);
|
||||
if (!tgResponse.ok) {
|
||||
try {
|
||||
return await createGetObjectResponse({
|
||||
reqId,
|
||||
contentType: file.mimeType,
|
||||
etag: file.fileHash || parts.map((p) => p.etag).join('-'),
|
||||
lastModified: file.createdAt instanceof Date ? file.createdAt : new Date(file.createdAt),
|
||||
totalSize,
|
||||
parts: sources,
|
||||
range,
|
||||
});
|
||||
} catch (error) {
|
||||
logger.warn('Telegram multipart content fetch failed', {
|
||||
status: tgResponse.status,
|
||||
uploadId: file.multipartUploadId,
|
||||
error: getErrorMessage(error),
|
||||
});
|
||||
return s3ErrorResponse(
|
||||
'InternalError',
|
||||
@@ -416,23 +449,6 @@ const handleGetMultipartObject = async (
|
||||
reqId,
|
||||
);
|
||||
}
|
||||
|
||||
const totalSize = parts.reduce((sum, p) => sum + p.sizeBytes, 0);
|
||||
return new Response(tgResponse.body, {
|
||||
status: 200,
|
||||
headers: {
|
||||
'content-type': file.mimeType,
|
||||
'content-length': String(totalSize || file.sizeBytes),
|
||||
etag: `"${file.fileHash || ''}"`,
|
||||
'last-modified':
|
||||
file.createdAt instanceof Date
|
||||
? file.createdAt.toUTCString()
|
||||
: new Date(file.createdAt).toUTCString(),
|
||||
'x-amz-request-id': reqId,
|
||||
'accept-ranges': 'bytes',
|
||||
'cache-control': 'public, max-age=31536000',
|
||||
},
|
||||
});
|
||||
};
|
||||
|
||||
const handleHeadObject = async (bucket: string, key: string, reqId: string): Promise<Response> => {
|
||||
|
||||
@@ -393,11 +393,53 @@ describe('S3 API (production, SigV4)', () => {
|
||||
expect(xml).toContain('InvalidRange');
|
||||
});
|
||||
|
||||
it('Multipart GetObject — returns complete concatenated body', async () => {
|
||||
const create = await s3Request('POST', `/${bucketName}/multipart-full.txt`, {
|
||||
query: { uploads: '' },
|
||||
});
|
||||
expect(create.status).toBe(200);
|
||||
const createXml = await create.text();
|
||||
const uploadId = createXml.match(/<UploadId>([^<]+)<\/UploadId>/)?.[1];
|
||||
expect(uploadId).toBeTruthy();
|
||||
|
||||
const part1 = new TextEncoder().encode('hello ');
|
||||
const part2 = new TextEncoder().encode('multipart');
|
||||
const p1 = await s3Request('PUT', `/${bucketName}/multipart-full.txt`, {
|
||||
query: { partNumber: '1', uploadId: uploadId! },
|
||||
body: part1,
|
||||
});
|
||||
const p2 = await s3Request('PUT', `/${bucketName}/multipart-full.txt`, {
|
||||
query: { partNumber: '2', uploadId: uploadId! },
|
||||
body: part2,
|
||||
});
|
||||
expect(p1.status).toBe(200);
|
||||
expect(p2.status).toBe(200);
|
||||
|
||||
const completeBody = `<CompleteMultipartUpload><Part><PartNumber>1</PartNumber><ETag>${p1.headers.get('etag')}</ETag></Part><Part><PartNumber>2</PartNumber><ETag>${p2.headers.get('etag')}</ETag></Part></CompleteMultipartUpload>`;
|
||||
const complete = await s3Request('POST', `/${bucketName}/multipart-full.txt`, {
|
||||
query: { uploadId: uploadId! },
|
||||
body: new TextEncoder().encode(completeBody),
|
||||
});
|
||||
expect(complete.status).toBe(200);
|
||||
|
||||
const full = await s3Request('GET', `/${bucketName}/multipart-full.txt`);
|
||||
expect(full.status).toBe(200);
|
||||
expect(await full.text()).toBe('hello multipart');
|
||||
|
||||
const partial = await s3Request('GET', `/${bucketName}/multipart-full.txt`, {
|
||||
headers: { range: 'bytes=3-9' },
|
||||
});
|
||||
expect(partial.status).toBe(206);
|
||||
expect(partial.headers.get('content-range')).toBe('bytes 3-9/15');
|
||||
expect(await partial.text()).toBe('lo mult');
|
||||
});
|
||||
|
||||
it('Delete bucket — must be empty first', async () => {
|
||||
// Clean up remaining objects
|
||||
await s3Request('DELETE', `/${bucketName}/test-file.txt`);
|
||||
await s3Request('DELETE', `/${bucketName}/copy-dest.txt`);
|
||||
await s3Request('DELETE', `/${bucketName}/presigned-test.txt`);
|
||||
await s3Request('DELETE', `/${bucketName}/multipart-full.txt`);
|
||||
|
||||
const r = await s3Request('DELETE', `/${bucketName}`);
|
||||
expect(r.status).toBe(204);
|
||||
|
||||
Reference in New Issue
Block a user