1"use client";
2
3
31import { WS_URL } from "./constants";
32import type { OwnPresence, RealtimeServerEvent } from "./types";
33
34export type RealtimeStatus = "disconnected" | "connecting" | "connected" | "reconnecting";
35
36export type RealtimeEvent =
37 | "status"
38 | "presence"
39 | "presence-leave"
40 | "block:upsert"
41 | "block:delete"
42 | "block:move"
43 | "block:changed"
44 | "comment.event"
45 | "notification.push"
46 | "ack";
47
48
59export function realtimeUrl(
60 token: string | null | undefined,
61 base: string = WS_URL,
62): string {
63 const root = base.replace(/\/+$/, "");
64 return token ? `${root}/realtime?token=${encodeURIComponent(token)}` : `${root}/realtime`;
65}
66
67type Listener = (event: RealtimeServerEvent) => void;
68type StatusListener = (status: RealtimeStatus) => void;
69
70const RECONNECT_BASE_MS = 250;
71const RECONNECT_MAX_MS = 15_000;
72
73class RealtimeClient {
74 private ws: WebSocket | null = null;
75 private status: RealtimeStatus = "disconnected";
76 private channels = new Set<string>();
77 private reconnectAttempts = 0;
78 private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
79 private closedByCaller = false;
80 private token: string | null = null;
81 private listeners = new Map<RealtimeEvent, Set<Listener>>();
82 private statusListeners = new Set<StatusListener>();
83
84 get connectionStatus(): RealtimeStatus {
85 return this.status;
86 }
87
88 get isAvailable(): boolean {
89 return typeof WebSocket !== "undefined";
90 }
91
92
93 connect(getToken: () => string | null): void {
94 if (!this.isAvailable) return;
95 this.closedByCaller = false;
96 this.token = getToken();
97 if (
98 this.ws &&
99 (this.ws.readyState === WebSocket.OPEN || this.ws.readyState === WebSocket.CONNECTING)
100 ) {
101 return;
102 }
103 this.setStatus("connecting");
104 const url = realtimeUrl(this.token);
105 const ws = new WebSocket(url);
106 this.ws = ws;
107
108 ws.onopen = () => {
109 this.reconnectAttempts = 0;
110 this.setStatus("connected");
111
112 for (const channel of this.channels) {
113 this.send({ type: "subscribe", channel });
114 }
115 };
116 ws.onmessage = (msg: MessageEvent) => this.handleMessage(msg);
117 ws.onerror = () => {
118
119 };
120 ws.onclose = () => {
121 this.ws = null;
122 if (this.closedByCaller) {
123 this.setStatus("disconnected");
124 return;
125 }
126 this.setStatus("reconnecting");
127 this.scheduleReconnect(getToken);
128 };
129 }
130
131
132 subscribe(channel: string): void {
133 this.channels.add(channel);
134 if (this.ws?.readyState === WebSocket.OPEN) {
135 this.send({ type: "subscribe", channel });
136 }
137 }
138
139 unsubscribe(channel: string): void {
140 this.channels.delete(channel);
141 if (this.ws?.readyState === WebSocket.OPEN) {
142 this.send({ type: "unsubscribe", channel });
143 }
144 }
145
146
154 sendPresence(state: OwnPresence): void {
155 const payload: Record<string, unknown> = { type: "presence" };
156 if (state.cursor !== undefined) payload.cursor = state.cursor;
157 if (state.viewing !== undefined) payload.viewing = state.viewing;
158 this.send(payload);
159 }
160
161
163 sendPresenceLeave(): void {
164 this.send({ type: "presence-leave" });
165 }
166
167 disconnect(): void {
168 this.closedByCaller = true;
169 if (this.reconnectTimer) {
170 clearTimeout(this.reconnectTimer);
171 this.reconnectTimer = null;
172 }
173 this.channels.clear();
174 if (this.ws) {
175 const ws = this.ws;
176 this.ws = null;
177 ws.onclose = null;
178 ws.onerror = null;
179 ws.onmessage = null;
180 ws.close();
181 }
182 this.setStatus("disconnected");
183 }
184
185
186 on(event: RealtimeEvent, listener: Listener): () => void {
187 let set = this.listeners.get(event);
188 if (!set) {
189 set = new Set();
190 this.listeners.set(event, set);
191 }
192 set.add(listener);
193 return () => set.delete(listener);
194 }
195
196 onStatus(listener: StatusListener): () => void {
197 this.statusListeners.add(listener);
198 return () => this.statusListeners.delete(listener);
199 }
200
201
202 emit(event: RealtimeServerEvent): void {
203 const key = event.type as RealtimeEvent;
204 this.listeners.get(key)?.forEach((l) => l(event));
205 }
206
207
208
209 private send(payload: unknown): void {
210 if (this.ws?.readyState === WebSocket.OPEN) {
211 this.ws.send(JSON.stringify(payload));
212 }
213 }
214
215 private handleMessage(msg: MessageEvent): void {
216 let event: RealtimeServerEvent;
217 try {
218 event = JSON.parse(String(msg.data)) as RealtimeServerEvent;
219 } catch {
220 return;
221 }
222 this.emit(event);
223 }
224
225 private scheduleReconnect(getToken: () => string | null): void {
226 this.reconnectAttempts += 1;
227 const base = Math.min(RECONNECT_BASE_MS * 2 ** (this.reconnectAttempts - 1), RECONNECT_MAX_MS);
228 const delay = base / 2 + Math.random() * (base / 2);
229 this.reconnectTimer = setTimeout(() => {
230 this.reconnectTimer = null;
231 this.connect(getToken);
232 }, delay);
233 }
234
235 private setStatus(status: RealtimeStatus): void {
236 if (this.status === status) return;
237 this.status = status;
238 this.statusListeners.forEach((l) => l(status));
239 }
240}
241
242
243export const realtime = new RealtimeClient();
244
245
246export const channelForPage = (pageId: string): string => `page:${pageId}`;
247export const channelForWorkspace = (workspaceId: string): string => `workspace:${workspaceId}`;
248
