basic websocket and room with messages implementation

Author: brent-broeckx

apps/backend/dev.db

@@ -21,10 +21,13 @@
     "@prisma/client": "^7.4.0",
     "cors": "^2.8.5",
     "dotenv": "^17.3.1",
-    "express": "^5.2.1"
+    "express": "^5.2.1",
+    "ws": "^8.18.3"
   },
   "devDependencies": {
+    "@types/cors": "^2.8.19",
     "@types/express": "^5.0.6",
+    "@types/ws": "^8.18.1",
     "prisma": "^7.4.0",
     "tsx": "^4.20.5"
   }

apps/backend/package.json

@@ -0,0 +1,53 @@
+/*
+  Warnings:
+
+  - Added the required column `projectId` to the `Room` table without a default value. This is not possible if the table is not empty.
+
+*/
+-- CreateTable
+CREATE TABLE "Project" (
+    "id" TEXT NOT NULL PRIMARY KEY,
+    "name" TEXT NOT NULL,
+    "slug" TEXT NOT NULL,
+    "description" TEXT,
+    "createdById" TEXT NOT NULL,
+    "createdAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" DATETIME NOT NULL,
+    CONSTRAINT "Project_createdById_fkey" FOREIGN KEY ("createdById") REFERENCES "User" ("id") ON DELETE CASCADE ON UPDATE CASCADE
+);
+
+-- CreateTable
+CREATE TABLE "ProjectMember" (
+    "id" TEXT NOT NULL PRIMARY KEY,
+    "projectId" TEXT NOT NULL,
+    "userId" TEXT NOT NULL,
+    "role" TEXT NOT NULL DEFAULT 'member',
+    "joinedAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    CONSTRAINT "ProjectMember_projectId_fkey" FOREIGN KEY ("projectId") REFERENCES "Project" ("id") ON DELETE CASCADE ON UPDATE CASCADE,
+    CONSTRAINT "ProjectMember_userId_fkey" FOREIGN KEY ("userId") REFERENCES "User" ("id") ON DELETE CASCADE ON UPDATE CASCADE
+);
+
+-- RedefineTables
+PRAGMA defer_foreign_keys=ON;
+PRAGMA foreign_keys=OFF;
+CREATE TABLE "new_Room" (
+    "id" TEXT NOT NULL PRIMARY KEY,
+    "projectId" TEXT NOT NULL,
+    "name" TEXT NOT NULL,
+    "repositoryId" TEXT,
+    "createdAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" DATETIME NOT NULL,
+    CONSTRAINT "Room_projectId_fkey" FOREIGN KEY ("projectId") REFERENCES "Project" ("id") ON DELETE CASCADE ON UPDATE CASCADE,
+    CONSTRAINT "Room_repositoryId_fkey" FOREIGN KEY ("repositoryId") REFERENCES "Repository" ("id") ON DELETE SET NULL ON UPDATE CASCADE
+);
+INSERT INTO "new_Room" ("createdAt", "id", "name", "repositoryId", "updatedAt") SELECT "createdAt", "id", "name", "repositoryId", "updatedAt" FROM "Room";
+DROP TABLE "Room";
+ALTER TABLE "new_Room" RENAME TO "Room";
+PRAGMA foreign_keys=ON;
+PRAGMA defer_foreign_keys=OFF;
+
+-- CreateIndex
+CREATE UNIQUE INDEX "Project_slug_key" ON "Project"("slug");
+
+-- CreateIndex
+CREATE UNIQUE INDEX "ProjectMember_projectId_userId_key" ON "ProjectMember"("projectId", "userId");

apps/backend/prisma/migrations/20260218185247_add_project_model_and_room_project_relation/migration.sql

