diff --git a/js/reactionCounter.js b/js/reactionCounter.js index 7bed133a..102ed6a3 100644 --- a/js/reactionCounter.js +++ b/js/reactionCounter.js @@ -1,9 +1,15 @@ import { publishVideoReaction } from "./nostr.js"; +import { nostrClient } from "./nostrClientFacade.js"; import { normalizePointerInput, pointerKey } from "./nostr/watchHistory.js"; import { devLogger, userLogger } from "./utils/logger.js"; const pointerStates = new Map(); const pointerHandlers = new Map(); +const pointerHydrationPromises = new Map(); +const pointerSubscriptions = new Map(); + +const REACTION_EVENT_KIND = 7; +const REACTION_HYDRATION_LIMIT = 400; function clonePointerDescriptor(pointer) { if (!pointer || typeof pointer !== "object") { @@ -43,6 +49,9 @@ function ensurePointerState(key, pointer) { totals: new Map(), reactionsByPubkey: new Map(), lastUpdatedAt: 0, + latestCreatedAt: 0, + lastSyncedAt: 0, + hydrated: false, }; pointerStates.set(key, state); return state; @@ -164,6 +173,12 @@ function applyReactionToState(key, pointer, event) { ) { state.reactionsByPubkey.set(normalized.pubkey, normalized); state.lastUpdatedAt = Date.now(); + if (Number.isFinite(normalized.created_at)) { + const created = Math.max(0, Math.floor(normalized.created_at)); + if (!Number.isFinite(state.latestCreatedAt) || created > state.latestCreatedAt) { + state.latestCreatedAt = created; + } + } if (!changed) { changed = true; } @@ -187,12 +202,18 @@ function snapshotState(state) { eventId: record.eventId, }; } + const latestCreatedAt = Number.isFinite(state.latestCreatedAt) + ? Math.max(0, Math.floor(state.latestCreatedAt)) + : 0; return { pointer: state.pointer ? { ...state.pointer } : null, total, counts, reactions, lastUpdatedAt: state.lastUpdatedAt, + latestCreatedAt, + lastSyncedAt: state.lastSyncedAt, + hydrated: Boolean(state.hydrated), }; } @@ -245,6 +266,61 @@ function subscribeToPointer(pointerInput, handler) { } } + const hydrationPromise = ensureReactionHydration(key, pointer, state); + if (hydrationPromise && typeof hydrationPromise.then === "function") { + hydrationPromise + .catch((error) => { + devLogger.warn( + "[reactionCounter] Failed to hydrate reaction history:", + error, + ); + }) + .finally(() => { + if (hasHandlers(key)) { + ensureReactionLiveSubscription(key, pointer); + } + }); + } else { + const liveToken = ensureReactionLiveSubscription(key, pointer); + if ( + !liveToken && + !hydrationPromise && + !nostrClient?.pool && + typeof nostrClient?.ensurePool === "function" + ) { + Promise.resolve() + .then(() => nostrClient.ensurePool()) + .then(() => { + if (!hasHandlers(key)) { + return; + } + const followUp = ensureReactionHydration(key, pointer, state); + if (followUp && typeof followUp.then === "function") { + followUp + .catch((error) => { + devLogger.warn( + "[reactionCounter] Failed to hydrate reaction history after pool init:", + error, + ); + }) + .finally(() => { + if (hasHandlers(key)) { + ensureReactionLiveSubscription(key, pointer); + } + }); + } else if (hasHandlers(key)) { + ensureReactionLiveSubscription(key, pointer); + } + }) + .catch((error) => { + devLogger.warn( + "[reactionCounter] Failed to initialize nostr pool for reactions:", + error, + ); + }); + } + } + return () => { unsubscribeFromPointer(pointer, handler); }; @@ -270,6 +346,7 @@ function unsubscribeFromPointer(pointerInput, handler) { if (handlers.size === 0) { pointerHandlers.delete(key); + teardownLiveSubscription(key); } } @@ -305,6 +382,291 @@ export function ingestLocalReaction({ event, pointer }) { } } +function hasHandlers(key) { + const handlers = pointerHandlers.get(key); + return Boolean(handlers && handlers.size); +} + +function getPointerSignature(pointer) { + if (!pointer || typeof pointer !== "object") { + return ""; + } + const type = pointer.type === "a" ? "a" : pointer.type === "e" ? "e" : ""; + const value = + typeof pointer.value === "string" && pointer.value.trim() + ? pointer.value.trim() + : ""; + if (!type || !value) { + return ""; + } + const relay = + typeof pointer.relay === "string" && pointer.relay.trim() + ? pointer.relay.trim() + : ""; + return `${type}:${value}:${relay}`; +} + +function gatherRelayCandidates(pointer) { + const relays = []; + const seen = new Set(); + + const addRelay = (candidate) => { + if (typeof candidate !== "string") { + return; + } + const trimmed = candidate.trim(); + if (!trimmed || seen.has(trimmed)) { + return; + } + seen.add(trimmed); + relays.push(trimmed); + }; + + if (pointer && typeof pointer.relay === "string") { + addRelay(pointer.relay); + } + + const client = nostrClient || null; + const candidateLists = [client?.readRelays, client?.relays]; + + for (const list of candidateLists) { + if (!list) { + continue; + } + if (Array.isArray(list)) { + for (const relay of list) { + addRelay(relay); + } + } else if (list instanceof Set) { + for (const relay of list) { + addRelay(relay); + } + } + } + + return relays; +} + +function buildReactionFilters(pointer, { limit, since } = {}) { + if (!pointer || typeof pointer !== "object") { + return []; + } + + const type = pointer.type === "a" ? "a" : pointer.type === "e" ? "e" : ""; + const value = + typeof pointer.value === "string" && pointer.value.trim() + ? pointer.value.trim() + : ""; + if (!type || !value) { + return []; + } + + const filter = { kinds: [REACTION_EVENT_KIND] }; + + if (type === "a") { + filter["#a"] = [value]; + } else { + filter["#e"] = [value]; + } + + if (Number.isFinite(limit) && limit > 0) { + filter.limit = Math.max(1, Math.floor(limit)); + } + + if (Number.isFinite(since) && since >= 0) { + filter.since = Math.max(0, Math.floor(since)); + } + + return [filter]; +} + +function buildPointerQuery(pointer, options = {}) { + const client = nostrClient || null; + const pool = client?.pool || null; + if (!pool || typeof pool.list !== "function") { + return null; + } + + const relays = gatherRelayCandidates(pointer); + if (!relays.length) { + return null; + } + + const filters = buildReactionFilters(pointer, options); + if (!filters.length) { + return null; + } + + return { pool, relays, filters }; +} + +async function hydratePointerState(key, pointer, state) { + const query = buildPointerQuery(pointer, { limit: REACTION_HYDRATION_LIMIT }); + if (!query) { + return null; + } + + let events = []; + try { + events = await query.pool.list(query.relays, query.filters); + } catch (error) { + throw error; + } + + const eventList = Array.isArray(events) ? events : []; + const ordered = eventList + .filter((event) => event && typeof event === "object") + .sort((a, b) => { + const aTime = Number.isFinite(a?.created_at) ? Number(a.created_at) : 0; + const bTime = Number.isFinite(b?.created_at) ? Number(b.created_at) : 0; + return aTime - bTime; + }); + + let changed = false; + for (const event of ordered) { + if (applyReactionToState(key, pointer, event)) { + changed = true; + } + } + + state.lastSyncedAt = Date.now(); + state.hydrated = true; + + if (changed && hasHandlers(key)) { + notifyHandlers(key); + } else if (!ordered.length && hasHandlers(key)) { + notifyHandlers(key); + } + + return null; +} + +function ensureReactionHydration(key, pointer, state) { + if (!state || state.hydrated) { + return null; + } + + if (pointerHydrationPromises.has(key)) { + return pointerHydrationPromises.get(key); + } + + const query = buildPointerQuery(pointer, { limit: REACTION_HYDRATION_LIMIT }); + if (!query) { + return null; + } + + const promise = hydratePointerState(key, pointer, state) + .catch((error) => { + devLogger.warn("[reactionCounter] Failed to hydrate reaction history:", error); + }) + .finally(() => { + pointerHydrationPromises.delete(key); + }); + + pointerHydrationPromises.set(key, promise); + return promise; +} + +function teardownLiveSubscription(key) { + const existing = pointerSubscriptions.get(key); + if (!existing) { + return; + } + pointerSubscriptions.delete(key); + const unsub = existing.unsub; + if (typeof unsub === "function") { + try { + unsub(); + } catch (error) { + devLogger.warn("[reactionCounter] Failed to tear down reaction stream:", error); + } + } +} + +function ensureReactionLiveSubscription(key, pointer) { + const signature = getPointerSignature(pointer); + if (!signature) { + teardownLiveSubscription(key); + return null; + } + + const existing = pointerSubscriptions.get(key); + if (existing && existing.signature === signature) { + return existing.unsub; + } + + teardownLiveSubscription(key); + + const query = buildPointerQuery(pointer, {}); + if (!query || typeof query.pool.sub !== "function") { + return null; + } + + const state = pointerStates.get(key); + const sinceSeconds = + Number.isFinite(state?.latestCreatedAt) && state.latestCreatedAt > 0 + ? Math.max(0, Math.floor(state.latestCreatedAt)) + : null; + if (Number.isFinite(sinceSeconds)) { + for (const filter of query.filters) { + if (filter && typeof filter === "object") { + filter.since = sinceSeconds; + } + } + } + + let subscription; + try { + subscription = query.pool.sub(query.relays, query.filters); + } catch (error) { + devLogger.warn("[reactionCounter] Failed to subscribe to reaction events:", error); + return null; + } + + const handleEvent = (event) => { + if (!event) { + return; + } + try { + const changed = applyReactionToState(key, pointer, event); + if (changed && hasHandlers(key)) { + notifyHandlers(key); + } + } catch (error) { + devLogger.warn("[reactionCounter] Reaction event handler failed:", error); + } + }; + + if (subscription && typeof subscription.on === "function") { + subscription.on("event", handleEvent); + } + + const originalUnsub = + subscription && typeof subscription.unsub === "function" + ? subscription.unsub.bind(subscription) + : null; + + const unsubscribe = () => { + if (originalUnsub) { + try { + originalUnsub(); + } catch (error) { + devLogger.warn( + "[reactionCounter] Failed to unsubscribe from reaction events:", + error + ); + } + } + }; + + pointerSubscriptions.set(key, { + signature, + unsub: unsubscribe, + }); + + return unsubscribe; +} + export const reactionCounter = { async publish(pointer, options = {}) { try {