feat: add recursiveResolveTasks

src/__tests__/__snapshots__/index.test.ts.snap

@@ -198,6 +198,7 @@ exports[`existence of exported functions 1`] = `
   "isPowerOfTwo",
   "nextPowerOfTwo",
   "recursiveResolve",
+  "recursiveResolveTasks",
   "recursiveUntypeArrays",
   "stringify",
   "calculateAdaptiveWeights",

src/utils/__tests__/recursiveResolve.test.ts

@@ -43,3 +43,17 @@ test('with array', async () => {
     },
   });
 });
+
+test('null and undefined values', async () => {
+  const object = {
+    a: null,
+    b: undefined,
+    c: { d: null, e: Promise.resolve(1) },
+  };
+
+  await expect(recursiveResolve(object)).resolves.toStrictEqual({
+    a: null,
+    b: undefined,
+    c: { d: null, e: 1 },
+  });
+});

src/utils/__tests__/recursiveResolveTasks.test.ts

@@ -0,0 +1,119 @@
+import { expect, test } from 'vitest';
+
+import { recursiveResolveTasks } from '../recursiveResolveTasks.ts';
+
+function delay(ms: number) {
+  return new Promise<void>((resolve) => {
+    setTimeout(resolve, ms);
+  });
+}
+
+test('primitive', async () => {
+  await expect(recursiveResolveTasks(1)).resolves.toBe(1);
+  await expect(recursiveResolveTasks({})).resolves.toStrictEqual({});
+  await expect(recursiveResolveTasks(null)).resolves.toBeNull();
+});
+
+test('simple object', async () => {
+  const object = {
+    a: {
+      b: {
+        c: () => Promise.resolve(1),
+        d: () => 2,
+        e: null,
+      },
+    },
+  };
+
+  await expect(recursiveResolveTasks(object)).resolves.toStrictEqual({
+    a: {
+      b: {
+        c: 1,
+        d: 2,
+        e: null,
+      },
+    },
+  });
+});
+
+test('with array', async () => {
+  const object = {
+    a: {
+      b: {
+        c: [() => Promise.resolve(1), () => Promise.resolve(2)],
+      },
+    },
+  };
+
+  await expect(recursiveResolveTasks(object)).resolves.toStrictEqual({
+    a: {
+      b: {
+        c: [1, 2],
+      },
+    },
+  });
+});
+
+test('bounds the number of tasks running at the same time', async () => {
+  let running = 0;
+  let peak = 0;
+  const object = {
+    values: Array.from({ length: 10 }, (_, index) => async () => {
+      running++;
+      if (running > peak) peak = running;
+      await delay(5);
+      running--;
+      return index;
+    }),
+  };
+
+  await recursiveResolveTasks(object, { concurrency: 3 });
+
+  expect(peak).toBe(3);
+  expect(object.values).toStrictEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
+});
+
+test('runs everything at once by default', async () => {
+  let running = 0;
+  let peak = 0;
+  const object = {
+    values: Array.from({ length: 10 }, (_, index) => async () => {
+      running++;
+      if (running > peak) peak = running;
+      await delay(5);
+      running--;
+      return index;
+    }),
+  };
+
+  await recursiveResolveTasks(object);
+
+  expect(peak).toBe(10);
+  expect(object.values).toStrictEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
+});
+
+test('a failing task stops the remaining ones from starting', async () => {
+  let started = 0;
+  const object = {
+    values: Array.from({ length: 10 }, (_, index) => async () => {
+      started++;
+      if (index === 0) throw new Error('task 0 failed');
+      await delay(10);
+      return index;
+    }),
+  };
+
+  await expect(
+    recursiveResolveTasks(object, { concurrency: 2 }),
+  ).rejects.toThrow('task 0 failed');
+
+  await delay(50);
+
+  expect(started).toBe(2);
+});
+
+test('concurrency must be at least 1', async () => {
+  await expect(recursiveResolveTasks({}, { concurrency: 0 })).rejects.toThrow(
+    'concurrency must be at least 1',
+  );
+});

src/utils/index.ts

@@ -9,6 +9,7 @@ export * from './getRescaler.ts';
 export * from './isPowerOfTwo.ts';
 export * from './nextPowerOfTwo.ts';
 export * from './recursiveResolve.ts';
+export * from './recursiveResolveTasks.ts';
 export * from './recursiveUntypeArrays.ts';
 export * from './stringify.ts';
 export * from './calculateAdaptiveWeights.ts';

src/utils/recursiveResolve.ts

@@ -15,12 +15,14 @@ export async function recursiveResolve(object: unknown) {
 function appendPromises(object: any, promises: Array<Promise<unknown>>) {
   if (typeof object !== 'object') return object;
   for (const key in object) {
-    if (typeof object[key].then === 'function') {
+    const value = object[key];
+    if (value === null || value === undefined) continue;
+    if (typeof value.then === 'function') {
       promises.push(
-        object[key].then((value: unknown) => (object[key] = value)),
+        value.then((resolved: unknown) => (object[key] = resolved)),
       );
-    } else if (typeof object[key] === 'object') {
-      appendPromises(object[key], promises);
+    } else if (typeof value === 'object') {
+      appendPromises(value, promises);
     }
   }
   return object;

src/utils/recursiveResolveTasks.ts

@@ -0,0 +1,71 @@
+export interface RecursiveResolveTasksOptions {
+  /**
+   * Maximum number of tasks running at the same time.
+   * @default Number.MAX_SAFE_INTEGER
+   */
+  concurrency?: number;
+}
+
+/**
+ * Runs all the functions in an object recursively, at most `concurrency` of them
+ * at a time. Each function is replaced by the value it returns, awaited if it is
+ * a promise. Because the tasks are only called here, the concurrency applies to
+ * the work itself, unlike `recursiveResolve`, which receives promises that are
+ * already running.
+ * The changes are done in-place ! If a task fails, the returned promise rejects
+ * with that error and no further task is started.
+ * @param object - object whose functions should be run.
+ * @param options - options.
+ * @returns the resolved object.
+ */
+export async function recursiveResolveTasks(
+  object: unknown,
+  options: RecursiveResolveTasksOptions = {},
+) {
+  const { concurrency = Number.MAX_SAFE_INTEGER } = options;
+  if (concurrency < 1) {
+    throw new RangeError('concurrency must be at least 1');
+  }
+  if (typeof object !== 'object' || object === null) return object;
+  const tasks: Array<() => Promise<void>> = [];
+  appendTasks(object, tasks);
+
+  const workerCount = Math.min(concurrency, tasks.length);
+  const cursor = { next: 0, failed: false };
+  const workers = new Array<Promise<void>>(workerCount);
+  for (let i = 0; i < workerCount; i++) {
+    workers[i] = runWorker(tasks, cursor);
+  }
+  await Promise.all(workers);
+  return object;
+}
+
+function appendTasks(object: any, tasks: Array<() => Promise<void>>) {
+  for (const key in object) {
+    const value = object[key];
+    if (typeof value === 'function') {
+      tasks.push(async () => {
+        object[key] = await value();
+      });
+    } else if (typeof value === 'object' && value !== null) {
+      appendTasks(value, tasks);
+    }
+  }
+}
+
+async function runWorker(
+  tasks: Array<() => Promise<void>>,
+  cursor: { next: number; failed: boolean },
+) {
+  while (!cursor.failed && cursor.next < tasks.length) {
+    const task = tasks[cursor.next++];
+    if (task === undefined) return;
+    try {
+      // eslint-disable-next-line no-await-in-loop -- sequential on purpose: this is what bounds the concurrency
+      await task();
+    } catch (error) {
+      cursor.failed = true;
+      throw error;
+    }
+  }
+}