feat(file-storage): emit 'done' event once file is fully written

packages/file-storage/src/file-storage-fs.class.ts

@@ -13,7 +13,7 @@ import {
 } from 'fs';
 import { readdir, readFile, rm, writeFile } from 'node:fs/promises';
 import { resolve as resolvePath } from 'node:path';
-import { Readable, Writable } from 'node:stream';
+import { Readable } from 'node:stream';
 
 import { MethodTypes } from './constants';
 import {
@@ -23,6 +23,7 @@ import {
   FileStorageConfigFactory,
   FileStorageDirBaseArgs,
 } from './file-storage.class';
+import { FileStorageWritable } from './types';
 
 export type StreamOptions = {
   flags?: string;
@@ -115,10 +116,13 @@ export class FileStorageLocal implements FileStorage {
     return writeFile(fileName, content, options);
   }
 
-  async uploadStream(args: FileStorageLocalUploadStream): Promise<Writable> {
+  async uploadStream(args: FileStorageLocalUploadStream): Promise<FileStorageWritable> {
     const { filePath, options, request } = args;
     const fileName = await this.transformFilePath(filePath, MethodTypes.WRITE, request, options);
-    return createWriteStream(fileName, options);
+    const writeStream = createWriteStream(fileName, options);
+    writeStream.prependOnceListener('finish', () => writeStream.emit('done'));
+    writeStream.prependOnceListener('error', (err) => writeStream.emit('done', err));
+    return writeStream;
   }
 
   downloadFile(args: {

packages/file-storage/src/file-storage-s3.class.ts

@@ -9,7 +9,7 @@ import {
 } from '@aws-sdk/client-s3';
 import { Upload } from '@aws-sdk/lib-storage';
 import type { Request } from 'express';
-import { PassThrough, Readable, Writable } from 'node:stream';
+import { PassThrough, Readable } from 'node:stream';
 
 import { MethodTypes } from './constants';
 import {
@@ -19,6 +19,7 @@ import {
   FileStorageConfigFactory,
   FileStorageDirBaseArgs,
 } from './file-storage.class';
+import { FileStorageWritable } from './types';
 
 /**
  * Either region or endpoint must be provided
@@ -140,7 +141,7 @@ export class FileStorageS3 implements FileStorage {
     }).done();
   }
 
-  async uploadStream(args: FileStorageS3UploadStream): Promise<Writable> {
+  async uploadStream(args: FileStorageS3UploadStream): Promise<FileStorageWritable> {
     const { filePath, options = {}, request } = args;
     const Key = await this.transformFilePath(filePath, MethodTypes.WRITE, request, options);
     const { s3, bucket: Bucket } = this.config;
@@ -155,8 +156,11 @@ export class FileStorageS3 implements FileStorage {
       },
     })
       .done()
+      .then(() => {
+        writeStream.emit('done');
+      })
       .catch((err) => {
-        writeStream.destroy(err);
+        writeStream.emit('done', err);
       });
     return writeStream;
   }

packages/file-storage/test/file-storage.e2e-spec.ts

@@ -4,7 +4,7 @@ import { Test, TestingModule } from '@nestjs/testing';
 import * as dotenv from 'dotenv';
 import { mkdir, rm } from 'node:fs/promises';
 import { resolve } from 'node:path';
-import { Readable } from 'node:stream';
+import { once, Readable } from 'node:stream';
 import { pipeline } from 'node:stream/promises';
 
 import { FileStorage, FileStorageModule, FileStorageModuleOptions, StorageType } from '../src';
@@ -111,8 +111,9 @@ testMap.forEach((testSuite) => {
       const upload = await fileStorage.uploadStream({ filePath: testFileName });
       const entry = Readable.from(Buffer.from(testFileContent));
       await pipeline(entry, upload);
-      // add delay, otherwise test is flaky
-      await new Promise<void>((resolve) => setTimeout(resolve, 100));
+      const ac = new AbortController();
+      const t = setTimeout(() => ac.abort(), 200);
+      await once(upload, 'done', { signal: ac.signal }).finally(() => clearTimeout(t));
       const result = await fileStorage.readDir({ dirPath });
       expect(result.length).toBe(1);
     });