diff --git a/js/nostr/watchHistory.js b/js/nostr/watchHistory.js index 4bdf89bc..fb2fe308 100644 --- a/js/nostr/watchHistory.js +++ b/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 { diff --git a/js/utils/asyncUtils.js b/js/utils/asyncUtils.js new file mode 100644 index 00000000..f6ceb271 --- /dev/null +++ b/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} - 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(); + }); +} diff --git a/tests/benchmark_watch_history_publish.mjs b/tests/benchmark_watch_history_publish.mjs new file mode 100644 index 00000000..0079d4ac --- /dev/null +++ b/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);