chore: update implementation for AM import

Author: naman-contentstack

.talismanrc

@@ -1,24 +1,6 @@
 fileignoreconfig:
-  - filename: packages/contentstack-export/src/export/modules/assets.ts
-    checksum: 1eacc8e86cb50fe283febe6688965854f420e02cf1b49555a15661fa0c3e3c7a
-  - filename: packages/contentstack-import/src/utils/import-config-handler.ts
-    checksum: f831cef1b7c3bd97bdbc170cff452350cee0f448d97df02e25aa41d6c4d64ad3
   - filename: packages/contentstack-asset-management/src/import/asset-types.ts
-    checksum: a39caa373b2a736d1e57063326cfb2073ae78376efa931b27d2c7110997708a5
-  - filename: packages/contentstack-asset-management/src/export/spaces.ts
-    checksum: 3ec11c8f710b60ae495c69344025587df2e6195c872a0c82feaf04ac044ecefa
-  - filename: packages/contentstack-import/src/import/modules/assets.ts
-    checksum: d9f4a29a29e8b8a2a36e498f2380d39e1c5c0ec13ff894ef450abd817f2a646e
+    checksum: 78f9334477ccb4b59a2b14b09214377f2eaf271e2dc68d1976321910d3b4e91c
   - filename: packages/contentstack-asset-management/src/import/fields.ts
-    checksum: bbae69c28ec69bf67c2c7b4df3620380ef3fca488b3288e137b65a60ee738b9e
-  - filename: packages/contentstack-asset-management/src/import/spaces.ts
-    checksum: 79cf2f1b55523d28c218d970155f887255a00dc095a941556b709d1f19c6a8a0
-  - filename: packages/contentstack-asset-management/src/types/asset-management-api.ts
-    checksum: 716df03dcba70b2cc0f77b1f6338524553ba740080d7087a8699147c3ce8f0ba
-  - filename: packages/contentstack-asset-management/src/utils/asset-management-api-adapter.ts
-    checksum: 92bcad2feabc1954ead89b370d284b7af5f38ec1dca60a41752371977ef106ff
-  - filename: packages/contentstack-asset-management/src/import/base.ts
-    checksum: 513b55cf7e92bdbe8b815141822ba10579b6a728bec4424fc664794246ad33bb
-  - filename: packages/contentstack-asset-management/src/import/assets.ts
-    checksum: 2d34fa57f5ab269f6c535dff3242cc1135dbe1decd84fa0bc8997d0410d520b2
+    checksum: 2a079cb7b58c11cf69068a9373f4b9a871487bc7ad03d2bc1efb45a3c1e333be
 version: '1.0'

packages/contentstack-asset-management/src/constants/index.ts

@@ -1,6 +1,20 @@
 export const BATCH_SIZE = 50;
 export const CHUNK_FILE_SIZE_MB = 1;
 
