feat(core-realtime): scaffold realtime optional core

Generator-emitted scaffold (pnpm turbo gen core-package realtime) plus
the story-00-precedent coverage repairs (coverage provider devDep,
symbols.ts exclude + tested allowlist mirror) and three minimal tests
covering generator-emitted realtime code the template suite misses.
Squash of 31d85e0 + review-fix cf11b38.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016j8z4VHjedXDTjEDNg7qHK

packages/core-eslint/base.js

@@ -8,6 +8,8 @@ import conformancePlugin from "./plugin.js";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
 // <gen:realtime-rules-imports>
+import noDirectSocketIO from "./rules/no-direct-socket-io.js";
+import noRealtimeHandlerReexport from "./rules/no-realtime-handler-reexport.js";
 
 const __dirname = path.dirname(fileURLToPath(import.meta.url));
 const repoRoot = path.resolve(__dirname, "..", "..");
@@ -82,6 +84,8 @@ export default [
         { type: "tooling", pattern: "packages/core-testing" },
         { type: "core-composition", pattern: "packages/core-api" },
         { type: "core-composition", pattern: "packages/core-cms" },
+        { type: "core", pattern: "packages/core-realtime", mode: "folder" },
+
         { type: "core", pattern: "packages/core-*" },
         { type: "feature", pattern: "packages/!(core-*)" },
       ],
@@ -245,4 +249,19 @@ export default [
   // no-realtime-handler-reexport) are added here when @repo/core-realtime is
   // scaffolded via `pnpm turbo gen core-package realtime`.
   // <gen:realtime-rules>
+  {
+    files: ["**/*.{ts,tsx,mjs,cjs,js}"],
+    plugins: {
+      "repo-rules": {
+        rules: {
+          "no-direct-socket-io": noDirectSocketIO,
+          "no-realtime-handler-reexport": noRealtimeHandlerReexport,
+        },
+      },
+    },
+    rules: {
+      "repo-rules/no-direct-socket-io": "error",
+      "repo-rules/no-realtime-handler-reexport": "error",
+    },
+  },
 ];

packages/core-eslint/rules/no-direct-socket-io.js

@@ -0,0 +1,39 @@
+// packages/core-eslint/rules/no-direct-socket-io.js
+const ALLOWED = [
+  /\/packages\/core-realtime\/src\//,
+  /\/apps\/[^/]+\/server\.ts$/,
+  /\/apps\/[^/]+\/src\/.*\.test\.ts$/,
+];
+
+export default {
+  meta: {
+    type: "problem",
+    docs: {
+      description:
+        "Block direct socket.io imports outside core-realtime + app servers",
+    },
+    messages: {
+      noDirectSocketIO:
+        'Import from "@repo/core-realtime" instead of "socket.io". Direct imports allowed only in packages/core-realtime/src/ and apps/*/server.ts.',
+      noDirectSocketIOClient:
+        'Use the realtime helpers from "@repo/core-realtime" / "@repo/core-testing/instrumentation" instead of "socket.io-client".',
+    },
+    schema: [],
+  },
+  create(context) {
+    const filename = context.filename ?? context.getFilename();
+    const allowed = ALLOWED.some((re) => re.test(filename));
+    if (allowed) return {};
+
+    return {
+      ImportDeclaration(node) {
+        const source = node.source.value;
+        if (source === "socket.io") {
+          context.report({ node, messageId: "noDirectSocketIO" });
+        } else if (source === "socket.io-client") {
+          context.report({ node, messageId: "noDirectSocketIOClient" });
+        }
+      },
+    };
+  },
+};

packages/core-eslint/rules/no-direct-socket-io.test.js

@@ -0,0 +1,48 @@
+// packages/core-eslint/rules/no-direct-socket-io.test.js
+import { RuleTester } from "eslint";
+import rule from "./no-direct-socket-io.js";
+
+const tester = new RuleTester({
+  languageOptions: { ecmaVersion: 2022, sourceType: "module" },
+});
+
+tester.run("no-direct-socket-io", rule, {
+  valid: [
+    // Allowed inside core-realtime
+    {
+      code: 'import { Server } from "socket.io";',
+      filename: "/repo/packages/core-realtime/src/socket-io-realtime-server.ts",
+    },
+    // Allowed in app servers
+    {
+      code: 'import { Server } from "socket.io";',
+      filename: "/repo/apps/web-next/server.ts",
+    },
+    // Allowed in app integration tests (e.g. realtime-ping e2e)
+    {
+      code: 'import { Server } from "socket.io";',
+      filename: "/repo/apps/web-next/src/__tests__/realtime-ping.test.ts",
+    },
+    {
+      code: 'import { io } from "socket.io-client";',
+      filename: "/repo/apps/web-next/src/__tests__/realtime-ping.test.ts",
+    },
+    // Allowed elsewhere when not importing socket.io
+    {
+      code: 'import { foo } from "bar";',
+      filename: "/repo/packages/blog/src/foo.ts",
+    },
+  ],
+  invalid: [
+    {
+      code: 'import { Server } from "socket.io";',
+      filename: "/repo/packages/blog/src/foo.ts",
+      errors: [{ messageId: "noDirectSocketIO" }],
+    },
+    {
+      code: 'import { io } from "socket.io-client";',
+      filename: "/repo/packages/blog/src/ui/Component.tsx",
+      errors: [{ messageId: "noDirectSocketIOClient" }],
+    },
+  ],
+});

packages/core-eslint/rules/no-realtime-handler-reexport.js

