added realtime stuff (NOT TESTED)

apps/server/lib/db/schema/index.ts

@@ -1 +1,12 @@
-export * from "./auth";
+import { integer, pgTable, text, serial, boolean } from "drizzle-orm/pg-core";
+
+export * from "./auth";
+
+export const message = pgTable("message", {
+  id: serial("id").primaryKey(),
+  roomId: text("room_id").notNull(),
+  username: text("username").notNull(),
+  content: text("content").notNull(),
+  timestamp: integer("timestamp").notNull(),
+  deleted: boolean("deleted").notNull().default(false),
+});

apps/server/src/index.ts

@@ -1,11 +1,13 @@
 import { auth } from "../lib/auth";
 import { Elysia, t } from "elysia";
+import { realtime } from "./routes/realtime";
 
 const app = new Elysia()
   .mount("/auth", auth.handler)
   .get("/health", () => "OK", {
     response: t.String(),
   })
+  .use(realtime)
   .listen(3000);
 
 export type App = typeof app;

apps/server/src/routes/realtime/classes.ts

@@ -0,0 +1,79 @@
+import { ElysiaWS } from "elysia/ws";
+import { ClientList, ServerMessage } from "./types";
+
+export class Clients {
+  private clients: ClientList;
+
+  constructor() {
+    this.clients = new Map();
+  }
+
+  userInRoom(username: string): boolean {
+    this.clients.forEach((room) => {
+      if (room.some((client) => client.username === username)) {
+        return true;
+      }
+    });
+
+    return false;
+  }
+
+  get clientList(): ClientList {
+    return this.clients;
+  }
+
+  addClient(roomId: string, username: string, ws: ElysiaWS) {
+    if (!this.clients.has(roomId)) {
+      this.clients.set(roomId, []);
+    }
+
+    const room = this.clients.get(roomId);
+    if (room) {
+      room.push({ username, ws });
+    }
+  }
+
+  leaveRoom(roomId: string, username: string) {
+    const room = this.clients.get(roomId);
+    if (room) {
+      this.clients.set(
+        roomId,
+        room.filter((client) => client.username !== username),
+      );
+
+      if (room.length === 0) {
+        this.clients.delete(roomId);
+      }
+    }
+  }
+
+  // Find and remove a client from all rooms based on the WebSocket instance
+  removeClient(ws: ElysiaWS) {
+    this.clients.forEach((room, roomId) => {
+      const updatedRoom = room.filter((client) => client.ws !== ws);
+      if (updatedRoom.length === 0) {
+        this.clients.delete(roomId);
+      } else {
+        this.clients.set(roomId, updatedRoom);
+      }
+    });
+  }
+
+  broadcast = {
+    all: (message: ServerMessage) => {
+      this.clients.forEach((room) => {
+        room.forEach((client) => {
+          client.ws.send(JSON.stringify(message));
+        });
+      });
+    },
+    room: (roomId: string, message: ServerMessage) => {
+      const room = this.clients.get(roomId);
+      if (room) {
+        room.forEach((client) => {
+          client.ws.send(JSON.stringify(message));
+        });
+      }
+    },
+  };
+}

apps/server/src/routes/realtime/index.ts

@@ -1,10 +1,82 @@
 import Elysia from "elysia";
