feat(core): Implement AssetImportStrategy, enable asset import from urls

docs/content/developer-guide/importing-product-data.md

@@ -35,7 +35,7 @@ Here's an explanation of each column:
 * `name`: The name of the product. Rows with an empty "name" are interpreted as variants of the preceeding product row.
 * `slug`: The product's slug. Can be omitted, in which case will be generated from the name.
 * `description`: The product description.
-* `assets`: One or more asset file names separated by the pipe (`|`) character. The files must be located on the local file system, and the path is interpreted as being relative to the [`importAssetsDir`]({{< relref "/docs/typescript-api/import-export/import-export-options" >}}#importassetsdir) as defined in the VendureConfig. The first asset will be set as the featuredAsset.
+* `assets`: One or more asset file names separated by the pipe (`|`) character. The files can be located on the local file system, in which case the path is interpreted as being relative to the [`importAssetsDir`]({{< relref "/docs/typescript-api/import-export/import-export-options" >}}#importassetsdir) as defined in the VendureConfig. Files can also be urls which will be fetched from a remote http/https url. If you need more control over how assets are imported, you can implement a custom [AssetImportStrategy]({{< relref "asset-import-strategy" >}}). The first asset will be set as the featuredAsset.
 * `facets`: One or more facets to apply to the product separated by the pipe (`|`) character. A facet has the format `<facet-name>:<facet-value>`.
 * `optionGroups`: OptionGroups define what variants make up the product. Applies only to products with more than one variant. 
 * `optionValues`: For each optionGroup defined, a corresponding value must be specified for each variant. Applies only to products with more than one variant.

packages/core/e2e/import.e2e-spec.ts

@@ -1,7 +1,9 @@
 import { omit } from '@vendure/common/lib/omit';
 import { User } from '@vendure/core';
 import { createTestEnvironment } from '@vendure/testing';
+import * as fs from 'fs';
 import gql from 'graphql-tag';
+import http from 'http';
 import path from 'path';
 
 import { initialData } from '../../../e2e-common/e2e-initial-data';
@@ -407,4 +409,97 @@ describe('Import resolver', () => {
         // Import localeString custom fields
         expect(paperStretcher.customFields.localName).toEqual('纸张拉伸器');
     }, 20000);
+
+    describe('asset urls', () => {
+        let staticServer: http.Server;
+
+        beforeAll(() => {
+            // Set up minimal static file server
+            staticServer = http
+                .createServer((req, res) => {
+                    const filePath = path.join(__dirname, 'fixtures/assets', req?.url ?? '');
+                    fs.readFile(filePath, (err, data) => {
+                        if (err) {
+                            res.writeHead(404);
+                            res.end(JSON.stringify(err));
+                            return;
+                        }
+                        res.writeHead(200);
+                        res.end(data);
+                    });
+                })
+                .listen(3456);
+        });
+
+        afterAll(() => {
+            if (staticServer) {
+                return new Promise<void>((resolve, reject) => {
+                    staticServer.close(err => {
+                        if (err) {
+                            reject(err);
+                        } else {
+                            resolve();
+                        }
+                    });
+                });
+            }
+        });
+
+        it('imports assets with url paths', async () => {
+            const timeout = process.env.CI ? 2000 : 1000;
+            await new Promise(resolve => {
+                setTimeout(resolve, timeout);
+            });
+
+            const csvFile = path.join(__dirname, 'fixtures', 'e2e-product-import-asset-urls.csv');
+            const result = await adminClient.fileUploadMutation({
+                mutation: gql`
+                    mutation ImportProducts($csvFile: Upload!) {
+                        importProducts(csvFile: $csvFile) {
+                            imported
+                            processed
+                            errors
+                        }
+                    }
+                `,
+                filePaths: [csvFile],
+                mapVariables: () => ({ csvFile: null }),
+            });
+
+            expect(result.importProducts.errors).toEqual([]);
+            expect(result.importProducts.imported).toBe(1);
+            expect(result.importProducts.processed).toBe(1);
+
+            const productResult = await adminClient.query(
+                gql`
+                    query GetProducts($options: ProductListOptions) {
+                        products(options: $options) {
+                            totalItems
+                            items {
+                                id
+                                name
+                                featuredAsset {
+                                    id
+                                    name
+                                    preview
+                                }
+                            }
+                        }
+                    }
+                `,
+                {
+                    options: {
+                        filter: {
+                            name: { contains: 'guitar' },
+                        },
+                    },
+                },
+            );
+
+            expect(productResult.products.items.length).toBe(1);
+            expect(productResult.products.items[0].featuredAsset.preview).toBe(
+                'test-url/test-assets/guitar__preview.png',
+            );
+        });
+    });
 });

