From af69f985833a75d712930b744419f9bf73b37927 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 21 Jun 2026 01:54:39 +0000 Subject: [PATCH] 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) Claude-Session: https://claude.ai/code/session_01V8gsmdVd7R8PJtn4UFnC4J --- js/desktop/src/hooks/use-stream-playback.ts | 158 ++++++++----- .../hooks/use-windowed-stream-particles.ts | 214 ++++++++++++++++++ 2 files changed, 316 insertions(+), 56 deletions(-) create mode 100644 js/desktop/src/hooks/use-windowed-stream-particles.ts diff --git a/js/desktop/src/hooks/use-stream-playback.ts b/js/desktop/src/hooks/use-stream-playback.ts index b4a1405..4cf9cec 100644 --- a/js/desktop/src/hooks/use-stream-playback.ts +++ b/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(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(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>(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, diff --git a/js/desktop/src/hooks/use-windowed-stream-particles.ts b/js/desktop/src/hooks/use-windowed-stream-particles.ts new file mode 100644 index 0000000..5961c28 --- /dev/null +++ b/js/desktop/src/hooks/use-windowed-stream-particles.ts @@ -0,0 +1,214 @@ +import { useCallback, useEffect, useRef, useState } from 'react'; +import { subscribeToParticleChildren } from '@/lib/firestore-particles'; +import type { Particle } from '@/api/types'; +import { + toFirestoreChildrenPath, + type ParticlePath, +} from '@/lib/particle-path'; + +const DEFAULT_PAGE_SIZE = 30; + +interface UseWindowedStreamParticlesParams { + /** + * Resume anchor (the viewer's playback marker). The window grows backward + * until it covers this timestamp so the resume particle is always loaded. + * Captured once per stream — advancing the marker during playback does not + * re-window. + */ + marker?: Date | null; + /** How many particles to add per backward growth step. */ + pageSize?: number; +} + +export interface UseWindowedStreamParticlesResult { + /** + * Loaded window, ascending (oldest → newest). The newest particle in the + * stream is always present — the window only ever grows backward. + */ + children: Particle[]; + isLoading: boolean; + error: Error | null; + /** Whether older particles likely exist before the loaded window. */ + hasMoreOlder: boolean; + /** Extend the window backward (older history). No-op when nothing remains. */ + loadOlder: () => void; + isLoadingOlder: boolean; +} + +/** Per-stream mutable tracking that must survive limit-driven re-subscriptions. */ +interface WindowTracking { + path: string | null; + /** Newest created_at (ms) seen — tells new tail particles from backfill. */ + newestMs: number | null; + /** Oldest created_at (ms) currently loaded. */ + oldestMs: number | null; + /** Backward growth target (the marker, ms), frozen on first capture. */ + coverageMs: number | null; +} + +/** + * Live, windowed view of a stream's particles. + * + * Instead of subscribing to every child (the old behaviour), this keeps a + * `orderBy(created_at desc) limit(N)` window anchored at the newest particle + * and grows it backward on demand. Because the window is anchored at the tail + * it always contains the most recent particles, so new arrivals stream in and + * forward playback never needs a fetch. The window grows backward to: + * 1. cover the resume marker, so playback can start where the user left off; + * 2. service `loadOlder()` when the list view scrolls up. + * + * Output is reversed to ascending order to match the rest of the playback code. + */ +export function useWindowedStreamParticles( + path: ParticlePath | undefined, + { + marker = null, + pageSize = DEFAULT_PAGE_SIZE, + }: UseWindowedStreamParticlesParams = {}, +): UseWindowedStreamParticlesResult { + const [children, setChildren] = useState([]); + const [isLoading, setIsLoading] = useState(true); + const [error, setError] = useState(null); + const [hasMoreOlder, setHasMoreOlder] = useState(false); + const [isLoadingOlder, setIsLoadingOlder] = useState(false); + const [limit, setLimit] = useState(pageSize); + + const collectionPath = path ? toFirestoreChildrenPath(path) : null; + + const trackingRef = useRef({ + path: null, + newestMs: null, + oldestMs: null, + coverageMs: null, + }); + + // Latest marker, read lazily inside the snapshot callback so a late-resolving + // marker (e.g. auth after first paint) still seeds backward coverage. + const markerRef = useRef(marker); + useEffect(() => { + markerRef.current = marker; + }, [marker]); + + // Reset window state when the stream changes (render-phase adjustment — the + // blessed alternative to a reset effect, avoids cascading effect renders). + const [trackedPath, setTrackedPath] = useState(collectionPath); + if (trackedPath !== collectionPath) { + setTrackedPath(collectionPath); + setLimit(pageSize); + setChildren([]); + setIsLoading(true); + setError(null); + setHasMoreOlder(false); + setIsLoadingOlder(false); + } + + useEffect(() => { + if (!collectionPath) return; + + // Reset per-stream tracking on a genuine stream change, but keep it across + // limit-driven re-subscriptions (newest/coverage must persist). + const tracking = trackingRef.current; + if (tracking.path !== collectionPath) { + tracking.path = collectionPath; + tracking.newestMs = null; + tracking.oldestMs = null; + tracking.coverageMs = null; + } + + const unsubscribe = subscribeToParticleChildren(collectionPath, { + orderByField: 'created_at', + orderDirection: 'desc', + limit, + onData: (descData) => { + const t = trackingRef.current; + + // Lazily freeze the backward-coverage target from the marker. + if (t.coverageMs === null && markerRef.current) { + t.coverageMs = markerRef.current.getTime(); + } + + // Firestore caps results at `limit`; a full window means more older + // particles may exist beyond it. + const saturated = descData.length === limit; + const newest = descData[0]; + const newestMs = newest ? newest.created_at.getTime() : null; + + // Anti-eviction: if the window is full and genuinely newer particles + // arrived at the tail, grow the limit so the oldest loaded particles + // aren't pushed out. Skip rendering the evicted snapshot — the regrown + // query delivers the complete window a beat later. + const prevNewest = t.newestMs; + if ( + saturated && + newestMs !== null && + prevNewest !== null && + newestMs > prevNewest + ) { + const newerCount = descData.filter( + (d) => d.created_at.getTime() > prevNewest, + ).length; + if (newerCount > 0) { + t.newestMs = newestMs; + setLimit((l) => l + newerCount); + return; + } + } + if (newestMs !== null) t.newestMs = newestMs; + + const ascData = descData.slice().reverse(); + t.oldestMs = + ascData.length > 0 ? ascData[0].created_at.getTime() : null; + + // Marker coverage: keep growing backward until the resume marker falls + // within the window (or we reach the start of the stream). + if ( + saturated && + t.coverageMs !== null && + t.oldestMs !== null && + t.oldestMs > t.coverageMs + ) { + setLimit((l) => l + pageSize); + } + + setChildren(ascData); + setHasMoreOlder(saturated); + setIsLoading(false); + setIsLoadingOlder(false); + }, + onError: (err) => { + console.warn(err); + setError(err); + setIsLoading(false); + setIsLoadingOlder(false); + }, + }); + + return () => unsubscribe(); + }, [collectionPath, limit, pageSize]); + + const loadOlder = useCallback(() => { + if (!hasMoreOlder || isLoadingOlder) return; + setIsLoadingOlder(true); + setLimit((l) => l + pageSize); + }, [hasMoreOlder, isLoadingOlder, pageSize]); + + if (!path) { + return { + children: [], + isLoading: false, + error: null, + hasMoreOlder: false, + loadOlder: () => {}, + isLoadingOlder: false, + }; + } + + return { + children, + isLoading, + error, + hasMoreOlder, + loadOlder, + isLoadingOlder, + }; +}