feat(desktop): paginate stream particles via windowed Firestore subscription

Streams previously prefetched every particle through an unbounded Firestore
subscription. This adds a windowed source that anchors a `created_at desc`
limit query at the newest particle and grows it backward on demand, so the
already-seen history before a viewer's playback marker is no longer loaded.

- `useWindowedStreamParticles`: tail-anchored live window that grows backward
  to cover the resume marker and to service `loadOlder()` (list scroll-up).
  Because the window always includes the newest particle, new arrivals stream
  in and forward playback never needs a fetch. Includes anti-eviction growth
  so a new tail particle never pushes loaded particles out of the window.
- `useStreamPlayback`: consumes the windowed source instead of loading all
  children. New-tail and removal handling are derived from the children array
  (Firestore change events can't tell a genuine arrival from pagination
  backfill). Resume-from-marker waits for backward growth to reach an older
  marker; `prev` at the window edge pulls in older history.

Playback resume, forward-only marker persistence, auto-advance-on-new, and
removal fallback are all preserved. List view and progress indicator continue
to render off the (now windowed) children; their pagination UX is a follow-up.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01V8gsmdVd7R8PJtn4UFnC4J

js/desktop/src/hooks/use-stream-playback.ts

@@ -8,7 +8,7 @@ import {
 } from 'react';
 import { useAuthStore } from '@/stores/auth-store';
 import type { Particle } from '@/api/types';
-import { useLiveParticleChildren } from '@/hooks/use-particle';
+import { useWindowedStreamParticles } from '@/hooks/use-windowed-stream-particles';
 import { toFirestoreDocPath, type ParticlePath } from '@/lib/particle-path';
 import { updateStreamPlaybackMarker } from '@/lib/firestore-particles';
 