packages/core/src/config/asset-import-strategy/asset-import-strategy.ts

@@ -0,0 +1,25 @@
+import { Readable } from 'stream';
+
+import { InjectableStrategy } from '../../common/types/injectable-strategy';
+
+/**
+ * @description
+ * The AssetImportStrategy determines how asset files get imported based on the path given in the
+ * import CSV or via the {@link AssetImporter} `getAssets()` method.
+ *
+ * The {@link DefaultAssetImportStrategy} is able to load files from either the local filesystem
+ * or from a remote URL.
+ *
+ * A custom strategy could be created which could e.g. get the asset file from an S3 bucket.
+ *
+ * @since 1.7.0
+ * @docsCategory import-export
+ */
+export interface AssetImportStrategy extends InjectableStrategy {
+    /**
+     * @description
+     * Given an asset path, this method should return a Stream of file data. This could
+     * e.g. be read from a file system or fetch from a remote location.
+     */
+    getStreamFromPath(assetPath: string): Readable | Promise<Readable>;
+}

packages/core/src/config/asset-import-strategy/default-asset-import-strategy.ts

@@ -0,0 +1,106 @@
+import fs from 'fs-extra';
+import http from 'http';
+import https from 'https';
+import path from 'path';
+import { from } from 'rxjs';
+import { delay, retryWhen, take, tap } from 'rxjs/operators';
+import { Readable } from 'stream';
+import { URL } from 'url';
+
+import { Injector } from '../../common/index';
+import { ConfigService } from '../config.service';
+import { Logger } from '../logger/vendure-logger';
+
+import { AssetImportStrategy } from './asset-import-strategy';
+
+function fetchUrl(urlString: string): Promise<Readable> {
+    return new Promise((resolve, reject) => {
+        const url = new URL(urlString);
+        const get = url.protocol.startsWith('https') ? https.get : http.get;
+        get(
+            url,
+            {
+                timeout: 5000,
+            },
+            res => {
+                const { statusCode } = res;
+                if (statusCode !== 200) {
+                    Logger.error(`Failed to fetch "${urlString.substr(0, 100)}", statusCode: ${statusCode}`);
+                    reject(new Error(`Request failed. Status code: ${statusCode}`));
+                } else {
+                    resolve(res);
+                }
+            },
+        );
+    });
+}
+
+/**
+ * @description
+ * The DefaultAssetImportStrategy is able to import paths from the local filesystem (taking into account the
+ * `importExportOptions.importAssetsDir` setting) as well as remote http/https urls.
+ *
+ * @since 1.7.0
+ * @docsCategory import-export
+ */
+export class DefaultAssetImportStrategy implements AssetImportStrategy {
+    private configService: ConfigService;
+
+    constructor(
+        private options?: {
+            retryDelayMs: number;
+            retryCount: number;
+        },
+    ) {}
+
+    init(injector: Injector) {
+        this.configService = injector.get(ConfigService);
+    }
+
+    getStreamFromPath(assetPath: string) {
+        if (/^https?:\/\//.test(assetPath)) {
+            return this.getStreamFromUrl(assetPath);
+        } else {
+            return this.getStreamFromLocalFile(assetPath);
+        }
+    }
+
+    private getStreamFromUrl(assetUrl: string): Promise<Readable> {
+        const { retryCount, retryDelayMs } = this.options ?? {};
+        return from(fetchUrl(assetUrl))
+            .pipe(
+                retryWhen(errors =>
+                    errors.pipe(
+                        tap(value => {
+                            Logger.verbose(value);
+                            Logger.verbose(`DefaultAssetImportStrategy: retrying fetchUrl for ${assetUrl}`);
+                        }),
+                        delay(retryDelayMs ?? 200),
+                        take(retryCount ?? 3),
+                    ),
+                ),
+            )
+            .toPromise();
+    }
+
+    private getStreamFromLocalFile(assetPath: string): Readable {
+        const { importAssetsDir } = this.configService.importExportOptions;
+        const filename = path.join(importAssetsDir, assetPath);
+
+        if (fs.existsSync(filename)) {
+            const fileStat = fs.statSync(filename);
+            if (fileStat.isFile()) {
+                try {
+                    const stream = fs.createReadStream(filename);
+                    return stream;
+                } catch (err) {
+                    throw err;
+                }
+            } else {
+                throw new Error(`Could not find file "${filename}"`);
+            }
+        } else {
+            throw new Error(`File "${filename}" does not exist`);
+        }
+    }
+}

