From a025dd210f49cb1833daf3266486ddb92bb0061d Mon Sep 17 00:00:00 2001 From: thePR0M3TH3AN <53631862+PR0M3TH3AN@users.noreply.github.com> Date: Wed, 1 Oct 2025 17:07:39 -0400 Subject: [PATCH] Refine watch history snapshot publishing --- js/nostr.js | 115 +++++++++++++---------- tests/watch-history.test.mjs | 171 +++++++++++++++++++++++++++++++++-- 2 files changed, 231 insertions(+), 55 deletions(-) diff --git a/js/nostr.js b/js/nostr.js index 9f11c393..dd97a339 100644 --- a/js/nostr.js +++ b/js/nostr.js @@ -2314,49 +2314,7 @@ class NostrClient { } const baseTimestamp = Math.max(nowSeconds, lastCreatedAt + totalChunks); - const chunkResults = []; - let overallSuccess = true; - - const indexEvent = buildWatchHistoryIndexEvent({ - pubkey: normalizedActor, - created_at: baseTimestamp + totalChunks, - snapshotId, - totalChunks, - chunkAddresses, - }); - - let signedIndexEvent = null; - try { - if (useExtension) { - signedIndexEvent = await window.nostr.signEvent(indexEvent); - } else { - signedIndexEvent = signEventWithPrivateKey( - indexEvent, - this.sessionActor.privateKey - ); - } - } catch (error) { - console.warn("[nostr] Failed to sign watch history index event:", error); - signedIndexEvent = null; - } - - let indexPublishResults = []; - let indexSuccess = false; - if (signedIndexEvent) { - indexPublishResults = await Promise.all( - relays.map((url) => publishEventToRelay(this.pool, url, signedIndexEvent)) - ); - indexSuccess = indexPublishResults.some((result) => result.success); - if (!indexSuccess) { - overallSuccess = false; - console.warn( - "[nostr] Failed to publish watch history index:", - indexPublishResults - ); - } - } else { - overallSuccess = false; - } + const preparedChunks = []; for (let index = 0; index < totalChunks; index++) { const chunkItems = Array.isArray(chunkItemsList[index]) @@ -2436,8 +2394,18 @@ class NostrClient { } } + preparedChunks.push({ + chunkIndex: index, + event: signedEvent, + }); + } + + const chunkResults = []; + let overallSuccess = true; + + for (const chunk of preparedChunks) { const publishResults = await Promise.all( - relays.map((url) => publishEventToRelay(this.pool, url, signedEvent)) + relays.map((url) => publishEventToRelay(this.pool, url, chunk.event)) ); const chunkSuccess = publishResults.some((result) => result.success); @@ -2450,13 +2418,62 @@ class NostrClient { } chunkResults.push({ - chunkIndex: index, - event: signedEvent, + chunkIndex: chunk.chunkIndex, + event: chunk.event, results: publishResults, success: chunkSuccess, }); } + const allChunksSucceeded = chunkResults.every((chunk) => chunk.success); + + let signedIndexEvent = null; + let indexPublishResults = []; + let indexSuccess = false; + + if (allChunksSucceeded) { + const indexEvent = buildWatchHistoryIndexEvent({ + pubkey: normalizedActor, + created_at: baseTimestamp + totalChunks, + snapshotId, + totalChunks, + chunkAddresses, + }); + + try { + if (useExtension) { + signedIndexEvent = await window.nostr.signEvent(indexEvent); + } else { + signedIndexEvent = signEventWithPrivateKey( + indexEvent, + this.sessionActor.privateKey + ); + } + } catch (error) { + console.warn("[nostr] Failed to sign watch history index event:", error); + signedIndexEvent = null; + } + + if (signedIndexEvent) { + indexPublishResults = await Promise.all( + relays.map((url) => publishEventToRelay(this.pool, url, signedIndexEvent)) + ); + indexSuccess = indexPublishResults.some((result) => result.success); + if (!indexSuccess) { + overallSuccess = false; + console.warn( + "[nostr] Failed to publish watch history index:", + indexPublishResults + ); + signedIndexEvent = null; + } + } else { + overallSuccess = false; + } + } else { + overallSuccess = false; + } + if (overallSuccess) { this.cancelWatchHistoryRepublish(normalizedActor); } else if (allowRetry && persistedItems.length) { @@ -2466,7 +2483,11 @@ class NostrClient { const baselineEntry = existingEntry || this.watchHistoryCache.get(normalizedActor) || null; - const pointerEvent = signedIndexEvent || chunkResults[0]?.event || null; + const pointerEvent = + signedIndexEvent || + chunkResults.find((chunk) => chunk.success)?.event || + chunkResults[0]?.event || + null; const newEntry = this.createWatchHistoryEntry( pointerEvent, persistedItems, diff --git a/tests/watch-history.test.mjs b/tests/watch-history.test.mjs index f39c1291..2d8a646a 100644 --- a/tests/watch-history.test.mjs +++ b/tests/watch-history.test.mjs @@ -34,18 +34,41 @@ delete globalThis.window.nostr; const { nostrClient } = await import("../js/nostr.js"); -function createPublishingClient(actorPubkey) { +function createPublishingClient(actorPubkey, options = {}) { const client = new nostrClient.constructor(); const publishedEvents = []; const payloads = []; + const scheduledRepublishes = []; + const cancelledRepublishes = []; + const persistedEntries = []; + const onPublish = + typeof options.onPublish === "function" ? options.onPublish : null; + client.pool = { publish(urls, event) { publishedEvents.push(event); + const publishDecision = onPublish ? onPublish(event) : null; + const shouldSucceed = publishDecision?.ok !== false; return { on(eventName, handler) { - if (eventName === "ok" || eventName === "seen") { - setTimeout(handler, 0); - } + setTimeout(() => { + if (shouldSucceed) { + if (eventName === "ok" || eventName === "seen") { + handler(); + } + } else if (eventName === "failed") { + const providedError = publishDecision?.error; + const error = + providedError instanceof Error + ? providedError + : new Error( + providedError + ? String(providedError) + : "mock publish failure" + ); + handler(error); + } + }, 0); }, }; }, @@ -59,10 +82,23 @@ function createPublishingClient(actorPubkey) { payloads.push(payload); return { ok: true, ciphertext: JSON.stringify(payload) }; }; - client.persistWatchHistoryEntry = () => {}; - client.cancelWatchHistoryRepublish = () => {}; - client.scheduleWatchHistoryRepublish = () => {}; - return { client, publishedEvents, payloads }; + client.persistWatchHistoryEntry = (_actor, entry) => { + persistedEntries.push(entry); + }; + client.cancelWatchHistoryRepublish = (actor) => { + cancelledRepublishes.push(actor); + }; + client.scheduleWatchHistoryRepublish = (actor, items) => { + scheduledRepublishes.push({ actor, items }); + }; + return { + client, + publishedEvents, + payloads, + scheduledRepublishes, + cancelledRepublishes, + persistedEntries, + }; } function createDecryptClient(actorPubkey) { @@ -328,6 +364,125 @@ expectedChunkAddresses.forEach((address) => { ); }); +const CHUNK_FAILURE_ACTOR = `${ACTOR}-chunk-failure`; +const { + client: chunkFailureClient, + publishedEvents: chunkFailureEvents, + payloads: chunkFailurePayloads, + scheduledRepublishes: chunkFailureRepublishes, + cancelledRepublishes: chunkFailureCancels, +} = createPublishingClient(CHUNK_FAILURE_ACTOR, { + onPublish(event) { + const chunkTag = event.tags.find( + (tag) => Array.isArray(tag) && tag[0] === "chunk" + ); + if (chunkTag) { + const chunkIndex = Number.parseInt(chunkTag[1], 10); + const total = Number.parseInt(chunkTag[2], 10); + if ( + Number.isInteger(chunkIndex) && + Number.isInteger(total) && + (total > 1 ? chunkIndex === total - 1 : chunkIndex === 0) + ) { + return { ok: false, error: new Error("chunk publish failed") }; + } + } + return { ok: true }; + }, +}); + +chunkFailureClient.watchHistoryPayloadMaxBytes = + publishingClient.watchHistoryPayloadMaxBytes; +chunkFailureClient.watchHistoryFetchEventLimit = 10; + +const chunkFailureResult = await chunkFailureClient.publishWatchHistorySnapshot( + CHUNK_FAILURE_ACTOR, + longPointers +); + +assert.equal( + chunkFailureResult.ok, + false, + "chunk failure should report overall failure" +); +assert.equal( + chunkFailureResult.indexEvent, + null, + "chunk failure should omit index event" +); +assert.equal( + chunkFailureResult.events.length, + chunkFailureResult.chunks.length, + "chunk failure should not include index event in combined events" +); +assert.equal( + chunkFailureResult.indexResults.length, + 0, + "chunk failure should skip index publish attempts" +); +assert( + chunkFailureResult.chunks.some((chunk) => chunk.success === false), + "chunk failure results should flag the failed chunk" +); +assert.equal( + chunkFailureRepublishes.length, + 1, + "chunk failure should schedule a retry" +); +assert.equal( + chunkFailureCancels.length, + 0, + "chunk failure should not cancel retries" +); +assert.equal( + chunkFailureRepublishes[0].actor, + CHUNK_FAILURE_ACTOR, + "chunk failure retry should target the affected actor" +); +assert( + chunkFailureEvents.every((event) => + event.tags.some((tag) => Array.isArray(tag) && tag[0] === "chunk") + ), + "chunk failure should avoid publishing index events" +); +assert.equal( + chunkFailurePayloads.length, + chunkFailureResult.chunks.length, + "chunk failure should still prepare payloads for each chunk" +); + +const chunkFailureCache = chunkFailureClient.watchHistoryCache.get( + CHUNK_FAILURE_ACTOR +); + +assert(chunkFailureCache, "chunk failure should cache a watch history entry"); +assert.equal( + chunkFailureCache.chunkPointers.index, + null, + "chunk failure cache should not retain an index pointer" +); +assert( + Array.from(chunkFailureCache.chunkEvents.values()).every((event) => + event.tags.some((tag) => Array.isArray(tag) && tag[0] === "chunk") + ), + "chunk failure cache should only persist chunk events" +); +assert.deepEqual( + chunkFailureRepublishes[0].items.map((item) => ({ + type: item.type, + value: item.value, + relay: item.relay || null, + watchedAt: item.watchedAt, + })), + chunkFailureResult.items.map((item) => ({ + type: item.type, + value: item.value, + relay: item.relay || null, + watchedAt: item.watchedAt, + })), + "republish scheduler should receive the remaining items" +); + delete globalThis.window.nostr; if (!globalThis.window.NostrTools.nip04) {