@@ -0,0 +1,48 @@
+// packages/core-eslint/rules/no-realtime-handler-reexport.js
+// Realtime handlers are private. A feature's realtime/handlers/*.handler.ts
+// must only be wired in the feature's own bind-production / bind-dev-seed files.
+// They must never be re-exported from barrel files or other public surfaces.
+
+const BIND_FILE = /\bdi\/bind-(?:production|dev-seed)\b/;
+const REALTIME_HANDLERS_IN_SOURCE = /\/realtime\/handlers\//;
+const HANDLERS_IN_SOURCE = /\/handlers\//;
+const REALTIME_IN_FILENAME = /\/realtime\//;
+
+export default {
+  meta: {
+    type: "problem",
+    docs: {
+      description:
+        "Block re-exports of realtime/handlers/** outside feature bind-* files (ADR-016 R1)",
+    },
+    messages: {
+      noRealtimeHandlerReexport:
+        "Realtime handlers (realtime/handlers/*.handler.ts) must not be re-exported (ADR-016 R1). " +
+        "Wire them only inside the feature's own bind-production / bind-dev-seed files.",
+    },
+    schema: [],
+  },
+  create(context) {
+    const filename = context.filename ?? context.getFilename();
+
+    // Bind-* files are the only allowed place for these exports/imports
+    if (BIND_FILE.test(filename)) return {};
+
+    function checkExportSource(node) {
+      if (!node.source) return;
+      const source = node.source.value;
+      const isRealtimeHandler =
+        REALTIME_HANDLERS_IN_SOURCE.test(source) ||
+        (HANDLERS_IN_SOURCE.test(source) &&
+          REALTIME_IN_FILENAME.test(filename));
+      if (isRealtimeHandler) {
+        context.report({ node, messageId: "noRealtimeHandlerReexport" });
+      }
+    }
+
+    return {
+      ExportNamedDeclaration: checkExportSource,
+      ExportAllDeclaration: checkExportSource,
+    };
+  },
+};

packages/core-eslint/rules/no-realtime-handler-reexport.test.js

@@ -0,0 +1,35 @@
+// packages/core-eslint/rules/no-realtime-handler-reexport.test.js
+import { RuleTester } from "eslint";
+import rule from "./no-realtime-handler-reexport.js";
+
+const tester = new RuleTester({
+  languageOptions: { ecmaVersion: 2022, sourceType: "module" },
+});
+
+tester.run("no-realtime-handler-reexport", rule, {
+  valid: [
+    // Importing a handler from inside a feature's bind-* file is allowed.
+    {
+      code: 'import { onPingHandler } from "../realtime/handlers/on-ping.handler";',
+      filename: "/repo/packages/blog/src/di/bind-production.ts",
+    },
+    // Re-exporting a channel descriptor is allowed.
+    {
+      code: 'export { presenceChannel } from "./realtime/presence.channel";',
+      filename: "/repo/packages/blog/src/index.ts",
+    },
+  ],
+  invalid: [
+    // Re-exporting a handler from any non-bind file is forbidden.
+    {
+      code: 'export { onPingHandler } from "./realtime/handlers/on-ping.handler";',
+      filename: "/repo/packages/blog/src/index.ts",
+      errors: [{ messageId: "noRealtimeHandlerReexport" }],
+    },
+    {
+      code: 'export * from "./handlers/on-ping.handler";',
+      filename: "/repo/packages/blog/src/realtime/index.ts",
+      errors: [{ messageId: "noRealtimeHandlerReexport" }],
+    },
+  ],
+});

packages/core-realtime/AGENTS.md

@@ -0,0 +1,19 @@
+# @repo/core-realtime
+
+Vendor-isolated realtime abstractions over Socket.IO. Feature packages depend only on the interfaces; only this package imports `socket.io`.
+
+ADR-016 (`docs/decisions/adr-016-realtime-layer.md`).
+
+## Public exports
+
+- `IRealtimeBroadcaster` — server → client broadcasts
+- `IRealtimeServer` — lifecycle, used at app boot only
+- `IRealtimeAuthenticator` — connect-time identity attachment (cookie / header → user)
+- `IRealtimeHandlerRegistry` + `RealtimeHandlerRegistry` — inbound handler registration
+- `defineRealtimeChannel`, `RealtimeChannelDescriptor`, `ChannelScope`
+- `InMemoryRealtimeBroadcaster` (test/dev), `SocketIORealtimeBroadcaster`, `SocketIORealtimeServer` (production)
+- `CORE_REALTIME_SYMBOLS`
+
+## Boundary
+
+Tagged `core`. The only place in the repo where `import "socket.io"` is allowed is `src/socket-io-*.ts` here, plus `apps/*/server.ts`. Enforced by the ESLint rule `core-eslint/no-direct-socket-io`.

packages/core-realtime/docs/library-decisions/2026-05-14-socket.io.md

@@ -0,0 +1,72 @@
+---
+package: socket.io
+version: "^4.7.0"
+tier: core
+decision: approved
+date: 2026-05-14
+deciders: [scaffolded]
+adr: adr-016
+filter-results:
+  license: MIT
+  types: native
+  maintenance: active
+  boundary-fit: pass
+  shadow-check: pass
+  eu-residency: self-hostable
+  cve-scan: clean
+  named-consumer: pass
+verification-commands:
+  - pnpm audit --audit-level=moderate
+  - npm view socket.io license
+accepted-cves: []
+---
+
+## Filter: license
+
+MIT — on the workspace allowlist.
+
+## Filter: types
+
+Ships first-party TypeScript types in its distribution.
+
+## Filter: maintenance
+
+Active. Maintained by the Socket.IO team; frequent releases and active issue tracker.
+
+## Filter: maintenance
+
+Active. Regular releases; widely deployed in production.
+
+## Filter: boundary-fit
+
+ADR-016 §R2 explicitly designates `core-realtime` as the sole allowed home for `socket.io`. Boundary rule `no-direct-socket-io` enforces this in ESLint.
+
+## Filter: shadow-check
+
+No competing realtime transport in the workspace. No shadow.
+
+## Filter: eu-residency
+
+Self-hosted server; the library itself does not transmit data to any vendor endpoint.
+
+## Filter: cve-scan
+
+No advisories at adoption time.
+
+## Filter: named-consumer
+
+`core-realtime` wraps socket.io to provide the `IRealtimeServer` abstraction (ADR-016).
+
+## Prompt: replaces
+
+Nothing — this is the initial realtime scaffolding. No prior transport to retire.
+
+## Prompt: migration-cost-out
+
+Hard: channel descriptors, handler signatures, and server-side broadcast API are all shaped around socket.io semantics. Replacing requires re-implementing the abstraction layer.
+
+## Prompt: alternatives-considered
+
+1. **ws** — lower-level, no rooms or namespaces; would require significant protocol work.
+2. **Ably / Pusher** — vendor-hosted; eu-residency risk and ongoing cost.
+   Socket.IO is the established standard for this use-case and is fully self-hostable.