+/** Default parallel AM API calls when import caller does not set apiConcurrency. */
+export const DEFAULT_AM_API_CONCURRENCY = 5;
+
+/**
+ * Mapper output paths — must stay aligned with contentstack-import `PATH_CONSTANTS`
+ * (`mapper` / `assets` / uid, url, space-uid file names).
+ */
+export const IMPORT_ASSETS_MAPPER_DIR_SEGMENTS = ['mapper', 'assets'] as const;
+export const IMPORT_ASSETS_MAPPER_FILES = {
+  UID_MAPPING: 'uid-mapping.json',
+  URL_MAPPING: 'url-mapping.json',
+  SPACE_UID_MAPPING: 'space-uid-mapping.json',
+} as const;
+
 /**
  * Main process name for Asset Management 2.0 export (single progress bar).
  * Use this when adding/starting the process and for all ticks.

packages/contentstack-asset-management/src/import/asset-types.ts

@@ -5,6 +5,7 @@ import { log } from '@contentstack/cli-utilities';
 import type { AssetManagementAPIConfig, ImportContext } from '../types/asset-management-api';
 import { AssetManagementImportAdapter } from './base';
 import { PROCESS_NAMES, PROCESS_STATUS } from '../constants/index';
+import { runInBatches } from '../utils/concurrent-batch';
 
 const STRIP_KEYS = ['created_at', 'created_by', 'updated_at', 'updated_by', 'is_system', 'category', 'preview_image_url', 'category_detail'];
 
@@ -50,6 +51,9 @@ export default class ImportAssetTypes extends AssetManagementImportAdapter {
 
     this.updateStatus(PROCESS_STATUS[PROCESS_NAMES.AM_IMPORT_ASSET_TYPES].IMPORTING, PROCESS_NAMES.AM_IMPORT_ASSET_TYPES);
 
+    type ToCreate = { uid: string; payload: Record<string, unknown> };
+    const toCreate: ToCreate[] = [];
+
     for (const assetType of items) {
       const uid = assetType.uid as string;
 
@@ -74,15 +78,23 @@ export default class ImportAssetTypes extends AssetManagementImportAdapter {
         continue;
       }
 
-      const payload = omit(assetType, STRIP_KEYS);
+      toCreate.push({ uid, payload: omit(assetType, STRIP_KEYS) as Record<string, unknown> });
+    }
+
+    await runInBatches(toCreate, this.apiConcurrency, async ({ uid, payload }) => {
       try {
         await this.createAssetType(payload as any);
         this.tick(true, `asset-type: ${uid}`, null, PROCESS_NAMES.AM_IMPORT_ASSET_TYPES);
         log.debug(`Imported asset type: ${uid}`, this.importContext.context);
       } catch (e) {
-        this.tick(false, `asset-type: ${uid}`, (e as Error)?.message ?? PROCESS_STATUS[PROCESS_NAMES.AM_IMPORT_ASSET_TYPES].FAILED, PROCESS_NAMES.AM_IMPORT_ASSET_TYPES);
+        this.tick(
+          false,
+          `asset-type: ${uid}`,
+          (e as Error)?.message ?? PROCESS_STATUS[PROCESS_NAMES.AM_IMPORT_ASSET_TYPES].FAILED,
+          PROCESS_NAMES.AM_IMPORT_ASSET_TYPES,
+        );
         log.debug(`Failed to import asset type ${uid}: ${e}`, this.importContext.context);
       }
-    }
+    });
   }
 }

packages/contentstack-asset-management/src/import/assets.ts

@@ -5,6 +5,7 @@ import { log } from '@contentstack/cli-utilities';
 import type { AssetManagementAPIConfig, ImportContext } from '../types/asset-management-api';
 import { AssetManagementImportAdapter } from './base';
 import { getArrayFromResponse } from '../utils/export-helpers';
+import { runInBatches } from '../utils/concurrent-batch';
 import { PROCESS_NAMES, PROCESS_STATUS } from '../constants/index';
 
 type FolderRecord = {
@@ -123,6 +124,14 @@ export default class ImportAssets extends AssetManagementImportAdapter {
     this.updateStatus(PROCESS_STATUS[PROCESS_NAMES.AM_IMPORT_ASSETS].IMPORTING, PROCESS_NAMES.AM_IMPORT_ASSETS);
     log.debug(`Uploading ${assetItems.length} asset(s) for space ${newSpaceUid}`, this.importContext.context);
 
+    type UploadJob = {
+      asset: AssetRecord;
+      filePath: string;
+      mappedParentUid: string | undefined;
+      oldUid: string;
+    };
+    const uploadJobs: UploadJob[] = [];
+
     for (const asset of assetItems) {
       const oldUid = asset.uid;
       const filename = asset.filename ?? asset.file_name ?? 'asset';
@@ -137,6 +146,11 @@ export default class ImportAssets extends AssetManagementImportAdapter {
       const assetParent = asset.parent_uid && asset.parent_uid !== 'root' ? asset.parent_uid : undefined;
       const mappedParentUid = assetParent ? folderUidMap[assetParent] ?? undefined : undefined;
 
+      uploadJobs.push({ asset, filePath, mappedParentUid, oldUid });
+    }
+
+    await runInBatches(uploadJobs, this.apiConcurrency, async ({ asset, filePath, mappedParentUid, oldUid }) => {
+      const filename = asset.filename ?? asset.file_name ?? 'asset';
       try {
         const { asset: created } = await this.uploadAsset(newSpaceUid, filePath, {
           title: asset.title ?? filename,
@@ -146,7 +160,6 @@ export default class ImportAssets extends AssetManagementImportAdapter {
 
         uidMap[oldUid] = created.uid;
 
-        // Map old AM direct URL → new AM direct URL.
         if (asset.url && created.url) {
           urlMap[asset.url] = created.url;
         }
@@ -162,7 +175,7 @@ export default class ImportAssets extends AssetManagementImportAdapter {
         );
         log.debug(`Failed to upload asset ${oldUid}: ${e}`, this.importContext.context);
       }
-    }
+    });
 
     return { uidMap, urlMap };
   }
@@ -181,24 +194,29 @@ export default class ImportAssets extends AssetManagementImportAdapter {
 
     while (remaining.length > 0 && remaining.length !== prevLength) {
       prevLength = remaining.length;
+      const ready: FolderRecord[] = [];
       const nextPass: FolderRecord[] = [];
 
       for (const folder of remaining) {
         const { parent_uid: parentUid } = folder;
-        // "root" is the AM API sentinel for a top-level folder
         const isRootParent = !parentUid || parentUid === 'root';
         const parentMapped = isRootParent || folderUidMap[parentUid] !== undefined;
 
         if (!parentMapped) {
           nextPass.push(folder);
-          continue;
+        } else {
+          ready.push(folder);
         }
+      }
 
+      await runInBatches(ready, this.apiConcurrency, async (folder) => {
+        const { parent_uid: parentUid } = folder;
+        const isRootParent = !parentUid || parentUid === 'root';
         try {
           const { folder: created } = await this.createFolder(newSpaceUid, {
             title: folder.title,
             description: folder.description,
-            parent_uid: isRootParent ? undefined : folderUidMap[parentUid],
+            parent_uid: isRootParent ? undefined : folderUidMap[parentUid!],
           });
           folderUidMap[folder.uid] = created.uid;
           this.tick(true, `folder: ${folder.uid}`, null, PROCESS_NAMES.AM_IMPORT_FOLDERS);
@@ -212,7 +230,7 @@ export default class ImportAssets extends AssetManagementImportAdapter {
           );
           log.debug(`Failed to create folder ${folder.uid}: ${e}`, this.importContext.context);
         }
-      }
+      });
 
       remaining = nextPass;
     }

packages/contentstack-asset-management/src/import/base.ts

@@ -3,7 +3,7 @@ import { FsUtility, log, CLIProgressManager, configHandler } from '@contentstack
 
 import type { AssetManagementAPIConfig, ImportContext } from '../types/asset-management-api';
 import { AssetManagementAdapter } from '../utils/asset-management-api-adapter';
-import { AM_MAIN_PROCESS_NAME } from '../constants/index';
+import { AM_MAIN_PROCESS_NAME, DEFAULT_AM_API_CONCURRENCY } from '../constants/index';
 
 export type { ImportContext };
 
@@ -61,6 +61,11 @@ export class AssetManagementImportAdapter extends AssetManagementAdapter {
     return this.importContext.spacesRootPath;
   }
 
+  /** Parallel AM API limit for import batches. */
+  protected get apiConcurrency(): number {
+    return this.importContext.apiConcurrency ?? DEFAULT_AM_API_CONCURRENCY;
+  }
+
   protected getAssetTypesDir(): string {
     return pResolve(this.importContext.spacesRootPath, 'asset_types');
   }

