feat(core-realtime): SocketIORealtimeServer (4 lifecycle gates)

packages/core-realtime/package.json

@@ -27,6 +27,7 @@
     "@repo/core-testing": "workspace:*",
     "@repo/core-typescript": "workspace:*",
     "@types/node": "^22.0.0",
+    "socket.io-client": "^4.7.0",
     "typescript": "^5.8.0",
     "vitest": "^3.0.0"
   }

packages/core-realtime/src/socket-io-realtime-server.test.ts

@@ -0,0 +1,124 @@
+import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import { z } from "zod";
+import { createServer, type Server as HttpServer } from "node:http";
+import { Server as IOServer } from "socket.io";
+import { io as ioClient, type Socket as ClientSocket } from "socket.io-client";
+import type { AddressInfo } from "node:net";
+import { SocketIORealtimeServer } from "@/socket-io-realtime-server";
+import { RealtimeHandlerRegistry } from "@/realtime-handler-registry";
+import { defineRealtimeChannel } from "@/realtime-channel";
+
+const pingChannel = defineRealtimeChannel(
+  "test.ping",
+  z.object({ at: z.string() }).strict(),
+  { scope: "authenticated" },
+);
+
+describe("SocketIORealtimeServer", () => {
+  let httpServer: HttpServer;
+  let io: IOServer;
+  let server: SocketIORealtimeServer;
+  let port: number;
+
+  beforeEach(async () => {
+    httpServer = createServer();
+    io = new IOServer(httpServer);
+    const registry = new RealtimeHandlerRegistry();
+
+    let received: { input: unknown; ctx: unknown } | null = null;
+    registry.register({
+      descriptor: pingChannel,
+      handler: async (input, ctx) => {
+        received = { input, ctx };
+      },
+    });
+
+    server = new SocketIORealtimeServer({
+      httpServer,
+      io,
+      authenticator: {
+        authenticate: async ({ cookies }) => {
+          if (cookies.session === "valid") return { userId: "u1", roles: [] };
+          return null;
+        },
+      },
+      registry,
+    });
+    await server.start();
+
+    await new Promise<void>((resolve) => httpServer.listen(0, resolve));
+    port = (httpServer.address() as AddressInfo).port;
+
+    // expose received via a closure for the assertions
+    (server as unknown as { received: typeof received }).received = received;
+  });
+
+  afterEach(async () => {
+    await server.stop();
+    httpServer.close();
+  });
+
+  it("rejects subscribe to authenticated channel from anonymous socket", async () => {
+    const client = ioClient(`http://localhost:${port}`);
+    await new Promise<void>((r) => client.on("connect", () => r()));
+
+    const ack = await new Promise<{ ok: boolean; error?: string }>((r) =>
+      client.emit("subscribe", "test.ping", r),
+    );
+
+    expect(ack.ok).toBe(false);
+    expect(ack.error).toBe("forbidden");
+    client.disconnect();
+  });
+
+  it("allows subscribe + invokes handler with ctx for authenticated socket", async () => {
+    const client = ioClient(`http://localhost:${port}`, {
+      extraHeaders: { Cookie: "session=valid" },
+    });
+    await new Promise<void>((r) => client.on("connect", () => r()));
+
+    const subAck = await new Promise<{ ok: boolean }>((r) =>
+      client.emit("subscribe", "test.ping", r),
+    );
+    expect(subAck.ok).toBe(true);
+
+    const sentAt = new Date().toISOString();
+    const ack = await new Promise<{ ok: boolean }>((r) =>
+      client.emit("test.ping", { at: sentAt }, r),
+    );
+
+    expect(ack.ok).toBe(true);
+    client.disconnect();
+  });
+
+  it("rejects unknown channel subscribe", async () => {
+    const client = ioClient(`http://localhost:${port}`);
+    await new Promise<void>((r) => client.on("connect", () => r()));
+
+    const ack = await new Promise<{ ok: boolean; error?: string }>((r) =>
+      client.emit("subscribe", "does.not.exist", r),
+    );
+
+    expect(ack.ok).toBe(false);
+    expect(ack.error).toBe("unknown_channel");
+    client.disconnect();
+  });
+
+  it("rejects malformed inbound input", async () => {
+    const client = ioClient(`http://localhost:${port}`, {
+      extraHeaders: { Cookie: "session=valid" },
+    });
+    await new Promise<void>((r) => client.on("connect", () => r()));
+    await new Promise<{ ok: boolean }>((r) =>
+      client.emit("subscribe", "test.ping", r),
+    );
+
+    const ack = await new Promise<{ ok: boolean; error?: string }>((r) =>
+      client.emit("test.ping", { at: 123 } as never, r),
+    );
+
+    expect(ack.ok).toBe(false);
+    expect(ack.error).toBe("invalid_input");
+    client.disconnect();
+  });
+});