packages/core-realtime/docs/library-decisions/2026-05-14-zod.md

@@ -0,0 +1,66 @@
+---
+package: zod
+version: "^3.23.0"
+tier: core
+decision: approved
+date: 2026-05-14
+deciders: [scaffolded]
+adr: adr-016
+filter-results:
+  license: MIT
+  types: native
+  maintenance: active
+  boundary-fit: pass
+  shadow-check: pass
+  eu-residency: n/a
+  cve-scan: clean
+  named-consumer: pass
+verification-commands:
+  - pnpm audit --audit-level=moderate
+  - npm view zod license
+accepted-cves: []
+---
+
+## Filter: license
+
+MIT — on the workspace allowlist.
+
+## Filter: types
+
+Ships first-party TypeScript types in its distribution (`.d.ts` included).
+
+## Filter: maintenance
+
+Active. Regular releases by Colin McDonnell; widely adopted.
+
+## Filter: boundary-fit
+
+Core package. Zod is the workspace-canonical validation library locked in `core-shared` (ADR-016).
+
+## Filter: shadow-check
+
+Zod is already the workspace-locked validation library. No shadow.
+
+## Filter: eu-residency
+
+Pure computation; no network calls or vendor data transmission. n/a.
+
+## Filter: cve-scan
+
+No advisories at adoption time.
+
+## Filter: named-consumer
+
+`core-realtime` uses zod for channel descriptor and payload schema validation.
+
+## Prompt: replaces
+
+Nothing — zod is the pre-existing workspace validation library.
+
+## Prompt: migration-cost-out
+
+Mechanical: swap schema definitions at call sites. No data-format lock-in.
+
+## Prompt: alternatives-considered
+
+Zod is workspace-locked (see `core-shared`). A replacement would require a workspace-wide ADR; no alternative was evaluated here.

packages/core-realtime/eslint.config.js

@@ -0,0 +1,3 @@
+import baseConfig from "@repo/core-eslint/base";
+
+export default baseConfig;

packages/core-realtime/package.json

@@ -0,0 +1,38 @@
+{
+  "name": "@repo/core-realtime",
+  "version": "0.0.1",
+  "private": true,
+  "type": "module",
+  "exports": {
+    ".": "./src/index.ts"
+  },
+  "scripts": {
+    "build": "tsc --noEmit",
+    "lint": "eslint .",
+    "typecheck": "tsc --noEmit",
+    "test": "vitest run"
+  },
+  "dependencies": {
+    "@repo/core-shared": "workspace:*",
+    "socket.io": "^4.7.0",
+    "zod": "^3.23.0"
+  },
+  "peerDependencies": {
+    "payload": "^3.0.0"
+  },
+  "peerDependenciesMeta": {
+    "payload": {
+      "optional": true
+    }
+  },
+  "devDependencies": {
+    "@repo/core-eslint": "workspace:*",
+    "@repo/core-testing": "workspace:*",
+    "@repo/core-typescript": "workspace:*",
+    "@types/node": "^22.0.0",
+    "@vitest/coverage-v8": "^3.2.4",
+    "socket.io-client": "^4.7.0",
+    "typescript": "^5.8.0",
+    "vitest": "^3.0.0"
+  }
+}

packages/core-realtime/src/authorize.test.ts

@@ -0,0 +1,73 @@
+import { describe, it, expect } from "vitest";
+import { z } from "zod";
+import { authorize } from "@/authorize";
+import { defineRealtimeChannel } from "@/realtime-channel";
+
+const schema = z.object({}).strict();
+
+describe("authorize", () => {
+  describe("public", () => {
+    const ch = defineRealtimeChannel("a", schema, { scope: "public" });
+    it("allows anonymous", async () => {
+      expect(await authorize(ch, {}, null)).toBe(true);
+    });
+    it("allows authenticated", async () => {
+      expect(await authorize(ch, {}, { userId: "u1", roles: [] })).toBe(true);
+    });
+  });
+
+  describe("authenticated", () => {
+    const ch = defineRealtimeChannel("a", schema, { scope: "authenticated" });
+    it("rejects anonymous", async () => {
+      expect(await authorize(ch, {}, null)).toBe(false);
+    });
+    it("allows any user", async () => {
+      expect(await authorize(ch, {}, { userId: "u1", roles: [] })).toBe(true);
+    });
+  });
+
+  describe("{ role }", () => {
+    const ch = defineRealtimeChannel("a", schema, { scope: { role: "admin" } });
+    it("rejects anonymous", async () => {
+      expect(await authorize(ch, {}, null)).toBe(false);
+    });
+    it("rejects user without role", async () => {
+      expect(await authorize(ch, {}, { userId: "u1", roles: ["editor"] })).toBe(
+        false,
+      );
+    });
+    it("allows user with role", async () => {
+      expect(
+        await authorize(ch, {}, { userId: "u1", roles: ["admin", "editor"] }),
+      ).toBe(true);
+    });
+  });
+
+  describe("{ userScoped }", () => {
+    const ch = defineRealtimeChannel("a", schema, {
+      scope: { userScoped: true, template: "notifications.user.{userId}" },
+    });
+    it("rejects anonymous", async () => {
+      expect(await authorize(ch, { userId: "u1" }, null)).toBe(false);
+    });
+    it("rejects user requesting someone else's channel", async () => {
+      expect(
+        await authorize(ch, { userId: "u_other" }, { userId: "u1", roles: [] }),
+      ).toBe(false);
+    });
+    it("allows user requesting own channel", async () => {
+      expect(
+        await authorize(ch, { userId: "u1" }, { userId: "u1", roles: [] }),
+      ).toBe(true);
+    });
+  });
+
+  describe("unknown scope shape", () => {
+    const ch = defineRealtimeChannel("a", schema, {
+      scope: {} as never,
+    });
+    it("falls through to deny", async () => {
+      expect(await authorize(ch, {}, { userId: "u1", roles: [] })).toBe(false);
+    });
+  });
+});

