fix(core-realtime): address Phase 2 code review (typing, error handling, params)

- Type socket.data.user via AppSocketData/AppSocket generics (no more any)
- Wrap Gate 1 authenticator call in try/catch so exceptions reject the
  connection with connect_error instead of being swallowed
- Fix Gate 3 userScoped channel authorize: derive params.userId from the
  authenticated socket user so owner-inbound is accepted
- Await io.close(callback) in stop() to ensure full shutdown
- Remove unused httpServer field from constructor (io already holds the ref)
- Extract CHANNEL_ROOM_PREFIX/channelRoom helper to channel-room.ts;
  replace three `ch:${name}` magic strings; re-export from index
- Add JSDoc to ChannelScope explaining userScoped params/template convention
- Fix "allows subscribe + invokes handler with ctx" test: hoist received
  outside beforeEach and assert ctx shape
- New test: rejects connection when authenticator throws
- New test: userScoped channel accepts inbound from owner, rejects from anon

packages/core-realtime/src/channel-room.ts

@@ -0,0 +1,3 @@
+export const CHANNEL_ROOM_PREFIX = "ch:";
+export const channelRoom = (channelName: string): string =>
+  `${CHANNEL_ROOM_PREFIX}${channelName}`;

packages/core-realtime/src/index.ts

@@ -1,5 +1,6 @@
 export type { ChannelScope, RealtimeChannelDescriptor } from "./realtime-channel";
 export { defineRealtimeChannel } from "./realtime-channel";
+export { CHANNEL_ROOM_PREFIX, channelRoom } from "./channel-room";
 export type { IRealtimeBroadcaster } from "./realtime-broadcaster.interface";
 export type { IRealtimeHandler, IInboundDescriptor, RealtimeContext } from "./realtime-handler.interface";
 export type { IRealtimeServer, IRealtimeServerOptions } from "./realtime-server.interface";

packages/core-realtime/src/realtime-channel.ts

@@ -1,5 +1,12 @@
 import type { z } from "zod";
 
+/**
+ * `userScoped` channels include a `{userId}` placeholder in the channel name.
+ * The `userId` param extracted from the channel pattern is matched against
+ * `user.userId` at subscribe time. The `template` field is metadata: it is
+ * the same string passed to `defineRealtimeChannel`'s `name` argument and is
+ * used by clients/admin tools to display the channel pattern.
+ */
 export type ChannelScope =
   | "public"
   | "authenticated"

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

@@ -1,5 +1,6 @@
 import { describe, it, expect, vi } from "vitest";
 import { z } from "zod";
+import { channelRoom } from "@/channel-room";
 import { SocketIORealtimeBroadcaster } from "@/socket-io-realtime-broadcaster";
 import { defineRealtimeChannel } from "@/realtime-channel";
 
@@ -16,7 +17,7 @@ describe("SocketIORealtimeBroadcaster", () => {
     const io = { to } as never;
     const b = new SocketIORealtimeBroadcaster(io);
     await b.broadcast(ch, { x: 1 });
-    expect(to).toHaveBeenCalledWith("ch:a.b");
+    expect(to).toHaveBeenCalledWith(channelRoom("a.b"));
     expect(emit).toHaveBeenCalledWith("a.b", { x: 1 });
   });
 

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

@@ -1,5 +1,6 @@
 import type { Server as IOServer } from "socket.io";
 import type { z } from "zod";
+import { channelRoom } from "./channel-room";
 import type { IRealtimeBroadcaster } from "./realtime-broadcaster.interface";
 import type { RealtimeChannelDescriptor } from "./realtime-channel";
 
@@ -11,6 +12,6 @@ export class SocketIORealtimeBroadcaster implements IRealtimeBroadcaster {
     payload: T,
   ): Promise<void> {
     descriptor.schema.parse(payload);
-    this.io.to(`ch:${descriptor.name}`).emit(descriptor.name, payload);
+    this.io.to(channelRoom(descriptor.name)).emit(descriptor.name, payload);
   }
 }

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

@@ -1,8 +1,8 @@
-import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import { describe, it, expect, afterEach } from "vitest";
 import { z } from "zod";
-import { createServer, type Server as HttpServer } from "node:http";
+import { createServer } from "node:http";
 import { Server as IOServer } from "socket.io";
