feat: bucket stream route handler

packages/server-analysis-manager/src/components/builder/helpers/container-pack.ts

@@ -30,10 +30,7 @@ export async function packContainerWithTrain(container: Container, context: Cont
     const client = useStorageClient();
 
     return new Promise<void>((resolve, reject) => {
-        client.request({
-            url: client.bucket.getDownloadPath(context.train.id),
-            responseType: 'stream',
-        })
+        client.bucket.stream(context.train.id)
             .then((response) => {
                 const extract = tar.extract();
 

packages/server-storage/src/http/controllers/bucket/handlers/index.ts

@@ -8,5 +8,6 @@
 export * from './create';
 export * from './delete';
 export * from './read';
+export * from './stream';
 export * from './update';
 export * from './upload';

packages/server-storage/src/http/controllers/bucket/handlers/stream.ts

@@ -0,0 +1,81 @@
+/*
+ * Copyright (c) 2022-2024.
+ * Author Peter Placzek (tada5hi)
+ * For the full copyright and license information,
+ * view the LICENSE file that was distributed with this source code.
+ */
+
+import { NotFoundError } from '@ebec/http';
+import type { Request, Response } from 'routup';
+import { useRequestParam } from 'routup';
+import type { Pack } from 'tar-stream';
+import tar from 'tar-stream';
+import { useDataSource } from 'typeorm-extension';
+import { streamToBuffer, useMinio } from '../../../../core';
+import {
+    BucketEntity, BucketFileEntity,
+} from '../../../../domains';
+
+async function packFile(
+    pack: Pack,
+    name: string,
+    file: BucketFileEntity,
+) : Promise<void> {
+    const minio = useMinio();
+    return new Promise<void>((resolve, reject) => {
+        minio.getObject(name, file.hash)
+            .then((stream) => streamToBuffer(stream))
+            .then((data) => {
+                pack.entry({
+                    name: file.path,
+                    size: data.byteLength,
+                }, data, (err) => {
+                    if (err) {
+                        reject(err);
+
+                        return;
+                    }
+
+                    resolve();
+                });
+            })
+            .catch((e) => reject(e));
+    });
+}
+
+async function packFiles(
+    pack: Pack,
+    name: string,
+    files: BucketFileEntity[],
+) {
+    const promises : Promise<void>[] = [];
+
+    for (let i = 0; i < files.length; i++) {
+        promises.push(packFile(pack, name, files[i]));
+    }
+
+    await Promise.all(promises);
+}
+
+export async function executeBucketRouteStreamHandler(req: Request, res: Response) : Promise<any> {
+    const id = useRequestParam(req, 'id');
+
+    const dataSource = await useDataSource();
+    const repository = dataSource.getRepository(BucketEntity);
+    const entity = await repository.findOneBy({ id });
+    if (!entity) {
+        throw new NotFoundError();
+    }
+
+    const fileRepository = dataSource.getRepository(BucketFileEntity);
+    const files = await fileRepository.findBy({
+        bucket_id: entity.id,
+    });
+
+    const pack = tar.pack();
+    pack.pipe(res);
+
+    await packFiles(pack, entity.name, files);
+
+    pack.finalize();
+}

packages/server-storage/src/http/controllers/bucket/index.ts

@@ -19,6 +19,7 @@ import {
     executeBucketRouteDeleteHandler,
     executeBucketRouteGetManyHandler,
     executeBucketRouteGetOneHandler,
+    executeBucketRouteStreamHandler,
     executeBucketRouteUpdateHandler,
     executeBucketRouteUploadHandler,
 } from './handlers';
@@ -36,6 +37,15 @@ export class BucketController {
         return await executeBucketRouteGetManyHandler(req, res) as PartialBucket[];
     }
 
+    @DGet('/:id/stream', [ForceLoggedInMiddleware])
+    async stream(
+        @DPath('id') id: string,
+            @DRequest() req: any,
+            @DResponse() res: any,
+    ): Promise<ReadableStream> {
+        return await executeBucketRouteStreamHandler(req, res) as ReadableStream;
+    }
+
     @DPost('/:id/upload', [ForceLoggedInMiddleware])
     async upload(
         @DPath('id') id: string,

packages/storage-kit/src/domains/bucket-file/api.ts

@@ -5,6 +5,7 @@
  * view the LICENSE file that was distributed with this source code.
  */
 
+import type { Analysis } from '@privateaim/core/src';
 import type { BuildInput } from 'rapiq';
 import { buildQuery } from 'rapiq';
 import { nullifyEmptyObjectProperties } from '@privateaim/core';
@@ -39,4 +40,17 @@ export class BucketFileAPI extends BaseAPI {
         const response = await this.client.post(`bucket-files/${id}`, nullifyEmptyObjectProperties(data));
         return response.data;
     }
+
+    getStreamPath(id: Analysis['id']): string {
+        return `bucket-files/${id}/stream`;
+    }
+
+    async stream(id: BucketFile['id']) : Promise<ReadableStream<any>> {
+        const response = await this.client.request({
+            url: this.getStreamPath(id),
+            responseType: 'stream',
+        });
+
+        return response.data;
+    }
 }

packages/storage-kit/src/domains/bucket/api.ts

@@ -52,7 +52,16 @@ export class BucketAPI extends BaseAPI {
         return response.data;
     }
 
-    getDownloadPath(id: Analysis['id']): string {
-        return `buckets/${id}/files/download`;
+    getStreamPath(id: Analysis['id']): string {
+        return `buckets/${id}/stream`;
+    }
+
+    async stream(id: Bucket['id']) : Promise<ReadableStream<any>> {
+        const response = await this.client.request({
+            url: this.getStreamPath(id),
+            responseType: 'stream',
+        });
+
+        return response.data;
     }
 }