Stream download for transcode, multipart upload for large output

backend/src/services/rustfs.ts

@@ -11,7 +11,8 @@ import {
 import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
 import { tmpdir } from "os";
 import { join } from "path";
-import { unlink, writeFile } from "fs/promises";
+import { unlink } from "fs/promises";
+import { createWriteStream, createReadStream } from "fs";
 
 const RUSTFS_ENDPOINT = process.env.RUSTFS_ENDPOINT || "http://localhost:9000";
 const RUSTFS_ACCESS_KEY = process.env.RUSTFS_ACCESS_KEY || "minioadmin";
@@ -231,8 +232,33 @@ export async function transcodeVideo(originalKey: string): Promise<string> {
         if (!response.Body) {
             throw new Error("Empty response body from RustFS");
         }
-        const bytes = await response.Body.transformToByteArray();
-        await writeFile(inputPath, bytes);
+
+        const sdkStream = response.Body as any;
+        const webStream: ReadableStream<Uint8Array> =
+            sdkStream.transformToWebStream
+                ? sdkStream.transformToWebStream()
+                : sdkStream;
+
+        const fileWriter = createWriteStream(inputPath);
+        const reader = webStream.getReader();
+
+        try {
+            while (true) {
+                const { done, value } = await reader.read();
+                if (done) break;
+                if (!fileWriter.write(value)) {
+                    await new Promise<void>((resolve) =>
+                        fileWriter.once("drain", resolve),
+                    );
+                }
+            }
+        } finally {
+            fileWriter.end();
+            reader.releaseLock();
+            await new Promise<void>((resolve) =>
+                fileWriter.on("finish", resolve),
+            );
+        }
 
         const proc = Bun.spawn(
             [
@@ -272,9 +298,45 @@ export async function transcodeVideo(originalKey: string): Promise<string> {
                 ? originalKey.substring(0, lastDot) + ".mp4"
                 : originalKey + ".mp4";
 
-        const file = Bun.file(outputPath);
-        const buffer = Buffer.from(await file.arrayBuffer());
+        const outputFile = Bun.file(outputPath);
+        const fileSize = outputFile.size;
+        const CHUNK_SIZE = 80 * 1024 * 1024;
+
+        if (fileSize < CHUNK_SIZE) {
+            const buffer = Buffer.from(await outputFile.arrayBuffer());
             await uploadObject(transcodedKey, buffer, "video/mp4");
+        } else {
+            const { uploadId } = await createMultipartUpload(
+                transcodedKey,
+                "video/mp4",
+            );
+            const numParts = Math.ceil(fileSize / CHUNK_SIZE);
+            const parts: { PartNumber: number; ETag: string }[] = [];
+
+            for (let i = 0; i < numParts; i++) {
+                const start = i * CHUNK_SIZE;
+                const end = Math.min(start + CHUNK_SIZE, fileSize);
+                const chunk = Buffer.from(
+                    await outputFile.slice(start, end).arrayBuffer(),
+                );
+
+                const partNumber = i + 1;
+                const uploadCmd = new UploadPartCommand({
+                    Bucket: RUSTFS_BUCKET,
+                    Key: transcodedKey,
+                    UploadId: uploadId,
+                    PartNumber: partNumber,
+                    Body: chunk,
+                });
+                const result = await s3.send(uploadCmd);
+                parts.push({
+                    PartNumber: partNumber,
+                    ETag: result.ETag!,
+                });
+            }
+
+            await completeMultipartUpload(transcodedKey, uploadId, parts);
+        }
 
         return transcodedKey;
     } finally {