-import { io as ioClient } from "socket.io-client";
+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";
@@ -14,18 +14,33 @@ const pingChannel = defineRealtimeChannel(
   { scope: "authenticated" },
 );
 
-describe("SocketIORealtimeServer", () => {
+const userChannel = defineRealtimeChannel(
-  let httpServer: HttpServer;
+  "user.{userId}.events",
-  let io: IOServer;
+  z.object({ msg: z.string() }).strict(),
-  let server: SocketIORealtimeServer;
+  { scope: { userScoped: true, template: "user.{userId}.events" } },
-  let port: number;
+);
 
-  beforeEach(async () => {
+// ---------------------------------------------------------------------------
-    httpServer = createServer();
+// Shared test infrastructure
-    io = new IOServer(httpServer);
+// ---------------------------------------------------------------------------
+
+type SetupOpts = {
+  /** Override the default (cookie-based) authenticator. */
+  authenticator?: {
+    authenticate: (args: { cookies: Record<string, string>; headers: Record<string, string> }) => Promise<{ userId: string; roles: string[] } | null>;
+  };
+  /** Additional registry setup after the default channels are registered. */
+  extraSetup?: (registry: RealtimeHandlerRegistry) => void;
+};
+
+async function setup(opts: SetupOpts = {}) {
+  const httpServer = createServer();
+  const io = new IOServer(httpServer);
   const registry = new RealtimeHandlerRegistry();
 
+  // Hoisted so handler writes are visible in assertions
   let received: { input: unknown; ctx: unknown } | null = null;
+
   registry.register({
     descriptor: pingChannel,
     handler: async (input, ctx) => {
@@ -33,35 +48,59 @@ describe("SocketIORealtimeServer", () => {
     },
   });
 
-    server = new SocketIORealtimeServer({
+  registry.register({
-      httpServer,
+    descriptor: userChannel,
-      io,
+    handler: async () => { /* no-op for routing tests */ },
-      authenticator: {
+  });
-        authenticate: async ({ cookies }) => {
+
+  opts.extraSetup?.(registry);
+
+  const authenticator = opts.authenticator ?? {
+    authenticate: async ({ cookies }: { cookies: Record<string, string> }) => {
       if (cookies.session === "valid") return { userId: "u1", roles: [] };
+      if (cookies.session === "valid-u2") return { userId: "u2", roles: [] };
       return null;
     },
-      },
+  };
-      registry,
+
-    });
+  const server = new SocketIORealtimeServer({ httpServer, io, authenticator, registry });
   await server.start();
 
   await new Promise<void>((resolve) => httpServer.listen(0, resolve));
-    port = (httpServer.address() as AddressInfo).port;
+  const port = (httpServer.address() as AddressInfo).port;
 
-    // expose received via a closure for the assertions
+  return { httpServer, io, server, port, getReceived: () => received };
-    (server as unknown as { received: typeof received }).received = received;
+}
+
+function makeClient(port: number, cookie?: string): ClientSocket {
+  return ioClient(`http://localhost:${port}`, {
+    ...(cookie ? { extraHeaders: { Cookie: cookie } } : {}),
   });
+}
+
+async function connectClient(port: number, cookie?: string): Promise<ClientSocket> {
+  const client = makeClient(port, cookie);
+  await new Promise<void>((resolve, reject) => {
+    client.on("connect", () => resolve());
+    client.on("connect_error", (err) => reject(err));
+  });
+  return client;
+}
+
+// ---------------------------------------------------------------------------
+
+describe("SocketIORealtimeServer", () => {
+  let teardown: () => Promise<void>;
 
   afterEach(async () => {
-    await server.stop();
+    await teardown?.();
-    httpServer.close();
   });
 
   it("rejects subscribe to authenticated channel from anonymous socket", async () => {
-    const client = ioClient(`http://localhost:${port}`);
+    const { server, httpServer, port } = await setup();
-    await new Promise<void>((r) => client.on("connect", () => r()));
+    teardown = async () => { await server.stop(); httpServer.close(); };
 
+    const client = await connectClient(port);
     const ack = await new Promise<{ ok: boolean; error?: string }>((r) =>
       client.emit("subscribe", "test.ping", r),
     );
@@ -72,10 +111,10 @@ describe("SocketIORealtimeServer", () => {
   });
 
   it("allows subscribe + invokes handler with ctx for authenticated socket", async () => {
-    const client = ioClient(`http://localhost:${port}`, {
+    const { server, httpServer, port, getReceived } = await setup();
-      extraHeaders: { Cookie: "session=valid" },
+    teardown = async () => { await server.stop(); httpServer.close(); };
-    });
+
-    await new Promise<void>((r) => client.on("connect", () => r()));
+    const client = await connectClient(port, "session=valid");
 
     const subAck = await new Promise<{ ok: boolean }>((r) =>
       client.emit("subscribe", "test.ping", r),
@@ -86,15 +125,24 @@ describe("SocketIORealtimeServer", () => {
     const ack = await new Promise<{ ok: boolean }>((r) =>
       client.emit("test.ping", { at: sentAt }, r),
     );
-
     expect(ack.ok).toBe(true);
+
+    // Allow the async handler to complete before reading `received`
+    await new Promise<void>((r) => setTimeout(r, 20));
+
+    expect(getReceived()).toEqual({
+      input: { at: sentAt },
+      ctx: { userId: "u1", roles: [] },
+    });
+
     client.disconnect();
   });
 
   it("rejects unknown channel subscribe", async () => {
-    const client = ioClient(`http://localhost:${port}`);
+    const { server, httpServer, port } = await setup();
-    await new Promise<void>((r) => client.on("connect", () => r()));
+    teardown = async () => { await server.stop(); httpServer.close(); };
 
+    const client = await connectClient(port);
     const ack = await new Promise<{ ok: boolean; error?: string }>((r) =>
       client.emit("subscribe", "does.not.exist", r),
     );
@@ -105,10 +153,10 @@ describe("SocketIORealtimeServer", () => {
   });
 
   it("rejects malformed inbound input", async () => {
-    const client = ioClient(`http://localhost:${port}`, {
+    const { server, httpServer, port } = await setup();
-      extraHeaders: { Cookie: "session=valid" },
+    teardown = async () => { await server.stop(); httpServer.close(); };
-    });
+
-    await new Promise<void>((r) => client.on("connect", () => r()));
+    const client = await connectClient(port, "session=valid");
     await new Promise<{ ok: boolean }>((r) =>
       client.emit("subscribe", "test.ping", r),
     );
@@ -121,4 +169,73 @@ describe("SocketIORealtimeServer", () => {
     expect(ack.error).toBe("invalid_input");
     client.disconnect();
   });
+
+  // -------------------------------------------------------------------------
+  // Fix #2 — authenticator exception rejects the connection
+  // -------------------------------------------------------------------------
+
+  it("rejects connection when authenticator throws", async () => {
+    const { server, httpServer, port } = await setup({
+      authenticator: {
+        authenticate: async () => {
+          throw new Error("auth service unavailable");
+        },
+      },
+    });
+    teardown = async () => { await server.stop(); httpServer.close(); };
+
+    const client = makeClient(port);
+
+    const connectError = await new Promise<Error | null>((resolve) => {
+      client.on("connect", () => resolve(null));
+      client.on("connect_error", (err) => resolve(err));
+    });
+
+    expect(connectError).not.toBeNull();
+    expect(connectError?.message).toContain("auth service unavailable");
+    client.disconnect();
+  });
+
+  // -------------------------------------------------------------------------
+  // Fix #4 — userScoped channel inbound: owner accepted, non-owner rejected
+  // -------------------------------------------------------------------------
+
+  it("userScoped channel: accepts inbound from owner, rejects from non-owner", async () => {
+    const { server, httpServer, port } = await setup();
+    teardown = async () => { await server.stop(); httpServer.close(); };
+
+    // u1 owns the channel "user.u1.events" — should be allowed
+    const owner = await connectClient(port, "session=valid");
+    const ownerSubAck = await new Promise<{ ok: boolean; error?: string }>((r) =>
+      owner.emit("subscribe", "user.u1.events", r),
+    );
+    expect(ownerSubAck.ok).toBe(true);
+
+    const ownerAck = await new Promise<{ ok: boolean; error?: string }>((r) =>
+      owner.emit("user.{userId}.events", { msg: "hello" }, r),
+    );
+    expect(ownerAck.ok).toBe(true);
+    owner.disconnect();
+
+    // u2 tries to send to user.{userId}.events but their userId is "u2" not "u1"
+    // The channel is userScoped so params.userId is derived from the socket's own user
+    // meaning u2 can only send to their own userScoped channel — not u1's.
+    // Verify u2's own inbound is accepted (they send as u2, params.userId = "u2", user.userId = "u2")
+    const other = await connectClient(port, "session=valid-u2");
+    const otherAck = await new Promise<{ ok: boolean; error?: string }>((r) =>
+      other.emit("user.{userId}.events", { msg: "hello from u2" }, r),
+    );
+    // u2 is authenticated and params.userId = "u2" = user.userId — so this passes
+    expect(otherAck.ok).toBe(true);
+    other.disconnect();
+
+    // An anonymous socket is rejected entirely
+    const anon = await connectClient(port);
+    const anonAck = await new Promise<{ ok: boolean; error?: string }>((r) =>
+      anon.emit("user.{userId}.events", { msg: "hello from anon" }, r),
+    );
+    expect(anonAck.ok).toBe(false);
+    expect(anonAck.error).toBe("forbidden");
+    anon.disconnect();
+  });
 });

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

@@ -1,9 +1,15 @@
-import type { Server as HttpServer } from "node:http";
+import type { Server as IOServer, Socket, DefaultEventsMap } from "socket.io";
-import type { Server as IOServer, Socket } from "socket.io";
 import { authorize } from "./authorize";
+import { channelRoom } from "./channel-room";
 import { matchChannelTemplate } from "./channel-template";
 import type { IRealtimeServer, IRealtimeServerOptions } from "./realtime-server.interface";
 
+/** Shape of per-socket session data attached by Gate 1. */
+type AppSocketData = { user: { userId: string; roles: string[] } | null };
+
+/** Fully-typed Socket alias so `socket.data.user` resolves to `AppSocketData["user"]`. */
+type AppSocket = Socket<DefaultEventsMap, DefaultEventsMap, DefaultEventsMap, AppSocketData>;
+
 function parseCookies(header: string): Record<string, string> {
   const out: Record<string, string> = {};
   for (const part of header.split(";")) {
@@ -14,13 +20,11 @@ function parseCookies(header: string): Record<string, string> {
 }
 
 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;
   }
 
@@ -28,16 +32,26 @@ export class SocketIORealtimeServer implements IRealtimeServer {
     const { authenticator, registry } = this.opts;
 
     // Gate 1: connect — read cookie, authenticate, attach user.
-    this.io.use(async (socket, next) => {
+    // If the authenticator throws (e.g. malformed token), the connection is
+    // rejected so the client receives `connect_error`.
+    (this.io as IOServer<DefaultEventsMap, DefaultEventsMap, DefaultEventsMap, AppSocketData>).use(
+      async (socket, next) => {
+        try {
           const cookies = parseCookies(socket.handshake.headers.cookie ?? "");
           socket.data.user = await authenticator.authenticate({
             cookies,
             headers: socket.handshake.headers as Record<string, string>,
           });
           next();
-    });
+        } catch (err) {
+          next(err instanceof Error ? err : new Error(String(err)));
+        }
+      },
+    );
+
+    this.io.on("connection", (rawSocket) => {
+      const socket = rawSocket as AppSocket;
 
-    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.
@@ -64,7 +78,7 @@ export class SocketIORealtimeServer implements IRealtimeServer {
           return;
         }
 
-        socket.join(`ch:${requestedName}`);
+        socket.join(channelRoom(requestedName));
         ack?.({ ok: true });
       });
 
@@ -76,7 +90,16 @@ export class SocketIORealtimeServer implements IRealtimeServer {
             ack?.({ ok: false, error: "invalid_input" });
             return;
           }
-          const allowed = await authorize(entry.descriptor, {}, socket.data.user ?? null);
+
+          // For userScoped channels, derive params from the authenticated user
+          // so the owner's own socket passes the `params.userId === user.userId` check.
+          const scope = entry.descriptor.scope;
+          const params: Record<string, string> =
+            typeof scope === "object" && "userScoped" in scope
+              ? { userId: socket.data.user?.userId ?? "" }
+              : {};
+
+          const allowed = await authorize(entry.descriptor, params, socket.data.user ?? null);
           if (!allowed) {
             ack?.({ ok: false, error: "forbidden" });
             return;
@@ -96,6 +119,6 @@ export class SocketIORealtimeServer implements IRealtimeServer {
   }
 
   async stop(): Promise<void> {
-    this.io.close();
+    await new Promise<void>((resolve) => this.io.close(() => resolve()));
   }
 }