Merge pull request #15 from orus-dev/websocket

Websocket #14

package.json

@@ -3,9 +3,10 @@
   "version": "0.1.0",
   "private": true,
   "scripts": {
-    "dev": "next dev",
+    "dev": "next dev --turbopack",
-    "build": "next build",
+    "build": "npm run build:lib && next build",
-    "start": "next start"
+    "postinstall": "npm run ws -- -y",
+    "ws": "npx --yes next-ws-cli@latest patch"
   },
   "dependencies": {
     "@codemirror/lang-go": "^6.0.1",
@@ -55,13 +56,15 @@
     "ioredis": "^5.9.2",
     "lucide-react": "^0.563.0",
     "next": "16.1.6",
+    "next-ws": "^2.1.16",
     "radix-ui": "^1.4.3",
     "react": "19.2.3",
     "react-day-picker": "^9.13.0",
     "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",

src/app/api/run/move/route.ts

@@ -4,11 +4,11 @@ import * as Core from "@/modules/live-run/core";
 import { getSession } from "@/modules/account/core";
 
 export async function POST(req: NextRequest) {
-  const { runId, moves } = await req.json();
+  const { runId, moves, file, language } = await req.json();
 
-  if (!runId || !moves)
+  if (!runId || !moves || file === undefined || language === undefined)
     return NextResponse.json(
-      { error: "body requires runId and moves" },
+      { error: "body requires runId, moves, file and language" },
       { status: 400 },
     );
 
@@ -27,9 +27,7 @@ export async function POST(req: NextRequest) {
       { status: 401 },
     );
 
-  console.log(moves);
+  const run = await Core.addLiveRunEvent(liveRun.id, file, language, moves);
-
-  const run = await Core.addLiveRunMoves(liveRun.id, moves);
 
   return NextResponse.json({ ok: true, run });
 }

src/app/api/view-run/route.ts

@@ -0,0 +1,70 @@
+import { WebSocketServer, WebSocket } from "ws";
+import Redis from "ioredis";
+import { NextRequest, NextResponse } from "next/server";
+import { LiveRunEvent } from "@/modules/live-run/types";
+import { getLiveRun, updateLiveRunViews } from "@/modules/live-run/core";
+
+export function GET(req: NextRequest) {
+  const headers = new Headers();
+  headers.set("Connection", "Upgrade");
+  headers.set("Upgrade", "websocket");
+  return new Response("Upgrade Required", { status: 426, headers });
+}
+
+export function UPGRADE(client: WebSocket, server: WebSocketServer) {
+  const subscriber = new Redis();
+  let runId: string;
+
+  client.once("message", async (message) => {
+    runId = message.toString();
+
+    await updateLiveRunViews(runId, 1);
+
+    let lastEvent: LiveRunEvent | null = null;
+
+    const sendText = async () => {
+      const text = await subscriber.get(`liveRunText:${runId}`);
+
+      if (client.readyState === WebSocket.OPEN) {
+        client.send(
+          JSON.stringify({
+            file: lastEvent?.file,
+            language: lastEvent?.language,
+            moves: [],
+            text,
+          }),
+        );
+      }
+    };
+
+    sendText();
+
+    client.on("message", sendText);
+
+    subscriber.subscribe(`liveRunEvent:${runId}`, (err, count) => {
+      if (err) {
+        console.error("Failed to subscribe", err);
+        client.close(1011, "Redis subscription failed");
+      }
+    });
+
+    subscriber.on("message", (channel, message) => {
+      const data: LiveRunEvent = JSON.parse(message);
+
+      lastEvent = data;
+      if (client.readyState === WebSocket.OPEN) {
+        client.send(JSON.stringify({ ...data, text: null }));
+      }
+    });
+  });
+
+  client.once("close", async () => {
+    if (runId) await subscriber.unsubscribe(`liveRun:${runId}`);
+    subscriber.quit();
+    await updateLiveRunViews(runId, -1);
+  });
+
+  client.on("error", (err) => {
+    console.error("WebSocket error:", err);
+  });
+}

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

@@ -8,10 +8,10 @@ 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 { LiveRun, LiveRunEvent, LiveRunMove } from "@/modules/live-run/types";
 import useAction from "@/hook/use-action";
-import { getLiveRunMoves } from "@/modules/live-run/actions";
+import { getOrigin } from "@/modules/getOrigin";
-import { LiveRun, LiveRunMove } from "@/modules/live-run/types";
 
 type Language =
   | "javascript"
@@ -22,7 +22,7 @@ type Language =
   | "java"
   | "json";
 
-function getLanguageExtension(lang: Language) {
+function getLanguageExtension(lang: Language | null) {
   switch (lang) {
     case "javascript":
       return javascript();
@@ -41,76 +41,74 @@ function getLanguageExtension(lang: Language) {
   }
 }
 
-export default function Editor({ run }: { run: LiveRun | null | undefined }) {
+export default function Editor({
+  run,
+  onEvent,
+}: {
+  run: LiveRun | null | undefined;
+  onEvent: (s: LiveRunEvent) => void;
+}) {
+  const [origin] = useAction(getOrigin);
   const [editorView, setEditorView] = useState<EditorView>();
-  const [rawMoves] = useAction<LiveRunMove[]>(
+  const scheduledTimeouts = useRef<NodeJS.Timeout[]>([]);
-    async () => (run ? await getLiveRunMoves(run.id) : []),
+  const [text, setText] = useState("");
-    [run],
+  const [language, setLanguage] = useState<Language | null>(null);
-  );
-
-  const [_, setLastMoveId] = useState<number>(-1);
-  const [moves, setMoves] = useState<LiveRunMove[]>([]);
-  const [movesLoading, setMovesLoading] = useState(false);
 
+  // WebSocket connection
   useEffect(() => {
-    if (!rawMoves || rawMoves.length === 0 || movesLoading) return;
+    if (!run || !origin?.host) return;
 
-    setLastMoveId((prevLastMoveId) => {
+    const ws = new WebSocket(`ws://${origin.host}/api/view-run`);
-      const newMoves = rawMoves.filter((move) => move.moveId > prevLastMoveId);
 
-      if (newMoves.length === 0) return prevLastMoveId;
+    ws.onopen = () => {
+      ws.send(run.id);
+    };
 
-      setMoves(newMoves);
+    ws.onmessage = (m) => {
+      const data: LiveRunEvent = JSON.parse(m.data);
 
-      return Math.max(...newMoves.map((m) => m.moveId));
+      onEvent(data);
-    });
-  }, [rawMoves, movesLoading]);
 
-  useEffect(() => {
+      if (data.language !== language) {
-    if (!editorView || !moves) return;
+        setLanguage(data.language as Language);
+      }
+
+      if (data.text) {
+        setText(data.text);
+      }
 
       let cancelled = false;
 
-    const executeMove = async (move: LiveRunMove) => {
+      data.moves.forEach((move) => {
-      return new Promise<void>((resolve) => {
         const timeout = setTimeout(() => {
-          if (!editorView || cancelled) return resolve();
+          if (!editorView || cancelled) return;
 
+          try {
             editorView.dispatch({
               selection: { anchor: move.cursor },
               scrollIntoView: true,
               changes: move.changes,
             });
-
+          } catch {
-          resolve();
+            cancelled = true;
+            ws.send("getText");
+          }
         }, move.latency);
 
-        return () => clearTimeout(timeout);
+        scheduledTimeouts.current.push(timeout);
       });
     };
 
-    (async () => {
+    return () => ws.close();
-      if (!moves) return;
+  }, [run, ]);
-      setMovesLoading(true);
-      for (const move of moves) {
-        if (cancelled) break;
-        await executeMove(move);
-      }
-      setMovesLoading(false);
-    })();
-
-    return () => {
-      cancelled = true;
-    };
-  }, [moves]);
 
   return (
     <CodeMirror
       readOnly
       className="h-full w-full"
-      value={``}
+      value={text}
       extensions={[
-        getLanguageExtension("javascript"),
+        getLanguageExtension(language),
         EditorState.transactionFilter.of((tr) => {
           if (tr.selection && tr.isUserEvent("select")) {
             return [];

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

@@ -4,7 +4,13 @@ import { LiveRun } from "@/modules/live-run/types";
 import { Eye } from "lucide-react";
 import { useEffect, useState } from "react";
 
-export default function RunTimer({ run }: { run: LiveRun }) {
+export default function RunTimer({
+  run,
+  views,
+}: {
+  run: LiveRun;
+  views: number | null;
+}) {
   const [time, setTime] = useState("00:00:00");
 
   useEffect(() => {
@@ -29,7 +35,7 @@ export default function RunTimer({ run }: { run: LiveRun }) {
           </p>
           <p className="text-muted-foreground flex gap-1 items-center">
             <Eye className="w-4 h-4" />
-            {run.views}
+            {views || run.views}
           </p>
         </span>
       </CardContent>

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

@@ -7,16 +7,18 @@ import { FileTree } from "./components/FileTree";
 import { File } from "lucide-react";
 import Editor from "./components/Editor";
 import useAction, { useActionInterval } from "@/hook/use-action";
-import { getLiveRun } from "@/modules/live-run/actions";
+import { getLiveRun, getLiveRunViews } from "@/modules/live-run/actions";
 import ProblemCard from "@/app/problems/components/ProblemCard";
 import ProfileCard from "./components/ProfileCard";
 import { getProblem } from "@/modules/problems/actions";
 import { useParams } from "next/navigation";
+import { useState } from "react";
 
 export default function Run() {
   const params = useParams<{ runId: string }>();
 
-  const [run] = useActionInterval(() => getLiveRun(params.runId), 1000, [
+  const [run] = useAction(() => getLiveRun(params.runId), [params]);
+  const [views] = useActionInterval(() => getLiveRunViews(params.runId), 2500, [
     params,
   ]);
 
@@ -25,6 +27,11 @@ export default function Run() {
     [run],
   );
 
+  const [file, setFile] = useState<{
+    file: string | null;
+    language: string | null;
+  }>({ file: null, language: null });
+
   const chatMessages: ChatMessage[] = [
     { user: "Alice", message: "Nice split!", time: "1:53" },
     { user: "Bob", message: "Watch out for the string problem!", time: "1:54" },
@@ -51,7 +58,7 @@ export default function Run() {
           </div>
 
           <div className="flex-1 animate-fade-in opacity-0 stagger-2">
-            {run && <RunTimer run={run} />}
+            {run && <RunTimer run={run} views={views} />}
           </div>
 
           <div className="flex-2">
@@ -83,10 +90,15 @@ export default function Run() {
             {/* Code */}
             <div className="flex-1 w-full max-h-svh md:max-h-full h-full pl-0 md:pl-5 flex flex-col">
               <header className="text-sm font-medium text-muted-foreground flex items-center gap-1">
-                <File size={16} /> myfile.js
+                <File size={16} /> {file.file}
               </header>
               <div className="flex-1 pt-3 pb-4 md:pb-5 w-full h-full">
-                <Editor run={run} />
+                <Editor
+                  run={run}
+                  onEvent={(event) => {
+                    setFile({ file: event.file, language: event.language });
+                  }}
+                />
               </div>
             </div>
           </CardContent>

src/components/auth/ContinueWith.tsx

@@ -6,7 +6,7 @@ import { signInWithGitHub } from "@/modules/account/actions";
 
 export default function ContinueWith() {
   const continueWithGithub = () => {
-    signInWithGitHub(window.location.origin);
+    signInWithGitHub();
   };
   const continueWithGoogle = () => {};
 

src/lib/move.ts

@@ -0,0 +1,16 @@
+import { LiveRunMove } from "@/modules/live-run/types";
+
+export function applyMoves(text: string, moves: LiveRunMove[]) {
+  let result = text;
+
+  for (const move of moves) {
+    if (!move.changes) continue;
+
+    result =
+      result.slice(0, move.changes.from) +
+      move.changes.insert +
+      result.slice(move.changes.to);
+  }
+
+  return result;
+}

src/lib/supabase/server.ts

@@ -1,3 +1,5 @@
+"use server";
+
 import "server-only";
 
 import { createServerClient, type CookieOptions } from "@supabase/ssr";

src/modules/account/actions.ts

@@ -2,15 +2,20 @@
 
 import * as Core from "./core";
 import { createClient } from "@/lib/supabase/server";
+import { headers } from "next/headers";
 import { redirect } from "next/navigation";
 
-export async function signInWithGitHub(origin: string) {
+export async function signInWithGitHub() {
   const supabase = await createClient();
 
+  const h = await headers();
+  const host = h.get("host");
+  const protocol = h.get("x-forwarded-proto") ?? "https";
+
   const { data, error } = await supabase.auth.signInWithOAuth({
     provider: "github",
     options: {
-      redirectTo: `${origin}/api/auth/callback`,
+      redirectTo: `${protocol}://${host}/api/auth/callback`,
     },
   });
 

src/modules/getOrigin.ts

@@ -0,0 +1,11 @@
+"use server";
+
+import { headers } from "next/headers";
+
+export async function getOrigin(): Promise<{ protocol: string; host: string }> {
+  const h = await headers();
+  const host = h.get("host") || "localhost:3000";
+  const protocol = h.get("x-forwarded-proto") ?? "https";
+
+  return { protocol, host };
+}

src/modules/live-run/actions.ts

@@ -11,6 +11,6 @@ export async function getLiveRun(id: string): Promise<LiveRun | null> {
   return Core.getLiveRun(id);
 }
 
-export async function getLiveRunMoves(id: string) {
+export async function getLiveRunViews(id: string): Promise<number | null> {
-  return Core.getLiveRunMoves(id);
+  return Core.getLiveRunViews(id);
 }

src/modules/live-run/core.ts

@@ -4,7 +4,8 @@ import "server-only";
 import { LiveRun, LiveRunMove } from "./types";
 import { SupabaseClient } from "@supabase/supabase-js";
 import { redis } from "@/lib/redis";
-import { randomInt } from "crypto";
+import Redis from "ioredis";
+import { applyMoves } from "@/lib/move";
 
 /** -------------------- Live Runs in Redis -------------------- */
 
@@ -27,6 +28,29 @@ export async function getLiveRun(id: string): Promise<LiveRun | null> {
   return JSON.parse(data) as LiveRun;
 }
 
+/**
+ * Get live run views from Redis
+ */
+export async function getLiveRunViews(id: string): Promise<number | null> {
+  const run = await getLiveRun(id);
+  if (!run) return null;
+  return run.views;
+}
+
+/**
+ * Get live run views from Redis
+ */
+export async function updateLiveRunViews(
+  id: string,
+  increment: number,
+): Promise<number | null> {
+  const run = await getLiveRun(id);
+  if (!run) return null;
+  run.views += increment;
+  await redis.set(`liveRun:${run.id}`, JSON.stringify(run));
+  return run.views;
+}
+
 /**
  * Get all live runs
  */
@@ -41,7 +65,7 @@ export async function getLiveRuns(): Promise<LiveRun[]> {
  * Remove a live run from Redis
  */
 export async function removeLiveRun(id: string) {
-  await redis.del(`liveRun:${id}`, `liveRunMoves:${id}`);
+  await redis.del(`liveRun:${id}`, `liveRunEvent:${id}`);
 }
 
 /** -------------------- Live Run Moves -------------------- */
@@ -49,28 +73,34 @@ export async function removeLiveRun(id: string) {
 /**
  * Add moves to a live run (ephemeral)
  */
-export async function addLiveRunMoves(id: string, moves: LiveRunMove[]) {
+export async function addLiveRunEvent(
-  const runExists = await redis.exists(`liveRun:${id}`);
+  runId: string,
-  if (!runExists) throw new Error("Invalid run");
+  file: string,
+  language: string,
+  moves: LiveRunMove[],
+) {
+  const redis = new Redis();
 
-  const pipeline = redis.pipeline();
+  // Make sure the live run exists
-  moves.forEach((move) => {
+  const runExists = await redis.exists(`liveRun:${runId}`);
-    pipeline.rpush(`liveRunMoves:${id}`, JSON.stringify(move));
+  if (!runExists) throw "Invalid live run";
-  });
 
-  setTimeout(() => {
+  // Publish the new moves to subscribers
-    redis.del(`liveRunMoves:${id}`);
+  await redis.publish(
-  }, 3000);
+    `liveRunEvent:${runId}`,
+    JSON.stringify({ file, language, moves }),
+  );
 
-  await pipeline.exec();
+  // Get the last text (from Redis)
-}
+  const lastText = (await redis.get(`liveRunText:${runId}`)) || "";
 
-/**
+  // Apply the new moves to get the updated text
- * Get live moves for a run
+  const updatedText = applyMoves(lastText, moves);
- */
+
-export async function getLiveRunMoves(id: string): Promise<LiveRunMove[]> {
+  // Store the updated text back in Redis
-  const moves = await redis.lrange(`liveRunMoves:${id}`, 0, -1);
+  await redis.set(`liveRunText:${runId}`, updatedText);
-  return moves.map((m) => JSON.parse(m)) as LiveRunMove[];
+
+  return updatedText;
 }
 
 /** -------------------- Submit Run -------------------- */

src/modules/live-run/types.ts

@@ -18,3 +18,10 @@ export interface LiveRunMove {
     insert: string;
   };
 }
+
+export interface LiveRunEvent {
+  moves: LiveRunMove[];
+  file: string | null;
+  language: string | null;
+  text: string | null;
+}