refactor(resources): remove composeObject

Author: ycdzj

src/minio/minio.service.ts

@@ -6,9 +6,6 @@ import {
   GetObjectCommand,
   HeadObjectCommand,
   DeleteObjectCommand,
-  CreateMultipartUploadCommand,
-  UploadPartCopyCommand,
-  CompleteMultipartUploadCommand,
   HeadBucketCommand,
   CreateBucketCommand,
 } from '@aws-sdk/client-s3';
@@ -80,7 +77,7 @@ export class MinioService implements OnModuleInit {
     this.bucket = s3Bucket;
   }
 
-  async putObject(objectName: string, buffer: Buffer, mimetype: string) {
+  async putObject(objectName: string, buffer: Buffer, mimetype?: string) {
     const command = new PutObjectCommand({
       Bucket: this.bucket,
       Key: objectName,
@@ -90,56 +87,6 @@ export class MinioService implements OnModuleInit {
     return await this.s3Client.send(command);
   }
 
-  async putChunkObject(objectName: string, chunk: Buffer) {
-    const command = new PutObjectCommand({
-      Bucket: this.bucket,
-      Key: objectName,
-      Body: chunk,
-    });
-    return await this.s3Client.send(command);
-  }
-
-  async composeObject(objectName: string, chunksName: Array<string>) {
-    const createCommand = new CreateMultipartUploadCommand({
-      Bucket: this.bucket,
-      Key: objectName,
-      ContentType: 'application/octet-stream',
-    });
-    const { UploadId } = await this.s3Client.send(createCommand);
-    if (!UploadId) {
-      throw new Error('Failed to initiate multipart upload');
-    }
-
-    const uploadedParts: { ETag: string; PartNumber: number }[] = [];
-    for (let i = 0; i < chunksName.length; i++) {
-      const uploadPartCommand = new UploadPartCopyCommand({
-        Bucket: this.bucket,
-        Key: objectName,
-        UploadId,
-        PartNumber: i + 1,
-        CopySource: `${this.bucket}/${chunksName[i]}`,
-      });
-      const response = await this.s3Client.send(uploadPartCommand);
-      if (!response.CopyPartResult?.ETag) {
-        throw new Error(`Failed to upload part`);
-      }
-      uploadedParts.push({
-        ETag: response.CopyPartResult.ETag,
-        PartNumber: i + 1,
-      });
-    }
-
-    const completeCommand = new CompleteMultipartUploadCommand({
-      Bucket: this.bucket,
-      Key: objectName,
-      UploadId,
-      MultipartUpload: {
-        Parts: uploadedParts,
-      },
-    });
-    return await this.s3Client.send(completeCommand);
-  }
-
   async getObject(objectName: string): Promise<Readable> {
     const command = new GetObjectCommand({
       Bucket: this.bucket,

src/namespace-resources/namespace-resources.service.ts

@@ -802,14 +802,6 @@ export class NamespaceResourcesService {
     return `resources/${resourceId}`;
   }
 
-  chunkPath(
-    namespaceId: string,
-    fileHash: string,
-    chunkNumber: string | number,
-  ) {
-    return `chunks/${namespaceId}/${fileHash}/${chunkNumber}`;
-  }
-
   async uploadFile(
     userId: string,
     namespaceId: string,

src/wizard/chunk-manager.service.ts

@@ -1,6 +1,7 @@
 import { Injectable, Logger } from '@nestjs/common';
 import { MinioService } from 'omniboxd/minio/minio.service';
 import { ConfigService } from '@nestjs/config';
+import { buffer } from 'node:stream/consumers';
 
 @Injectable()
 export class ChunkManagerService {
@@ -26,7 +27,7 @@ export class ChunkManagerService {
     const buffer = Buffer.from(data, 'base64');
 
     try {
-      await this.minioService.putChunkObject(chunkPath, buffer);
+      await this.minioService.putObject(chunkPath, buffer);
       this.logger.debug(
         `Stored chunk ${chunkIndex + 1}/${totalChunks} for task ${taskId}`,
       );
@@ -40,32 +41,13 @@ export class ChunkManagerService {
   }
 
   async assembleChunks(taskId: string, totalChunks: number): Promise<string> {
-    const assembledPath = this.getAssembledPath(taskId);
-    const chunkPaths = Array.from({ length: totalChunks }, (_, i) =>
-      this.getChunkPath(taskId, i),
-    );
-
-    try {
-      // Use MinIO's composeObject to merge all chunks
-      await this.minioService.composeObject(assembledPath, chunkPaths);
-
-      // Retrieve the assembled data
-      const stream = await this.minioService.getObject(assembledPath);
-      const chunks: Buffer[] = [];
-
-      return new Promise((resolve, reject) => {
-        stream.on('data', (chunk) => chunks.push(chunk));
-        stream.on('end', () => {
-          const assembledBuffer = Buffer.concat(chunks);
-          const assembledData = assembledBuffer.toString('utf-8');
-          resolve(assembledData);
-        });
-        stream.on('error', reject);
-      });
-    } catch (error) {
-      this.logger.error(`Failed to assemble chunks for task ${taskId}:`, error);
-      throw error;
+    const buffers: Buffer[] = [];
+    for (let i = 0; i < totalChunks; i++) {
+      const chunkPath = this.getChunkPath(taskId, i);
+      const stream = await this.minioService.getObject(chunkPath);
+      buffers.push(await buffer(stream));
     }
+    return Buffer.concat(buffers).toString('utf-8');
   }
 
   cleanupChunks(taskId: string, totalChunks: number): void {
@@ -87,15 +69,10 @@ export class ChunkManagerService {
     try {
       const objectsToRemove: string[] = [];
 
-      // Add chunk paths
       for (let i = 0; i < totalChunks; i++) {
         objectsToRemove.push(this.getChunkPath(taskId, i));
       }
 
-      // Add assembled path
-      objectsToRemove.push(this.getAssembledPath(taskId));
-
-      // Remove all objects
       await Promise.all(
         objectsToRemove.map((objectName) =>
           this.minioService.removeObject(objectName).catch((error) => {
@@ -113,8 +90,4 @@ export class ChunkManagerService {
   private getChunkPath(taskId: string, chunkIndex: number): string {
     return `wizard-chunks/${taskId}/chunk-${chunkIndex.toString().padStart(6, '0')}`;
   }
-
-  private getAssembledPath(taskId: string): string {
-    return `wizard-chunks/${taskId}/assembled`;
-  }
 }