+import type { ElysiaWS } from "elysia/ws";
+import { ClientMessage, ClientMessageSchema } from "./types";
+import { Clients } from "./classes";
+import { db, message as messageTable } from "@/lib/db";
+
+let clients = new Clients();
+
+async function handleMessage(ws: ElysiaWS, message: ClientMessage) {
+  switch (message.type) {
+    case "join": {
+      clients.addClient(message.roomId, message.username, ws);
+      break;
+    }
+    case "leave": {
+      clients.leaveRoom(message.roomId, message.username);
+      break;
+    }
+    case "message": {
+      clients.broadcast.room(message.roomId, {
+        type: "message",
+        username: message.username,
+        m: message.m,
+      });
+
+      // No await because it don't want clients to wait on a stupid db
+      db.insert(messageTable)
+        .values({
+          id: message.m.id,
+          roomId: message.m.roomId,
+          username: message.m.username,
+          content: message.m.content,
+          timestamp: message.m.timestamp,
+          deleted: false,
+        })
+        .onConflictDoNothing()
+        .returning();
+
+      break;
+    }
+    case "sync": {
+      // This time we have to await :(
+      const rows = await db.select().from(messageTable);
+
+      ws.send({
+        type: "sync",
+        roomId: message.roomId,
+        messages: rows.map((row) => ({
+          id: row.id,
+          roomId: row.roomId,
+          username: row.username,
+          content: row.content,
+          timestamp: row.timestamp,
+          deleted: row.deleted,
+        })),
+      });
+
+      break;
+    }
+  }
+}
 
 export const realtime = new Elysia().ws("/realtime", {
-  open: (ws) => {
-    console.log("WebSocket connection opened");
+  open: (ws: ElysiaWS) => {
+    console.log("[+] Client connected:", ws.id);
   },
-  message: (ws, message) => {
-    console.log("Received message:", message);
+
+  body: ClientMessageSchema,
+
+  message: async (ws: ElysiaWS, message: ClientMessage) => {
+    console.log("[*] Received message:", message);
+
+    await handleMessage(ws, message);
+  },
+
+  close: (ws: ElysiaWS) => {
+    // Remove the client from all rooms they are in
+    clients.removeClient(ws);
+
+    console.log("[-] Client disconnected:", ws.id);
   },
 });

apps/server/src/routes/realtime/types.ts

@@ -1,22 +1,73 @@
-export type Message = {
-  id: number;
-  roomId: string;
-  username: string;
-  content: string;
-  timestamp: number;
-  deleted: boolean;
-};
+import { t } from "elysia";
+import { ElysiaWS } from "elysia/ws";
+
+const MessageSchema = t.Object({
+  id: t.Number(),
+  roomId: t.String(),
+  username: t.String(),
+  content: t.String(),
+  timestamp: t.Number(),
+  deleted: t.Boolean(),
+});
 
 // Client -> Server
-export type ClientMessage =
-  | { type: "join"; roomId: string }
-  | { type: "leave"; roomId: string }
-  | { type: "message"; roomId: string; m: Message }
-  | { type: "sync"; roomId: string };
+const ClientMessageSchema = t.Union([
+  t.Object({
+    type: t.Literal("join"),
+    username: t.String(),
+    roomId: t.String(),
+  }),
+  t.Object({
+    type: t.Literal("leave"),
+    username: t.String(),
+    roomId: t.String(),
+  }),
+  t.Object({
+    type: t.Literal("message"),
+    username: t.String(),
+    roomId: t.String(),
+    m: MessageSchema,
+  }),
+  t.Object({
+    type: t.Literal("sync"),
+    roomId: t.String(),
+    // No need for username as we can just directly send the sync through the WS
+  }),
+]);
 
 // Server -> Client(s)
-export type ServerMessage =
-  | { type: "joined"; username: string }
-  | { type: "left"; username: string }
-  | { type: "message"; username: string; m: Message }
-  | { type: "sync"; roomId: string; messages: Message[] };
+const ServerMessageSchema = t.Union([
+  t.Object({
+    type: t.Literal("joined"),
+    username: t.String(),
+  }),
+  t.Object({
+    type: t.Literal("left"),
+    username: t.String(),
+  }),
+  t.Object({
+    type: t.Literal("message"),
+    username: t.String(),
+    m: MessageSchema,
+  }),
+  t.Object({
+    type: t.Literal("sync"),
+    roomId: t.String(),
+    messages: t.Array(MessageSchema),
+  }),
+]);
+
+type ClientList = Map<
+  string, // Room ID
+  Array<{
+    username: string; // Username of the client
+    ws: ElysiaWS; // WebSocket connection of the client
+  }>
+>;
+
+type ServerMessage = typeof ServerMessageSchema.static;
+type Message = typeof MessageSchema.static;
+type ClientMessage = typeof ClientMessageSchema.static;
+
+export { ServerMessageSchema, MessageSchema, ClientMessageSchema };
+export type { ServerMessage, Message, ClientMessage, ClientList };

apps/server/tsconfig.json

@@ -23,6 +23,7 @@
     // "moduleDetection": "auto",                        /* Control what method is used to detect module-format JS files. */
     /* Modules */
     "module": "ES2022", /* Specify what module code is generated. */
+    "moduleResolution": "bundler", /* Use package.json exports for Bun/ESM. */
     "rootDir": "./", /* Specify the root folder within your source files. */
     "paths": {
       "@/*": [
@@ -96,4 +97,4 @@
     // "skipDefaultLibCheck": true,                      /* Skip type checking .d.ts files that are included with TypeScript. */
     "skipLibCheck": true /* Skip type checking all .d.ts files. */
   }
-}
+}

bruno/connect.yml

@@ -0,0 +1,11 @@
+info:
+  name: connect
+  type: websocket
+  seq: 2
+
+websocket:
+  url: ws://localhost:3000/realtime
+  message:
+    type: json
+    data: "{}"
+  auth: inherit