packages/core-realtime/src/authorize.ts

@@ -0,0 +1,22 @@
+import type { z } from "zod";
+import type { RealtimeChannelDescriptor } from "./realtime-channel";
+
+export async function authorize(
+  descriptor: RealtimeChannelDescriptor<string, z.ZodType>,
+  params: Record<string, string>,
+  user: { userId: string; roles: string[] } | null,
+): Promise<boolean> {
+  const scope = descriptor.scope;
+
+  if (scope === "public") return true;
+  if (scope === "authenticated") return user !== null;
+
+  if (typeof scope === "object" && "role" in scope) {
+    return user !== null && user.roles.includes(scope.role);
+  }
+  if (typeof scope === "object" && "userScoped" in scope) {
+    return user !== null && params.userId === user.userId;
+  }
+
+  return false;
+}

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/channel-template.test.ts

@@ -0,0 +1,35 @@
+import { describe, it, expect } from "vitest";
+import { matchChannelTemplate } from "@/channel-template";
+
+describe("matchChannelTemplate", () => {
+  it("matches a plain channel name exactly", () => {
+    expect(matchChannelTemplate("blog.feed", "blog.feed")).toEqual({
+      params: {},
+    });
+    expect(matchChannelTemplate("blog.feed", "blog.other")).toBeNull();
+  });
+
+  it("matches a templated channel and extracts params", () => {
+    expect(
+      matchChannelTemplate(
+        "notifications.user.{userId}",
+        "notifications.user.user_42",
+      ),
+    ).toEqual({ params: { userId: "user_42" } });
+  });
+
+  it("returns null when a templated channel doesn't match the shape", () => {
+    expect(
+      matchChannelTemplate("notifications.user.{userId}", "notifications.user"),
+    ).toBeNull();
+    expect(
+      matchChannelTemplate("notifications.user.{userId}", "blog.feed"),
+    ).toBeNull();
+  });
+
+  it("supports multiple placeholders", () => {
+    expect(
+      matchChannelTemplate("rooms.{roomId}.user.{userId}", "rooms.r1.user.u1"),
+    ).toEqual({ params: { roomId: "r1", userId: "u1" } });
+  });
+});

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

@@ -0,0 +1,29 @@
+export function matchChannelTemplate(
+  template: string,
+  candidate: string,
+): { params: Record<string, string> } | null {
+  // No placeholders: exact match.
+  if (!template.includes("{")) {
+    return template === candidate ? { params: {} } : null;
+  }
+
+  // Build a regex from the template, replacing {name} with named groups.
+  const names: string[] = [];
+  const escaped = template.replace(/[.+?^${}()|[\]\\]/g, "\\$&"); // escape regex specials
+  // The above escapes `{` and `}` too — restore them around placeholders.
+  const pattern = escaped.replace(
+    /\\\{([a-zA-Z_][a-zA-Z0-9_]*)\\\}/g,
+    (_m, name) => {
+      names.push(name);
+      return `([^.]+)`;
+    },
+  );
+  const re = new RegExp(`^${pattern}$`);
+  const match = candidate.match(re);
+  if (!match) return null;
+  const params: Record<string, string> = {};
+  names.forEach((name, i) => {
+    params[name] = match[i + 1]!;
+  });
+  return { params };
+}

packages/core-realtime/src/in-memory-realtime-broadcaster.test.ts

@@ -0,0 +1,34 @@
+import { describe, it, expect } from "vitest";
+import { z } from "zod";
+import { InMemoryRealtimeBroadcaster } from "@/in-memory-realtime-broadcaster";
+import { defineRealtimeChannel } from "@/realtime-channel";
+
+const ch = defineRealtimeChannel("a.b", z.object({ x: z.number() }).strict(), {
+  scope: "public",
+});
+
+describe("InMemoryRealtimeBroadcaster", () => {
+  it("validates payload via the descriptor schema", async () => {
+    const b = new InMemoryRealtimeBroadcaster();
+    await expect(
+      b.broadcast(ch, { x: "not a number" } as never),
+    ).rejects.toThrow();
+  });
+
+  it("delivers to subscribers in order", async () => {
+    const b = new InMemoryRealtimeBroadcaster();
+    const got: number[] = [];
+    b.subscribe(ch, async (p) => {
+      got.push(p.x);
+    });
+    await b.broadcast(ch, { x: 1 });
+    await b.broadcast(ch, { x: 2 });
+    expect(got).toEqual([1, 2]);
+  });
+
+  it("does nothing when no subscribers", async () => {
+    const b = new InMemoryRealtimeBroadcaster();
+    await b.broadcast(ch, { x: 1 });
+    // does not throw
+  });
+});

packages/core-realtime/src/in-memory-realtime-broadcaster.ts

@@ -0,0 +1,28 @@
+import type { z } from "zod";
+import type { IRealtimeBroadcaster } from "./realtime-broadcaster.interface";
+import type { RealtimeChannelDescriptor } from "./realtime-channel";
+
+type Listener<T> = (payload: T) => Promise<void> | void;
+
+export class InMemoryRealtimeBroadcaster implements IRealtimeBroadcaster {
+  private readonly listeners = new Map<string, Listener<unknown>[]>();
+
+  async broadcast<T>(
+    descriptor: RealtimeChannelDescriptor<string, z.ZodType<T>>,
+    payload: T,
+  ): Promise<void> {
+    descriptor.schema.parse(payload);
+    const arr = this.listeners.get(descriptor.name) ?? [];
+    for (const l of arr) await l(payload);
+  }
+
+  // Test-friendly: lets unit tests subscribe directly without a Socket.IO server.
+  subscribe<T>(
+    descriptor: RealtimeChannelDescriptor<string, z.ZodType<T>>,
+    listener: Listener<T>,
+  ): void {
+    const arr = this.listeners.get(descriptor.name) ?? [];
+    arr.push(listener as Listener<unknown>);
+    this.listeners.set(descriptor.name, arr);
+  }
+}

