chore: update logic to stay consistent with CLI export package

Author: naman-contentstack

.talismanrc

@@ -1,4 +1,6 @@
 fileignoreconfig:
 - filename: pnpm-lock.yaml
   checksum: cdead0797199d22bbc55b9e5b6b86983f28eb760fabe5e1f2d5139c4456a9131
+- filename: packages/contentstack-asset-management/test/unit/utils/concurrent-batch.test.ts
+  checksum: d6e5995e0bbdc862a2d21c0ebfead30d9939021da6f7f62c82027de1eed7fd76
 version: '1.0'

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

@@ -1,3 +1,11 @@
+/**
+ * Batched concurrency primitive for the package. All concurrent work — pagination
+ * fan-out, asset downloads, uploads, and folder/asset-type/field creation — runs
+ * through these helpers. Do not hand-roll `Promise.all`/`Promise.allSettled`
+ * batching elsewhere; use `runInBatches` (void, fault-tolerant) for bulk
+ * side-effects and `mapInBatches` (collects results, fail-fast) for fetches.
+ */
+
 /**
  * Split an array into chunks of at most `size` elements.
  */
@@ -32,3 +40,29 @@ export async function runInBatches<T>(
     offset += batch.length;
   }
 }
+
+/**
+ * Run async work in batches of at most `concurrency` tasks, collecting results
+ * in input order. Uses Promise.all per batch (fail-fast: a rejected task aborts
+ * the whole call). Use this for fetches where a dropped result would silently
+ * yield incomplete data (e.g. pagination fan-out).
+ */
+export async function mapInBatches<T, R>(
+  items: T[],
+  concurrency: number,
+  fn: (item: T, index: number) => Promise<R>,
+): Promise<R[]> {
+  if (items.length === 0) {
+    return [];
+  }
+  const limit = Math.max(1, concurrency);
+  const batches = chunkArray(items, limit);
+  const results: R[] = [];
+  let offset = 0;
+  for (const batch of batches) {
+    const settled = await Promise.all(batch.map((item, j) => fn(item, offset + j)));
+    results.push(...settled);
+    offset += batch.length;
+  }
+  return results;
+}

packages/contentstack-asset-management/src/utils/cs-assets-api-adapter.ts

@@ -2,7 +2,7 @@ import { readFileSync } from 'node:fs';
 import { basename } from 'node:path';
 import { HttpClient, log, authenticationHandler, handleAndLogError } from '@contentstack/cli-utilities';
 
