WebSocket receiver

package-lock.json

@@ -61,7 +61,8 @@
         "react-dom": "19.2.3",
         "recharts": "^2.15.4",
         "tailwind-merge": "^3.4.0",
-        "tailwindcss-animate": "^1.0.7"
+        "tailwindcss-animate": "^1.0.7",
+        "ws": "^8.19.0"
       },
       "devDependencies": {
         "@tailwindcss/postcss": "^4",

package.json

@@ -3,9 +3,9 @@
   "version": "0.1.0",
   "private": true,
   "scripts": {
-    "dev": "next dev",
+    "dev": "node server.mjs",
     "build": "next build",
-    "start": "next start"
+    "start": "NODE_ENV=production node server.mjs"
   },
   "dependencies": {
     "@codemirror/lang-go": "^6.0.1",
@@ -61,7 +61,8 @@
     "react-dom": "19.2.3",
     "recharts": "^2.15.4",
     "tailwind-merge": "^3.4.0",
-    "tailwindcss-animate": "^1.0.7"
+    "tailwindcss-animate": "^1.0.7",
+    "ws": "^8.19.0"
   },
   "devDependencies": {
     "@tailwindcss/postcss": "^4",

server.mjs

@@ -0,0 +1,60 @@
+import { createServer } from 'http';
+import next from 'next';
+import { WebSocketServer } from 'ws';
+
+const dev = process.env.NODE_ENV !== 'production';
+const app = next({ dev });
+const handle = app.getRequestHandler();
+
+app.prepare().then(() => {
+  const server = createServer((req, res) => {
+    handle(req, res);
+  });
+  
+  const clients = {};
+
+  const wss = new WebSocketServer({ noServer: true });
+
+  // Handle WebSocket connections
+  wss.on('connection', (ws, req) => {
+    ws.on('message', (msg) => {
+      const runId = msg.toString();
+      if (!clients[runId]) clients[runId] = []
+      clients[runId].push(ws);
+    });
+  });
+
+  // Handle POST requests
+  server.on('request', async (req, res) => {
+    console.log(req.method, req.url)
+    if (req.method === 'POST' && req.url === '/api/run/move') {
+      let body = '';
+      req.on('data', chunk => body += chunk);
+      req.on('end', () => {
+        const move = JSON.parse(body);
+
+        clients[move.runId]?.forEach(client => {
+          if (client.readyState === client.OPEN) {
+            client.send(JSON.stringify(move.moves));
+          }
+        });
+
+        res.writeHead(200, { 'Content-Type': 'application/json' });
+        res.end(JSON.stringify({ status: 'ok' }));
+      });
+    }
+  });
+
+  // Upgrade HTTP connections to WebSocket when needed
+  server.on('upgrade', (req, socket, head) => {
+    if (req.url === '/api/view-run') {
+      wss.handleUpgrade(req, socket, head, (ws) => {
+        wss.emit('connection', ws, req);
+      });
+    }
+  });
+
+  server.listen(3000, () => {
+    console.log('Server running on http://localhost:3000');
+  });
+});

src/app/runs/[runId]/components/Editor.tsx

@@ -8,7 +8,7 @@ import { go } from "@codemirror/lang-go";
 import { json } from "@codemirror/lang-json";
 import { python } from "@codemirror/lang-python";
 import { java } from "@codemirror/lang-java";
-import { useEffect, useState } from "react";
+import { useEffect, useRef, useState } from "react";
 import useAction from "@/hook/use-action";
 import { getLiveRunMoves } from "@/modules/live-run/actions";
 import { LiveRun, LiveRunMove } from "@/modules/live-run/types";
@@ -43,66 +43,70 @@ function getLanguageExtension(lang: Language) {
 
 export default function Editor({ run }: { run: LiveRun | null | undefined }) {
   const [editorView, setEditorView] = useState<EditorView>();
-  const [rawMoves] = useAction<LiveRunMove[]>(
-    async () => (run ? await getLiveRunMoves(run.id) : []),
-    [run],
-  );
+  const [isConnected, setConnected] = useState(false);
 
-  const [_, setLastMoveId] = useState<number>(-1);
-  const [moves, setMoves] = useState<LiveRunMove[]>([]);
+  const [rawMoves, setRawMoves] = useState<LiveRunMove[]>([]);
   const [movesLoading, setMovesLoading] = useState(false);
+  const scheduledTimeouts = useRef<NodeJS.Timeout[]>([]);
 
+  // WebSocket connection
   useEffect(() => {
-    if (!rawMoves || rawMoves.length === 0 || movesLoading) return;
+    if (!run) return;
 
-    setLastMoveId((prevLastMoveId) => {
-      const newMoves = rawMoves.filter((move) => move.moveId > prevLastMoveId);
+    const ws = new WebSocket(`ws://${window.location.host}/api/view-run`);
 
-      if (newMoves.length === 0) return prevLastMoveId;
+    ws.onopen = () => {
+      ws.send(run.id);
+    };
 
-      setMoves(newMoves);
+    ws.onmessage = (m) => {
+      const data: LiveRunMove[] = JSON.parse(m.data);
+      setRawMoves(data);
+    };
 
-      return Math.max(...newMoves.map((m) => m.moveId));
-    });
-  }, [rawMoves, movesLoading]);
+    return () => ws.close();
+  }, [run]);
 
+  // Scheduler
   useEffect(() => {
-    if (!editorView || !moves) return;
+    if (!editorView || !rawMoves.length) return;
 
-    let cancelled = false;
+    setMovesLoading(true);
 
-    const executeMove = async (move: LiveRunMove) => {
-      return new Promise<void>((resolve) => {
+    // Clear any existing scheduled timeouts
+    scheduledTimeouts.current.forEach((t) => clearTimeout(t));
+    scheduledTimeouts.current = [];
+
+    const startTime = Date.now();
+
+    rawMoves.forEach((move) => {
       const timeout = setTimeout(() => {
-          if (!editorView || cancelled) return resolve();
+        if (!editorView) return;
 
         editorView.dispatch({
           selection: { anchor: move.cursor },
           scrollIntoView: true,
           changes: move.changes,
         });
-
-          resolve();
       }, move.latency);
 
-        return () => clearTimeout(timeout);
+      scheduledTimeouts.current.push(timeout);
     });
-    };
 
-    (async () => {
-      if (!moves) return;
-      setMovesLoading(true);
-      for (const move of moves) {
-        if (cancelled) break;
-        await executeMove(move);
-      }
-      setMovesLoading(false);
-    })();
+    // Stop loading after the last move
+    const lastMoveLatency = rawMoves[rawMoves.length - 1].latency;
+    const finishTimeout = setTimeout(
+      () => setMovesLoading(false),
+      lastMoveLatency + 50,
+    );
+    scheduledTimeouts.current.push(finishTimeout);
 
     return () => {
-      cancelled = true;
+      scheduledTimeouts.current.forEach((t) => clearTimeout(t));
+      scheduledTimeouts.current = [];
+      setMovesLoading(false);
     };
-  }, [moves]);
+  }, [rawMoves, editorView]);
 
   return (
     <CodeMirror

src/app/runs/[runId]/page.tsx

@@ -16,7 +16,7 @@ import { useParams } from "next/navigation";
 export default function Run() {
   const params = useParams<{ runId: string }>();
 
-  const [run] = useActionInterval(() => getLiveRun(params.runId), 1000, [
+  const [run] = useActionInterval(() => getLiveRun(params.runId), 5000, [
     params,
   ]);
 

src/modules/live-run/core.ts

@@ -4,7 +4,6 @@ import "server-only";
 import { LiveRun, LiveRunMove } from "./types";
 import { SupabaseClient } from "@supabase/supabase-js";
 import { redis } from "@/lib/redis";
-import { randomInt } from "crypto";
 
 /** -------------------- Live Runs in Redis -------------------- */