feat(P5A.T3): useWatchSession hook with realtime + throttled push

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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<void>;
+} {
+  const [session, setSession] = useState<WatchSession | null>(null);
+  const [loading, setLoading] = useState(true);
+  const [error, setError] = useState<string | null>(null);
+  const pendingRef = useRef<WatchSessionState | null>(null);
+  const lastPushAtRef = useRef<number>(0);
+  const pushTimerRef = useRef<number | null>(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 };
+}