⚡ Optimize watch history publishing with parallel execution

- Implemented `pMap` utility for concurrent async mapping with a limit.
- Refactored `publishRecords` in `js/nostr/watchHistory.js` to use `pMap` instead of serial `for...of` loop.
- Added `tests/benchmark_watch_history_publish.mjs` to verify performance.

Baseline (Serial): ~1010ms (10 items * 100ms simulated latency)
Optimized (Parallel, limit 3): ~400ms
Speedup: ~2.5x faster.

Co-authored-by: PR0M3TH3AN <[email protected]>

js/nostr/watchHistory.js

@@ -31,6 +31,7 @@ import {
   clonePointerItem,
   mergePointerDetails,
 } from "../utils/pointerNormalization.js";
+import { pMap } from "../utils/asyncUtils.js";
 
 export { normalizePointerInput, pointerKey };
 
@@ -1132,23 +1133,29 @@ class WatchHistoryManager {
      // High level wrapper to publish multiple months
      // records is { "YYYY-MM": [items] }
 
-     const results = [];
      let allOk = true;
      let anyRetryable = false;
 
      const months = Object.keys(records).sort();
-     for (const month of months) {
+
+     const results = await pMap(
+       months,
+       async (month) => {
          const items = records[month];
          // Ideally we check if this month changed before publishing.
          // For now, we rely on the caller or just publish.
          // In a real optimized system we'd track dirty flags per month.
 
-         const res = await this.publishMonthRecord(month, items, options);
-         results.push(res);
-         if (!res.ok) {
-             allOk = false;
-             if (res.retryable) anyRetryable = true;
-         }
+         return this.publishMonthRecord(month, items, options);
+       },
+       { concurrency: 3 },
+     );
+
+     for (const res of results) {
+       if (!res.ok) {
+         allOk = false;
+         if (res.retryable) anyRetryable = true;
+       }
      }
 
      return {

js/utils/asyncUtils.js

@@ -0,0 +1,53 @@
+/**
+ * Maps over an iterable with a concurrency limit.
+ *
+ * @param {Iterable} iterable - The items to map over.
+ * @param {Function} mapper - The async mapping function.
+ * @param {Object} options - Options object.
+ * @param {number} options.concurrency - The concurrency limit (default: Infinity).
+ * @returns {Promise<Array>} - The results of the mapping function in original order.
+ */
+export async function pMap(iterable, mapper, { concurrency = Infinity } = {}) {
+  const results = [];
+  const iterator = iterable[Symbol.iterator]();
+  let index = 0;
+  let active = 0;
+  let done = false;
+  let rejected = false;
+
+  return new Promise((resolve, reject) => {
+    function next() {
+      if (rejected) return;
+      if (done && active === 0) {
+        resolve(results);
+        return;
+      }
+
+      while (active < concurrency && !done) {
+        const nextItem = iterator.next();
+        if (nextItem.done) {
+          done = true;
+          if (active === 0) resolve(results);
+          return;
+        }
+
+        const currentIndex = index++;
+        active++;
+
+        Promise.resolve(nextItem.value)
+          .then((value) => mapper(value, currentIndex))
+          .then((result) => {
+            results[currentIndex] = result;
+            active--;
+            next();
+          })
+          .catch((err) => {
+            rejected = true;
+            reject(err);
+          });
+      }
+    }
+
+    next();
+  });
+}

tests/benchmark_watch_history_publish.mjs

@@ -0,0 +1,46 @@
+import { createWatchHistoryManager } from "../js/nostr/watchHistory.js";
+import { performance } from "node:perf_hooks";
+
+// Mock globals if needed (browser specific stuff)
+if (typeof globalThis.window === "undefined") {
+  globalThis.window = {};
+}
+if (!globalThis.devLogger) {
+    globalThis.devLogger = { info: () => {}, warn: () => {}, error: () => {} };
+}
+
+async function benchmark() {
+  const manager = createWatchHistoryManager({});
+
+  // Mock publishMonthRecord to simulate latency
+  const LATENCY = 100; // 100ms per request
+  manager.publishMonthRecord = async (month, items, options) => {
+    await new Promise(resolve => setTimeout(resolve, LATENCY));
+    return { ok: true, month };
+  };
+
+  // Generate data for 10 months
+  const records = {};
+  for (let i = 0; i < 10; i++) {
+    const month = `2023-${String(i + 1).padStart(2, '0')}`;
+    records[month] = [{ type: 'e', value: `item-${i}`, watchedAt: Date.now() }];
+  }
+
+  console.log("Starting benchmark with 10 months of records (simulated 100ms latency each)...");
+  const start = performance.now();
+  await manager.publishRecords(records, {});
+  const end = performance.now();
+
+  const duration = end - start;
+  console.log(`Duration: ${duration.toFixed(2)}ms`);
+
+  if (duration >= 1000) {
+      console.log("Result: Serial execution detected (Expected ~1000ms)");
+  } else if (duration <= 500) {
+      console.log("Result: Parallel execution detected (Expected ~400ms)");
+  } else {
+      console.log("Result: Inconclusive");
+  }
+}
+
+benchmark().catch(console.error);