packages/contentstack-asset-management/src/import/fields.ts

@@ -5,6 +5,7 @@ import { log } from '@contentstack/cli-utilities';
 import type { AssetManagementAPIConfig, ImportContext } from '../types/asset-management-api';
 import { AssetManagementImportAdapter } from './base';
 import { PROCESS_NAMES, PROCESS_STATUS } from '../constants/index';
+import { runInBatches } from '../utils/concurrent-batch';
 
 const STRIP_KEYS = ['created_at', 'created_by', 'updated_at', 'updated_by', 'is_system', 'asset_types_count'];
 
@@ -50,6 +51,9 @@ export default class ImportFields extends AssetManagementImportAdapter {
 
     this.updateStatus(PROCESS_STATUS[PROCESS_NAMES.AM_IMPORT_FIELDS].IMPORTING, PROCESS_NAMES.AM_IMPORT_FIELDS);
 
+    type ToCreate = { uid: string; payload: Record<string, unknown> };
+    const toCreate: ToCreate[] = [];
+
     for (const field of items) {
       const uid = field.uid as string;
 
@@ -74,15 +78,23 @@ export default class ImportFields extends AssetManagementImportAdapter {
         continue;
       }
 
-      const payload = omit(field, STRIP_KEYS);
+      toCreate.push({ uid, payload: omit(field, STRIP_KEYS) as Record<string, unknown> });
+    }
+
+    await runInBatches(toCreate, this.apiConcurrency, async ({ uid, payload }) => {
       try {
         await this.createField(payload as any);
         this.tick(true, `field: ${uid}`, null, PROCESS_NAMES.AM_IMPORT_FIELDS);
         log.debug(`Imported field: ${uid}`, this.importContext.context);
       } catch (e) {
-        this.tick(false, `field: ${uid}`, (e as Error)?.message ?? PROCESS_STATUS[PROCESS_NAMES.AM_IMPORT_FIELDS].FAILED, PROCESS_NAMES.AM_IMPORT_FIELDS);
+        this.tick(
+          false,
+          `field: ${uid}`,
+          (e as Error)?.message ?? PROCESS_STATUS[PROCESS_NAMES.AM_IMPORT_FIELDS].FAILED,
+          PROCESS_NAMES.AM_IMPORT_FIELDS,
+        );
         log.debug(`Failed to import field ${uid}: ${e}`, this.importContext.context);
       }
-    }
+    });
   }
 }