packages/core/src/config/config.module.ts

@@ -83,6 +83,7 @@ export class ConfigModule implements OnApplicationBootstrap, OnApplicationShutdo
         const { entityIdStrategy: entityIdStrategyDeprecated } = this.configService;
         const { entityIdStrategy } = this.configService.entityOptions;
         const { healthChecks } = this.configService.systemOptions;
+        const { assetImportStrategy } = this.configService.importExportOptions;
         return [
             ...adminAuthenticationStrategy,
             ...shopAuthenticationStrategy,
@@ -109,6 +110,7 @@ export class ConfigModule implements OnApplicationBootstrap, OnApplicationShutdo
             stockAllocationStrategy,
             stockDisplayStrategy,
             ...healthChecks,
+            assetImportStrategy,
         ];
     }
 

packages/core/src/config/default-config.ts

@@ -5,9 +5,11 @@ import {
     SUPER_ADMIN_USER_PASSWORD,
 } from '@vendure/common/lib/shared-constants';
 
+import { TypeORMHealthCheckStrategy } from '../health-check/typeorm-health-check-strategy';
 import { InMemoryJobQueueStrategy } from '../job-queue/in-memory-job-queue-strategy';
 import { InMemoryJobBufferStorageStrategy } from '../job-queue/job-buffer/in-memory-job-buffer-storage-strategy';
 
+import { DefaultAssetImportStrategy } from './asset-import-strategy/default-asset-import-strategy';
 import { DefaultAssetNamingStrategy } from './asset-naming-strategy/default-asset-naming-strategy';
 import { NoAssetPreviewStrategy } from './asset-preview-strategy/no-asset-preview-strategy';
 import { NoAssetStorageStrategy } from './asset-storage-strategy/no-asset-storage-strategy';
@@ -32,7 +34,6 @@ import { defaultPromotionActions, defaultPromotionConditions } from './promotion
 import { InMemorySessionCacheStrategy } from './session-cache/in-memory-session-cache-strategy';
 import { defaultShippingCalculator } from './shipping-method/default-shipping-calculator';
 import { defaultShippingEligibilityChecker } from './shipping-method/default-shipping-eligibility-checker';
-import { TypeORMHealthCheckStrategy } from '../health-check/typeorm-health-check-strategy';
 import { DefaultTaxLineCalculationStrategy } from './tax/default-tax-line-calculation-strategy';
 import { DefaultTaxZoneStrategy } from './tax/default-tax-zone-strategy';
 import { RuntimeVendureConfig } from './vendure-config';
