Refine watch history chunk publishing and fetching

js/nostr.js

@@ -1090,6 +1090,31 @@ class NostrClient {
       }
     }
 
+    const previousChunkPointers = previousEntry?.chunkPointers;
+    const headPointer = previousChunkPointers?.head
+      ? clonePointerItem(previousChunkPointers.head)
+      : null;
+    const chunkPointerList = Array.isArray(previousChunkPointers?.chunks)
+      ? previousChunkPointers.chunks
+          .map((pointer) => clonePointerItem(pointer))
+          .filter(Boolean)
+      : [];
+
+    entry.chunkPointers = {
+      head: headPointer,
+      chunks: chunkPointerList,
+    };
+
+    entry.chunkEvents = new Map();
+    if (previousEntry?.chunkEvents instanceof Map) {
+      for (const [key, value] of previousEntry.chunkEvents.entries()) {
+        const clonedEvent = cloneEventForCache(value);
+        if (clonedEvent) {
+          entry.chunkEvents.set(key, clonedEvent);
+        }
+      }
+    }
+
     return entry;
   }
 
@@ -1528,6 +1553,14 @@ class NostrClient {
     }
 
     const totalChunks = chunkItemsList.length || 1;
+    const chunkIdentifiers = chunkItemsList.map((_, index) =>
+      index === 0
+        ? WATCH_HISTORY_LIST_IDENTIFIER
+        : `watch-history:${snapshotId}:${index}`
+    );
+    const chunkAddresses = chunkIdentifiers.map(
+      (identifier) => `${WATCH_HISTORY_KIND}:${normalizedActor}:${identifier}`
+    );
 
     const nowSeconds = Math.floor(Date.now() / 1000);
     let lastCreatedAt = 0;
@@ -1593,8 +1626,10 @@ class NostrClient {
         return tag;
       });
 
