From 2939ff8bb55fa1fc8fb6920a87811ba71bf3ac0f Mon Sep 17 00:00:00 2001 From: Danijel Martinek Date: Fri, 8 May 2026 21:16:03 +0200 Subject: [PATCH] feat(core-realtime): SocketIORealtimeServer (4 lifecycle gates) --- packages/core-realtime/package.json | 1 + .../src/socket-io-realtime-server.test.ts | 124 ++++++++++++++++++ .../src/socket-io-realtime-server.ts | 101 ++++++++++++++ pnpm-lock.yaml | 39 ++++++ 4 files changed, 265 insertions(+) create mode 100644 packages/core-realtime/src/socket-io-realtime-server.test.ts create mode 100644 packages/core-realtime/src/socket-io-realtime-server.ts diff --git a/packages/core-realtime/package.json b/packages/core-realtime/package.json index 4d01694..9407e4d 100644 --- a/packages/core-realtime/package.json +++ b/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" } diff --git a/packages/core-realtime/src/socket-io-realtime-server.test.ts b/packages/core-realtime/src/socket-io-realtime-server.test.ts new file mode 100644 index 0000000..8cf2e90 --- /dev/null +++ b/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((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((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((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((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((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(); + }); +}); diff --git a/packages/core-realtime/src/socket-io-realtime-server.ts b/packages/core-realtime/src/socket-io-realtime-server.ts new file mode 100644 index 0000000..5fcedd7 --- /dev/null +++ b/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 { + const out: Record = {}; + 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 { + 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, + }); + 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 } | 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 { + this.io.close(); + } +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 999c0e2..3e7e0bc 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -580,6 +580,9 @@ importers: '@types/node': specifier: ^22.0.0 version: 22.19.17 + socket.io-client: + specifier: ^4.7.0 + version: 4.8.3 typescript: specifier: ^5.8.0 version: 5.9.3 @@ -4790,6 +4793,9 @@ packages: end-of-stream@1.4.5: resolution: {integrity: sha512-ooEGc6HP26xXq/N+GCGOT0JKCLDGrq2bQUZrQ7gyrJiZANJ/8YDTxTpQBXGMn+WbIQXNVpyWymm7KYVICQnyOg==} + engine.io-client@6.6.4: + resolution: {integrity: sha512-+kjUJnZGwzewFDw951CDWcwj35vMNf2fcj7xQWOctq1F2i1jkDdVvdFG9kM/BEChymCH36KgjnW0NsL58JYRxw==} + engine.io-parser@5.2.3: resolution: {integrity: sha512-HqD3yTBfnBxIrbnM1DoD6Pcq8NECnh8d4As1Qgh0z5Gg3jRRIqijury0CL3ghu/edArpUYiYqQiDUQBIs4np3Q==} engines: {node: '>=10.0.0'} @@ -6910,6 +6916,10 @@ packages: socket.io-adapter@2.5.6: resolution: {integrity: sha512-DkkO/dz7MGln0dHn5bmN3pPy+JmywNICWrJqVWiVOyvXjWQFIv9c2h24JrQLLFJ2aQVQf/Cvl1vblnd4r2apLQ==} + socket.io-client@4.8.3: + resolution: {integrity: sha512-uP0bpjWrjQmUt5DTHq9RuoCBdFJF10cdX9X+a368j/Ft0wmaVgxlrjvK3kjvgCODOMMOz9lcaRzxmso0bTWZ/g==} + engines: {node: '>=10.0.0'} + socket.io-parser@4.2.6: resolution: {integrity: sha512-asJqbVBDsBCJx0pTqw3WfesSY0iRX+2xzWEWzrpcH7L6fLzrhyF8WPI8UaeM4YCuDfpwA/cgsdugMsmtz8EJeg==} engines: {node: '>=10.0.0'} @@ -7614,6 +7624,10 @@ packages: xmlchars@2.2.0: resolution: {integrity: sha512-JZnDKK8B0RCDw84FNdDAIpZK+JuJw+s7Lz8nksI7SIuU3UXJJslUthsi+uWBUYOwPFwW7W7PRLRfUKpxjtjFCw==} + xmlhttprequest-ssl@2.1.2: + resolution: {integrity: sha512-TEU+nJVUUnA4CYJFLvK5X9AOeH4KvDvhIfm0vV1GaQRtchnG0hgK5p8hw/xjv8cunWYCsiPCSDzObPyhEwq3KQ==} + engines: {node: '>=0.4.0'} + xss@1.0.15: resolution: {integrity: sha512-FVdlVVC67WOIPvfOwhoMETV72f6GbW7aOabBC3WxN/oUdoEMDyLz4OgRv5/gck2ZeNqEQu+Tb0kloovXOfpYVg==} engines: {node: '>= 0.10.0'} @@ -11828,6 +11842,18 @@ snapshots: dependencies: once: 1.4.0 + engine.io-client@6.6.4: + dependencies: + '@socket.io/component-emitter': 3.1.2 + debug: 4.4.3 + engine.io-parser: 5.2.3 + ws: 8.18.3 + xmlhttprequest-ssl: 2.1.2 + transitivePeerDependencies: + - bufferutil + - supports-color + - utf-8-validate + engine.io-parser@5.2.3: {} engine.io@6.6.7: @@ -14499,6 +14525,17 @@ snapshots: - supports-color - utf-8-validate + socket.io-client@4.8.3: + dependencies: + '@socket.io/component-emitter': 3.1.2 + debug: 4.4.3 + engine.io-client: 6.6.4 + socket.io-parser: 4.2.6 + transitivePeerDependencies: + - bufferutil + - supports-color + - utf-8-validate + socket.io-parser@4.2.6: dependencies: '@socket.io/component-emitter': 3.1.2 @@ -15282,6 +15319,8 @@ snapshots: xmlchars@2.2.0: {} + xmlhttprequest-ssl@2.1.2: {} + xss@1.0.15: dependencies: commander: 2.20.3