packages/core-realtime/src/index.ts

@@ -0,0 +1,32 @@
+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";
+export type { IRealtimeAuthenticator } from "./realtime-authenticator.interface";
+export type { IRealtimeHandlerRegistry } from "./realtime-handler-registry";
+export { RealtimeHandlerRegistry } from "./realtime-handler-registry";
+export { CORE_REALTIME_SYMBOLS } from "./symbols";
+export { InMemoryRealtimeBroadcaster } from "./in-memory-realtime-broadcaster";
+export { SocketIORealtimeBroadcaster } from "./socket-io-realtime-broadcaster";
+export { SocketIORealtimeServer } from "./socket-io-realtime-server";
+export { authorize } from "./authorize";
+export { matchChannelTemplate } from "./channel-template";
+export {
+  realtimePingChannel,
+  realtimePongChannel,
+  realtimePingInboundDescriptor,
+  type PingPayload,
+  type PongPayload,
+} from "./realtime-ping";

packages/core-realtime/src/realtime-authenticator.interface.ts

@@ -0,0 +1,6 @@
+export interface IRealtimeAuthenticator {
+  authenticate(handshake: {
+    cookies: Record<string, string>;
+    headers: Record<string, string>;
+  }): Promise<{ userId: string; roles: string[] } | null>;
+}

packages/core-realtime/src/realtime-broadcaster.interface.ts

@@ -0,0 +1,10 @@
+import type { z } from "zod";
+import type { RealtimeBroadcasterProtocol } from "@repo/core-shared/di/bind-protocols";
+import type { RealtimeChannelDescriptor } from "./realtime-channel";
+
+export interface IRealtimeBroadcaster extends RealtimeBroadcasterProtocol {
+  broadcast<T>(
+    descriptor: RealtimeChannelDescriptor<string, z.ZodType<T>>,
+    payload: T,
+  ): Promise<void>;
+}

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

@@ -0,0 +1,35 @@
+import { describe, it, expect } from "vitest";
+import { z } from "zod";
+import { defineRealtimeChannel } from "@/realtime-channel";
+
+describe("defineRealtimeChannel", () => {
+  it("returns a descriptor with name, schema, and scope", () => {
+    const ch = defineRealtimeChannel(
+      "test.channel",
+      z.object({ id: z.string() }).strict(),
+      { scope: "public" },
+    );
+    expect(ch.name).toBe("test.channel");
+    expect(ch.scope).toBe("public");
+    expect(() => ch.schema.parse({ id: "x" })).not.toThrow();
+  });
+
+  it("preserves all four scope shapes", () => {
+    expect(
+      defineRealtimeChannel("a", z.object({}), { scope: "public" }).scope,
+    ).toBe("public");
+    expect(
+      defineRealtimeChannel("a", z.object({}), { scope: "authenticated" })
+        .scope,
+    ).toBe("authenticated");
+    expect(
+      defineRealtimeChannel("a", z.object({}), { scope: { role: "admin" } })
+        .scope,
+    ).toEqual({ role: "admin" });
+    expect(
+      defineRealtimeChannel("a", z.object({}), {
+        scope: { userScoped: true, template: "x.{id}" },
+      }).scope,
+    ).toEqual({ userScoped: true, template: "x.{id}" });
+  });
+});

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

@@ -0,0 +1,34 @@
+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"
+  | { role: string }
+  | { userScoped: true; template: string };
+
+export type RealtimeChannelDescriptor<
+  TName extends string,
+  TSchema extends z.ZodType,
+> = {
+  readonly name: TName;
+  readonly schema: TSchema;
+  readonly scope: ChannelScope;
+};
+
+export function defineRealtimeChannel<
+  TName extends string,
+  TSchema extends z.ZodType,
+>(
+  name: TName,
+  schema: TSchema,
+  options: { scope: ChannelScope },
+): RealtimeChannelDescriptor<TName, TSchema> {
+  return { name, schema, scope: options.scope };
+}

packages/core-realtime/src/realtime-handler-registry.test.ts

