diff --git a/js/src/App.tsx b/js/src/App.tsx index ffbfe62..514f73b 100644 --- a/js/src/App.tsx +++ b/js/src/App.tsx @@ -14,6 +14,7 @@ import ParticleViewResolver from "@/features/particles/particle-view-resolver"; import Layout from "@/features/layout"; import NetworkSettingsPage from "@/features/network-settings"; import { Toaster } from "@/components/ui/sonner"; +import { PusherProvider } from "@/lib/pusher-provider"; const queryClient = new QueryClient(); @@ -37,7 +38,11 @@ const App = () => { return ; } - return ; + return ( + + + + ); }; function AutoplayNavigationListener() { diff --git a/js/src/hooks/use-channel.ts b/js/src/hooks/use-channel.ts new file mode 100644 index 0000000..0cc1fd7 --- /dev/null +++ b/js/src/hooks/use-channel.ts @@ -0,0 +1,91 @@ +import { useEffect, useState, useCallback, useRef } from "react"; +import { usePusherClient } from "@/lib/pusher-provider"; +import type { ChannelMessage } from "@/lib/pusher-client"; + +interface UseChannelResult { + /** Current set of humanIds present in the channel */ + presence: string[]; + /** Messages received on this channel (since the hook mounted) */ + messages: ChannelMessage[]; + /** Send a message to the channel */ + sendMessage: (payload: unknown) => void; +} + +/** + * Subscribe to a pusher channel. Manages presence tracking and message delivery. + * Subscribes on mount, unsubscribes on unmount. + * + * @param channelId - The channel to subscribe to, or null to skip. + */ +export function useChannel(channelId: string | null): UseChannelResult { + const client = usePusherClient(); + const [presence, setPresence] = useState([]); + const [messages, setMessages] = useState([]); + + // Keep a ref to avoid re-subscribing when sendMessage changes + const clientRef = useRef(client); + const channelRef = useRef(channelId); + clientRef.current = client; + channelRef.current = channelId; + + useEffect(() => { + if (!client || !channelId) { + setPresence([]); + setMessages([]); + return; + } + + client.subscribe(channelId); + + const onSubscribed = (msg: { presence?: string[] }) => { + setPresence(msg.presence ?? []); + }; + + const onJoin = (msg: { humanId?: string }) => { + if (msg.humanId) { + setPresence((prev) => + prev.includes(msg.humanId!) ? prev : [...prev, msg.humanId!], + ); + } + }; + + const onLeave = (msg: { humanId?: string }) => { + if (msg.humanId) { + setPresence((prev) => prev.filter((id) => id !== msg.humanId)); + } + }; + + const onMessage = (msg: { humanId?: string; payload?: unknown }) => { + if (msg.humanId) { + setMessages((prev) => [ + ...prev, + { humanId: msg.humanId!, payload: msg.payload }, + ]); + } + }; + + client.on(channelId, "subscribed", onSubscribed); + client.on(channelId, "join", onJoin); + client.on(channelId, "leave", onLeave); + client.on(channelId, "message", onMessage); + + return () => { + client.off(channelId, "subscribed", onSubscribed); + client.off(channelId, "join", onJoin); + client.off(channelId, "leave", onLeave); + client.off(channelId, "message", onMessage); + client.unsubscribe(channelId); + }; + }, [client, channelId]); + + const sendMessage = useCallback( + (payload: unknown) => { + if (clientRef.current && channelRef.current) { + clientRef.current.sendMessage(channelRef.current, payload); + } + }, + [], + ); + + return { presence, messages, sendMessage }; +} diff --git a/js/src/lib/pusher-client.ts b/js/src/lib/pusher-client.ts new file mode 100644 index 0000000..8a4c707 --- /dev/null +++ b/js/src/lib/pusher-client.ts @@ -0,0 +1,282 @@ +/** + * PusherClient manages a WebSocket connection to the pusher service. + * Handles authentication, reconnection with exponential backoff, + * channel subscriptions, and event dispatching. + */ + +export type ConnectionState = + | "disconnected" + | "connecting" + | "connected" + | "reconnecting"; + +export interface ChannelMessage { + humanId: string; + payload: unknown; +} + +// Server → Client message shape +interface ServerMessage { + type: "subscribed" | "join" | "leave" | "message" | "error"; + channel?: string; + humanId?: string; + presence?: string[]; + payload?: unknown; + message?: string; +} + +type ChannelEventType = "subscribed" | "join" | "leave" | "message"; +type ChannelEventCallback = (msg: ServerMessage) => void; + +interface PusherClientConfig { + url: string; + getToken: () => string | null; +} + +const INITIAL_RECONNECT_DELAY = 1000; +const MAX_RECONNECT_DELAY = 30000; +const PING_INTERVAL = 20000; // 20s — keeps alive through GKE gateway timeout + +export class PusherClient { + private config: PusherClientConfig; + private ws: WebSocket | null = null; + private state: ConnectionState = "disconnected"; + private stateListeners = new Set<(state: ConnectionState) => void>(); + + // Channel event listeners: channelId → eventType → callbacks + private listeners = new Map< + string, + Map> + >(); + + // Active subscriptions for re-subscribe on reconnect + private activeSubscriptions = new Set(); + + // Reconnection state + private reconnectDelay = INITIAL_RECONNECT_DELAY; + private reconnectTimer: ReturnType | null = null; + private shouldReconnect = false; + + // Keep-alive ping + private pingTimer: ReturnType | null = null; + + constructor(config: PusherClientConfig) { + this.config = config; + } + + get connectionState(): ConnectionState { + return this.state; + } + + connect(): void { + if (this.ws) return; + + const token = this.config.getToken(); + if (!token) { + console.warn("[pusher] no token available, cannot connect"); + return; + } + + this.shouldReconnect = true; + this.setState( + this.state === "reconnecting" ? "reconnecting" : "connecting", + ); + + const url = `${this.config.url}?token=${encodeURIComponent(token)}`; + this.ws = new WebSocket(url); + + this.ws.onopen = () => { + this.setState("connected"); + this.reconnectDelay = INITIAL_RECONNECT_DELAY; + this.startPing(); + this.resubscribeAll(); + }; + + this.ws.onclose = () => { + this.cleanup(); + if (this.shouldReconnect) { + this.scheduleReconnect(); + } + }; + + this.ws.onerror = (event) => { + console.warn("[pusher] websocket error", event); + // onclose will fire after onerror, so reconnection is handled there + }; + + this.ws.onmessage = (event) => { + this.handleMessage(event.data as string); + }; + } + + disconnect(): void { + this.shouldReconnect = false; + this.clearReconnectTimer(); + this.cleanup(); + this.activeSubscriptions.clear(); + this.setState("disconnected"); + } + + subscribe(channelId: string): void { + this.activeSubscriptions.add(channelId); + this.send({ type: "subscribe", channel: channelId }); + } + + unsubscribe(channelId: string): void { + this.activeSubscriptions.delete(channelId); + this.send({ type: "unsubscribe", channel: channelId }); + } + + sendMessage(channelId: string, payload: unknown): void { + this.send({ type: "message", channel: channelId, payload }); + } + + on( + channelId: string, + event: ChannelEventType, + callback: ChannelEventCallback, + ): void { + if (!this.listeners.has(channelId)) { + this.listeners.set(channelId, new Map()); + } + const channelListeners = this.listeners.get(channelId)!; + if (!channelListeners.has(event)) { + channelListeners.set(event, new Set()); + } + channelListeners.get(event)!.add(callback); + } + + off( + channelId: string, + event: ChannelEventType, + callback: ChannelEventCallback, + ): void { + const channelListeners = this.listeners.get(channelId); + if (!channelListeners) return; + const eventListeners = channelListeners.get(event); + if (!eventListeners) return; + eventListeners.delete(callback); + + // Cleanup empty maps + if (eventListeners.size === 0) channelListeners.delete(event); + if (channelListeners.size === 0) this.listeners.delete(channelId); + } + + onStateChange(callback: (state: ConnectionState) => void): () => void { + this.stateListeners.add(callback); + return () => this.stateListeners.delete(callback); + } + + // --- Private --- + + private send(msg: { type: string; channel?: string; payload?: unknown }): void { + if (this.ws?.readyState === WebSocket.OPEN) { + this.ws.send(JSON.stringify(msg)); + } + } + + private handleMessage(data: string): void { + let msg: ServerMessage; + try { + msg = JSON.parse(data); + } catch { + console.warn("[pusher] failed to parse message", data); + return; + } + + if (msg.type === "error") { + console.warn("[pusher] server error:", msg.message); + return; + } + + if (!msg.channel) return; + + const channelListeners = this.listeners.get(msg.channel); + if (!channelListeners) return; + + const eventListeners = channelListeners.get(msg.type as ChannelEventType); + if (!eventListeners) return; + + for (const cb of eventListeners) { + try { + cb(msg); + } catch (err) { + console.error("[pusher] listener error", err); + } + } + } + + private resubscribeAll(): void { + for (const channelId of this.activeSubscriptions) { + this.send({ type: "subscribe", channel: channelId }); + } + } + + private scheduleReconnect(): void { + this.setState("reconnecting"); + + // Exponential backoff with jitter + const jitter = Math.random() * 0.5 + 0.75; // 0.75 - 1.25x + const delay = Math.min( + this.reconnectDelay * jitter, + MAX_RECONNECT_DELAY, + ); + + this.reconnectTimer = setTimeout(() => { + this.reconnectDelay = Math.min( + this.reconnectDelay * 2, + MAX_RECONNECT_DELAY, + ); + this.connect(); + }, delay); + } + + private cleanup(): void { + this.stopPing(); + if (this.ws) { + this.ws.onopen = null; + this.ws.onclose = null; + this.ws.onerror = null; + this.ws.onmessage = null; + if ( + this.ws.readyState === WebSocket.OPEN || + this.ws.readyState === WebSocket.CONNECTING + ) { + this.ws.close(); + } + this.ws = null; + } + } + + private clearReconnectTimer(): void { + if (this.reconnectTimer) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } + } + + private startPing(): void { + this.stopPing(); + this.pingTimer = setInterval(() => { + // Send an empty message as a keep-alive + if (this.ws?.readyState === WebSocket.OPEN) { + this.ws.send("ping"); + } + }, PING_INTERVAL); + } + + private stopPing(): void { + if (this.pingTimer) { + clearInterval(this.pingTimer); + this.pingTimer = null; + } + } + + private setState(state: ConnectionState): void { + if (this.state === state) return; + this.state = state; + for (const cb of this.stateListeners) { + cb(state); + } + } +} diff --git a/js/src/lib/pusher-provider.tsx b/js/src/lib/pusher-provider.tsx new file mode 100644 index 0000000..8174d36 --- /dev/null +++ b/js/src/lib/pusher-provider.tsx @@ -0,0 +1,76 @@ +import { + createContext, + useContext, + useEffect, + useRef, + useState, + type ReactNode, +} from "react"; +import { PusherClient, type ConnectionState } from "./pusher-client"; +import { useSessionStore } from "@/stores/session-store"; + +const PusherContext = createContext(null); +const PusherStateContext = createContext("disconnected"); + +// TODO: make this configurable per environment +const PUSHER_URL = "wss://pusher.dev.flowy.live/ws"; + +export function PusherProvider({ children }: { children: ReactNode }) { + const token = useSessionStore((s) => s.token); + const clientRef = useRef(null); + const [connectionState, setConnectionState] = + useState("disconnected"); + + useEffect(() => { + if (!token) { + // Disconnect if token is cleared (logout) + if (clientRef.current) { + clientRef.current.disconnect(); + clientRef.current = null; + setConnectionState("disconnected"); + } + return; + } + + const client = new PusherClient({ + url: PUSHER_URL, + getToken: () => useSessionStore.getState().token, + }); + + clientRef.current = client; + + const unsubscribeState = client.onStateChange((state) => { + setConnectionState(state); + }); + + client.connect(); + + return () => { + unsubscribeState(); + client.disconnect(); + clientRef.current = null; + }; + }, [token]); + + return ( + + + {children} + + + ); +} + +/** + * Returns the PusherClient instance, or null if not connected. + */ +export function usePusherClient(): PusherClient | null { + return useContext(PusherContext); +} + +/** + * Returns the current WebSocket connection state. + */ +export function usePusherConnectionState(): ConnectionState { + return useContext(PusherStateContext); +}