From ac3873e5d704153659c8e90b0d74242894faa9c5 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Mon, 5 Jan 2026 15:42:48 +0000 Subject: [PATCH] Refactor watch history service to use idempotent monthly upsert logic. Removes legacy snapshot/chunk retry scheduling in favor of a monthly upsert approach. Updates state persistence to remove `pendingSnapshotId`. Fixes race condition in `scheduleRepublishForQueue` by using selective queue pruning. Ensures data is not lost on publish failure in `snapshot`. Updates tests to align with new architecture. --- js/nostr/watchHistory.js | 1 + js/watchHistoryService.js | 78 ++++++++++++++++-------------------- tests/watch-history.test.mjs | 21 ++++++++-- 3 files changed, 54 insertions(+), 46 deletions(-) diff --git a/js/nostr/watchHistory.js b/js/nostr/watchHistory.js index d8aaf773..d6a61453 100644 --- a/js/nostr/watchHistory.js +++ b/js/nostr/watchHistory.js @@ -1451,6 +1451,7 @@ class WatchHistoryManager { ok: allOk, retryable: anyRetryable, results, + items: results.flatMap(r => r.items || []), snapshotId: "", // No single snapshot ID // Return last event as pointerEvent? Or null? pointerEvent: results.length > 0 ? results[results.length-1].pointerEvent : null diff --git a/js/watchHistoryService.js b/js/watchHistoryService.js index 12583bbd..75ecc9d4 100644 --- a/js/watchHistoryService.js +++ b/js/watchHistoryService.js @@ -291,7 +291,6 @@ function getOrCreateQueue(actorKey) { queue = { items: new Map(), throttle: new Map(), - pendingSnapshotId: null, lastSnapshotReason: null, republishScheduled: false, }; @@ -382,11 +381,6 @@ function restoreQueueState() { }); } - if (typeof entry?.pendingSnapshotId === "string" && entry.pendingSnapshotId) { - queue.pendingSnapshotId = entry.pendingSnapshotId; - } else { - queue.pendingSnapshotId = null; - } if ( typeof entry?.lastSnapshotReason === "string" && entry.lastSnapshotReason @@ -396,7 +390,8 @@ function restoreQueueState() { queue.lastSnapshotReason = null; } - if (queue.pendingSnapshotId && isFeatureEnabled(actorKey)) { + // If there are items in the queue, schedule a sync + if (queue.items.size > 0 && isFeatureEnabled(actorKey)) { scheduleRepublishForQueue(actorKey); } } @@ -423,12 +418,11 @@ function persistQueueState() { for (const [actorKey, queue] of state.queues.entries()) { const items = collectQueueItems(actorKey); - if (!items.length && !queue.pendingSnapshotId) { + if (!items.length) { continue; } payload.actors[actorKey] = { items, - pendingSnapshotId: queue.pendingSnapshotId, lastSnapshotReason: queue.lastSnapshotReason, }; hasEntries = true; @@ -547,7 +541,6 @@ function clearQueue(actorKey) { } queue.items.clear(); queue.throttle.clear(); - queue.pendingSnapshotId = null; queue.republishScheduled = false; persistQueueState(); notifyQueueChange(actorKey); @@ -586,11 +579,6 @@ function pruneQueueAfterSnapshot(actorKey, queue, keysToClear, snapshotStart) { } } - if (queue.pendingSnapshotId) { - queue.pendingSnapshotId = null; - mutated = true; - } - if (queue.republishScheduled) { queue.republishScheduled = false; mutated = true; @@ -626,51 +614,55 @@ function scheduleRepublishForQueue(actorKey) { return; } const queue = state.queues.get(actorKey); - if (!queue || !queue.pendingSnapshotId || queue.republishScheduled) { + if (!queue || !queue.items.size || queue.republishScheduled) { return; } - const snapshotId = queue.pendingSnapshotId; + + const taskId = `watch-history-sync:${actorKey}`; queue.republishScheduled = true; + nostrClient.scheduleWatchHistoryRepublish( - snapshotId, + taskId, async (attempt) => { const latestItems = collectQueueItems(actorKey); if (!latestItems.length) { - queue.pendingSnapshotId = null; queue.republishScheduled = false; persistQueueState(); return { ok: true, skipped: true, actor: actorKey, - snapshotId, }; } - const publishResult = await nostrClient.publishWatchHistorySnapshot( + + const snapshotStartTime = Date.now(); + const keysToClear = new Set( + latestItems + .map((pointer) => pointerKey(pointer)) + .filter((keyValue) => typeof keyValue === "string" && keyValue) + ); + + const publishResult = await nostrClient.updateWatchHistoryList( latestItems, { actorPubkey: actorKey, - snapshotId, // kept for signature compatibility but unused attempt, - source: `${queue.lastSnapshotReason || "session"}-retry`, } ); - // snapshotId is no longer single-valued, so we don't update pendingSnapshotId with it if (publishResult?.ok) { - queue.pendingSnapshotId = null; - queue.republishScheduled = false; - queue.items.clear(); - queue.throttle.clear(); + pruneQueueAfterSnapshot( + actorKey, + queue, + keysToClear, + snapshotStartTime + ); await updateFingerprintCache( actorKey, publishResult.items || latestItems, ); - persistQueueState(); - notifyQueueChange(actorKey); } else if (!publishResult?.retryable) { - // If not retryable, we clear the pending state - queue.pendingSnapshotId = null; + // If not retryable, we clear the scheduled flag to allow new attempts later queue.republishScheduled = false; persistQueueState(); } @@ -680,7 +672,6 @@ function scheduleRepublishForQueue(actorKey) { onSchedule: ({ delay }) => { emit("republish-scheduled", { actor: actorKey, - snapshotId, delayMs: Number.isFinite(delay) ? delay : null, }); }, @@ -887,6 +878,9 @@ async function publishView(pointerInput, createdAt, metadata = {}) { watchedAt: normalizedPointer.watchedAt, } ); + } else { + // Automatically schedule a republish if enabled + scheduleRepublishForQueue(actorKey); } return viewResult; @@ -960,11 +954,10 @@ 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( + const publishResult = await nostrClient.updateWatchHistoryList( payloadItems, { actorPubkey: actorKey, - source: reason, } ); @@ -981,8 +974,6 @@ async function snapshot(items, options = {}) { if (!publishResult?.ok) { if (publishResult?.retryable) { if (queue) { - // Use a dummy ID or just a flag since we don't track single snapshot IDs anymore - queue.pendingSnapshotId = "pending-retry"; queue.lastSnapshotReason = reason; queue.republishScheduled = false; persistQueueState(); @@ -995,12 +986,13 @@ async function snapshot(items, options = {}) { throw error; } - if (!items) { - pruneQueueAfterSnapshot(actorKey, queue, queuePayloadKeys, snapshotStartTime); - } else if (queue) { - queue.pendingSnapshotId = null; - queue.republishScheduled = false; - persistQueueState(); + if (publishResult?.ok) { + if (!items) { + pruneQueueAfterSnapshot(actorKey, queue, queuePayloadKeys, snapshotStartTime); + } else if (queue) { + queue.republishScheduled = false; + persistQueueState(); + } } await updateFingerprintCache(actorKey, publishResult.items || payloadItems); diff --git a/tests/watch-history.test.mjs b/tests/watch-history.test.mjs index 26673f19..c611feb5 100644 --- a/tests/watch-history.test.mjs +++ b/tests/watch-history.test.mjs @@ -1129,14 +1129,25 @@ async function testWatchHistoryPartialRelayRetry() { let attemptIndex = 0; let currentAttempt = 0; - nostrClient.publishWatchHistorySnapshot = async function publishWithTracking( + const trackingWrapper = async function publishWithTracking( ...args ) { attemptIndex += 1; currentAttempt = attemptIndex; + // We must call the original method because it contains the logic that uses poolHarness + // However, we need to ensure we call the method that the service actually uses. + // Since service now calls updateWatchHistoryList, we should ideally wrap that. + // But updateWatchHistoryList calls publishRecords -> publishMonthRecord. + // If we wrap publishWatchHistorySnapshot, it might be bypassed if service calls updateWatchHistoryList directly. + // The service code I wrote calls updateWatchHistoryList. + // But wait, the previous code called updateWatchHistoryList. + // Let's hook into updateWatchHistoryList instead. return originalPublishSnapshot.apply(this, args); }; + nostrClient.publishWatchHistorySnapshot = trackingWrapper; + nostrClient.updateWatchHistoryList = trackingWrapper; + nostrClient.recordVideoView = async () => ({ ok: true, event: { @@ -1365,12 +1376,14 @@ async function testWatchHistorySnapshotRetainsNewQueueEntriesDuringPublish() { releasePublish = resolve; }); - nostrClient.publishWatchHistorySnapshot = async (items, options = {}) => { + const publishMock = async (items, options = {}) => { publishCalled = true; publishItems = Array.isArray(items) ? items.map((item) => ({ ...item })) : []; await publishGate; return { ok: true, items, snapshotId: "inflight-snapshot" }; }; + nostrClient.publishWatchHistorySnapshot = publishMock; + nostrClient.updateWatchHistoryList = publishMock; const snapshotPromise = watchHistoryService.snapshot(null, { actor, @@ -1768,7 +1781,7 @@ async function testWatchHistorySnapshotMergesQueuedWithCachedItems() { }); const publishedPayloads = []; - nostrClient.publishWatchHistorySnapshot = async (items, options = {}) => { + const payloadMock = async (items, options = {}) => { const clonedItems = Array.isArray(items) ? items.map((entry) => ({ ...entry })) : []; @@ -1783,6 +1796,8 @@ async function testWatchHistorySnapshotMergesQueuedWithCachedItems() { publishResults: {}, }; }; + nostrClient.publishWatchHistorySnapshot = payloadMock; + nostrClient.updateWatchHistoryList = payloadMock; nostrClient.getWatchHistoryFingerprint = async (_actor, items = []) => `fingerprint-${Array.isArray(items) ? items.length : 0}-${Date.now()}`;