Merge pull request #63 from s1seven/62-fixfile-storage-allow-to-properly-handle-upload-error-with-s3

fix: file storage allow to properly handle upload error with S3

Author: getlarge

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 { finished, 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,12 @@ 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);
+    finished(writeStream, (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/src/file-storage.class.ts

@@ -1,8 +1,9 @@
 /* eslint-disable @typescript-eslint/no-unused-vars */
 import type { Request } from 'express';
-import type { Readable, Writable } from 'node:stream';
+import type { Readable } from 'node:stream';
 
 import { MethodTypes } from './constants';
+import { FileStorageWritable } from './types';
 
 // TODO: extend configuration
 export interface FileStorageConfig {
@@ -71,7 +72,7 @@ export abstract class FileStorage {
     args: FileStorageBaseArgs & {
       options?: string | any;
     },
-  ): Promise<Writable> {
+  ): Promise<FileStorageWritable> {
     throw new Error(defaultErrorMessage);
   }
 

packages/file-storage/src/file-storage.service.ts

@@ -1,5 +1,5 @@
 import { Injectable } from '@nestjs/common';
-import type { Readable, Writable } from 'node:stream';
+import type { Readable } from 'node:stream';
 
 import { InjectFileStorageStrategy } from './decorators';
 import { FileStorage, FileStorageBaseArgs, FileStorageDirBaseArgs } from './file-storage.class';
@@ -18,6 +18,7 @@ import {
   FileStorageS3UploadFile,
   FileStorageS3UploadStream,
 } from './file-storage-s3.class';
+import type { FileStorageWritable } from './types';
 
 @Injectable()
 export class FileStorageService implements Omit<FileStorage, 'transformFilePath'> {
@@ -31,7 +32,7 @@ export class FileStorageService implements Omit<FileStorage, 'transformFilePath'
     return this.fileStorage.uploadFile(args);
   }
 
-  uploadStream(args: FileStorageLocalUploadStream | FileStorageS3UploadStream): Promise<Writable> {
+  uploadStream(args: FileStorageLocalUploadStream | FileStorageS3UploadStream): Promise<FileStorageWritable> {
     return this.fileStorage.uploadStream(args);
   }
 

packages/file-storage/src/types.ts

@@ -1,4 +1,5 @@
 import { InjectionToken, ModuleMetadata } from '@nestjs/common';
+import { Writable } from 'node:stream';
 
 import { FileStorage, FileStorageConfigFactory } from './file-storage.class';
 import { FileStorageLocalSetup } from './file-storage-fs.class';
@@ -30,3 +31,15 @@ export interface FileStorageModuleAsyncOptions extends Pick<ModuleMetadata, 'imp
   // useExisting?: FileStorage;
   inject?: InjectionToken[];
 }
+
+interface WritableWithDoneEvent {
+  emit(event: 'done', error?: Error): boolean;
+  addListener(event: 'done', listener: (error?: Error) => void): this;
+  on(event: 'done', listener: (error?: Error) => void): this;
+  once(event: 'done', listener: (error?: Error) => void): this;
+  prependOnceListener(event: 'done', listener: (error?: Error) => void): this;
+  prependListener(event: 'done', listener: (error?: Error) => void): this;
+  removeListener(event: 'done', listener: () => void): this;
+}
+
+export type FileStorageWritable = Writable & WritableWithDoneEvent;

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';
@@ -110,9 +110,11 @@ testMap.forEach((testSuite) => {
     it('uploadStream uploads a file', async () => {
       const upload = await fileStorage.uploadStream({ filePath: testFileName });
       const entry = Readable.from(Buffer.from(testFileContent));
+      const ac = new AbortController();
+      const t = setTimeout(() => ac.abort(), 300);
+      const listener = once(upload, 'done', { signal: ac.signal });
       await pipeline(entry, upload);
-      // add delay, otherwise test is flaky
-      await new Promise<void>((resolve) => setTimeout(resolve, 100));
+      await listener.finally(() => clearTimeout(t));
       const result = await fileStorage.readDir({ dirPath });
       expect(result.length).toBe(1);
     });