packages/contentstack-asset-management/src/import/spaces.ts

@@ -1,5 +1,6 @@
 import { resolve as pResolve, join } from 'node:path';
-import { readdirSync, statSync } from 'node:fs';
+import { mkdirSync, readdirSync, statSync } from 'node:fs';
+import { writeFile } from 'node:fs/promises';
 import { log, CLIProgressManager, configHandler } from '@contentstack/cli-utilities';
 
 import type {
@@ -9,7 +10,11 @@ import type {
   ImportResult,
   SpaceMapping,
 } from '../types/asset-management-api';
-import { AM_MAIN_PROCESS_NAME } from '../constants/index';
+import {
+  AM_MAIN_PROCESS_NAME,
+  IMPORT_ASSETS_MAPPER_DIR_SEGMENTS,
+  IMPORT_ASSETS_MAPPER_FILES,
+} from '../constants/index';
 import { AssetManagementAdapter } from '../utils/asset-management-api-adapter';
 import ImportAssetTypes from './asset-types';
 import ImportFields from './fields';
@@ -34,7 +39,16 @@ export class ImportSpaces {
   }
 
   async start(): Promise<ImportResult> {
-    const { contentDir, assetManagementUrl, org_uid, apiKey, host, sourceApiKey, context } = this.options;
+    const {
+      contentDir,
+      assetManagementUrl,
+      org_uid,
+      apiKey,
+      host,
+      sourceApiKey,
+      context,
+      apiConcurrency,
+    } = this.options;
 
     const spacesRootPath = pResolve(contentDir, 'spaces');
 
@@ -45,6 +59,7 @@ export class ImportSpaces {
       host,
       org_uid,
       context,
+      apiConcurrency,
     };
 
     const apiConfig: AssetManagementAPIConfig = {
@@ -142,6 +157,19 @@ export class ImportSpaces {
         }
       }
 
+      if (this.options.backupDir) {
+        const mapperDir = join(this.options.backupDir, ...IMPORT_ASSETS_MAPPER_DIR_SEGMENTS);
+        mkdirSync(mapperDir, { recursive: true });
+        await writeFile(join(mapperDir, IMPORT_ASSETS_MAPPER_FILES.UID_MAPPING), JSON.stringify(allUidMap), 'utf8');
+        await writeFile(join(mapperDir, IMPORT_ASSETS_MAPPER_FILES.URL_MAPPING), JSON.stringify(allUrlMap), 'utf8');
+        await writeFile(
+          join(mapperDir, IMPORT_ASSETS_MAPPER_FILES.SPACE_UID_MAPPING),
+          JSON.stringify(allSpaceUidMap),
+          'utf8',
+        );
+        log.debug('Wrote AM 2.0 mapper files (uid, url, space-uid)', context);
+      }
+
       progress.completeProcess(AM_MAIN_PROCESS_NAME, !hasFailures);
       log.debug('Asset Management 2.0 import completed', context);
     } catch (err) {

packages/contentstack-asset-management/src/types/asset-management-api.ts

@@ -171,6 +171,11 @@ export type ImportContext = {
   org_uid: string;
   /** Optional logging context (same shape as ExportConfig.context). */
   context?: Record<string, unknown>;
+  /**
+   * Max parallel AM API calls for import (fields, asset types, folders batch, uploads).
+   * Set from `AssetManagementImportOptions.apiConcurrency`.
+   */
+  apiConcurrency?: number;
 };
 
 /**
@@ -191,6 +196,12 @@ export type AssetManagementImportOptions = {
   sourceApiKey?: string;
   /** Optional logging context. */
   context?: Record<string, unknown>;
+  /**
+   * When set, mapper files are written under `{backupDir}/mapper/assets/` after import.
+   */
+  backupDir?: string;
+  /** Parallel AM API limit; defaults to package constant when omitted. */
+  apiConcurrency?: number;
 };
 
 /**
@@ -206,8 +217,8 @@ export type SpaceMapping = {
 
 /**
  * The value returned by `ImportSpaces.start()`.
- * Written to `mapper/assets/uid-mapping.json` and `mapper/assets/url-mapping.json`
- * by the bridge module so `entries.ts` can resolve asset references.
+ * When `backupDir` is set on options, the AM package also writes these maps under
+ * `mapper/assets/` for `entries.ts` to resolve asset references.
  */
 export type ImportResult = {
   uidMap: Record<string, string>;

packages/contentstack-asset-management/src/utils/concurrent-batch.ts

@@ -0,0 +1,34 @@
+/**
+ * Split an array into chunks of at most `size` elements.
+ */
+export function chunkArray<T>(items: T[], size: number): T[][] {
+  if (size <= 0) {
+    return items.length ? [items] : [];
+  }
+  const out: T[][] = [];
+  for (let i = 0; i < items.length; i += size) {
+    out.push(items.slice(i, i + size));
+  }
+  return out;
+}
+
+/**
+ * Run async work in batches of at most `concurrency` parallel tasks at a time.
+ * Uses Promise.allSettled per batch so one failure does not abort the batch.
+ */
+export async function runInBatches<T>(
+  items: T[],
+  concurrency: number,
+  fn: (item: T, index: number) => Promise<void>,
+): Promise<void> {
+  if (items.length === 0) {
+    return;
+  }
+  const limit = Math.max(1, concurrency);
+  const batches = chunkArray(items, limit);
+  let offset = 0;
+  for (const batch of batches) {
+    await Promise.allSettled(batch.map((item, j) => fn(item, offset + j)));
+    offset += batch.length;
+  }
+}

packages/contentstack-asset-management/src/utils/index.ts

@@ -6,3 +6,4 @@ export {
   getReadableStreamFromDownloadResponse,
   writeStreamToFile,
 } from './export-helpers';
+export { chunkArray, runInBatches } from './concurrent-batch';