import { createContext, useCallback, useContext, useEffect, useMemo, useRef, useState, type ReactNode, } from 'react'; import { useChannel } from '@/hooks/use-channel'; import { useAuthStore } from '@/stores/auth-store'; // --------------------------------------------------------------------------- // Types // --------------------------------------------------------------------------- export type ComposingMode = 'recording' | 'typing' | 'screen'; export interface ComposingUser { humanId: string; mode: ComposingMode; lastSeen: number; } interface StreamPresenceContextValue { onlineHumanIds: Set; composingUsers: ComposingUser[]; startComposing: (mode: ComposingMode) => void; stopComposing: () => void; } // --------------------------------------------------------------------------- // Constants // --------------------------------------------------------------------------- const COMPOSING_TIMEOUT_MS = 10_000; const COMPOSING_HEARTBEAT_MS = 5_000; const COMPOSING_CLEANUP_INTERVAL_MS = 2_000; // --------------------------------------------------------------------------- // Context // --------------------------------------------------------------------------- const StreamPresenceContext = createContext( null, ); // --------------------------------------------------------------------------- // Provider // --------------------------------------------------------------------------- interface StreamPresenceProviderProps { networkId: string; streamId: string; children: ReactNode; } export function StreamPresenceProvider({ networkId, streamId, children, }: StreamPresenceProviderProps) { const channelId = `stream:${networkId}:${streamId}`; const { presence, messages, sendMessage } = useChannel(channelId); const currentUserId = useAuthStore((s) => s.user?.id); // --- Online presence --- const onlineHumanIds = useMemo(() => new Set(presence), [presence]); // --- Composing state --- const [composingUsers, setComposingUsers] = useState([]); const composingMapRef = useRef(new Map()); const processedCountRef = useRef(0); // Process new messages incrementally useEffect(() => { if (messages.length <= processedCountRef.current) return; const newMessages = messages.slice(processedCountRef.current); processedCountRef.current = messages.length; let changed = false; const map = composingMapRef.current; for (const msg of newMessages) { const payload = msg.payload as | { type: string; mode?: string } | undefined; if (!payload?.type) continue; // Skip own events if (msg.humanId === currentUserId) continue; if (payload.type === 'composing_start' && payload.mode) { map.set(msg.humanId, { humanId: msg.humanId, mode: payload.mode as ComposingMode, lastSeen: Date.now(), }); changed = true; } else if (payload.type === 'composing_stop') { if (map.delete(msg.humanId)) changed = true; } } if (changed) { setComposingUsers(Array.from(map.values())); } }, [messages, currentUserId]); // Also clear composing when a user leaves the channel useEffect(() => { const map = composingMapRef.current; const onlineSet = new Set(presence); let changed = false; for (const humanId of map.keys()) { if (!onlineSet.has(humanId)) { map.delete(humanId); changed = true; } } if (changed) { setComposingUsers(Array.from(map.values())); } }, [presence]); // Cleanup stale composing entries useEffect(() => { const interval = setInterval(() => { const map = composingMapRef.current; const now = Date.now(); let changed = false; for (const [humanId, entry] of map) { if (now - entry.lastSeen > COMPOSING_TIMEOUT_MS) { map.delete(humanId); changed = true; } } if (changed) { setComposingUsers(Array.from(map.values())); } }, COMPOSING_CLEANUP_INTERVAL_MS); return () => clearInterval(interval); }, []); // --- Composing broadcast --- const heartbeatRef = useRef>(undefined); const startComposing = useCallback( (mode: ComposingMode) => { // Send immediately sendMessage({ type: 'composing_start', mode }); // Clear any existing heartbeat clearInterval(heartbeatRef.current); // Start heartbeat heartbeatRef.current = setInterval(() => { sendMessage({ type: 'composing_start', mode }); }, COMPOSING_HEARTBEAT_MS); }, [sendMessage], ); const stopComposing = useCallback(() => { clearInterval(heartbeatRef.current); heartbeatRef.current = undefined; sendMessage({ type: 'composing_stop' }); }, [sendMessage]); // Cleanup heartbeat on unmount useEffect(() => { return () => { clearInterval(heartbeatRef.current); }; }, []); const value = useMemo( () => ({ onlineHumanIds, composingUsers, startComposing, stopComposing, }), [onlineHumanIds, composingUsers, startComposing, stopComposing], ); return ( {children} ); } // --------------------------------------------------------------------------- // Hooks // --------------------------------------------------------------------------- function useStreamPresenceContext() { const ctx = useContext(StreamPresenceContext); if (!ctx) { throw new Error( 'useStreamPresence must be used within a StreamPresenceProvider', ); } return ctx; } export function useStreamPresence() { const { onlineHumanIds } = useStreamPresenceContext(); return { onlineHumanIds }; } export function useStreamComposing() { const { composingUsers } = useStreamPresenceContext(); return { composingUsers }; } export function useStreamComposingBroadcast() { const { startComposing, stopComposing } = useStreamPresenceContext(); return { startComposing, stopComposing }; }