packages/core-realtime/src/socket-io-realtime-server.ts

@@ -0,0 +1,101 @@
+import type { Server as HttpServer } from "node:http";
+import type { Server as IOServer, Socket } from "socket.io";
+import { authorize } from "./authorize";
+import { matchChannelTemplate } from "./channel-template";
+import type { IRealtimeServer, IRealtimeServerOptions } from "./realtime-server.interface";
+
+function parseCookies(header: string): Record<string, string> {
+  const out: Record<string, string> = {};
+  for (const part of header.split(";")) {
+    const [k, ...rest] = part.trim().split("=");
+    if (k) out[k] = decodeURIComponent(rest.join("="));
+  }
+  return out;
+}
+
+export class SocketIORealtimeServer implements IRealtimeServer {
+  private readonly httpServer: HttpServer;
+  private readonly io: IOServer;
+  private readonly opts: IRealtimeServerOptions;
+
+  constructor(opts: IRealtimeServerOptions) {
+    this.opts = opts;
+    this.httpServer = opts.httpServer;
+    this.io = opts.io;
+  }
+
+  async start(): Promise<void> {
+    const { authenticator, registry } = this.opts;
+
+    // Gate 1: connect — read cookie, authenticate, attach user.
+    this.io.use(async (socket, next) => {
+      const cookies = parseCookies(socket.handshake.headers.cookie ?? "");
+      socket.data.user = await authenticator.authenticate({
+        cookies,
+        headers: socket.handshake.headers as Record<string, string>,
+      });
+      next();
+    });
+
+    this.io.on("connection", (socket: Socket) => {
+      // Gate 2: subscribe.
+      socket.on("subscribe", async (requestedName: string, ack?: (r: unknown) => void) => {
+        // Find a registered descriptor whose name (or template) matches requestedName.
+        let matched: { descriptor: { name: string; scope: unknown }; params: Record<string, string> } | null = null;
+        for (const entry of registry.list()) {
+          const m = matchChannelTemplate(entry.descriptor.name, requestedName);
+          if (m) {
+            matched = { descriptor: entry.descriptor, params: m.params };
+            break;
+          }
+        }
+        if (!matched) {
+          ack?.({ ok: false, error: "unknown_channel" });
+          return;
+        }
+
+        const allowed = await authorize(
+          matched.descriptor as never,
+          matched.params,
+          socket.data.user ?? null,
+        );
+        if (!allowed) {
+          ack?.({ ok: false, error: "forbidden" });
+          return;
+        }
+
+        socket.join(`ch:${requestedName}`);
+        ack?.({ ok: true });
+      });
+
+      // Gate 3: inbound — one listener per registered channel.
+      for (const entry of registry.list()) {
+        socket.on(entry.descriptor.name, async (payload: unknown, ack?: (r: unknown) => void) => {
+          const parsed = entry.descriptor.schema.safeParse(payload);
+          if (!parsed.success) {
+            ack?.({ ok: false, error: "invalid_input" });
+            return;
+          }
+          const allowed = await authorize(entry.descriptor, {}, socket.data.user ?? null);
+          if (!allowed) {
+            ack?.({ ok: false, error: "forbidden" });
+            return;
+          }
+          try {
+            await entry.handler(parsed.data, {
+              userId: socket.data.user?.userId ?? null,
+              roles: socket.data.user?.roles ?? [],
+            });
+            ack?.({ ok: true });
+          } catch {
+            ack?.({ ok: false, error: "handler_error" });
+          }
+        });
+      }
+    });
+  }
+
+  async stop(): Promise<void> {
+    this.io.close();
+  }
+}