mirror of
https://github.com/PR0M3TH3AN/bitvid.git
synced 2026-10-27 07:52:24 +00:00
Refactor subscribeVideos buffering logic to VideoEventBuffer class
Separated the complex event buffering and flushing logic from `NostrClient.subscribeVideos` into a dedicated `VideoEventBuffer` utility class. This improves readability and maintainability of the `client.js` file. - Created `js/nostr/videoEventBuffer.js` - Updated `js/nostr/client.js` to use `VideoEventBuffer` - Added unit tests in `tests/nostr/videoEventBuffer.test.mjs` Co-authored-by: PR0M3TH3AN <[email protected]>
This commit is contained in:
google-labs-jules[bot]andPR0M3TH3AN committed
1 parent
11df8f79a3
commit
0b35e2639b
3 files changed
+320
-149
No files matched your search
+5
-149
@@ -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) {
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user