Merge pull request #220 from PR0M3TH3AN/codex/refactor-publishwatchhistorysnapshot-method

Refine watch history snapshot publishing for chunk failures

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,

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) {