feat: add concurrency in export and update logs in AM

.talismanrc

@@ -1,4 +1,4 @@
 fileignoreconfig:
   - filename: pnpm-lock.yaml
-    checksum: 8b5a2f43585d3191cdc71ad611f50c94b6d13fb7442cf4218ee0851a068af178
+    checksum: 7a2d08a029dd995917883504dd816fc7a579aca7d3e39fc5368959f0e766c7b2
 version: '1.0'

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

@@ -13,15 +13,17 @@ export default class ExportAssetTypes extends AssetManagementExportAdapter {
 
   async start(spaceUid: string): Promise<void> {
     await this.init();
+
+    log.debug('Starting shared asset types export process...', this.exportContext.context);
+
     const assetTypesData = await this.getWorkspaceAssetTypes(spaceUid);
     const items = getArrayFromResponse(assetTypesData, 'asset_types');
     const dir = this.getAssetTypesDir();
-    log.debug(
-      items.length === 0
-        ? 'No asset types, wrote empty asset-types'
-        : `Writing ${items.length} shared asset types`,
-      this.exportContext.context,
-    );
+    if (items.length === 0) {
+      log.info('No asset types to export, writing empty asset-types', this.exportContext.context);
+    } else {
+      log.debug(`Writing ${items.length} shared asset types`, this.exportContext.context);
+    }
     await this.writeItemsToChunkedJson(dir, 'asset-types.json', 'asset_types', ['uid', 'title', 'category', 'file_extension'], items);
     this.tick(true, PROCESS_NAMES.AM_ASSET_TYPES, null);
   }

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

@@ -7,6 +7,7 @@ import type { AssetManagementAPIConfig, LinkedWorkspace } from '../types/asset-m
 import type { ExportContext } from '../types/export-types';
 import { AssetManagementExportAdapter } from './base';
 import { getAssetItems, writeStreamToFile } from '../utils/export-helpers';
+import { runInBatches } from '../utils/concurrent-batch';
 import { PROCESS_NAMES, PROCESS_STATUS } from '../constants/index';
 
 export default class ExportAssets extends AssetManagementExportAdapter {
@@ -16,8 +17,14 @@ export default class ExportAssets extends AssetManagementExportAdapter {
 
   async start(workspace: LinkedWorkspace, spaceDir: string): Promise<void> {
     await this.init();
+
+    log.debug(`Starting assets export for space ${workspace.space_uid}`, this.exportContext.context);
+    log.info(`Exporting asset folders, metadata, and files for space ${workspace.space_uid}`, this.exportContext.context);
+
     const assetsDir = pResolve(spaceDir, 'assets');
     await mkdir(assetsDir, { recursive: true });
+    log.debug(`Assets directory ready: ${assetsDir}`, this.exportContext.context);
+
     log.debug(`Fetching folders and assets for space ${workspace.space_uid}`, this.exportContext.context);
 
     const [folders, assetsData] = await Promise.all([
@@ -43,37 +50,61 @@ export default class ExportAssets extends AssetManagementExportAdapter {
       ['uid', 'url', 'filename', 'file_name', 'parent_uid'],
       assetItems,
     );
+    log.debug(
+      `Finished writing chunked assets metadata (${assetItems.length} item(s)) under ${assetsDir}`,
+      this.exportContext.context,
+    );
+    log.info(
+      assetItems.length === 0
+        ? `Wrote empty asset metadata for space ${workspace.space_uid}`
+        : `Wrote ${assetItems.length} asset metadata record(s) for space ${workspace.space_uid}`,
+      this.exportContext.context,
+    );
     this.tick(true, `assets: ${workspace.space_uid} (${assetItems.length})`, null);
 
+    log.debug(`Starting binary downloads for space ${workspace.space_uid}`, this.exportContext.context);
     await this.downloadWorkspaceAssets(assetsData, assetsDir, workspace.space_uid);
   }
 
-  private async downloadWorkspaceAssets(
-    assetsData: unknown,
-    assetsDir: string,
-    spaceUid: string,
-  ): Promise<void> {
+  private async downloadWorkspaceAssets(assetsData: unknown, assetsDir: string, spaceUid: string): Promise<void> {
     const items = getAssetItems(assetsData);
     if (items.length === 0) {
+      log.info(`No asset files to download for space ${spaceUid}`, this.exportContext.context);
       log.debug('No assets to download', this.exportContext.context);
       return;
     }
 
     this.updateStatus(PROCESS_STATUS[PROCESS_NAMES.AM_DOWNLOADS].DOWNLOADING);
+    log.info(`Downloading asset files for space ${spaceUid} (${items.length} in metadata)`, this.exportContext.context);
     log.debug(`Downloading ${items.length} asset file(s) for space ${spaceUid}...`, this.exportContext.context);
     const filesDir = pResolve(assetsDir, 'files');
     await mkdir(filesDir, { recursive: true });
+    log.debug(`Asset files directory ready: ${filesDir}`, this.exportContext.context);
 
     const securedAssets = this.exportContext.securedAssets ?? false;
     const authtoken = securedAssets ? configHandler.get('authtoken') : null;
+    log.debug(
+      `Asset downloads: securedAssets=${securedAssets}, concurrency=${this.downloadAssetsBatchConcurrency}`,
+      this.exportContext.context,
+    );
     let lastError: string | null = null;
     let allSuccess = true;
+    let downloadOk = 0;
+    let downloadFail = 0;
 
-    for (const asset of items) {
+    const validItems = items.filter((asset) => Boolean(asset.url && (asset.uid ?? asset._uid)));
+    const skipped = items.length - validItems.length;
+    if (skipped > 0) {
+      log.debug(
+        `Skipping ${skipped} asset row(s) without url or uid (${validItems.length} file download(s) scheduled)`,
+        this.exportContext.context,
+      );
+    }
+    await runInBatches(validItems, this.downloadAssetsBatchConcurrency, async (asset) => {
       const uid = asset.uid ?? asset._uid;
       const url = asset.url;
       const filename = asset.filename ?? asset.file_name ?? 'asset';
-      if (!url || !uid) continue;
+      if (!url || !uid) return;
       try {
         const separator = url.includes('?') ? '&' : '?';
         const downloadUrl = securedAssets && authtoken ? `${url}${separator}authtoken=${authtoken}` : url;
@@ -86,15 +117,26 @@ export default class ExportAssets extends AssetManagementExportAdapter {
         await mkdir(assetFolderPath, { recursive: true });
         const filePath = pResolve(assetFolderPath, filename);
         await writeStreamToFile(nodeStream, filePath);
-        log.debug(`Downloaded asset ${uid}`, this.exportContext.context);
+        downloadOk += 1;
+        log.debug(`Downloaded asset ${uid} → ${filePath}`, this.exportContext.context);
       } catch (e) {
         allSuccess = false;
+        downloadFail += 1;
         lastError = (e as Error)?.message ?? PROCESS_STATUS[PROCESS_NAMES.AM_DOWNLOADS].FAILED;
         log.debug(`Failed to download asset ${uid}: ${e}`, this.exportContext.context);
       }
-    }
+    });
 
     this.tick(allSuccess, `downloads: ${spaceUid}`, lastError);
-    log.debug('Asset downloads completed', this.exportContext.context);
+    log.info(
+      allSuccess
+        ? `Finished downloading ${downloadOk} asset file(s) for space ${spaceUid}`
+        : `Asset downloads for space ${spaceUid} completed with errors: ${downloadOk} succeeded, ${downloadFail} failed`,
+      this.exportContext.context,
+    );
+    log.debug(
+      `Asset downloads finished for space ${spaceUid}: ok=${downloadOk}, failed=${downloadFail}, allSuccess=${allSuccess}`,
+      this.exportContext.context,
+    );
   }
 }

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

@@ -5,7 +5,7 @@ import { FsUtility, log, CLIProgressManager, configHandler } from '@contentstack
 import type { AssetManagementAPIConfig } from '../types/asset-management-api';
 import type { ExportContext } from '../types/export-types';
 import { AssetManagementAdapter } from '../utils/asset-management-api-adapter';
-import { AM_MAIN_PROCESS_NAME, FALLBACK_AM_CHUNK_FILE_SIZE_MB } from '../constants/index';
+import { AM_MAIN_PROCESS_NAME, FALLBACK_AM_API_CONCURRENCY, FALLBACK_AM_CHUNK_FILE_SIZE_MB } from '../constants/index';
 
 export type { ExportContext };
 
@@ -63,6 +63,16 @@ export class AssetManagementExportAdapter extends AssetManagementAdapter {
     return this.exportContext.spacesRootPath;
   }
 
+  /** Parallel AM export limit for bootstrap and default batch operations. */
+  protected get apiConcurrency(): number {
+    return this.exportContext.apiConcurrency ?? FALLBACK_AM_API_CONCURRENCY;
+  }
+
+  /** Asset download batch size; falls back to {@link apiConcurrency}. */
+  protected get downloadAssetsBatchConcurrency(): number {
+    return this.exportContext.downloadAssetsConcurrency ?? this.apiConcurrency;
+  }
+
   protected getAssetTypesDir(): string {
     return pResolve(this.exportContext.spacesRootPath, 'asset_types');
   }

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

@@ -13,13 +13,17 @@ export default class ExportFields extends AssetManagementExportAdapter {
 
   async start(spaceUid: string): Promise<void> {
     await this.init();
+
+    log.debug('Starting shared fields export process...', this.exportContext.context);
+
     const fieldsData = await this.getWorkspaceFields(spaceUid);
     const items = getArrayFromResponse(fieldsData, 'fields');
     const dir = this.getFieldsDir();
-    log.debug(
-      items.length === 0 ? 'No field items, wrote empty fields' : `Writing ${items.length} shared fields`,
-      this.exportContext.context,
-    );
+    if (items.length === 0) {
+      log.info('No field items to export, writing empty fields', this.exportContext.context);
+    } else {
+      log.debug(`Writing ${items.length} shared fields`, this.exportContext.context);
+    }
     await this.writeItemsToChunkedJson(dir, 'fields.json', 'fields', ['uid', 'title', 'display_type'], items);
     this.tick(true, PROCESS_NAMES.AM_FIELDS, null);
   }

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

@@ -1,6 +1,6 @@
 import { resolve as pResolve } from 'node:path';
 import { mkdir } from 'node:fs/promises';
-import { log, CLIProgressManager, configHandler } from '@contentstack/cli-utilities';
+import { log, CLIProgressManager, configHandler, handleAndLogError } from '@contentstack/cli-utilities';
 
 import type { AssetManagementExportOptions, AssetManagementAPIConfig } from '../types/asset-management-api';
 import type { ExportContext } from '../types/export-types';
@@ -46,6 +46,8 @@ export class ExportSpaces {
       return;
     }
 
+    log.debug('Starting Asset Management export process...', context);
+    log.info('Started Asset Management export', context);
     log.debug(`Exporting Asset Management 2.0 (${linkedWorkspaces.length} space(s))`, context);
     log.debug(`Spaces: ${linkedWorkspaces.map((ws) => ws.space_uid).join(', ')}`, context);
 
@@ -70,6 +72,8 @@ export class ExportSpaces {
       context,
       securedAssets,
       chunkFileSizeMb,
+      apiConcurrency: this.options.apiConcurrency,
+      downloadAssetsConcurrency: this.options.downloadAssetsConcurrency,
     };
 
     const sharedFieldsDir = pResolve(spacesRootPath, 'fields');
@@ -81,11 +85,9 @@ export class ExportSpaces {
     try {
       const exportAssetTypes = new ExportAssetTypes(apiConfig, exportContext);
       exportAssetTypes.setParentProgressManager(progress);
-      await exportAssetTypes.start(firstSpaceUid);
-
       const exportFields = new ExportFields(apiConfig, exportContext);
       exportFields.setParentProgressManager(progress);
-      await exportFields.start(firstSpaceUid);
+      await Promise.all([exportAssetTypes.start(firstSpaceUid), exportFields.start(firstSpaceUid)]);
 
       for (const ws of linkedWorkspaces) {
         progress.updateStatus(`Exporting space: ${ws.space_uid}...`, AM_MAIN_PROCESS_NAME);
@@ -109,9 +111,11 @@ export class ExportSpaces {
       }
 
       progress.completeProcess(AM_MAIN_PROCESS_NAME, true);
+      log.info('Asset Management export completed successfully', context);
       log.debug('Asset Management 2.0 export completed', context);
     } catch (err) {
       progress.completeProcess(AM_MAIN_PROCESS_NAME, false);
+      handleAndLogError(err, { ...(context as Record<string, unknown>) }, 'Asset Management export failed');
       throw err;
     }
   }

packages/contentstack-asset-management/src/export/workspaces.ts

@@ -15,6 +15,9 @@ export default class ExportWorkspace extends AssetManagementExportAdapter {
 
   async start(workspace: LinkedWorkspace, spaceDir: string, branchName: string): Promise<void> {
     await this.init();
+
+    log.debug(`Starting export for AM space ${workspace.space_uid}`, this.exportContext.context);
+
     const spaceResponse = await this.getSpace(workspace.space_uid);
     const space = spaceResponse.space;
     await mkdir(spaceDir, { recursive: true });
@@ -25,7 +28,13 @@ export default class ExportWorkspace extends AssetManagementExportAdapter {
       is_default: workspace.is_default,
       branch: branchName || 'main',
     };
-    await writeFile(pResolve(spaceDir, 'metadata.json'), JSON.stringify(metadata, null, 2));
+    const metadataPath = pResolve(spaceDir, 'metadata.json');
+    try {
+      await writeFile(metadataPath, JSON.stringify(metadata, null, 2));
+    } catch (e) {
+      log.warn(`Could not write ${metadataPath}: ${e}`, this.exportContext.context);
+      throw e;
+    }
     this.tick(true, `space: ${workspace.space_uid}`, null);
     log.debug(`Space metadata written for ${workspace.space_uid}`, this.exportContext.context);
 

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

@@ -31,30 +31,33 @@ export default class ImportAssetTypes extends AssetManagementImportAdapter {
   async start(): Promise<void> {
     await this.init();
 
+    log.debug('Starting shared asset types import process...', this.importContext.context);
+
     const stripKeys = this.importContext.assetTypesImportInvalidKeys ?? [...FALLBACK_ASSET_TYPES_IMPORT_INVALID_KEYS];
     const dir = this.getAssetTypesDir();
     const indexName = this.importContext.assetTypesFileName ?? 'asset-types.json';
     const indexPath = join(dir, indexName);
 
     if (!existsSync(indexPath)) {
-      log.debug('No shared asset types to import (index missing)', this.importContext.context);
+      log.info('No shared asset types to import (index missing)', this.importContext.context);
       return;
     }
 
     const existingByUid = await this.loadExistingAssetTypesMap();
 
-    this.updateStatus(PROCESS_STATUS[PROCESS_NAMES.AM_IMPORT_ASSET_TYPES].IMPORTING, PROCESS_NAMES.AM_IMPORT_ASSET_TYPES);
+    this.updateStatus(
+      PROCESS_STATUS[PROCESS_NAMES.AM_IMPORT_ASSET_TYPES].IMPORTING,
+      PROCESS_NAMES.AM_IMPORT_ASSET_TYPES,
+    );
 
     await forEachChunkedJsonStore<Record<string, unknown>>(
       dir,
       indexName,
       {
         context: this.importContext.context,
         chunkReadLogLabel: 'asset-types',
-        onOpenError: (e) =>
-          log.debug(`Could not open chunked asset-types index: ${e}`, this.importContext.context),
-        onEmptyIndexer: () =>
-          log.debug('No shared asset types to import (empty indexer)', this.importContext.context),
+        onOpenError: (e) => log.warn(`Could not open chunked asset-types index: ${e}`, this.importContext.context),
+        onEmptyIndexer: () => log.debug('No shared asset types to import (empty indexer)', this.importContext.context),
       },
       async (records) => {
         const toCreate = this.buildAssetTypesToCreate(records, existingByUid, stripKeys);
@@ -103,7 +106,10 @@ export default class ImportAssetTypes extends AssetManagementImportAdapter {
             this.importContext.context,
           );
         } else {
-          log.debug(`Asset type "${uid}" already exists with matching definition, skipping`, this.importContext.context);
+          log.debug(
+            `Asset type "${uid}" already exists with matching definition, skipping`,
+            this.importContext.context,
+          );
         }
         this.tick(true, `asset-type: ${uid} (skipped, already exists)`, null, PROCESS_NAMES.AM_IMPORT_ASSET_TYPES);
         continue;