Hydrate reaction counters from nostr relays

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 {