-import { chunkArray } from './concurrent-batch';
+import { mapInBatches } from './concurrent-batch';
 import { FALLBACK_AM_API_FETCH_CONCURRENCY, FALLBACK_AM_API_PAGE_SIZE } from '../constants/index';
 
 import type {
@@ -265,21 +265,13 @@ export class CSAssetsAdapter implements ICSAssetsAdapter {
       (_, i) => (i + 1) * pageSize,
     );
 
-    const skipBatches = chunkArray(skips, concurrency);
-    const rest: unknown[] = [];
-
-    for (const batch of skipBatches) {
-      const pages = await Promise.all(
-        batch.map((skip) =>
-          this.getSpaceLevel<Record<string, unknown>>(spaceUid, path, {
-            ...baseParams, limit: String(pageSize), skip: String(skip),
-          }).then((r) => (Array.isArray(r?.[itemsKey]) ? (r[itemsKey] as unknown[]) : [])),
-        ),
-      );
-      rest.push(...pages.flat());
-    }
+    const pages = await mapInBatches(skips, concurrency, (skip) =>
+      this.getSpaceLevel<Record<string, unknown>>(spaceUid, path, {
+        ...baseParams, limit: String(pageSize), skip: String(skip),
+      }).then((r) => (Array.isArray(r?.[itemsKey]) ? (r[itemsKey] as unknown[]) : [])),
+    );
 
-    return [...firstItems, ...rest];
+    return [...firstItems, ...pages.flat()];
   }
 
   async getWorkspaceAssets(spaceUid: string, workspaceUid?: string, pageSize = FALLBACK_AM_API_PAGE_SIZE, fetchConcurrency = FALLBACK_AM_API_FETCH_CONCURRENCY): Promise<unknown> {

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

@@ -7,6 +7,6 @@ export {
   getReadableStreamFromDownloadResponse,
   writeStreamToFile,
 } from './export-helpers';
-export { chunkArray, runInBatches } from './concurrent-batch';
+export { chunkArray, runInBatches, mapInBatches } from './concurrent-batch';
 export { detectAssetManagementExportFromContentDir } from './detect-asset-management-export';
 export type { AssetManagementExportFlags } from '../types/asset-management-export-flags';

packages/contentstack-asset-management/test/unit/utils/concurrent-batch.test.ts

@@ -0,0 +1,112 @@
+import { expect } from 'chai';
+
+import { chunkArray, mapInBatches, runInBatches } from '../../../src/utils/concurrent-batch';
+
+describe('concurrent-batch', () => {
+  describe('chunkArray', () => {
+    it('should split an array into chunks of at most `size`', () => {
+      expect(chunkArray([1, 2, 3, 4, 5], 2)).to.deep.equal([[1, 2], [3, 4], [5]]);
+    });
+
+    it('should return a single chunk when size >= length', () => {
+      expect(chunkArray([1, 2, 3], 10)).to.deep.equal([[1, 2, 3]]);
+    });
+
+    it('should return the whole array as one chunk when size <= 0', () => {
+      expect(chunkArray([1, 2, 3], 0)).to.deep.equal([[1, 2, 3]]);
+    });
+
+    it('should return [] for an empty array', () => {
+      expect(chunkArray([], 3)).to.deep.equal([]);
+    });
+  });
+
+  describe('mapInBatches', () => {
+    it('should collect results in input order', async () => {
+      const results = await mapInBatches([1, 2, 3, 4, 5], 2, async (n) => n * 10);
+      expect(results).to.deep.equal([10, 20, 30, 40, 50]);
+    });
+
+    it('should pass the correct absolute index across batches', async () => {
+      const indexes: number[] = [];
+      await mapInBatches(['a', 'b', 'c', 'd', 'e'], 2, async (_item, index) => {
+        indexes.push(index);
+        return index;
+      });
+      expect([...indexes].sort((a, b) => a - b)).to.deep.equal([0, 1, 2, 3, 4]);
+    });
+
+    it('should never run more than `concurrency` tasks at once', async () => {
+      let inFlight = 0;
+      let maxInFlight = 0;
+      await mapInBatches(Array.from({ length: 10 }, (_, i) => i), 3, async (n) => {
+        inFlight += 1;
+        maxInFlight = Math.max(maxInFlight, inFlight);
+        await new Promise((resolve) => setImmediate(resolve));
+        inFlight -= 1;
+        return n;
+      });
+      expect(maxInFlight).to.be.at.most(3);
+    });
+
+    it('should return [] for an empty array without invoking fn', async () => {
+      let called = false;
+      const results = await mapInBatches([], 5, async (n) => {
+        called = true;
+        return n;
+      });
+      expect(results).to.deep.equal([]);
+      expect(called).to.equal(false);
+    });
+
+    it('should fail fast when a task rejects', async () => {
+      let error: Error | undefined;
+      try {
+        await mapInBatches([1, 2, 3], 2, async (n) => {
+          if (n === 2) throw new Error('boom');
+          return n;
+        });
+      } catch (e) {
+        error = e as Error;
+      }
+      expect(error).to.be.instanceOf(Error);
+      expect(error?.message).to.equal('boom');
+    });
+
+    it('should treat concurrency < 1 as 1', async () => {
+      const results = await mapInBatches([1, 2, 3], 0, async (n) => n);
+      expect(results).to.deep.equal([1, 2, 3]);
+    });
+  });
+
+  describe('runInBatches', () => {
+    it('should invoke fn for every item with the correct absolute index', async () => {
+      const seen: Array<{ item: string; index: number }> = [];
+      await runInBatches(['a', 'b', 'c'], 2, async (item, index) => {
+        seen.push({ item, index });
+      });
+      expect(seen.sort((a, b) => a.index - b.index)).to.deep.equal([
+        { item: 'a', index: 0 },
+        { item: 'b', index: 1 },
+        { item: 'c', index: 2 },
+      ]);
+    });
+
+    it('should not abort the batch when one task rejects (fault-tolerant)', async () => {
+      const completed: number[] = [];
+      await runInBatches([1, 2, 3, 4], 2, async (n) => {
+        if (n === 2) throw new Error('boom');
+        completed.push(n);
+      });
+      expect(completed.sort((a, b) => a - b)).to.deep.equal([1, 3, 4]);
+    });
+
+    it('should be a no-op for an empty array', async () => {
+      let called = false;
+      await runInBatches([], 5, async () => {
+        called = true;
+      });
+      expect(called).to.equal(false);
+    });
+  });
+});