+      const chunkIdentifier = chunkIdentifiers[index] || WATCH_HISTORY_LIST_IDENTIFIER;
+
       const tags = [
-        ["d", WATCH_HISTORY_LIST_IDENTIFIER],
+        ["d", chunkIdentifier],
         ["encrypted", "nip04"],
         ["snapshot", snapshotId],
         ["chunk", String(index), String(totalChunks)],
@@ -1602,6 +1637,14 @@ class NostrClient {
       ];
       if (index === 0) {
         tags.splice(2, 0, ["head", "1"]);
+        const pointerTagsForChunks = chunkAddresses
+          .map((address) =>
+            typeof address === "string" && address
+              ? ["a", address]
+              : null
+          )
+          .filter(Boolean);
+        tags.push(...pointerTagsForChunks);
       }
 
       // Ensure created_at stays monotonic so chunk 0 always outranks prior snapshots
@@ -2047,7 +2090,7 @@ class NostrClient {
       limit: fetchLimit,
     };
 
-    let fetchedEvents = [];
+    let fetchedHeadEvents = [];
     try {
       const events = await this.pool.list(relayList, [filter]);
       if (Array.isArray(events)) {
@@ -2069,7 +2112,7 @@ class NostrClient {
             dedupe.set(id, evt);
           }
         }
-        fetchedEvents = Array.from(dedupe.values());
+        fetchedHeadEvents = Array.from(dedupe.values());
       }
     } catch (error) {
       if (isDevMode) {
@@ -2077,6 +2120,114 @@ class NostrClient {
       }
     }
 
+    fetchedHeadEvents.sort((a, b) => {
+      const aCreated = typeof a?.created_at === "number" ? a.created_at : 0;
+      const bCreated = typeof b?.created_at === "number" ? b.created_at : 0;
+      return bCreated - aCreated;
+    });
+
+    const chunkIdentifiers = new Set();
+    const snapshotIds = new Set();
+
+    for (const event of fetchedHeadEvents) {
+      if (!event || typeof event !== "object") {
+        continue;
+      }
+
+      const tags = Array.isArray(event.tags) ? event.tags : [];
+      for (const tag of tags) {
+        if (!Array.isArray(tag) || tag.length < 2) {
+          continue;
+        }
+        if (tag[0] === "snapshot" && typeof tag[1] === "string" && tag[1]) {
+          snapshotIds.add(tag[1]);
+        } else if (tag[0] === "a") {
+          const pointer = normalizePointerTag(tag);
+          if (pointer?.type === "a") {
+            const [kindStr, pubkey, identifier] = pointer.value.split(":");
+            const kind = Number.parseInt(kindStr, 10);
+            if (
+              Number.isFinite(kind) &&
+              kind === WATCH_HISTORY_KIND &&
+              typeof pubkey === "string" &&
+              pubkey === actor &&
+              typeof identifier === "string" &&
+              identifier &&
+              identifier !== WATCH_HISTORY_LIST_IDENTIFIER
+            ) {
+              chunkIdentifiers.add(identifier);
+            }
+          }
+        }
+      }
+    }
+
+    const chunkFilters = [];
+    const chunkFetchLimit = Math.max(
+      fetchLimit,
+      chunkIdentifiers.size ? chunkIdentifiers.size + 1 : 0,
+      snapshotIds.size ? snapshotIds.size * 2 : 0
+    );
+
+    if (chunkIdentifiers.size) {
+      chunkFilters.push({
+        kinds: [WATCH_HISTORY_KIND],
+        authors: [actor],
+        "#d": Array.from(chunkIdentifiers),
+        limit: chunkFetchLimit,
+      });
+    }
+
+    if (snapshotIds.size) {
+      chunkFilters.push({
+        kinds: [WATCH_HISTORY_KIND],
+        authors: [actor],
+        "#snapshot": Array.from(snapshotIds),
+        limit: chunkFetchLimit,
+      });
+    }
+
+    let fetchedChunkEvents = [];
+    if (chunkFilters.length) {
+      try {
+        const events = await this.pool.list(relayList, chunkFilters);
+        if (Array.isArray(events)) {
+          fetchedChunkEvents = events;
+        }
+      } catch (error) {
+        if (isDevMode) {
+          console.warn(
+            "[nostr] Failed to fetch watch history chunks:",
+            error
+          );
+        }
+      }
+    }
+
+    const eventDedupe = new Map();
+    const registerEvent = (evt) => {
+      if (!evt || typeof evt !== "object") {
+        return;
+      }
+      const id = typeof evt.id === "string" ? evt.id : null;
+      if (!id) {
+        return;
+      }
+      const existing = eventDedupe.get(id);
+      if (
+        !existing ||
+        (typeof evt.created_at === "number" ? evt.created_at : 0) >
+          (typeof existing.created_at === "number" ? existing.created_at : 0)
+      ) {
+        eventDedupe.set(id, evt);
+      }
+    };
+
+    fetchedHeadEvents.forEach(registerEvent);
+    fetchedChunkEvents.forEach(registerEvent);
+
+    const fetchedEvents = Array.from(eventDedupe.values());
+
     fetchedEvents.sort((a, b) => {
       const aCreated = typeof a?.created_at === "number" ? a.created_at : 0;
       const bCreated = typeof b?.created_at === "number" ? b.created_at : 0;
@@ -2274,6 +2425,85 @@ class NostrClient {
       previousEntry
     );
 
+    if (!(entry.chunkEvents instanceof Map)) {
+      entry.chunkEvents = new Map();
+    } else {
+      entry.chunkEvents.clear();
+    }
+
+    const chunkPointerItemsForEntry = [];
+    let headPointerForEntry = null;
+    const chunkPointerKeys = new Set();
+
+    if (effectiveSnapshot?.snapshotId) {
+      const bucketForEntry = snapshotMap.get(effectiveSnapshot.snapshotId);
+      if (bucketForEntry) {
+        const chunkEntries = Array.from(bucketForEntry.chunks.values()).sort(
+          (a, b) => a.payload.chunkIndex - b.payload.chunkIndex
+        );
+        for (const chunk of chunkEntries) {
+          if (!chunk?.event) {
+            continue;
+          }
+          const addressValue = eventToAddressPointer(chunk.event);
+          if (!addressValue) {
+            continue;
+          }
+          const pointer = clonePointerItem({
+            type: "a",
+            value: addressValue,
+            relay: null,
+          });
+          if (!pointer) {
+            continue;
+          }
+          const key = pointerKey(pointer);
+          if (!key) {
+            continue;
+          }
+          const clonedChunkEvent = cloneEventForCache(chunk.event);
+          if (clonedChunkEvent) {
+            entry.chunkEvents.set(key, clonedChunkEvent);
+          }
+          chunkPointerKeys.add(key);
+          if (chunk.payload?.chunkIndex === 0) {
+            headPointerForEntry = pointer;
+          } else {
+            chunkPointerItemsForEntry.push(pointer);
+          }
+        }
+      }
+    }
+
+    entry.chunkPointers = {
+      head: headPointerForEntry,
+      chunks: chunkPointerItemsForEntry,
+    };
+
+    const headPointerKey = headPointerForEntry ? pointerKey(headPointerForEntry) : "";
+    if (headPointerKey) {
+      chunkPointerKeys.add(headPointerKey);
+      if (entry.pointerEvent) {
+        entry.chunkEvents.set(headPointerKey, cloneEventForCache(entry.pointerEvent));
+      }
+    }
+
+    for (const key of Array.from(entry.resolved.keys())) {
+      if (!key || !key.startsWith("a:")) {
+        continue;
+      }
+      if (!chunkPointerKeys.has(key)) {
+        entry.resolved.delete(key);
+      }
+    }
+
+    for (const key of chunkPointerKeys) {
+      const eventForKey = entry.chunkEvents.get(key);
+      if (eventForKey) {
+        entry.resolved.set(key, cloneEventForCache(eventForKey));
+      }
+    }
+
     this.watchHistoryCache.set(actor, entry);
     this.persistWatchHistoryEntry(actor, entry);
 
@@ -2320,7 +2550,87 @@ class NostrClient {
       ? Math.max(1, normalizedBatchSize)
       : 1;
 
-    const available = entry.items.filter((item) => {
+    const classifyWatchHistoryPointer = (pointer) => {
+      if (!pointer || pointer.type !== "a") {
+        return "none";
+      }
+      const [kindStr, pubkey, identifier] = pointer.value.split(":");
+      const kind = Number.parseInt(kindStr, 10);
+      if (!Number.isFinite(kind) || kind !== WATCH_HISTORY_KIND) {
+        return "none";
+      }
+      if (identifier === WATCH_HISTORY_LIST_IDENTIFIER) {
+        return "head";
+      }
+      if (typeof identifier === "string" && identifier) {
+        return "chunk";
+      }
+      return "none";
+    };
+
+    const addPointerToMap = (pointer, map) => {
+      const normalized = clonePointerItem(pointer);
+      if (!normalized) {
+        return null;
+      }
+      const key = pointerKey(normalized);
+      if (!key) {
+        return null;
+      }
+      map.set(key, normalized);
+      return normalized;
+    };
+
+    const headPointerMap = new Map();
+    const chunkPointerMap = new Map();
+    const videoCandidates = [];
+
+    if (entry.chunkPointers?.head) {
+      addPointerToMap(entry.chunkPointers.head, headPointerMap);
+    }
+    if (Array.isArray(entry.chunkPointers?.chunks)) {
+      for (const chunkPointer of entry.chunkPointers.chunks) {
+        addPointerToMap(chunkPointer, chunkPointerMap);
+      }
+    }
+
+    for (const item of entry.items) {
+      const pointer = clonePointerItem(item);
+      if (!pointer) {
+        continue;
+      }
+      const pointerType = classifyWatchHistoryPointer(pointer);
+      if (pointerType === "head") {
+        addPointerToMap(pointer, headPointerMap);
+        continue;
+      }
+      if (pointerType === "chunk") {
+        addPointerToMap(pointer, chunkPointerMap);
+        continue;
+      }
+      videoCandidates.push(pointer);
+    }
+
+    const pointerEventClone = entry.pointerEvent
+      ? cloneEventForCache(entry.pointerEvent)
+      : null;
+    for (const pointer of headPointerMap.values()) {
+      const key = pointerKey(pointer);
+      if (!key) {
+        continue;
+      }
+      if (pointerEventClone) {
+        entry.resolved.set(key, pointerEventClone);
+        if (entry.chunkEvents instanceof Map) {
+          entry.chunkEvents.set(key, pointerEventClone);
+        }
+      } else if (!entry.resolved.has(key)) {
+        entry.resolved.set(key, null);
+      }
+      entry.resolving.delete(key);
+    }
+
+    const available = videoCandidates.filter((item) => {
       const key = pointerKey(item);
       if (!key) {
         return false;
@@ -2334,12 +2644,43 @@ class NostrClient {
       return true;
     });
 
-    const batch = available.slice(0, effectiveBatchSize);
+    const chunkAvailable = [];
+    for (const [key, pointer] of chunkPointerMap.entries()) {
+      if (!key) {
+        continue;
+      }
+      if (entry.resolved.has(key)) {
+        continue;
+      }
+      if (entry.resolving.has(key)) {
+        continue;
+      }
+      const cachedChunk = entry.chunkEvents?.get?.(key);
+      if (cachedChunk) {
+        const cloned = cloneEventForCache(cachedChunk);
+        if (cloned) {
+          entry.chunkEvents.set(key, cloned);
+          entry.resolved.set(key, cloned);
+        } else {
+          entry.resolved.set(key, cachedChunk);
+        }
+        continue;
+      }
+      chunkAvailable.push(pointer);
+    }
+
+    const videoBatch = available.slice(0, effectiveBatchSize);
+    const batch = [...videoBatch, ...chunkAvailable];
     if (!batch.length) {
       return [];
     }
 
-    batch.forEach((item) => entry.resolving.add(pointerKey(item)));
+    batch.forEach((item) => {
+      const key = pointerKey(item);
+      if (key) {
+        entry.resolving.add(key);
+      }
+    });
 
     const relaySet = new Set(
       Array.isArray(this.relays) && this.relays.length ? this.relays : RELAY_URLS
@@ -2403,7 +2744,24 @@ class NostrClient {
 
     for (const pointer of batch) {
       const key = pointerKey(pointer);
+      if (!key) {
+        continue;
+      }
+      const pointerType = classifyWatchHistoryPointer(pointer);
       const event = pointerMatches.get(key);
+      if (pointerType === "chunk" || pointerType === "head") {
+        if (event) {
+          const cloned = cloneEventForCache(event) || event;
+          if (!entry.chunkEvents) {
+            entry.chunkEvents = new Map();
+          }
+          entry.chunkEvents.set(key, cloned);
+          entry.resolved.set(key, cloned);
+        }
+        entry.resolving.delete(key);
+        continue;
+      }
+
       if (event) {
         try {
           const video = convertEventToVideo(event);

tests/watch-history.test.mjs

@@ -1,6 +1,11 @@
 import "./test-helpers/setup-localstorage.mjs";
 import assert from "node:assert/strict";
 
+const {
+  WATCH_HISTORY_LIST_IDENTIFIER,
+  WATCH_HISTORY_KIND,
+} = await import("../js/config.js");
+
 if (typeof globalThis.window === "undefined") {
   globalThis.window = {};
 }
@@ -117,6 +122,60 @@ const headTag = publishResult.event.tags.find(
 );
 assert(headTag, "head chunk should include head tag");
 
+const chunkIdentifiers = publishResult.events.map((event) =>
+  event.tags.find((tag) => Array.isArray(tag) && tag[0] === "d")?.[1]
+);
+
+assert.equal(
+  chunkIdentifiers.length,
+  publishResult.events.length,
+  "each chunk should expose a d tag"
+);
+
+assert.equal(
+  chunkIdentifiers[0],
+  WATCH_HISTORY_LIST_IDENTIFIER,
+  "head chunk should retain canonical watch history identifier"
+);
+
+const uniqueChunkIdentifiers = new Set(chunkIdentifiers);
+assert.equal(
+  uniqueChunkIdentifiers.size,
+  publishResult.events.length,
+  "each chunk should have a unique d tag"
+);
+
+chunkIdentifiers.slice(1).forEach((identifier, index) => {
+  assert(
+    identifier && identifier.startsWith("watch-history:"),
+    `chunk ${index + 1} should use a watch-history namespace identifier`
+  );
+  assert.notEqual(
+    identifier,
+    WATCH_HISTORY_LIST_IDENTIFIER,
+    "only the head chunk should use the canonical identifier"
+  );
+});
+
+const headATags = publishResult.event.tags
+  .filter((tag) => Array.isArray(tag) && tag[0] === "a")
+  .map((tag) => tag[1]);
+
+const expectedChunkAddresses = publishResult.events.map((event) => {
+  const identifier = event.tags.find(
+    (tag) => Array.isArray(tag) && tag[0] === "d"
+  )?.[1];
+  assert(identifier, "chunk should include identifier tag");
+  return `${WATCH_HISTORY_KIND}:${ACTOR}:${identifier}`;
+});
+
+expectedChunkAddresses.forEach((address) => {
+  assert(
+    headATags.includes(address),
+    `head chunk should reference ${address}`
+  );
+});
+
 delete globalThis.window.nostr;
 
 if (!globalThis.window.NostrTools.nip04) {
@@ -158,6 +217,117 @@ assert.deepEqual(
   "chunked fetch should preserve pointer order"
 );
 
+if (typeof localStorage !== "undefined") {
+  localStorage.clear();
+}
+
+const headEvent = publishResult.event;
+const chunkEvents = publishResult.events.filter(
+  (event) => event.id !== headEvent.id
+);
+
+const snapshotId = headEvent.tags.find(
+  (tag) => Array.isArray(tag) && tag[0] === "snapshot"
+)?.[1];
+
+const freshClient = createDecryptClient(ACTOR);
+freshClient.relays = ["wss://unit.test"];
+freshClient.watchHistoryFetchEventLimit = 10;
+
+const observedFilters = [];
+freshClient.pool = {
+  list: async (_relays, filters) => {
+    observedFilters.push(filters);
+
+    const firstFilter = filters[0] || {};
+    const dFilter = Array.isArray(firstFilter["#d"])
+      ? firstFilter["#d"]
+      : [];
+
+    if (filters.length === 1 && dFilter.includes(WATCH_HISTORY_LIST_IDENTIFIER)) {
+      return [headEvent];
+    }
+
+    const requestedIdentifiers = new Set();
+    const requestedSnapshots = new Set();
+
+    for (const filter of filters) {
+      if (Array.isArray(filter?.["#d"])) {
+        for (const identifier of filter["#d"]) {
+          requestedIdentifiers.add(identifier);
+        }
+      }
+      if (Array.isArray(filter?.["#snapshot"])) {
+        for (const value of filter["#snapshot"]) {
+          requestedSnapshots.add(value);
+        }
+      }
+    }
+
+    const responses = [];
+
+    for (const event of publishResult.events) {
+      const identifier = event.tags.find(
+        (tag) => Array.isArray(tag) && tag[0] === "d"
+      )?.[1];
+      const eventSnapshot = event.tags.find(
+        (tag) => Array.isArray(tag) && tag[0] === "snapshot"
+      )?.[1];
+
+      if (identifier && requestedIdentifiers.has(identifier)) {
+        responses.push(event);
+        continue;
+      }
+      if (eventSnapshot && requestedSnapshots.has(eventSnapshot)) {
+        responses.push(event);
+      }
+    }
+
+    return responses;
+  },
+};
+
+const rebuilt = await freshClient.fetchWatchHistory(ACTOR);
+
+assert.equal(
+  rebuilt.items.length,
+  longPointers.length,
+  "fresh client should rebuild full watch history list"
+);
+
+assert(
+  observedFilters.length >= 2,
+  "fetch should include a follow-up query for chunk events"
+);
+
+const followUpFilters = observedFilters[observedFilters.length - 1];
+const identifiersRequested = new Set();
+let snapshotRequested = false;
+
+for (const filter of followUpFilters) {
+  if (Array.isArray(filter?.["#d"])) {
+    for (const identifier of filter["#d"]) {
+      identifiersRequested.add(identifier);
+    }
+  }
+  if (Array.isArray(filter?.["#snapshot"])) {
+    snapshotRequested = snapshotRequested || filter["#snapshot"].includes(snapshotId);
+  }
+}
+
+assert(
+  chunkEvents.every((event) => {
+    const identifier = event.tags.find(
+      (tag) => Array.isArray(tag) && tag[0] === "d"
+    )?.[1];
+    return (
+      identifiersRequested.has(identifier) ||
+      (snapshotId && snapshotRequested)
+    );
+  }),
+  "follow-up query should target chunk identifiers or snapshot"
+);
+
 const MONOTONIC_ACTOR = `${ACTOR}-monotonic`;
 const { client: monotonicPublishingClient, publishedEvents: monotonicEvents } =
   createPublishingClient(MONOTONIC_ACTOR);