diff --git a/js/watchHistoryService.js b/js/watchHistoryService.js index 4b909fd9..197eb378 100644 --- a/js/watchHistoryService.js +++ b/js/watchHistoryService.js @@ -901,6 +901,55 @@ function clearQueue(actorKey) { notifyQueueChange(actorKey); } +function pruneQueueAfterSnapshot(actorKey, queue, keysToClear, snapshotStart) { + if (!actorKey || !queue) { + return; + } + + const keySet = new Set(keysToClear || []); + const normalizedStart = Number(snapshotStart); + const cutoff = Number.isFinite(normalizedStart) ? normalizedStart : Date.now(); + + let mutated = false; + + if (keySet.size > 0) { + for (const key of keySet) { + const entry = queue.items.get(key); + if (!entry) { + continue; + } + const updatedAt = Number(entry?.updatedAt); + const addedAt = Number(entry?.addedAt); + const referenceTime = Number.isFinite(updatedAt) + ? updatedAt + : Number.isFinite(addedAt) + ? addedAt + : 0; + if (referenceTime > cutoff) { + continue; + } + queue.items.delete(key); + queue.throttle.delete(key); + mutated = true; + } + } + + if (queue.pendingSnapshotId) { + queue.pendingSnapshotId = null; + mutated = true; + } + + if (queue.republishScheduled) { + queue.republishScheduled = false; + mutated = true; + } + + if (mutated) { + persistQueueState(); + notifyQueueChange(actorKey); + } +} + function resolveWatchedAt(...candidates) { for (const candidate of candidates) { if (!Number.isFinite(candidate)) { @@ -1252,6 +1301,14 @@ async function snapshot(items, options = {}) { } const queue = ensureQueue(actorKey); + const queuePayloadKeys = + !items && queue && payloadItems.length + ? new Set( + payloadItems + .map((pointer) => pointerKey(pointer)) + .filter((keyValue) => typeof keyValue === "string" && keyValue) + ) + : new Set(); if (!items) { const cachedItems = getCachedSnapshotItems(actorKey); @@ -1271,6 +1328,7 @@ async function snapshot(items, options = {}) { ); const run = (async () => { + const snapshotStartTime = Date.now(); emit("snapshot-start", { actor: actorKey, reason, items: payloadItems }); const publishResult = await nostrClient.publishWatchHistorySnapshot( payloadItems, @@ -1308,7 +1366,7 @@ async function snapshot(items, options = {}) { } if (!items) { - clearQueue(actorKey); + pruneQueueAfterSnapshot(actorKey, queue, queuePayloadKeys, snapshotStartTime); } else if (queue) { queue.pendingSnapshotId = null; queue.republishScheduled = false; diff --git a/tests/watch-history.test.mjs b/tests/watch-history.test.mjs index 7c3a05bf..5f6559e5 100644 --- a/tests/watch-history.test.mjs +++ b/tests/watch-history.test.mjs @@ -1325,6 +1325,118 @@ async function testWatchHistoryPartialRelayRetry() { } } +async function testWatchHistorySnapshotRetainsNewQueueEntriesDuringPublish() { + console.log("Running watch history snapshot inflight queue retention test..."); + + const actor = "npub-snapshot-inflight"; + const originalPublishSnapshot = nostrClient.publishWatchHistorySnapshot; + const originalRecordView = nostrClient.recordVideoView; + const originalEnsure = nostrClient.ensureSessionActor; + const originalSession = nostrClient.sessionActor; + const originalPub = nostrClient.pubkey; + + try { + localStorage.clear(); + watchHistoryService.resetProgress(); + nostrClient.pubkey = actor; + nostrClient.sessionActor = { pubkey: actor, privateKey: "snapshot-priv" }; + nostrClient.ensureSessionActor = async () => actor; + nostrClient.watchHistoryLastCreatedAt = 0; + + let createdAt = 1_700_200_000; + nostrClient.recordVideoView = async (_pointer, options = {}) => ({ + ok: true, + event: { + id: `view-${options.created_at || createdAt}`, + pubkey: actor, + created_at: options.created_at || createdAt, + }, + }); + + await watchHistoryService.publishView( + { type: "e", value: "inflight-initial" }, + createdAt, + { actor }, + ); + + const queuedBefore = watchHistoryService.getQueuedPointers(actor); + assert.equal( + queuedBefore.length, + 1, + "queue should contain the initial pointer before snapshot", + ); + + let publishCalled = false; + let publishItems = []; + let releasePublish; + const publishGate = new Promise((resolve) => { + releasePublish = resolve; + }); + + nostrClient.publishWatchHistorySnapshot = async (items, options = {}) => { + publishCalled = true; + publishItems = Array.isArray(items) ? items.map((item) => ({ ...item })) : []; + await publishGate; + return { ok: true, items, snapshotId: "inflight-snapshot" }; + }; + + const snapshotPromise = watchHistoryService.snapshot(null, { + actor, + reason: "inflight-test", + }); + + await new Promise((resolve) => setTimeout(resolve, 0)); + assert.equal( + publishCalled, + true, + "snapshot should invoke publish before resolving", + ); + + createdAt += 60; + await watchHistoryService.publishView( + { type: "e", value: "inflight-new" }, + createdAt, + { actor }, + ); + + const queuedDuring = watchHistoryService.getQueuedPointers(actor); + assert.equal( + queuedDuring.some((entry) => entry?.value === "inflight-new"), + true, + "queue should include the new pointer while snapshot is pending", + ); + + assert.equal( + publishItems.some((entry) => entry?.value === "inflight-new"), + false, + "inflight publish should not include pointers queued after the snapshot started", + ); + + releasePublish(); + const snapshotResult = await snapshotPromise; + assert.ok(snapshotResult?.ok, "snapshot should resolve successfully"); + + const queuedAfter = watchHistoryService.getQueuedPointers(actor); + assert.equal( + queuedAfter.length, + 1, + "queue should retain only the pointer added during the inflight snapshot", + ); + assert.equal( + queuedAfter[0]?.value, + "inflight-new", + "new pointer should remain queued for the next snapshot", + ); + } finally { + nostrClient.publishWatchHistorySnapshot = originalPublishSnapshot; + nostrClient.recordVideoView = originalRecordView; + nostrClient.ensureSessionActor = originalEnsure; + nostrClient.sessionActor = originalSession; + nostrClient.pubkey = originalPub; + watchHistoryService.resetProgress(); + } +} + async function testResolveWatchHistoryBatchingWindow() { const actor = "b".repeat(64); const originalEnsure = nostrClient.ensureSessionActor; @@ -2345,6 +2457,7 @@ await testEnsureExtensionPermissionCaching(); await testFetchWatchHistoryExtensionDecryptsHexAndNpub(); await testPublishSnapshotFailureRetry(); await testWatchHistoryPartialRelayRetry(); +await testWatchHistorySnapshotRetainsNewQueueEntriesDuringPublish(); await testResolveWatchHistoryBatchingWindow(); await testWatchHistoryServiceIntegration(); await testWatchHistorySnapshotMergesQueuedWithCachedItems();