Using next-ws for better code structure

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

@@ -0,0 +1,35 @@
+import { NextRequest, NextResponse } from "next/server";
+import { createClient } from "@/lib/supabase/server";
+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();
+
+  if (!runId || !moves)
+    return NextResponse.json(
+      { error: "body requires runId and moves" },
+      { status: 400 },
+    );
+
+  const supabase = await createClient();
+
+  const session = await getSession(supabase);
+
+  if (!session)
+    return NextResponse.json({ error: "Invalid session" }, { status: 401 });
+
+  const liveRun = await Core.getLiveRun(runId);
+
+  if (!liveRun)
+    return NextResponse.json(
+      { error: "Run is not live or does not exist" },
+      { status: 401 },
+    );
+
+  console.log(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,47 @@
+import { WebSocketServer, WebSocket } from "ws";
+import Redis from "ioredis";
+import { LiveRunMove } from "@/modules/live-run/types";
+import { NextRequest, NextResponse } from "next/server";
+
+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) {
+  console.log("A client connected");
+
+  const subscriber = new Redis();
+  let runId: string;
+
+  client.once("message", (message) => {
+    runId = message.toString();
+
+    subscriber.subscribe(`liveRunMoves:${runId}`, (err, count) => {
+      if (err) {
+        console.error("Failed to subscribe", err);
+        client.close(1011, "Redis subscription failed");
+      } else {
+        console.log(`Subscribed to ${count} channels`);
+      }
+    });
+
+    subscriber.on("message", (channel, message) => {
+      if (client.readyState === WebSocket.OPEN) {
+        client.send(message);
+      }
+    });
+  });
+
+  client.once("close", async () => {
+    if (runId) await subscriber.unsubscribe(`liveRun:${runId}`);
+    subscriber.quit();
+    console.log("Client disconnected, Redis unsubscribed");
+  });
+
+  client.on("error", (err) => {
+    console.error("WebSocket error:", err);
+  });
+}

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

@@ -1,14 +0,0 @@
-import { WebSocketServer, WebSocket } from "ws";
-
-export function UPGRADE(client: WebSocket, server: WebSocketServer) {
-  console.log("A client connected");
-
-  client.on("message", (message) => {
-    console.log("Received message:", message);
-    client.send(message);
-  });
-
-  client.once("close", () => {
-    console.log("A client disconnected");
-  });
-}

src/modules/live-run/core.ts

@@ -4,6 +4,7 @@ import "server-only";
 import { LiveRun, LiveRunMove } from "./types";
 import { SupabaseClient } from "@supabase/supabase-js";
 import { redis } from "@/lib/redis";
+import Redis from "ioredis";
 
 /** -------------------- Live Runs in Redis -------------------- */
 
@@ -43,6 +44,22 @@ export async function removeLiveRun(id: string) {
   await redis.del(`liveRun:${id}`, `liveRunMoves:${id}`);
 }
 
+/** -------------------- Live Run Moves -------------------- */
+
+/**
+ * Add moves to a live run (ephemeral)
+ */
+export async function addLiveRunMoves(runId: string, moves: LiveRunMove[]) {
+  const runExists = await redis.exists(`liveRun:${runId}`);
+
+  if (!runExists) throw "Invalid live run";
+
+  const publisher = new Redis();
+
+  // Send a message
+  await publisher.publish(`liveRunMoves:${runId}`, JSON.stringify(moves));
+}
+
 /** -------------------- Submit Run -------------------- */
 
 /**