diff --git a/js/nostr/client.js b/js/nostr/client.js index 672f2bc8..5d33da9c 100644 --- a/js/nostr/client.js +++ b/js/nostr/client.js @@ -225,6 +225,7 @@ import { prepareVideoEditPayload, } from "./videoPayloadBuilder.js"; import { inferMimeTypeFromUrl } from "../utils/mime.js"; +import { VideoEventBuffer } from "./videoEventBuffer.js"; function normalizeProfileFromEvent(event) { if (!event || !event.content) return null; @@ -3171,161 +3172,21 @@ export class NostrClient { devLogger.log("[subscribeVideos] Subscribing with filter:", filter); const sub = this.pool.sub(this.getHealthyRelays(this.relays), [filter]); - const invalidDuringSub = []; // BUFFERING STATE // We collect events here instead of processing them instantly to avoid // 1000s of React re-renders during the initial relay dump. // The buffer acts as a pressure valve between the network and the UI. - let eventBuffer = []; - const EVENT_FLUSH_DEBOUNCE_MS = 75; - let flushTimerId = null; - - /** - * Flushes the pending event buffer. - * - * This batch processing is critical for performance during the initial - * relay dump (thousands of events). It prevents React from re-rendering - * for every single event. - * - * Algorithm: - * 1. Drain the buffer. - * 2. Validate & Convert events to Video objects. - * 3. Apply tombstones (filter out known deleted items). - * 4. Update `activeMap` (Last-Write-Wins based on created_at). - * 5. Persist the new state to cache. - */ - const flushEventBuffer = () => { - if (!eventBuffer.length) { - return; - } - - const toProcess = eventBuffer; - eventBuffer = []; - const updatedVideos = []; - - for (const evt of toProcess) { - try { - if (evt && evt.id) { - this.rawEvents.set(evt.id, evt); - } - const video = convertEventToVideo(evt); - - if (video.invalid) { - invalidDuringSub.push({ id: video.id, reason: video.reason }); - continue; - } - - // Merge any NIP-71 metadata we might already have cached for this video - this.mergeNip71MetadataIntoVideo(video); - // Determine the "true" creation time of the root video - this.applyRootCreatedAt(video); - - const activeKey = getActiveKey(video); - const wasDeletedEvent = video.deleted === true; - - // If this is a deletion event (Kind 5 or deletion marker), record a tombstone - // to prevent older versions from resurrecting. - if (wasDeletedEvent) { - this.recordTombstone(activeKey, video.created_at); - } else { - // Otherwise, check if this video is already known to be deleted - this.applyTombstoneGuard(video); - } - - // Store in allEvents (history preservation) - this.allEvents.set(evt.id, video); - this.dirtyEventIds.add(evt.id); - - // If it's a "deleted" note, remove from activeMap - if (video.deleted) { - if (activeKey) { - if (wasDeletedEvent) { - this.activeMap.delete(activeKey); - } else { - const currentActive = this.activeMap.get(activeKey); - if (currentActive?.id === video.id) { - this.activeMap.delete(activeKey); - } - } - } - continue; - } - - // LATEST-WINS LOGIC - // We only update the UI if the incoming video is newer than what we have. - // This handles the "Edit" case where multiple versions exist on relays. - const prevActive = this.activeMap.get(activeKey); - if (!prevActive || video.created_at > prevActive.created_at) { - this.activeMap.set(activeKey, video); - updatedVideos.push(video); - } - } catch (err) { - devLogger.error("[subscribeVideos] Error processing event:", err); - } - } - - if (updatedVideos.length > 0) { - // Trigger the callback once per batch to avoid UI thrashing - onVideo(updatedVideos); - - // Fetch NIP-71 metadata (categorization tags) in the background for the whole batch - this.populateNip71MetadataForVideos(updatedVideos) - .then(() => { - for (const video of updatedVideos) { - this.applyRootCreatedAt(video); - } - }) - .catch((error) => { - devLogger.warn( - "[nostr] Failed to hydrate NIP-71 metadata for live video batch:", - error - ); - }); - } - - // Persist processed events after each flush so reloads warm quickly. - this.saveLocalData("subscribeVideos:flush"); - }; - - const scheduleFlush = (immediate = false) => { - if (flushTimerId) { - if (!immediate) { - return; - } - clearTimeout(flushTimerId); - flushTimerId = null; - } - - if (immediate) { - flushEventBuffer(); - return; - } - - flushTimerId = setTimeout(() => { - flushTimerId = null; - flushEventBuffer(); - }, EVENT_FLUSH_DEBOUNCE_MS); - }; + const buffer = new VideoEventBuffer(this, onVideo); // 1) On each incoming event, just push to the buffer and schedule a flush sub.on("event", (event) => { - eventBuffer.push(event); - scheduleFlush(false); + buffer.push(event); }); // You can still use sub.on("eose") if needed sub.on("eose", () => { - if (isDevMode && invalidDuringSub.length > 0) { - userLogger.warn( - `[subscribeVideos] found ${invalidDuringSub.length} invalid video notes (with reasons):`, - invalidDuringSub - ); - } - devLogger.log( - "[subscribeVideos] Reached EOSE for all relays (historical load done)" - ); - scheduleFlush(true); + buffer.handleEose(); }); // Return the subscription object if you need to unsub manually later @@ -3337,12 +3198,7 @@ export class NostrClient { return; } unsubscribed = true; - if (flushTimerId) { - clearTimeout(flushTimerId); - flushTimerId = null; - } - // Ensure any straggling events are flushed before tearing down. - flushEventBuffer(); + buffer.cleanup(); try { return originalUnsub(); } catch (err) { diff --git a/js/nostr/videoEventBuffer.js b/js/nostr/videoEventBuffer.js new file mode 100644 index 00000000..1941b4a3 --- /dev/null +++ b/js/nostr/videoEventBuffer.js @@ -0,0 +1,161 @@ +/** + * VideoEventBuffer + * + * Handles buffering and batch processing of incoming video events + * to prevent UI thrashing during high-volume updates (e.g. initial load). + */ + +import { isDevMode } from "../config.js"; +import { devLogger, userLogger } from "../utils/logger.js"; +import { convertEventToVideo } from "./nip71.js"; + +export class VideoEventBuffer { + constructor(client, onVideo) { + this.client = client; + this.onVideo = onVideo; + this.buffer = []; + this.invalidEvents = []; + this.flushTimerId = null; + this.FLUSH_DEBOUNCE_MS = 75; + } + + push(event) { + this.buffer.push(event); + this.scheduleFlush(false); + } + + scheduleFlush(immediate = false) { + if (this.flushTimerId) { + if (!immediate) { + return; + } + clearTimeout(this.flushTimerId); + this.flushTimerId = null; + } + + if (immediate) { + this.flush(); + return; + } + + this.flushTimerId = setTimeout(() => { + this.flushTimerId = null; + this.flush(); + }, this.FLUSH_DEBOUNCE_MS); + } + + flush() { + if (!this.buffer.length) { + return; + } + + const toProcess = this.buffer; + this.buffer = []; + const updatedVideos = []; + + for (const evt of toProcess) { + try { + if (evt && evt.id) { + this.client.rawEvents.set(evt.id, evt); + } + const video = convertEventToVideo(evt); + + if (video.invalid) { + this.invalidEvents.push({ id: video.id, reason: video.reason }); + continue; + } + + // Merge any NIP-71 metadata we might already have cached for this video + this.client.mergeNip71MetadataIntoVideo(video); + // Determine the "true" creation time of the root video + this.client.applyRootCreatedAt(video); + + const activeKey = this.client.getActiveKey(video); + const wasDeletedEvent = video.deleted === true; + + // If this is a deletion event (Kind 5 or deletion marker), record a tombstone + // to prevent older versions from resurrecting. + if (wasDeletedEvent) { + this.client.recordTombstone(activeKey, video.created_at); + } else { + // Otherwise, check if this video is already known to be deleted + this.client.applyTombstoneGuard(video); + } + + // Store in allEvents (history preservation) + this.client.allEvents.set(evt.id, video); + this.client.dirtyEventIds.add(evt.id); + + // If it's a "deleted" note, remove from activeMap + if (video.deleted) { + if (activeKey) { + if (wasDeletedEvent) { + this.client.activeMap.delete(activeKey); + } else { + const currentActive = this.client.activeMap.get(activeKey); + if (currentActive?.id === video.id) { + this.client.activeMap.delete(activeKey); + } + } + } + continue; + } + + // LATEST-WINS LOGIC + // We only update the UI if the incoming video is newer than what we have. + // This handles the "Edit" case where multiple versions exist on relays. + const prevActive = this.client.activeMap.get(activeKey); + if (!prevActive || video.created_at > prevActive.created_at) { + this.client.activeMap.set(activeKey, video); + updatedVideos.push(video); + } + } catch (err) { + devLogger.error("[VideoEventBuffer] Error processing event:", err); + } + } + + if (updatedVideos.length > 0) { + // Trigger the callback once per batch to avoid UI thrashing + this.onVideo(updatedVideos); + + // Fetch NIP-71 metadata (categorization tags) in the background for the whole batch + this.client.populateNip71MetadataForVideos(updatedVideos) + .then(() => { + for (const video of updatedVideos) { + this.client.applyRootCreatedAt(video); + } + }) + .catch((error) => { + devLogger.warn( + "[nostr] Failed to hydrate NIP-71 metadata for live video batch:", + error + ); + }); + } + + // Persist processed events after each flush so reloads warm quickly. + this.client.saveLocalData("subscribeVideos:flush"); + } + + handleEose() { + if (isDevMode && this.invalidEvents.length > 0) { + userLogger.warn( + `[subscribeVideos] found ${this.invalidEvents.length} invalid video notes (with reasons):`, + this.invalidEvents + ); + } + devLogger.log( + "[subscribeVideos] Reached EOSE for all relays (historical load done)" + ); + this.scheduleFlush(true); + } + + cleanup() { + if (this.flushTimerId) { + clearTimeout(this.flushTimerId); + this.flushTimerId = null; + } + // Ensure any straggling events are flushed before tearing down. + this.flush(); + } +} diff --git a/tests/nostr/videoEventBuffer.test.mjs b/tests/nostr/videoEventBuffer.test.mjs new file mode 100644 index 00000000..025dc550 --- /dev/null +++ b/tests/nostr/videoEventBuffer.test.mjs @@ -0,0 +1,154 @@ + +import { test } from 'node:test'; +import { strict as assert } from 'node:assert'; +import { VideoEventBuffer } from "../../js/nostr/videoEventBuffer.js"; + +class MockClient { + constructor() { + this.rawEvents = new Map(); + this.allEvents = new Map(); + this.activeMap = new Map(); + this.dirtyEventIds = new Set(); + this.tombstones = new Map(); + this.saveLocalDataCalls = []; + this.populateNip71MetadataCalls = []; + } + + getActiveKey(video) { + return video.videoRootId || video.id; + } + + recordTombstone(key, timestamp) { + this.tombstones.set(key, timestamp); + } + + applyTombstoneGuard(video) { + const key = this.getActiveKey(video); + const ts = this.tombstones.get(key); + if (ts && video.created_at <= ts) { + video.deleted = true; + } + } + + mergeNip71MetadataIntoVideo(video) {} + applyRootCreatedAt(video) {} + + async populateNip71MetadataForVideos(videos) { + this.populateNip71MetadataCalls.push(videos); + return Promise.resolve(); + } + + saveLocalData(reason) { + this.saveLocalDataCalls.push(reason); + } +} + +const validEvent = { + id: "id1", + pubkey: "pubkey1", + created_at: 1000, + kind: 30078, + tags: [["t", "video"]], + content: JSON.stringify({ + title: "Test Video", + url: "https://example.com/video.mp4", + magnet: "magnet:?xt=urn:btih:1234567890abcdef1234567890abcdef12345678&dn=video", + thumbnail: "https://example.com/thumb.jpg", + videoRootId: "root1" + }) +}; + +const newerEvent = { + ...validEvent, + id: "id2", + created_at: 2000 +}; + +const deletedEvent = { + ...validEvent, + id: "id3", + created_at: 3000, + content: JSON.stringify({ + deleted: true, + videoRootId: "root1", + url: "https://example.com/video.mp4", + title: "Deleted Video" + }) +}; + +test('VideoEventBuffer', async (t) => { + await t.test('Push and Flush', () => { + const client = new MockClient(); + let onVideoCalls = []; + const onVideo = (videos) => onVideoCalls.push(videos); + const buffer = new VideoEventBuffer(client, onVideo); + + buffer.push(validEvent); + assert.equal(buffer.buffer.length, 1); + assert.notEqual(buffer.flushTimerId, null); + + // Force flush + buffer.scheduleFlush(true); + assert.equal(buffer.buffer.length, 0); + assert.equal(client.allEvents.size, 1); + assert.equal(client.activeMap.size, 1); + assert.equal(onVideoCalls.length, 1); + + // videoRootId is "root1", so key is "root1" (based on MockClient.getActiveKey) + assert.ok(client.activeMap.has("root1")); + assert.equal(client.activeMap.get("root1").id, validEvent.id); + }); + + await t.test('Latest Wins', () => { + const client = new MockClient(); + let onVideoCalls = []; + const onVideo = (videos) => onVideoCalls.push(videos); + const buffer = new VideoEventBuffer(client, onVideo); + + // Setup: activeMap has older event + client.activeMap.set("root1", { ...validEvent, created_at: 1000, videoRootId: "root1" }); + + buffer.push(newerEvent); + buffer.scheduleFlush(true); // Force flush + + // Should update because newerEvent.created_at (2000) > 1000 + assert.equal(client.activeMap.get("root1").id, newerEvent.id); + assert.equal(client.activeMap.get("root1").created_at, 2000); + }); + + await t.test('Tombstone Handling', () => { + const client = new MockClient(); + const buffer = new VideoEventBuffer(client, () => {}); + + // Process deleted event + buffer.push(deletedEvent); + buffer.scheduleFlush(true); + + // Should verify tombstone recorded + assert.ok(client.tombstones.has("root1")); + assert.equal(client.tombstones.get("root1"), deletedEvent.created_at); + + // Should NOT be in activeMap + assert.ok(!client.activeMap.has("root1")); + + // Now try pushing an older event + buffer.push(newerEvent); // created_at 2000 < 3000 + buffer.scheduleFlush(true); + + // Should be guarded by tombstone + assert.ok(!client.activeMap.has("root1")); + }); + + await t.test('Cleanup', () => { + const client = new MockClient(); + const buffer = new VideoEventBuffer(client, () => {}); + + buffer.push(validEvent); + assert.notEqual(buffer.flushTimerId, null); + + buffer.cleanup(); + assert.equal(buffer.flushTimerId, null); + assert.equal(buffer.buffer.length, 0); // Should have flushed + assert.equal(client.allEvents.size, 1); + }); +});