mirror of
https://github.com/PR0M3TH3AN/bitvid.git
synced 2026-10-27 07:52:24 +00:00
refactor(views): batch kind-30079 view counts via SubscriptionManager (P3)
Replaces the per-card view-count transport (each card: list + NIP-45 COUNT +
live sub, × relays — the ~880 REQ/s storm) with ONE shared batched kind-30079
subscription + ONE backfill list() over all visible cards, bucketed back to
per-pointer state by #a. Re-enables the VideoListView circuit-breaker.
Grid counts are convergent/approximate (cheap, persists via the existing view
cache); the watch page opts into an exact NIP-45 COUNT for its single pointer
(modalCoordinator passes { exact: true }) — accurate where it matters, one
request, no storm. Legacy per-pointer path retained for SubscriptionManager-less
environments (tests/mocks).
Verified: nip07-heat steady-idle with views on; flaky multi-relay harness still
all-PASS; new cheat-resistant batching+bucketing test. (Pre-existing
view-counter publish-session-tag failure is unrelated — fails at baseline.)
Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
1 parent
1ac03c49e9
commit
dde0c35477
4 files changed
+331
-8
No files matched your search
@@ -169,7 +169,7 @@ export function createModalCoordinator(deps) {
|
||||
}
|
||||
latestViewEl.dataset.viewCountState = status;
|
||||
}
|
||||
});
|
||||
}, { exact: true });
|
||||
|
||||
this.modalViewCountUnsub = () => {
|
||||
try {
|
||||
|
||||
@@ -1151,11 +1151,9 @@ export class VideoListView {
|
||||
}
|
||||
|
||||
ensureViewCountSubscription(pointerInfo) {
|
||||
// CIRCUIT-BREAKER (confirmed against real env): one live kind-30079 view
|
||||
// subscription per card, fanned across ~20 relays, contributed ~880 REQ/s.
|
||||
// Disabled until batched into a single count query. See refactor plan.
|
||||
return null;
|
||||
// eslint-disable-next-line no-unreachable
|
||||
// View counts now flow through viewCounter's batched transport (one shared
|
||||
// kind-30079 subscription + backfill over all visible cards), so this is
|
||||
// back on. See docs/architecture-refactor.md (P3).
|
||||
if (!pointerInfo || !pointerInfo.key || !pointerInfo.pointer) {
|
||||
return null;
|
||||
}
|
||||
|
||||
+246
-2
@@ -42,6 +42,221 @@ const pointerListeners = new Map();
|
||||
let persistTimer = null;
|
||||
let nextTokenId = 1;
|
||||
|
||||
// ---- Batched relay transport (P3 — docs/architecture-refactor.md) ----
|
||||
// All grid view-count lookups share ONE kind-30079 live subscription and ONE
|
||||
// backfill list() over the active pointer set, bucketed back to per-pointer
|
||||
// state. This replaces the per-card relay subscription that stormed O(cards ×
|
||||
// relays). Falls back to the legacy per-pointer path when no SubscriptionManager
|
||||
// is available (unit tests / mock clients). The watch page additionally runs an
|
||||
// exact NIP-45 COUNT for its single pointer (opt-in via options.exact).
|
||||
const VIEW_EVENT_KIND = 30079;
|
||||
const VIEW_FILTER_BATCH = 200;
|
||||
const activeKeys = new Set();
|
||||
const hydratedKeys = new Set();
|
||||
let batchedLiveHandle = null;
|
||||
let batchedLiveSignature = "";
|
||||
let liveRefreshTimer = null;
|
||||
const backfillPending = new Set();
|
||||
let backfillTimer = null;
|
||||
|
||||
function getSubscriptionManager() {
|
||||
try {
|
||||
return typeof nostrClient?.getSubscriptionManager === "function"
|
||||
? nostrClient.getSubscriptionManager()
|
||||
: null;
|
||||
} catch (_) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function buildViewFilters(keys, sinceSeconds) {
|
||||
const aVals = [];
|
||||
const eVals = [];
|
||||
for (const key of keys) {
|
||||
const pointer = pointerListeners.get(key)?.pointer;
|
||||
if (!pointer || typeof pointer.value !== "string") {
|
||||
continue;
|
||||
}
|
||||
(pointer.type === "a" ? aVals : eVals).push(pointer.value);
|
||||
}
|
||||
const filters = [];
|
||||
const pushChunks = (tag, vals) => {
|
||||
for (let i = 0; i < vals.length; i += VIEW_FILTER_BATCH) {
|
||||
const filter = { kinds: [VIEW_EVENT_KIND], "#t": ["view"] };
|
||||
filter[tag] = vals.slice(i, i + VIEW_FILTER_BATCH);
|
||||
if (Number.isFinite(sinceSeconds)) {
|
||||
filter.since = sinceSeconds;
|
||||
}
|
||||
filters.push(filter);
|
||||
}
|
||||
};
|
||||
pushChunks("#a", aVals);
|
||||
pushChunks("#e", eVals);
|
||||
return filters;
|
||||
}
|
||||
|
||||
// Map a batched kind-30079 event back to the active pointer key it belongs to.
|
||||
function resolveEventKey(event) {
|
||||
if (!event || !Array.isArray(event.tags)) {
|
||||
return null;
|
||||
}
|
||||
for (const tag of event.tags) {
|
||||
if (!Array.isArray(tag) || tag.length < 2) {
|
||||
continue;
|
||||
}
|
||||
const [label, value] = tag;
|
||||
if ((label === "a" || label === "e") && typeof value === "string") {
|
||||
const key = `${label}:${value.trim()}`;
|
||||
if (activeKeys.has(key)) {
|
||||
return key;
|
||||
}
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function applyBatchedEvent(event) {
|
||||
const key = resolveEventKey(event);
|
||||
if (!key) {
|
||||
return;
|
||||
}
|
||||
if (applyEventToState(key, event)) {
|
||||
schedulePersist();
|
||||
notifyHandlers(key);
|
||||
}
|
||||
}
|
||||
|
||||
function scheduleLiveRefresh() {
|
||||
if (liveRefreshTimer) {
|
||||
return;
|
||||
}
|
||||
liveRefreshTimer = setTimeout(() => {
|
||||
liveRefreshTimer = null;
|
||||
refreshBatchedLiveSubscription();
|
||||
}, 0);
|
||||
}
|
||||
|
||||
function refreshBatchedLiveSubscription() {
|
||||
const manager = getSubscriptionManager();
|
||||
if (!manager) {
|
||||
return;
|
||||
}
|
||||
if (!activeKeys.size) {
|
||||
if (batchedLiveHandle) {
|
||||
batchedLiveHandle.close();
|
||||
batchedLiveHandle = null;
|
||||
batchedLiveSignature = "";
|
||||
}
|
||||
return;
|
||||
}
|
||||
const since =
|
||||
Math.floor(Date.now() / 1000) - Math.max(1, VIEW_COUNT_DEDUPE_WINDOW_SECONDS);
|
||||
const filters = buildViewFilters(activeKeys, since);
|
||||
if (!filters.length) {
|
||||
return;
|
||||
}
|
||||
const signature = JSON.stringify(filters);
|
||||
if (batchedLiveHandle && signature === batchedLiveSignature) {
|
||||
return;
|
||||
}
|
||||
batchedLiveSignature = signature;
|
||||
if (batchedLiveHandle) {
|
||||
batchedLiveHandle.update({ filters });
|
||||
} else {
|
||||
batchedLiveHandle = manager.subscribe({
|
||||
key: "viewcounts:live",
|
||||
label: "view-counts",
|
||||
filters,
|
||||
onEvent: (event) => applyBatchedEvent(event),
|
||||
});
|
||||
}
|
||||
for (const key of activeKeys) {
|
||||
setPointerStatus(key, "live");
|
||||
}
|
||||
}
|
||||
|
||||
function scheduleBackfill(key) {
|
||||
backfillPending.add(key);
|
||||
if (backfillTimer) {
|
||||
return;
|
||||
}
|
||||
backfillTimer = setTimeout(() => {
|
||||
backfillTimer = null;
|
||||
runBatchedBackfill();
|
||||
}, 0);
|
||||
}
|
||||
|
||||
async function runBatchedBackfill() {
|
||||
const manager = getSubscriptionManager();
|
||||
if (!manager) {
|
||||
return;
|
||||
}
|
||||
const keys = Array.from(backfillPending).filter(
|
||||
(k) => activeKeys.has(k) && !hydratedKeys.has(k),
|
||||
);
|
||||
backfillPending.clear();
|
||||
if (!keys.length) {
|
||||
return;
|
||||
}
|
||||
for (const key of keys) {
|
||||
setPointerStatus(key, "hydrating");
|
||||
}
|
||||
const since =
|
||||
Math.floor(Date.now() / 1000) - VIEW_COUNT_BACKFILL_MAX_DAYS * SECONDS_PER_DAY;
|
||||
const filters = buildViewFilters(keys, since);
|
||||
let events = [];
|
||||
try {
|
||||
events = await manager.list({ filters });
|
||||
} catch (error) {
|
||||
userLogger.warn("[viewCounter] Batched view backfill failed:", error);
|
||||
}
|
||||
if (Array.isArray(events)) {
|
||||
for (const event of events) {
|
||||
applyBatchedEvent(event);
|
||||
}
|
||||
}
|
||||
for (const key of keys) {
|
||||
hydratedKeys.add(key);
|
||||
const state = ensurePointerState(key);
|
||||
state.lastSyncedAt = Date.now();
|
||||
setPointerStatus(key, batchedLiveHandle ? "live" : "idle");
|
||||
}
|
||||
schedulePersist();
|
||||
}
|
||||
|
||||
// Exact NIP-45 COUNT for a single pointer (watch page). Cheap — one pointer,
|
||||
// no storm — and accurate where a viewer might actually scrutinize the number.
|
||||
async function exactCountForPointer(key, pointer, options = {}) {
|
||||
try {
|
||||
const result = await countVideoViewEventsApi(pointer, {
|
||||
relays: options.relays,
|
||||
signal: options.signal,
|
||||
});
|
||||
const best = Number.isFinite(result?.best?.count)
|
||||
? Number(result.best.count)
|
||||
: Number.isFinite(result?.total)
|
||||
? Number(result.total)
|
||||
: null;
|
||||
if (best === null) {
|
||||
return;
|
||||
}
|
||||
const state = ensurePointerState(key);
|
||||
if (result && typeof result === "object") {
|
||||
state.partial = Boolean(result.partial);
|
||||
}
|
||||
if (best > state.total) {
|
||||
state.total = best;
|
||||
state.lastSyncedAt = Date.now();
|
||||
}
|
||||
schedulePersist();
|
||||
notifyHandlers(key);
|
||||
} catch (error) {
|
||||
if (error?.name !== "AbortError") {
|
||||
userLogger.warn("[viewCounter] Exact view count failed:", error);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
restoreCacheSnapshot();
|
||||
|
||||
/**
|
||||
@@ -537,6 +752,14 @@ async function hydratePointer(key, listeners) {
|
||||
}
|
||||
|
||||
function ensureHydration(key, listeners) {
|
||||
// Batched path: one shared backfill list() over the active pointer set.
|
||||
if (getSubscriptionManager()) {
|
||||
if (!hydratedKeys.has(key)) {
|
||||
scheduleBackfill(key);
|
||||
}
|
||||
return;
|
||||
}
|
||||
// Legacy per-pointer fallback (no SubscriptionManager — tests / mocks).
|
||||
if (!listeners.hydrationPromise) {
|
||||
listeners.hydrationPromise = hydratePointer(key, listeners).finally(() => {
|
||||
listeners.hydrationPromise = null;
|
||||
@@ -553,6 +776,13 @@ function ensureHydration(key, listeners) {
|
||||
}
|
||||
|
||||
function ensureLiveSubscription(key, listeners) {
|
||||
// Batched path: the single shared live subscription already covers this key
|
||||
// (added to activeKeys on subscribe); just (re)issue it, coalesced.
|
||||
if (getSubscriptionManager()) {
|
||||
scheduleLiveRefresh();
|
||||
return;
|
||||
}
|
||||
// Legacy per-pointer fallback (no SubscriptionManager — tests / mocks).
|
||||
if (listeners.liveUnsub || !listeners.handlers.size) {
|
||||
return;
|
||||
}
|
||||
@@ -603,6 +833,7 @@ export function subscribeToVideoViewCount(pointerInput, handler, options = {}) {
|
||||
const { key, pointer } = canonicalizePointer(pointerInput);
|
||||
const listeners = ensurePointerListeners(key, pointer);
|
||||
listeners.options = mergeListenerOptions(listeners.options, options);
|
||||
activeKeys.add(key);
|
||||
const token = `token-${nextTokenId++}`;
|
||||
listeners.handlers.set(token, handler);
|
||||
const state = ensurePointerState(key);
|
||||
@@ -618,6 +849,10 @@ export function subscribeToVideoViewCount(pointerInput, handler, options = {}) {
|
||||
}
|
||||
ensureHydration(key, listeners);
|
||||
ensureLiveSubscription(key, listeners);
|
||||
// Watch page (single video) opts into an exact NIP-45 COUNT for accuracy.
|
||||
if (options && options.exact && getSubscriptionManager()) {
|
||||
void exactCountForPointer(key, pointer, options);
|
||||
}
|
||||
return token;
|
||||
}
|
||||
|
||||
@@ -631,8 +866,17 @@ export function unsubscribeFromVideoViewCount(pointerInput, token) {
|
||||
return;
|
||||
}
|
||||
listeners.handlers.delete(token);
|
||||
if (!listeners.handlers.size && listeners.liveUnsub) {
|
||||
listeners.liveUnsub();
|
||||
if (!listeners.handlers.size) {
|
||||
if (listeners.liveUnsub) {
|
||||
listeners.liveUnsub(); // legacy per-pointer sub
|
||||
}
|
||||
// Batched path: drop from the active set and re-issue the shared sub so it
|
||||
// no longer covers this pointer. State stays cached in pointerStates.
|
||||
if (activeKeys.delete(key)) {
|
||||
hydratedKeys.delete(key);
|
||||
pointerListeners.delete(key);
|
||||
scheduleLiveRefresh();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
// Scenario (SCN-view-counts-batched):
|
||||
// Given multiple grid cards subscribe to view counts,
|
||||
// When a SubscriptionManager is available,
|
||||
// Then they share ONE batched kind-30079 subscription whose filter covers all
|
||||
// their addresses, and incoming view events are bucketed back to the correct
|
||||
// per-pointer count (one video's views never leak into another's).
|
||||
//
|
||||
// P3 fix for the per-card view-count storm. See docs/architecture-refactor.md.
|
||||
|
||||
import "./test-helpers/setup-localstorage.mjs";
|
||||
import { test } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
|
||||
const { nostrClient } = await import("../js/nostrClientFacade.js");
|
||||
const vc = await import("../js/viewCounter.js");
|
||||
|
||||
const tick = (ms = 20) => new Promise((r) => setTimeout(r, ms));
|
||||
|
||||
test("grid view counts share one batched sub and bucket by #a", async (t) => {
|
||||
const original = nostrClient.getSubscriptionManager;
|
||||
let subscribeCalls = 0;
|
||||
let listCalls = 0;
|
||||
let lastFilters = null;
|
||||
let onEvent = null;
|
||||
const handle = {
|
||||
update({ filters }) {
|
||||
lastFilters = filters;
|
||||
},
|
||||
close() {},
|
||||
};
|
||||
nostrClient.getSubscriptionManager = () => ({
|
||||
subscribe(opts) {
|
||||
subscribeCalls += 1;
|
||||
lastFilters = opts.filters;
|
||||
onEvent = opts.onEvent;
|
||||
return handle;
|
||||
},
|
||||
list() {
|
||||
listCalls += 1;
|
||||
return Promise.resolve([]);
|
||||
},
|
||||
});
|
||||
t.after(() => {
|
||||
nostrClient.getSubscriptionManager = original;
|
||||
});
|
||||
|
||||
const pk = "a".repeat(64);
|
||||
const A1 = `30078:${pk}:vid-batch-1`;
|
||||
const A2 = `30078:${pk}:vid-batch-2`;
|
||||
let total1 = null;
|
||||
let total2 = null;
|
||||
const tok1 = vc.subscribeToVideoViewCount({ type: "a", value: A1 }, ({ total }) => {
|
||||
total1 = total;
|
||||
});
|
||||
const tok2 = vc.subscribeToVideoViewCount({ type: "a", value: A2 }, ({ total }) => {
|
||||
total2 = total;
|
||||
});
|
||||
|
||||
await tick(); // let the coalesced live refresh + backfill run
|
||||
|
||||
assert.equal(subscribeCalls, 1, "exactly one batched subscription for both cards");
|
||||
const aVals = (lastFilters || []).flatMap((f) => f["#a"] || []);
|
||||
assert.ok(
|
||||
aVals.includes(A1) && aVals.includes(A2),
|
||||
"the single batched filter covers both video addresses",
|
||||
);
|
||||
assert.ok(typeof onEvent === "function", "batched sub wired an onEvent handler");
|
||||
|
||||
// Deliver a view event for A1 only — only A1's count should move.
|
||||
onEvent({
|
||||
id: "view-1",
|
||||
pubkey: "b".repeat(64),
|
||||
created_at: Math.floor(Date.now() / 1000),
|
||||
tags: [["a", A1], ["t", "view"]],
|
||||
});
|
||||
assert.equal(total1, 1, "A1 view count incremented");
|
||||
assert.equal(total2, 0, "A2 view count unaffected (correct per-#a bucketing)");
|
||||
|
||||
vc.unsubscribeFromVideoViewCount({ type: "a", value: A1 }, tok1);
|
||||
vc.unsubscribeFromVideoViewCount({ type: "a", value: A2 }, tok2);
|
||||
});
|
||||
Reference in new issue
Block a user