@@ -92,6 +92,11 @@ interface UseStreamPlaybackResult {
   currentIndex: number;
   status: PlaybackStatus;
   initialized: boolean;
+  /** Whether older particles exist before the loaded window (list scroll-up). */
+  hasMoreOlder: boolean;
+  /** Extend the loaded window backward. */
+  loadOlder: () => void;
+  isLoadingOlder: boolean;
   next: () => void;
   prev: () => void;
   goTo: (index: number) => void;
@@ -113,50 +118,26 @@ export function useStreamPlayback(
 ): UseStreamPlaybackResult {
   const userId = useAuthStore((s) => s.user?.id);
   const [state, dispatch] = useReducer(playbackReducer, initialState);
-  // Read via ref so the onAdded subscription callback stays stable.
+
+  const marker = userId
+    ? (streamParticle.playback_markers?.[userId] ?? null)
+    : null;
+
+  // Windowed source: only the tail (plus enough history to cover the marker)
+  // is loaded, instead of every particle in the stream.
+  const { children, hasMoreOlder, loadOlder, isLoadingOlder } =
+    useWindowedStreamParticles(path, { marker });
+
+  // Read via ref so the new-particle effect stays cheap to reason about.
   const autoAdvanceOnNewRef = useRef(autoAdvanceOnNew);
   useEffect(() => {
     autoAdvanceOnNewRef.current = autoAdvanceOnNew;
   }, [autoAdvanceOnNew]);
-  // Track the stream ID we've initialized for, to reset when navigating between streams
+
+  // Track the stream ID we've initialized for, to reset when navigating streams.
   const initializedForRef = useRef<string | null>(null);
-  // Latest currentIndex for onParticleRemoved, which is passed into
-  // useLiveParticleChildren. Reading it through a ref keeps the callback stable
-  // (no re-subscription) and breaks the declaration cycle
-  // children -> currentIndex -> callback -> children. useEffectEvent can't be
-  // used here — Effect Events may not be passed to another hook.
-  const currentIndexRef = useRef(0);
 
-  // --- Firestore change callbacks ---
-  const onParticleAdded = useCallback((particle: Particle) => {
-    if (!autoAdvanceOnNewRef.current) return;
-    dispatch({ type: 'PARTICLE_ADDED', particleId: particle.id });
-  }, []);
-
-  const onParticleRemoved = useCallback(
-    (removed: Particle, updatedChildren: Particle[]) => {
-      const fallbackIndex = Math.min(
-        currentIndexRef.current,
-        updatedChildren.length - 1,
-      );
-      const fallback = updatedChildren[Math.max(0, fallbackIndex)];
-      dispatch({
-        type: 'PARTICLE_REMOVED',
-        removedParticleId: removed.id,
-        fallbackParticleId: fallback?.id ?? null,
-      });
-    },
-    [],
-  );
-
-  const { children } = useLiveParticleChildren(path, {
-    orderByField: 'created_at',
-    orderDirection: 'asc',
-    onAdded: onParticleAdded,
-    onRemoved: onParticleRemoved,
-  });
-
-  // Derive current index and particle from ID
+  // Derive current index and particle from ID.
   const currentIndex = useMemo(() => {
     if (!state.currentParticleId) return -1;
     return children.findIndex((c) => c.id === state.currentParticleId);
@@ -164,12 +145,64 @@ export function useStreamPlayback(
 
   const currentParticle = currentIndex !== -1 ? children[currentIndex] : null;
 
-  // Keep the ref read by onParticleRemoved in sync with the derived index.
+  // Remember the last index the current particle was actually found at, so a
+  // removal can fall back to a sensible neighbour even though `currentIndex`
+  // has already gone to -1 by the time we notice.
+  const lastValidIndexRef = useRef(0);
   useEffect(() => {
-    currentIndexRef.current = currentIndex;
+    if (currentIndex >= 0) lastValidIndexRef.current = currentIndex;
   }, [currentIndex]);
 
-  // Fallback init — always sees latest children/state via useEffectEvent
+  // --- New tail particle → resume from end ---
+  // Derive arrivals from the children tail rather than Firestore change events,
+  // which can't distinguish a genuine new particle from pagination backfill.
+  const prevNewestIdRef = useRef<string | null>(null);
+  useEffect(() => {
+    if (children.length === 0) {
+      prevNewestIdRef.current = null;
+      return;
+    }
+    const newestId = children[children.length - 1].id;
+    const prevNewestId = prevNewestIdRef.current;
+    prevNewestIdRef.current = newestId;
+    if (prevNewestId === null || newestId === prevNewestId) return;
+    if (!autoAdvanceOnNewRef.current) return;
+    // Resume at the first particle added after where playback ended.
+    const prevIndex = children.findIndex((c) => c.id === prevNewestId);
+    const firstNew =
+      prevIndex >= 0
+        ? (children[prevIndex + 1] ?? children[children.length - 1])
+        : children[children.length - 1];
+    dispatch({ type: 'PARTICLE_ADDED', particleId: firstNew.id });
+  }, [children]);
+
+  // --- Current particle removed (deletion) → fall back to a neighbour ---
+  const prevIdsRef = useRef<Set<string>>(new Set());
+  useEffect(() => {
+    const id = state.currentParticleId;
+    const prevIds = prevIdsRef.current;
+    const currIds = new Set(children.map((c) => c.id));
+    prevIdsRef.current = currIds;
+
+    if (!id || children.length === 0) return;
+    if (currIds.has(id)) return;
+    // Only treat as a removal if it was present before — a not-yet-arrived id
+    // (e.g. optimistic goToParticle) should wait, not fall back.
+    if (!prevIds.has(id)) return;
+
+    const fallbackIndex = Math.min(
+      lastValidIndexRef.current,
+      children.length - 1,
+    );
+    const fallback = children[Math.max(0, fallbackIndex)];
+    dispatch({
+      type: 'PARTICLE_REMOVED',
+      removedParticleId: id,
+      fallbackParticleId: fallback?.id ?? null,
+    });
+  }, [children, state.currentParticleId]);
+
+  // Fallback init — always sees latest children/state via useEffectEvent.
   const initFallback = useEffectEvent(() => {
     if (state.initialized || children.length === 0) return;
     initializedForRef.current = streamParticle.id;
@@ -178,7 +211,7 @@ export function useStreamPlayback(
 
   // --- Init logic: runs on every children change until initialized ---
   useEffect(() => {
-    // Reset if we navigated to a different stream
+    // Reset if we navigated to a different stream.
     if (
       initializedForRef.current !== null &&
       initializedForRef.current !== streamParticle.id
@@ -186,7 +219,7 @@ export function useStreamPlayback(
       initializedForRef.current = null;
     }
 
-    // Already initialized for this stream
+    // Already initialized for this stream.
     if (state.initialized && initializedForRef.current === streamParticle.id)
       return;
 
@@ -195,35 +228,39 @@ export function useStreamPlayback(
     const playbackPosition = streamParticle.playback_markers?.[userId ?? ''];
 
     if (!playbackPosition) {
-      // No marker — start from the beginning
+      // No marker — start from the start of the loaded window.
       initializedForRef.current = streamParticle.id;
       dispatch({ type: 'INIT', particleId: children[0].id });
       return;
     }
 
-    // Try to find the marker's target particle
+    // Resume at the first particle after the marker.
     const found = children.find(
       (c) => c.created_at.getTime() > playbackPosition.getTime(),
     );
-
     if (found) {
       initializedForRef.current = streamParticle.id;
       dispatch({ type: 'INIT', particleId: found.id });
       return;
-    } else {
-      initializedForRef.current = streamParticle.id;
-      dispatch({ type: 'INIT', particleId: children[children.length - 1].id });
     }
 
-    // Marker target not found yet — fall back after timeout
-    const timeout = setTimeout(initFallback, INIT_FALLBACK_TIMEOUT_MS);
-    return () => clearTimeout(timeout);
+    // Marker is older than everything loaded so far. If the window is still
+    // growing backward to reach it, wait for more particles to arrive.
+    if (hasMoreOlder) {
+      const timeout = setTimeout(initFallback, INIT_FALLBACK_TIMEOUT_MS);
+      return () => clearTimeout(timeout);
+    }
+
+    // Reached the start with no particle after the marker → caught up.
+    initializedForRef.current = streamParticle.id;
+    dispatch({ type: 'INIT', particleId: children[children.length - 1].id });
   }, [
     children,
     streamParticle.id,
     streamParticle.playback_markers,
     userId,
     state.initialized,
+    hasMoreOlder,
   ]);
 
   // --- Persist playback marker (only advance forward, never backwards) ---
@@ -237,7 +274,7 @@ export function useStreamPlayback(
       lastPersistedMarkerRef.current ??
       streamParticle.playback_markers?.[userId];
 
-    // Only update if advancing beyond the current marker
+    // Only update if advancing beyond the current marker.
     if (existingMarker && currentTime.getTime() <= existingMarker.getTime())
       return;
 
@@ -266,12 +303,18 @@ export function useStreamPlayback(
   }, [children, currentIndex]);
 
   const prev = useCallback(() => {
-    if (currentIndex <= 0) return;
+    if (currentIndex < 0) return;
+    if (currentIndex === 0) {
+      // At the start of the loaded window — pull in older history so the user
+      // can keep going back.
+      if (hasMoreOlder) loadOlder();
+      return;
+    }
     dispatch({
       type: 'SET_PARTICLE',
       particleId: children[currentIndex - 1].id,
     });
-  }, [children, currentIndex]);
+  }, [children, currentIndex, hasMoreOlder, loadOlder]);
 
   const goTo = useCallback(
     (index: number) => {
@@ -295,6 +338,9 @@ export function useStreamPlayback(
     currentIndex,
     status: state.status,
     initialized: state.initialized,
+    hasMoreOlder,
+    loadOlder,
+    isLoadingOlder,
     next,
     prev,
     goTo,