@@ -0,0 +1,94 @@
+import { describe, it, expect, vi } from "vitest";
+import { z } from "zod";
+import { RealtimeHandlerRegistry } from "@/realtime-handler-registry";
+import { defineRealtimeChannel } from "@/realtime-channel";
+
+const ch = defineRealtimeChannel(
+  "test.ch",
+  z.object({ x: z.number() }).strict(),
+  { scope: "authenticated" },
+);
+
+describe("RealtimeHandlerRegistry", () => {
+  it("registers and retrieves a handler by channel name", () => {
+    const reg = new RealtimeHandlerRegistry();
+    const handler = vi.fn();
+    reg.register({ descriptor: ch, handler });
+    const got = reg.getInboundDescriptor("test.ch");
+    expect(got).not.toBeNull();
+    expect(got!.descriptor.name).toBe("test.ch");
+    expect(got!.handler).toBe(handler);
+  });
+
+  it("returns null for unknown channel name", () => {
+    const reg = new RealtimeHandlerRegistry();
+    expect(reg.getInboundDescriptor("unknown")).toBeNull();
+  });
+
+  it("list() returns all registered descriptors", () => {
+    const reg = new RealtimeHandlerRegistry();
+    reg.register({ descriptor: ch, handler: vi.fn() });
+    expect(reg.list()).toHaveLength(1);
+    expect(reg.list()[0]!.descriptor.name).toBe("test.ch");
+  });
+
+  it("re-registering the same channel replaces the previous entry", () => {
+    const reg = new RealtimeHandlerRegistry();
+    const h1 = vi.fn();
+    const h2 = vi.fn();
+    reg.register({ descriptor: ch, handler: h1 });
+    reg.register({ descriptor: ch, handler: h2 });
+    expect(reg.getInboundDescriptor("test.ch")!.handler).toBe(h2);
+    expect(reg.list()).toHaveLength(1);
+  });
+
+  it("registerChannel stores a descriptor that appears in listChannels() but not in list()", () => {
+    const reg = new RealtimeHandlerRegistry();
+    const outboundCh = defineRealtimeChannel(
+      "test.outbound",
+      z.object({ y: z.string() }).strict(),
+      { scope: "authenticated" },
+    );
+    reg.registerChannel(outboundCh);
+    expect(reg.listChannels()).toHaveLength(1);
+    expect(reg.listChannels()[0]!.name).toBe("test.outbound");
+    expect(reg.list()).toHaveLength(0);
+  });
+
+  it("register auto-populates listChannels()", () => {
+    const reg = new RealtimeHandlerRegistry();
+    reg.register({ descriptor: ch, handler: vi.fn() });
+    expect(reg.listChannels()).toHaveLength(1);
+    expect(reg.listChannels()[0]!.name).toBe("test.ch");
+  });
+
+  it("listChannels() returns both inbound and outbound-only channels when both are registered", () => {
+    const reg = new RealtimeHandlerRegistry();
+    const outboundCh = defineRealtimeChannel(
+      "test.outbound",
+      z.object({ y: z.string() }).strict(),
+      { scope: "authenticated" },
+    );
+    reg.register({ descriptor: ch, handler: vi.fn() });
+    reg.registerChannel(outboundCh);
+    expect(reg.listChannels()).toHaveLength(2);
+    const names = reg
+      .listChannels()
+      .map((c) => c.name)
+      .sort();
+    expect(names).toEqual(["test.ch", "test.outbound"]);
+  });
+
+  it("re-registering an outbound-only channel via registerChannel replaces the previous entry", () => {
+    const reg = new RealtimeHandlerRegistry();
+    const outboundCh = defineRealtimeChannel(
+      "test.outbound",
+      z.object({ y: z.string() }).strict(),
+      { scope: "authenticated" },
+    );
+    reg.registerChannel(outboundCh);
+    reg.registerChannel(outboundCh);
+    expect(reg.listChannels()).toHaveLength(1);
+    expect(reg.listChannels()[0]!.name).toBe("test.outbound");
+  });
+});

packages/core-realtime/src/realtime-handler-registry.ts

@@ -0,0 +1,57 @@
+import type { z } from "zod";
+import type { RealtimeRegistryProtocol } from "@repo/core-shared/di/bind-protocols";
+import type { RealtimeChannelDescriptor } from "./realtime-channel";
+import type { IInboundDescriptor } from "./realtime-handler.interface";
+
+export interface IRealtimeHandlerRegistry extends RealtimeRegistryProtocol {
+  register<T>(entry: IInboundDescriptor<string, z.ZodType<T>>): void;
+  getInboundDescriptor(
+    channelName: string,
+  ): IInboundDescriptor<string, z.ZodType> | null;
+  list(): IInboundDescriptor<string, z.ZodType>[];
+  /** Register an outbound-only channel so Gate 2 can authorize subscriptions to it. */
+  registerChannel(
+    descriptor: RealtimeChannelDescriptor<string, z.ZodType>,
+  ): void;
+  listChannels(): RealtimeChannelDescriptor<string, z.ZodType>[];
+}
+
+export class RealtimeHandlerRegistry implements IRealtimeHandlerRegistry {
+  private readonly entries = new Map<
+    string,
+    IInboundDescriptor<string, z.ZodType>
+  >();
+  private readonly channels = new Map<
+    string,
+    RealtimeChannelDescriptor<string, z.ZodType>
+  >();
+
+  register<T>(entry: IInboundDescriptor<string, z.ZodType<T>>): void {
+    this.entries.set(
+      entry.descriptor.name,
+      entry as IInboundDescriptor<string, z.ZodType>,
+    );
+    // Also add the descriptor to the channel map so Gate 2 can authorize subscriptions.
+    this.channels.set(entry.descriptor.name, entry.descriptor);
+  }
+
+  getInboundDescriptor(
+    channelName: string,
+  ): IInboundDescriptor<string, z.ZodType> | null {
+    return this.entries.get(channelName) ?? null;
+  }
+
+  list(): IInboundDescriptor<string, z.ZodType>[] {
+    return Array.from(this.entries.values());
+  }
+
+  registerChannel(
+    descriptor: RealtimeChannelDescriptor<string, z.ZodType>,
+  ): void {
+    this.channels.set(descriptor.name, descriptor);
+  }
+
+  listChannels(): RealtimeChannelDescriptor<string, z.ZodType>[] {
+    return Array.from(this.channels.values());
+  }
+}

packages/core-realtime/src/realtime-handler.interface.ts

@@ -0,0 +1,20 @@
+import type { z } from "zod";
+import type { RealtimeChannelDescriptor } from "./realtime-channel";
+
+export type RealtimeContext = {
+  userId: string | null;
+  roles: string[];
+};
+
+export type IRealtimeHandler<T> = (
+  input: T,
+  ctx: RealtimeContext,
+) => Promise<void>;
+
+export type IInboundDescriptor<
+  TName extends string,
+  TSchema extends z.ZodType,
+> = {
+  readonly descriptor: RealtimeChannelDescriptor<TName, TSchema>;
+  readonly handler: IRealtimeHandler<z.infer<TSchema>>;
+};

packages/core-realtime/src/realtime-ping.test.ts

