written clean architecture for socket + added throttle method to reduce server load

Author: Arunkoo

Client/src/hooks/useDocumentSocket.ts

@@ -1,67 +1,152 @@
-import { useEffect, useRef } from "react";
-import { io, Socket } from "socket.io-client";
-/* eslint-disable @typescript-eslint/no-explicit-any */
-interface UseDocumentSocketParams {
-  workspaceId: string;
-  documentId: string;
-  onContentChange: (content: string) => void;
-  onCursorChange: (cursor: any) => void;
-}
+import { useCallback, useEffect, useRef, useState } from "react";
+import type {
+  ContentChangePayload,
+  CursorChangePayload,
+  PresencePayload,
+} from "../types/realtime";
+import { createSocket, disconnectSocket, getSocket } from "../lib/socket";
+
+export function useDocumentSocket(options: {
+  workspaceId?: string | null;
+  documentId?: string | undefined;
+  onRemoteContent: (payload: ContentChangePayload) => void;
+  onRemoteCursor?: (payload: CursorChangePayload) => void;
+}) {
+  const { workspaceId, documentId, onRemoteContent, onRemoteCursor } = options;
 
-export const useDocumentSocket = ({
-  workspaceId,
-  documentId,
-  onContentChange,
-  onCursorChange,
-}: UseDocumentSocketParams) => {
-  const socketRef = useRef<Socket | null>(null);
+  //maintaing a list of online user....
+  const [onlineUsers, setOnlineUsers] = useState<string[]>([]);
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  const socketRef = useRef<any>(null);
 
+  // throttle state  to 200ms....
+  const lastSentAt = useRef<number>(0);
+  const pendingContent = useRef<string | null>(null);
+  const timerRef = useRef<number | null>(null);
+  const THROTTLE_MS = 200;
+
+  //connect and attach lstners..
   useEffect(() => {
-    if (!workspaceId || !documentId) return;
-    const socket = io("http://localhost:3000", {
-      transports: ["websocket"],
-      withCredentials: true,
-      query: {
-        workspaceId,
-        documentId,
-      },
-    });
+    if (!workspaceId || documentId) return;
 
+    const WS_URL = import.meta.env.VITE_API_URL
+      ? (import.meta.env.VITE_API_URL as string).replace(/\/api.*$/, "")
+      : "http://localhost:3000";
+
+    const socket = createSocket(WS_URL, { workspaceId, documentId });
     socketRef.current = socket;
 
-    //listning to the sockets....
-    socket.on("contentChange", (data) => {
-      onContentChange(data.content);
+    //connection logs....
+
+    socket.on("connect", () => {
+      console.log("[socket] connected", socket.id);
     });
-    socket.on("cursorChange", (data) => {
-      onCursorChange?.(data);
+    socket.on("connect_error", (err) => {
+      console.warn("[socket] connect_error", err?.message ?? err);
     });
-    socket.on("userJoined", (data) => {
-      console.log("User Joined:", data);
+    socket.on("disconnect", (reason) => {
+      console.log("[socket] disconnected", reason);
     });
-    socket.on("userLeft", (data) => {
-      console.log("User Left:", data);
+
+    socket.on("contentChange", (payload: ContentChangePayload) => {
+      if (!payload || typeof payload.content !== "string") return;
+      onRemoteContent(payload);
     });
-    return () => {
-      socket.disconnect();
-    };
-    // eslint-disable-next-line react-hooks/exhaustive-deps
-  }, [workspaceId, documentId]);
-
-  //send updates ..
-  const sendDocumentUpdate = (newContent: string) => {
-    socketRef.current?.emit("documentUpdate", {
-      documentId,
-      newContent,
+
+    socket.on("cursorChange", (payload: CursorChangePayload) => {
+      if (!payload) return;
+      onRemoteCursor?.(payload);
     });
-  };
 
-  const sendCursorUpdate = (position: number) => {
-    socketRef.current?.emit("cursorUpdate", {
-      documentId,
-      position,
+    socket.on("userJoined", (payload: PresencePayload) => {
+      if (!payload?.userId) return;
+      setOnlineUsers((prev) => {
+        if (prev.includes(payload.userId)) return prev;
+        return [...prev, payload.userId];
+      });
     });
-  };
 
-  return { sendDocumentUpdate, sendCursorUpdate };
-};
+    socket.on("userLeft", (payload: PresencePayload) => {
+      if (!payload?.userId) return;
+      setOnlineUsers((prev) => prev.filter((u) => u !== payload.userId));
+    });
+
+    return () => {
+      // cleanup: stop listeners and disconnect if this was the only usage
+      socket.off("contentChange");
+      socket.off("cursorChange");
+      socket.off("userJoined");
+      socket.off("userLeft");
+    };
+  }, [workspaceId, documentId, onRemoteContent, onRemoteCursor]);
+
+  const safeDisconnect = useCallback(() => {
+    // remove all listeners and disconnect
+    const s = getSocket();
+    if (s) {
+      s.off("contentChange");
+      s.off("cursorChange");
+      s.off("userJoined");
+      s.off("userLeft");
+    }
+    disconnectSocket();
+    socketRef.current = null;
+    setOnlineUsers([]);
+  }, []);
+
+  // Throttled document update: coalesce updates and send at most once/200ms
+  const sendDocumentUpdate = useCallback(
+    (newContent: string) => {
+      const s = getSocket();
+      if (!s) return;
+
+      const now = Date.now();
+      const since = now - (lastSentAt.current || 0);
+
+      if (since >= THROTTLE_MS) {
+        lastSentAt.current = now;
+        s.emit("documentUpdate", { documentId, newContent });
+        // clear pending
+        if (timerRef.current) {
+          window.clearTimeout(timerRef.current);
+          timerRef.current = null;
+        }
+        pendingContent.current = null;
+        return;
+      }
+
+      // otherwise schedule latest content to be sent later
+      pendingContent.current = newContent;
+      if (timerRef.current) return;
+
+      const wait = THROTTLE_MS - since;
+      timerRef.current = window.setTimeout(() => {
+        const latest = pendingContent.current;
+        if (latest != null) {
+          const sock = getSocket();
+          sock?.emit("documentUpdate", { documentId, newContent: latest });
+          lastSentAt.current = Date.now();
+        }
+        pendingContent.current = null;
+        timerRef.current = null;
+      }, wait) as unknown as number;
+    },
+    [documentId]
+  );
+
+  const sendCursorUpdate = useCallback(
+    (position: number) => {
+      const s = getSocket();
+      if (!s) return;
+      s.emit("cursorUpdate", { documentId, position });
+    },
+    [documentId]
+  );
+
+  return {
+    sendDocumentUpdate,
+    sendCursorUpdate,
+    onlineUsers,
+    safeDisconnect,
+  };
+}

Client/src/lib/socket.ts

@@ -0,0 +1,27 @@
+import { io, Socket } from "socket.io-client";
+
+let socket: Socket | null = null;
+
+export function createSocket(
+  url: string,
+  query: Record<string, string | undefined>
+) {
+  if (socket) return socket;
+  socket = io(url, {
+    transports: ["websocket"],
+    withCredentials: true,
+    query,
+    autoConnect: true,
+  });
+  return socket;
+}
+
+export function getSocket() {
+  return socket;
+}
+
+export function disconnectSocket() {
+  if (!socket) return;
+  socket.disconnect();
+  socket = null;
+}

Client/src/types/realtime.ts

@@ -0,0 +1,16 @@
+export type ContentChangePayload = {
+  userId?: string;
+  content: string;
+  timestamp?: string;
+};
+
+export type CursorChangePayload = {
+  userId?: string;
+  position: number;
+  timestamp?: string;
+};
+
+export type PresencePayload = {
+  userId: string;
+  message?: string;
+};