@@ -148,6 +149,7 @@ export const defaultConfig: RuntimeVendureConfig = {
     },
     importExportOptions: {
         importAssetsDir: __dirname,
+        assetImportStrategy: new DefaultAssetImportStrategy(),
     },
     jobQueueOptions: {
         jobQueueStrategy: new InMemoryJobQueueStrategy(),

packages/core/src/config/index.ts

@@ -1,3 +1,4 @@
+export * from './asset-import-strategy/asset-import-strategy';
 export * from './asset-naming-strategy/asset-naming-strategy';
 export * from './asset-naming-strategy/default-asset-naming-strategy';
 export * from './asset-preview-strategy/asset-preview-strategy';

packages/core/src/config/vendure-config.ts

@@ -9,6 +9,7 @@ import { Middleware } from '../common';
 import { PermissionDefinition } from '../common/permission-definition';
 import { JobBufferStorageStrategy } from '../job-queue/job-buffer/job-buffer-storage-strategy';
 
+import { AssetImportStrategy } from './asset-import-strategy/asset-import-strategy';
 import { AssetNamingStrategy } from './asset-naming-strategy/asset-naming-strategy';
 import { AssetPreviewStrategy } from './asset-preview-strategy/asset-preview-strategy';
 import { AssetStorageStrategy } from './asset-storage-strategy/asset-storage-strategy';
@@ -773,6 +774,14 @@ export interface ImportExportOptions {
      * @default __dirname
      */
     importAssetsDir?: string;
+    /**
+     * @description
+     * This strategy determines how asset files get imported based on the path given in the
+     * import CSV or via the {@link AssetImporter} `getAssets()` method.
+     *
+     * @since 1.7.0
+     */
+    assetImportStrategy?: AssetImportStrategy;
 }
 
 /**

packages/core/src/data-import/providers/asset-importer/asset-importer.ts

@@ -1,8 +1,8 @@
 import { Injectable } from '@nestjs/common';
-import fs from 'fs-extra';
 import path from 'path';
 
 import { RequestContext } from '../../../api/index';
+import { isGraphQlErrorResult } from '../../../common/index';
 import { ConfigService } from '../../../config/config.service';
 import { Asset } from '../../../entity/asset/asset.entity';
 import { AssetService } from '../../../service/services/asset.service';
@@ -33,32 +33,26 @@ export class AssetImporter {
     ): Promise<{ assets: Asset[]; errors: string[] }> {
         const assets: Asset[] = [];
         const errors: string[] = [];
-        const { importAssetsDir } = this.configService.importExportOptions;
+        const { assetImportStrategy } = this.configService.importExportOptions;
         const uniqueAssetPaths = new Set(assetPaths);
         for (const assetPath of uniqueAssetPaths.values()) {
             const cachedAsset = this.assetMap.get(assetPath);
             if (cachedAsset) {
                 assets.push(cachedAsset);
             } else {
-                const filename = path.join(importAssetsDir, assetPath);
-
-                if (fs.existsSync(filename)) {
-                    const fileStat = fs.statSync(filename);
-                    if (fileStat.isFile()) {
-                        try {
-                            const stream = fs.createReadStream(filename);
-                            const asset = (await this.assetService.createFromFileStream(
-                                stream,
-                                ctx,
-                            )) as Asset;
-                            this.assetMap.set(assetPath, asset);
-                            assets.push(asset);
-                        } catch (err) {
-                            errors.push(err.toString());
+                try {
+                    const stream = await assetImportStrategy.getStreamFromPath(assetPath);
+                    if (stream) {
+                        const asset = await this.assetService.createFromFileStream(stream, assetPath, ctx);
+                        if (isGraphQlErrorResult(asset)) {
+                            errors.push(asset.message);
+                        } else {
+                            this.assetMap.set(assetPath, asset as Asset);
+                            assets.push(asset as Asset);
                         }
                     }
-                } else {
-                    errors.push(`File "${filename}" does not exist`);
+                } catch (e: any) {
+                    errors.push(e.message);
                 }
             }
         }

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

@@ -17,6 +17,7 @@ import { notNullOrUndefined } from '@vendure/common/lib/shared-utils';
 import { unique } from '@vendure/common/lib/unique';
 import { ReadStream as FSReadStream } from 'fs';
 import { ReadStream } from 'fs-extra';
+import { IncomingMessage } from 'http';
 import mime from 'mime-types';
 import path from 'path';
 import { Readable, Stream } from 'stream';
@@ -433,26 +434,46 @@ export class AssetService {
      * Create an Asset from a file stream, for example to create an Asset during data import.
      */
     async createFromFileStream(stream: ReadStream, ctx?: RequestContext): Promise<CreateAssetResult>;
-    async createFromFileStream(stream: Readable, filePath: string): Promise<CreateAssetResult>;
+    async createFromFileStream(
+        stream: Readable,
+        filePath: string,
+        ctx?: RequestContext,
+    ): Promise<CreateAssetResult>;
     async createFromFileStream(
         stream: ReadStream | Readable,
         maybeFilePathOrCtx?: string | RequestContext,
+        maybeCtx?: RequestContext,
     ): Promise<CreateAssetResult> {
+        const { assetImportStrategy } = this.configService.importExportOptions;
         const filePathFromArgs =
             maybeFilePathOrCtx instanceof RequestContext ? undefined : maybeFilePathOrCtx;
         const filePath =
             stream instanceof ReadStream || stream instanceof FSReadStream ? stream.path : filePathFromArgs;
         if (typeof filePath === 'string') {
-            const filename = path.basename(filePath);
-            const mimetype = mime.lookup(filename) || 'application/octet-stream';
+            const filename = path.basename(filePath).split('?')[0];
+            const mimetype = this.getMimeType(stream, filename);
             const ctx =
-                maybeFilePathOrCtx instanceof RequestContext ? maybeFilePathOrCtx : RequestContext.empty();
+                maybeFilePathOrCtx instanceof RequestContext
+                    ? maybeFilePathOrCtx
+                    : maybeCtx instanceof RequestContext
+                    ? maybeCtx
+                    : RequestContext.empty();
             return this.createAssetInternal(ctx, stream, filename, mimetype);
         } else {
             throw new InternalServerError(`error.path-should-be-a-string-got-buffer`);
         }
     }
 
+    private getMimeType(stream: Readable, filename: string): string {
+        if (stream instanceof IncomingMessage) {
+            const contentType = stream.headers['content-type'];
+            if (contentType) {
+                return contentType;
+            }
+        }
+        return mime.lookup(filename) || 'application/octet-stream';
+    }
+
     /**
      * @description
      * Unconditionally delete given assets.