@@ -0,0 +1,53 @@
+import { describe, it, expect } from "vitest";
+import { InMemoryRealtimeBroadcaster } from "@/in-memory-realtime-broadcaster";
+import {
+  realtimePingChannel,
+  realtimePingInboundDescriptor,
+  realtimePongChannel,
+  type PongPayload,
+} from "@/realtime-ping";
+
+describe("realtimePingChannel / realtimePongChannel", () => {
+  it("declares authenticated-scoped channels", () => {
+    expect(realtimePingChannel.name).toBe("realtime.ping");
+    expect(realtimePingChannel.scope).toBe("authenticated");
+    expect(realtimePongChannel.name).toBe("realtime.pong");
+    expect(realtimePongChannel.scope).toBe("authenticated");
+  });
+
+  it("rejects a ping payload without an ISO datetime", () => {
+    expect(() =>
+      realtimePingChannel.schema.parse({ at: "not-a-date" }),
+    ).toThrow();
+  });
+});
+
+describe("realtimePingInboundDescriptor", () => {
+  const at = new Date().toISOString();
+
+  it("broadcasts a pong echoing the authenticated userId", async () => {
+    const broadcaster = new InMemoryRealtimeBroadcaster();
+    const got: PongPayload[] = [];
+    broadcaster.subscribe(realtimePongChannel, (p) => {
+      got.push(p);
+    });
+
+    const { descriptor, handler } = realtimePingInboundDescriptor(broadcaster);
+    expect(descriptor).toBe(realtimePingChannel);
+
+    await handler({ at }, { userId: "user-1", roles: [] });
+    expect(got).toEqual([{ at, echo: "user-1" }]);
+  });
+
+  it('echoes "anonymous" when the context has no userId', async () => {
+    const broadcaster = new InMemoryRealtimeBroadcaster();
+    const got: PongPayload[] = [];
+    broadcaster.subscribe(realtimePongChannel, (p) => {
+      got.push(p);
+    });
+
+    const { handler } = realtimePingInboundDescriptor(broadcaster);
+    await handler({ at }, { userId: null, roles: [] });
+    expect(got).toEqual([{ at, echo: "anonymous" }]);
+  });
+});

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

@@ -0,0 +1,44 @@
+import { z } from "zod";
+import { defineRealtimeChannel } from "./realtime-channel";
+import type { IRealtimeBroadcaster } from "./realtime-broadcaster.interface";
+import type {
+  IInboundDescriptor,
+  RealtimeContext,
+} from "./realtime-handler.interface";
+
+const pingSchema = z.object({ at: z.string().datetime() }).strict();
+const pongSchema = z
+  .object({ at: z.string().datetime(), echo: z.string() })
+  .strict();
+
+export type PingPayload = z.infer<typeof pingSchema>;
+export type PongPayload = z.infer<typeof pongSchema>;
+
+export const realtimePingChannel = defineRealtimeChannel(
+  "realtime.ping",
+  pingSchema,
+  { scope: "authenticated" },
+);
+
+export const realtimePongChannel = defineRealtimeChannel(
+  "realtime.pong",
+  pongSchema,
+  { scope: "authenticated" },
+);
+
+export function realtimePingInboundDescriptor(
+  broadcaster: IRealtimeBroadcaster,
+): IInboundDescriptor<"realtime.ping", z.ZodType<PingPayload>> {
+  return {
+    descriptor: realtimePingChannel,
+    handler: async (
+      input: PingPayload,
+      ctx: RealtimeContext,
+    ): Promise<void> => {
+      await broadcaster.broadcast(realtimePongChannel, {
+        at: input.at,
+        echo: ctx.userId ?? "anonymous",
+      });
+    },
+  };
+}

packages/core-realtime/src/realtime-server.interface.ts

@@ -0,0 +1,16 @@
+import type { Server as HttpServer } from "node:http";
+import type { Server as IOServer } from "socket.io";
+import type { IRealtimeAuthenticator } from "./realtime-authenticator.interface";
+import type { IRealtimeHandlerRegistry } from "./realtime-handler-registry";
+
+export type IRealtimeServerOptions = {
+  httpServer: HttpServer;
+  io: IOServer;
+  authenticator: IRealtimeAuthenticator;
+  registry: IRealtimeHandlerRegistry;
+};
+
+export interface IRealtimeServer {
+  start(): Promise<void>;
+  stop(): Promise<void>;
+}

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

@@ -0,0 +1,32 @@
+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";
+
+const ch = defineRealtimeChannel("a.b", z.object({ x: z.number() }).strict(), {
+  scope: "public",
+});
+
+describe("SocketIORealtimeBroadcaster", () => {
+  it("emits to the channel's room with the channel name as event", async () => {
+    const emit = vi.fn();
+    const to = vi.fn(() => ({ emit }));
+    const io = { to } as never;
+    const b = new SocketIORealtimeBroadcaster(io);
+    await b.broadcast(ch, { x: 1 });
+    expect(to).toHaveBeenCalledWith(channelRoom("a.b"));
+    expect(emit).toHaveBeenCalledWith("a.b", { x: 1 });
+  });
+
+  it("validates payload before emitting", async () => {
+    const emit = vi.fn();
+    const to = vi.fn(() => ({ emit }));
+    const io = { to } as never;
+    const b = new SocketIORealtimeBroadcaster(io);
+    await expect(
+      b.broadcast(ch, { x: "not a number" } as never),
+    ).rejects.toThrow();
+    expect(emit).not.toHaveBeenCalled();
+  });
+});

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

@@ -0,0 +1,17 @@
+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";
+
+export class SocketIORealtimeBroadcaster implements IRealtimeBroadcaster {
+  constructor(private readonly io: IOServer) {}
+
+  async broadcast<T>(
+    descriptor: RealtimeChannelDescriptor<string, z.ZodType<T>>,
+    payload: T,
+  ): Promise<void> {
+    descriptor.schema.parse(payload);
+    this.io.to(channelRoom(descriptor.name)).emit(descriptor.name, payload);
+  }
+}

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

