Spaces:
Running
Running
| "use client"; | |
| import { useEffect, useRef, useState, useCallback } from "react"; | |
| import type { QueryClient } from "@tanstack/react-query"; | |
| import type { AppRouter } from "@micprotocol/api"; | |
| import { createTRPCClient, httpBatchLink } from "@trpc/client"; | |
| export type RealtimeFrame = | |
| | { type: "hello"; userId: string; publicUrl: string; at: string } | |
| | { type: "auth_error"; message: string; at?: string } | |
| | { type: "snapshot"; conversationId: string; title: string; messages: any[]; sequence: number; members: any; presence: any } | |
| | { type: "message"; id: string; conversationId: string; userId: string; body: string; meta?: any; seq: number; at: string } | |
| | { type: "system"; conversationId: string; body: string; seq: number; at: string; meta?: any } | |
| | { type: "reaction"; conversationId: string; userId: string; messageId: string; emoji: string; seq: number; at: string } | |
| | { type: "typing"; conversationId: string; userId: string; at: string } | |
| | { type: "read"; conversationId: string; userId: string; seq: number; at: string } | |
| | { type: "presence"; conversationId: string; users: any[] } | |
| | { type: "subscribed"; conversationId: string } | |
| | { type: "ack"; message?: any; reaction?: any } | |
| | { type: "pong"; at: string } | |
| | { type: "error"; message: string } | |
| | { type: string; [key: string]: any }; | |
| export interface UseRealtimeOptions { | |
| /** Called for every inbound frame. */ | |
| onFrame?: (frame: RealtimeFrame) => void; | |
| /** Called once on successful handshake (after `hello`). */ | |
| onReady?: (info: { userId: string }) => void; | |
| /** Called when the connection is lost / errors out. */ | |
| onError?: (err: string) => void; | |
| /** Auto-reconnect on disconnect (default true). */ | |
| autoReconnect?: boolean; | |
| } | |
| export interface RealtimeClient { | |
| send: (frame: Record<string, unknown>) => void; | |
| subscribe: (conversationId: string) => void; | |
| unsubscribe: (conversationId: string) => void; | |
| sendMessage: (conversationId: string, body: string, meta?: Record<string, unknown>) => void; | |
| sendTyping: (conversationId: string) => void; | |
| sendRead: (conversationId: string, seq: number) => void; | |
| sendReaction: (conversationId: string, messageId: string, emoji: string) => void; | |
| disconnect: () => void; | |
| } | |
| /** | |
| * Resolve the realtime base URL for the CURRENT page context. | |
| * | |
| * Production safety: when the page is served over https, a bare `ws://` URL | |
| * is upgraded to `wss://` so a deployment that forgot to set a wss:// | |
| * NEXT_PUBLIC_MIC_REALTIME_URL still works — and never sends the realtime | |
| * JWT over an unencrypted channel. Local dev (http page) is left untouched. | |
| */ | |
| function resolveRealtimeBaseUrl(): string { | |
| const configured = process.env.NEXT_PUBLIC_MIC_REALTIME_URL; | |
| if ( | |
| configured && | |
| configured.startsWith("ws://") && | |
| typeof window !== "undefined" && | |
| window.location.protocol === "https:" | |
| ) { | |
| return configured.replace(/^ws:\/\//, "wss://"); | |
| } | |
| return configured ?? "ws://localhost:4001"; | |
| } | |
| /** | |
| * Subscribe to the mic-realtime WebSocket. Opens a token from the tRPC | |
| * router, connects to /ws?token=..., and re-emits every frame to | |
| * `onFrame`. Returns a stable client handle for sending frames. | |
| */ | |
| export function useRealtime(opts: UseRealtimeOptions = {}): { | |
| client: RealtimeClient | null; | |
| connected: boolean; | |
| authError: string | null; | |
| } { | |
| const [connected, setConnected] = useState(false); | |
| const [authError, setAuthError] = useState<string | null>(null); | |
| const wsRef = useRef<WebSocket | null>(null); | |
| const optsRef = useRef(opts); | |
| const reconnectAttemptRef = useRef(0); | |
| useEffect(() => { | |
| optsRef.current = opts; | |
| }, [opts]); | |
| useEffect(() => { | |
| let cancelled = false; | |
| let reconnectTimer: ReturnType<typeof setTimeout> | null = null; | |
| const connect = async () => { | |
| try { | |
| const apiUrl = | |
| process.env.NEXT_PUBLIC_API_URL || "http://localhost:4000"; | |
| const api = createTRPCClient<AppRouter>({ | |
| links: [ | |
| httpBatchLink({ | |
| url: `${apiUrl}/trpc`, | |
| fetch: (url, init) => | |
| fetch(url, { ...init, credentials: "include" }), | |
| }), | |
| ], | |
| }); | |
| const { token } = await api.chat.realtimeToken.query(); | |
| if (cancelled) return; | |
| const wsUrl = `${resolveRealtimeBaseUrl()}/ws?token=${encodeURIComponent(token)}`; | |
| const ws = new WebSocket(wsUrl); | |
| wsRef.current = ws; | |
| ws.onopen = () => { | |
| setConnected(true); | |
| setAuthError(null); | |
| reconnectAttemptRef.current = 0; | |
| }; | |
| ws.onmessage = (ev) => { | |
| try { | |
| const frame = JSON.parse(ev.data) as RealtimeFrame; | |
| if (frame.type === "hello") { | |
| optsRef.current.onReady?.({ | |
| userId: (frame as any).userId, | |
| }); | |
| } else if (frame.type === "auth_error") { | |
| setAuthError((frame as any).message ?? "auth error"); | |
| optsRef.current.onError?.((frame as any).message ?? "auth error"); | |
| } else { | |
| optsRef.current.onFrame?.(frame); | |
| } | |
| } catch { | |
| // ignore malformed frame | |
| } | |
| }; | |
| ws.onerror = () => { | |
| optsRef.current.onError?.("socket error"); | |
| }; | |
| ws.onclose = (ev) => { | |
| setConnected(false); | |
| wsRef.current = null; | |
| if ( | |
| !cancelled && | |
| (opts.autoReconnect ?? true) && | |
| ev.code !== 1000 && | |
| reconnectAttemptRef.current < 8 | |
| ) { | |
| const delay = Math.min( | |
| 30_000, | |
| 500 * Math.pow(2, reconnectAttemptRef.current), | |
| ); | |
| reconnectAttemptRef.current++; | |
| reconnectTimer = setTimeout(connect, delay); | |
| } | |
| }; | |
| } catch (err) { | |
| const message = | |
| err instanceof Error ? err.message : String(err); | |
| setAuthError(message); | |
| optsRef.current.onError?.(message); | |
| if (!cancelled && (opts.autoReconnect ?? true)) { | |
| reconnectTimer = setTimeout(connect, 2_000); | |
| } | |
| } | |
| }; | |
| connect(); | |
| return () => { | |
| cancelled = true; | |
| if (reconnectTimer) clearTimeout(reconnectTimer); | |
| wsRef.current?.close(); | |
| wsRef.current = null; | |
| }; | |
| }, [opts.autoReconnect]); | |
| const send = useCallback((frame: Record<string, unknown>) => { | |
| const ws = wsRef.current; | |
| if (ws && ws.readyState === WebSocket.OPEN) { | |
| ws.send(JSON.stringify(frame)); | |
| } | |
| }, []); | |
| const client: RealtimeClient | null = connected | |
| ? { | |
| send, | |
| subscribe: (conversationId) => | |
| send({ type: "subscribe", conversationId }), | |
| unsubscribe: (conversationId) => | |
| send({ type: "unsubscribe", conversationId }), | |
| sendMessage: (conversationId, body, meta) => | |
| send({ type: "message", conversationId, body, meta }), | |
| sendTyping: (conversationId) => | |
| send({ type: "typing", conversationId }), | |
| sendRead: (conversationId, seq) => | |
| send({ type: "read", conversationId, seq }), | |
| sendReaction: (conversationId, messageId, emoji) => | |
| send({ | |
| type: "reaction", | |
| conversationId, | |
| messageId, | |
| emoji, | |
| }), | |
| disconnect: () => wsRef.current?.close(), | |
| } | |
| : null; | |
| return { client, connected, authError }; | |
| } | |