From 0a3b13c147e476589d8d361efe96c776ec8ff2d9 Mon Sep 17 00:00:00 2001 From: Maaz Khokhar Date: Fri, 17 Jul 2026 23:34:03 -0500 Subject: [PATCH] added realtime stuff (NOT TESTED) --- apps/server/lib/db/schema/index.ts | 13 +++- apps/server/src/index.ts | 2 + apps/server/src/routes/realtime/classes.ts | 79 ++++++++++++++++++++ apps/server/src/routes/realtime/index.ts | 80 +++++++++++++++++++- apps/server/src/routes/realtime/types.ts | 87 +++++++++++++++++----- apps/server/tsconfig.json | 3 +- bruno/connect.yml | 11 +++ 7 files changed, 251 insertions(+), 24 deletions(-) create mode 100644 apps/server/src/routes/realtime/classes.ts create mode 100644 bruno/connect.yml diff --git a/apps/server/lib/db/schema/index.ts b/apps/server/lib/db/schema/index.ts index b5f0fae..76341c1 100644 --- a/apps/server/lib/db/schema/index.ts +++ b/apps/server/lib/db/schema/index.ts @@ -1 +1,12 @@ -export * from "./auth"; \ No newline at end of file +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), +}); diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index b0e175f..8f4299c 100644 --- a/apps/server/src/index.ts +++ b/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; diff --git a/apps/server/src/routes/realtime/classes.ts b/apps/server/src/routes/realtime/classes.ts new file mode 100644 index 0000000..50a671e --- /dev/null +++ b/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)); + }); + } + }, + }; +} diff --git a/apps/server/src/routes/realtime/index.ts b/apps/server/src/routes/realtime/index.ts index d0802ef..2c72a2b 100644 --- a/apps/server/src/routes/realtime/index.ts +++ b/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); }, }); diff --git a/apps/server/src/routes/realtime/types.ts b/apps/server/src/routes/realtime/types.ts index c3f3733..9b0d461 100644 --- a/apps/server/src/routes/realtime/types.ts +++ b/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 }; diff --git a/apps/server/tsconfig.json b/apps/server/tsconfig.json index 52a5b9f..de30112 100644 --- a/apps/server/tsconfig.json +++ b/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. */ } -} \ No newline at end of file +} diff --git a/bruno/connect.yml b/bruno/connect.yml new file mode 100644 index 0000000..4631bc3 --- /dev/null +++ b/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