fix(core): Handle fs-capacitor readstream creation error (#3646)

packages/core/src/service/services/asset.service.ts

@@ -292,32 +292,23 @@ export class AssetService {
      * See the [Uploading Files docs](/guides/developer-guide/uploading-files) for an example of usage.
      */
     async create(ctx: RequestContext, input: CreateAssetInput): Promise<CreateAssetResult> {
-        return new Promise(async (resolve, reject) => {
-            const { createReadStream, filename, mimetype } = await input.file;
-            const stream = createReadStream();
-            stream.on('error', (err: any) => {
-                reject(err);
-            });
-            let result: Asset | MimeTypeError;
-            try {
-                result = await this.createAssetInternal(ctx, stream, filename, mimetype, input.customFields);
-            } catch (e: any) {
-                reject(e);
-                return;
-            }
-            if (isGraphQlErrorResult(result)) {
-                resolve(result);
-                return;
-            }
-            await this.customFieldRelationService.updateRelations(ctx, Asset, input, result);
-            if (input.tags) {
-                const tags = await this.tagService.valuesToTags(ctx, input.tags);
-                result.tags = tags;
-                await this.connection.getRepository(ctx, Asset).save(result);
-            }
-            await this.eventBus.publish(new AssetEvent(ctx, result, 'created', input));
-            resolve(result);
-        });
+        const { createReadStream, filename, mimetype } = await input.file;
+        const { stream, errorPromise } = this.makeStreamGuard(createReadStream);
+        const result = await Promise.race([
+            this.createAssetInternal(ctx, stream, filename, mimetype, input.customFields),
+            errorPromise,
+        ]);
+        if (isGraphQlErrorResult(result)) {
+            return result;
+        }
+        await this.customFieldRelationService.updateRelations(ctx, Asset, input, result);
+        if (input.tags) {
+            const tags = await this.tagService.valuesToTags(ctx, input.tags);
+            result.tags = tags;
+            await this.connection.getRepository(ctx, Asset).save(result);
+        }
+        await this.eventBus.publish(new AssetEvent(ctx, result, 'created', input));
+        return result;
     }
 
     /**
@@ -703,4 +694,24 @@ export class AssetService {
         });
         return { products, variants, collections };
     }
+
+    private makeStreamGuard(createReadStream: () => ReadStream): {
+        stream: ReadStream;
+        errorPromise: Promise<never>;
+    } {
+        let onReject: (err: unknown) => void;
+        const errorPromise = new Promise<never>((_, rej) => {
+            onReject = rej;
+        });
+
+        // `fs-capacitor`'s `createReadStream` can throw if its `WriteStream` has already been destroyed
+        // sync error so will bubble to consumer immediately
+        const stream = createReadStream();
+
+        stream.on('error', err => {
+            onReject(err);
+        });
+
+        return { stream, errorPromise };
+    }
 }