diff --git a/apps/desktop/src/hooks/useWatchSession.ts b/apps/desktop/src/hooks/useWatchSession.ts new file mode 100644 index 0000000..c09eea2 --- /dev/null +++ b/apps/desktop/src/hooks/useWatchSession.ts @@ -0,0 +1,145 @@ +import { useCallback, useEffect, useRef, useState } from 'react'; + +import { + endWatchSession, + getWatchSession, + updateWatchSessionState, + type WatchSession, + type WatchSessionState, +} from '@chat-app/shared/chat'; + +import { supabase } from '../lib/supabase'; + +const PUSH_THROTTLE_MS = 500; + +export function useWatchSession(sessionId: string | null): { + session: WatchSession | null; + loading: boolean; + error: string | null; + pushState: (state: WatchSessionState) => void; + endSession: () => Promise; +} { + const [session, setSession] = useState(null); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(null); + const pendingRef = useRef(null); + const lastPushAtRef = useRef(0); + const pushTimerRef = useRef(null); + + useEffect(() => { + if (!sessionId) { + setSession(null); + setLoading(false); + setError(null); + return; + } + let cancelled = false; + void (async () => { + try { + setLoading(true); + setError(null); + const fresh = await getWatchSession(supabase, sessionId); + if (!cancelled) { + setSession(fresh); + setLoading(false); + } + } catch (err) { + if (!cancelled) { + setLoading(false); + setError(err instanceof Error ? err.message : 'failed to load session'); + } + } + })(); + + const channel = supabase + .channel('watch_session:' + sessionId) + .on( + 'postgres_changes', + { + event: 'UPDATE', + schema: 'public', + table: 'conversation_watch_sessions', + filter: 'id=eq.' + sessionId, + }, + (payload) => { + const row = payload.new as { + id?: string; + ended_at?: string | null; + current_state?: unknown; + } | null; + if (!row?.id) return; + const raw = (row.current_state ?? {}) as Partial<{ + playing: boolean; + position_seconds: number; + updated_at_ms: number; + }>; + setSession((cur) => { + if (!cur) return cur; + return { + ...cur, + endedAt: row.ended_at ?? null, + currentState: { + playing: typeof raw.playing === 'boolean' ? raw.playing : cur.currentState.playing, + positionSeconds: + typeof raw.position_seconds === 'number' + ? raw.position_seconds + : cur.currentState.positionSeconds, + updatedAtMs: + typeof raw.updated_at_ms === 'number' + ? raw.updated_at_ms + : cur.currentState.updatedAtMs, + }, + }; + }); + }, + ) + .subscribe(); + + return () => { + cancelled = true; + void supabase.removeChannel(channel); + if (pushTimerRef.current !== null) { + window.clearTimeout(pushTimerRef.current); + pushTimerRef.current = null; + } + }; + }, [sessionId]); + + // Throttled writer: keeps the latest state in pendingRef; fires at most + // once per PUSH_THROTTLE_MS. Trailing-edge push guarantees the final + // state is always sent even when a rapid burst stops before the leading- + // edge timeout expires. + const pushState = useCallback((state: WatchSessionState) => { + if (!sessionId) return; + pendingRef.current = state; + const now = Date.now(); + const elapsed = now - lastPushAtRef.current; + if (elapsed >= PUSH_THROTTLE_MS) { + lastPushAtRef.current = now; + const toPush = pendingRef.current; + pendingRef.current = null; + void updateWatchSessionState(supabase, sessionId, toPush).catch((err) => { + console.warn('updateWatchSessionState failed', err); + }); + return; + } + if (pushTimerRef.current !== null) window.clearTimeout(pushTimerRef.current); + pushTimerRef.current = window.setTimeout(() => { + pushTimerRef.current = null; + const toPush = pendingRef.current; + if (!toPush) return; + pendingRef.current = null; + lastPushAtRef.current = Date.now(); + void updateWatchSessionState(supabase, sessionId, toPush).catch((err) => { + console.warn('updateWatchSessionState trailing failed', err); + }); + }, PUSH_THROTTLE_MS - elapsed); + }, [sessionId]); + + const endSession = useCallback(async () => { + if (!sessionId) return; + await endWatchSession(supabase, sessionId); + }, [sessionId]); + + return { session, loading, error, pushState, endSession }; +}