@@ -43,6 +43,8 @@ model User {
   createdAt      DateTime     @default(now())
   updatedAt      DateTime     @updatedAt
   roomMembership RoomMember[]
+  projectMemberships ProjectMember[]
+  createdProjects Project[]
   sentMessages   Message[]
   accounts       Account[]
   sessions       Session[]
@@ -86,12 +88,39 @@ model VerificationToken {
   @@unique([identifier, token])
 }
 
+model Project {
+  id          String          @id @default(cuid())
+  name        String
+  slug        String          @unique
+  description String?
+  createdById String
+  createdAt   DateTime        @default(now())
+  updatedAt   DateTime        @updatedAt
+  createdBy   User            @relation(fields: [createdById], references: [id], onDelete: Cascade)
+  members     ProjectMember[]
+  rooms       Room[]
+}
+
+model ProjectMember {
+  id        String   @id @default(cuid())
+  projectId String
+  userId    String
+  role      String   @default("member")
+  joinedAt  DateTime @default(now())
+  project   Project  @relation(fields: [projectId], references: [id], onDelete: Cascade)
+  user      User     @relation(fields: [userId], references: [id], onDelete: Cascade)
+
+  @@unique([projectId, userId])
+}
+
 model Room {
   id             String       @id @default(cuid())
+  projectId      String
   name           String
   repositoryId   String?
   createdAt      DateTime     @default(now())
   updatedAt      DateTime     @updatedAt
+  project        Project      @relation(fields: [projectId], references: [id], onDelete: Cascade)
   roomMembers    RoomMember[]
   messages       Message[]
   repository     Repository?  @relation(fields: [repositoryId], references: [id])

apps/backend/prisma/schema.prisma

@@ -0,0 +1,72 @@
+import type { Request } from "express";
+
+export type AuthenticatedUser = {
+  id: string;
+  name?: string | null;
+  email?: string | null;
+  image?: string | null;
+};
+
+type SessionResponse = {
+  user?: {
+    id?: string;
+    name?: string | null;
+    email?: string | null;
+    image?: string | null;
+  };
+};
+
+function parseAuthenticatedUser(payload: SessionResponse | null): AuthenticatedUser | null {
+  if (!payload?.user?.id) {
+    return null;
+  }
+
+  return {
+    id: payload.user.id,
+    name: payload.user.name,
+    email: payload.user.email,
+    image: payload.user.image
+  };
+}
+
+export async function resolveAuthUserFromSession(params: {
+  origin: string;
+  cookieHeader?: string;
+}): Promise<AuthenticatedUser | null> {
+  if (!params.cookieHeader) {
+    return null;
+  }
+
+  try {
+    const response = await fetch(`${params.origin}/auth/session`, {
+      method: "GET",
+      headers: {
+        cookie: params.cookieHeader,
+        accept: "application/json"
+      }
+    });
+
+    if (!response.ok) {
+      return null;
+    }
+
+    const payload = (await response.json()) as SessionResponse | null;
+
+    return parseAuthenticatedUser(payload);
+  } catch {
+    return null;
+  }
+}
+
+export async function getAuthenticatedUserFromRequest(req: Request): Promise<AuthenticatedUser | null> {
+  const host = req.get("host");
+
+  if (!host) {
+    return null;
+  }
+
+  return resolveAuthUserFromSession({
+    origin: `${req.protocol}://${host}`,
+    cookieHeader: req.headers.cookie
+  });
+}

apps/backend/src/auth/session.ts

@@ -0,0 +1,165 @@
+import type { Server as HttpServer, IncomingMessage } from "node:http";
+
+import WebSocket, { WebSocketServer } from "ws";
+
+import { resolveAuthUserFromSession } from "../auth/session.js";
+
+import { type ChatMessage, isUserRoomMember } from "./service.js";
+
+type ClientContext = {
+  socket: WebSocket;
+  userId: string;
+  roomId?: string;
+};
+
+type IncomingClientMessage = {
+  type?: string;
+  roomId?: string;
+};
+
+export type ChatRealtimeBroadcaster = {
+  broadcastMessage: (message: ChatMessage) => void;
+};
+
+function getRequestOrigin(request: IncomingMessage): string | null {
+  const host = request.headers.host;
+
+  if (!host) {
+    return null;
+  }
+
+  const forwardedProtocol =
+    typeof request.headers["x-forwarded-proto"] === "string"
+      ? request.headers["x-forwarded-proto"]
+      : undefined;
+  const socketWithTlsFlag = request.socket as { encrypted?: boolean };
+  const protocol = forwardedProtocol ?? (socketWithTlsFlag.encrypted ? "https" : "http");
+
+  return `${protocol}://${host}`;
+}
+
+function safeSend(socket: WebSocket, payload: unknown): void {
+  if (socket.readyState !== WebSocket.OPEN) {
+    return;
+  }
+
+  socket.send(JSON.stringify(payload));
+}
+
+export class WsChatRealtimeGateway implements ChatRealtimeBroadcaster {
+  private readonly server: WebSocketServer;
+
+  private readonly clients = new Set<ClientContext>();
+
+  constructor(httpServer: HttpServer) {
+    this.server = new WebSocketServer({
+      server: httpServer,
+      path: "/ws/chat"
+    });
+
+    this.server.on("connection", (socket, request) => {
+      void this.handleConnection(socket, request);
+    });
+  }
+
+  broadcastMessage(message: ChatMessage): void {
+    for (const client of this.clients) {
+      if (client.roomId === message.roomId) {
+        safeSend(client.socket, {
+          type: "message:new",
+          message
+        });
+      }
+    }
+  }
+
+  private async handleConnection(socket: WebSocket, request: IncomingMessage): Promise<void> {
+    const origin = getRequestOrigin(request);
+
+    if (!origin) {
+      socket.close(1008, "Missing origin");
+
+      return;
+    }
+
+    const authUser = await resolveAuthUserFromSession({
+      origin,
+      cookieHeader: request.headers.cookie
+    });
+
+    if (!authUser) {
+      socket.close(1008, "Unauthorized");
+
+      return;
+    }
+
+    const context: ClientContext = {
+      socket,
+      userId: authUser.id
+    };
+
+    this.clients.add(context);
+
+    safeSend(socket, {
+      type: "connected",
+      userId: authUser.id
+    });
+
+    socket.on("message", (rawMessage) => {
+      void this.handleClientMessage(context, rawMessage.toString());
+    });
+
+    socket.on("close", () => {
+      this.clients.delete(context);
+    });
+
+    socket.on("error", () => {
+      this.clients.delete(context);
+    });
+  }
+
+  private async handleClientMessage(client: ClientContext, rawMessage: string): Promise<void> {
+    let payload: IncomingClientMessage;
+
+    try {
+      payload = JSON.parse(rawMessage) as IncomingClientMessage;
+    } catch {
+      safeSend(client.socket, {
+        type: "error",
+        message: "Invalid message payload"
+      });
+
+      return;
+    }
+
+    if (payload.type !== "subscribe" || !payload.roomId) {
+      safeSend(client.socket, {
+        type: "error",
+        message: "Unsupported event"
+      });
+
+      return;
+    }
+
+    const isMember = await isUserRoomMember({
+      userId: client.userId,
+      roomId: payload.roomId
+    });
+
+    if (!isMember) {
+      safeSend(client.socket, {
+        type: "error",
+        message: "Access denied for room"
+      });
+
+      return;
+    }
+
+    client.roomId = payload.roomId;
+
+    safeSend(client.socket, {
+      type: "subscribed",
+      roomId: payload.roomId
+    });
+  }
+}