@@ -0,0 +1,303 @@
+import { describe, it, expect, afterEach } from "vitest";
+import { z } from "zod";
+import { createServer } 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" },
+);
+
+const userChannel = defineRealtimeChannel(
+  "user.{userId}.events",
+  z.object({ msg: z.string() }).strict(),
+  { scope: { userScoped: true, template: "user.{userId}.events" } },
+);
+
+// ---------------------------------------------------------------------------
+// Shared test infrastructure
+// ---------------------------------------------------------------------------
+
+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) => {
+      received = { input, ctx };
+    },
+  });
+
+  registry.register({
+    descriptor: userChannel,
+    handler: async () => {
+      /* no-op for routing tests */
+    },
+  });
+
+  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;
+    },
+  };
+
+  const server = new SocketIORealtimeServer({
+    httpServer,
+    io,
+    authenticator,
+    registry,
+  });
+  await server.start();
+
+  await new Promise<void>((resolve) => httpServer.listen(0, resolve));
+  const port = (httpServer.address() as AddressInfo).port;
+
+  return { httpServer, io, server, port, getReceived: () => 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 teardown?.();
+  });
+
+  it("rejects subscribe to authenticated channel from anonymous socket", async () => {
+    const { server, httpServer, port } = await setup();
+    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),
+    );
+
+    expect(ack.ok).toBe(false);
+    expect(ack.error).toBe("forbidden");
+    client.disconnect();
+  });
+
+  it("allows subscribe + invokes handler with ctx for authenticated socket", async () => {
+    const { server, httpServer, port, getReceived } = await setup();
+    teardown = async () => {
+      await server.stop();
+      httpServer.close();
+    };
+
+    const client = await connectClient(port, "session=valid");
+
+    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);
+
+    // 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 { server, httpServer, port } = await setup();
+    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),
+    );
+
+    expect(ack.ok).toBe(false);
+    expect(ack.error).toBe("unknown_channel");
+    client.disconnect();
+  });
+
+  it("rejects malformed inbound input", async () => {
+    const { server, httpServer, port } = await setup();
+    teardown = async () => {
+      await server.stop();
+      httpServer.close();
+    };
+
+    const client = await connectClient(port, "session=valid");
+    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();
+  });
+
+  it("acks handler_error when the inbound handler throws", async () => {
+    const throwingChannel = defineRealtimeChannel(
+      "test.throws",
+      z.object({}).strict(),
+      { scope: "authenticated" },
+    );
+    const { server, httpServer, port } = await setup({
+      extraSetup: (registry) => {
+        registry.register({
+          descriptor: throwingChannel,
+          handler: async () => {
+            throw new Error("boom");
+          },
+        });
+      },
+    });
+    teardown = async () => {
+      await server.stop();
+      httpServer.close();
+    };
+
+    const client = await connectClient(port, "session=valid");
+    const ack = await new Promise<{ ok: boolean; error?: string }>((r) =>
+      client.emit("test.throws", {}, r),
+    );
+
+    expect(ack.ok).toBe(false);
+    expect(ack.error).toBe("handler_error");
+    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

@@ -0,0 +1,151 @@
+import type { Server as IOServer, Socket, DefaultEventsMap } 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(";")) {
+    const [k, ...rest] = part.trim().split("=");
+    if (k) out[k] = decodeURIComponent(rest.join("="));
+  }
+  return out;
+}
+
+export class SocketIORealtimeServer implements IRealtimeServer {
+  private readonly io: IOServer;
+  private readonly opts: IRealtimeServerOptions;
+
+  constructor(opts: IRealtimeServerOptions) {
+    this.opts = opts;
+    this.io = opts.io;
+  }
+
+  async start(): Promise<void> {
+    const { authenticator, registry } = this.opts;
+
+    // Gate 1: connect — read cookie, authenticate, attach user.
+    // 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;
+
+      // Gate 2: subscribe.
+      socket.on(
+        "subscribe",
+        async (requestedName: string, ack?: (r: unknown) => void) => {
+          // Find a registered descriptor whose name (or template) matches requestedName.
+          // listChannels() covers both inbound descriptors and outbound-only channels.
+          let matched: {
+            descriptor: { name: string; scope: unknown };
+            params: Record<string, string>;
+          } | null = null;
+          for (const descriptor of registry.listChannels()) {
+            const m = matchChannelTemplate(descriptor.name, requestedName);
+            if (m) {
+              matched = { 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(channelRoom(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;
+            }
+
+            // 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;
+            }
+            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> {
+    await new Promise<void>((resolve) => this.io.close(() => resolve()));
+  }
+}

packages/core-realtime/src/symbols.ts

@@ -0,0 +1,8 @@
+export const CORE_REALTIME_SYMBOLS = {
+  IRealtimeBroadcaster: Symbol.for("core-realtime:IRealtimeBroadcaster"),
+  IRealtimeServer: Symbol.for("core-realtime:IRealtimeServer"),
+  IRealtimeAuthenticator: Symbol.for("core-realtime:IRealtimeAuthenticator"),
+  IRealtimeHandlerRegistry: Symbol.for(
+    "core-realtime:IRealtimeHandlerRegistry",
+  ),
+} as const;

packages/core-realtime/tsconfig.json

@@ -0,0 +1,12 @@
+{
+  "extends": "@repo/core-typescript/base.json",
+  "compilerOptions": {
+    "outDir": "dist",
+    "rootDir": ".",
+    "paths": {
+      "@/*": ["./src/*"]
+    }
+  },
+  "include": ["src/**/*", "*.config.ts"],
+  "exclude": ["node_modules", "dist"]
+}

packages/core-realtime/turbo.json

@@ -0,0 +1,5 @@
+{
+  "$schema": "https://turborepo.dev/schema.json",
+  "extends": ["//"],
+  "tags": ["core"]
+}

packages/core-realtime/vitest.config.ts

@@ -0,0 +1,18 @@
+import path from "node:path";
+import { mergeConfig } from "vitest/config";
+import { nodeVitestConfig } from "@repo/core-typescript/vitest.base.node";
+
+export default mergeConfig(nodeVitestConfig, {
+  test: {
+    coverage: {
+      exclude: [
+        // DI symbol constants — boilerplate, covered implicitly by bind-* tests
+        // (mirrors core-shared/vitest.config.ts)
+        "src/**/symbols.ts",
+      ],
+    },
+  },
+  resolve: {
+    alias: { "@": path.resolve(__dirname, "./src") },
+  },
+});