From 500d835f0e6b17e0d3a426f21b555f131d8265bd Mon Sep 17 00:00:00 2001 From: Danijel Martinek Date: Fri, 8 May 2026 19:51:06 +0200 Subject: [PATCH] docs(spec): realtime layer (Socket.IO) design MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Long-form design for the bidirectional realtime layer the user brainstormed on 2026-05-08. Builds on ADR-015 (events and jobs); hybrid pattern where IRealtimeBroadcaster is layered alongside IEventBus, with a bridge for forwarding allowlisted bus events to realtime channels. v1 ships: @repo/core-realtime package, custom Node server in apps/web-next, cookie-session auth, four scope kinds, two generators (gen realtime channel|handler), three new anchors per feature, RecordingRealtimeBroadcaster test helper, realtime-ping proof-of-life. Dashboard, DB-backed roles/permissions, multi-instance fanout, and a generic core-package generator are deferred — captured in §13 / §14 / §15. Companion HTML explainer at realtime-auth-explainer.html (project root, uncommitted) walks the four-checkpoint auth model visually. ADR-016 will be created during implementation. --- .../specs/2026-05-08-realtime-design.md | 879 ++++++++++++++++++ 1 file changed, 879 insertions(+) create mode 100644 docs/superpowers/specs/2026-05-08-realtime-design.md diff --git a/docs/superpowers/specs/2026-05-08-realtime-design.md b/docs/superpowers/specs/2026-05-08-realtime-design.md new file mode 100644 index 0000000..38dec4c --- /dev/null +++ b/docs/superpowers/specs/2026-05-08-realtime-design.md @@ -0,0 +1,879 @@ +# Realtime layer (Socket.IO) — Design + +**Date:** 2026-05-08 +**Status:** Draft (pending user review) +**Companion ADR:** ADR-016 (to be created during implementation; this spec is the long-form design that ADR-016 distills). +**Builds on:** ADR-015 (cross-feature events and background jobs). The bridge in §8 is a third consumer of the `IEventBus` shipped in ADR-015. + +--- + +## 1. Context and motivation + +This template currently has no convention for **realtime communication between server and browser**. tRPC covers request/response (`query` / `mutation`); the event bus from ADR-015 covers in-process cross-feature publish/subscribe. Neither delivers server state to a connected browser tab without polling. + +Two distinct concerns drive this work: + +- **Generic realtime seam.** Establish the abstraction now (interface, vendor-neutral adapter, DI binding, scaffolding) so individual features can adopt realtime on demand without each feature having to design its own socket integration. Same logic that motivated `IEventBus` and `IJobQueue` in ADR-015: define the seam once. +- **Live observability dashboard (deferred to a follow-up PR).** An admin view that streams live event/job traffic from the existing bus + queue. The dashboard is the first concrete consumer of the seam, but its UI work is large enough to ship separately. v1 instead carries a built-in `realtime-ping` channel as the proof-of-life. + +The architectural principle from prior work (ADR-014 instrumentation, ADR-015 events/jobs) is preserved: the wire-protocol library (`socket.io`) is hidden behind an `IRealtimeBroadcaster` / `IRealtimeServer` interface. Feature packages MUST NOT import `socket.io` directly. Only the new `@repo/core-realtime` package and the apps' bootstrap layer touch the SDK. + +--- + +## 2. Conceptual model and rules + +### 2.1 What the realtime layer is *for* + +| | Realtime broadcast | Realtime handler (inbound) | +|---|---|---| +| **Direction** | Server → connected clients | Client → server | +| **Caller** | A feature's use case (direct), or the bridge (forwarding bus events) | A connected client emits a payload | +| **Audience** | Subscribers to a named channel | The feature that registered the handler | +| **Transport** | Socket.IO (vendor-isolated behind `IRealtimeBroadcaster`) | Socket.IO (vendor-isolated; handler factory pattern) | +| **Why it exists** | Push state changes to browser tabs without polling | Receive ephemeral / high-frequency client signals (presence, cursors) | + +### 2.2 The three rules + +Mirroring ADR-015's rule pattern. These are the discipline; mechanical wiring follows from them. + +- **Rule R0 — Realtime is for state delivery, not for replacing tRPC.** Persistent operations with request/response semantics belong on tRPC procedures. Use realtime when (a) the server needs to push without a request, or (b) the data is too high-frequency for HTTP. +- **Rule R1 — Channel descriptors are exported; handlers are private.** A feature's `realtime/.channel.ts` is re-exported from the package root barrel. A feature's `realtime/handlers/.handler.ts` is wired only inside that feature's own `bind-production` / `bind-dev-seed` and is never re-exported from any subpath. Mirrors ADR-015's E1 for events. +- **Rule R2 — Vendor isolation: `socket.io` lives in one package only.** Feature packages MUST NOT `import "socket.io"` or `import "socket.io-client"`. The single allowlist entry is `packages/core-realtime/src/socket-io-*.ts`. Apps' custom Node servers are the second allowlist entry (they instantiate the server). ESLint rule `no-direct-socket-io` enforces this — parallel to `no-direct-payload-jobs` from ADR-015. + +These rules go verbatim into ADR-016 and into `AGENTS.md` § Per-Package Conventions. + +### 2.3 The seam (server → client flow) + +``` +feature/use-case + │ + │ realtime.broadcast(channelDescriptor, payload) + ▼ +IRealtimeBroadcaster ←── vendor-neutral + │ + │ implements + ▼ +SocketIORealtimeBroadcaster ←── lives in core-realtime + │ + │ this.io.to("ch:").emit(name, payload) + ▼ +Socket.IO room "ch:" + │ + ▼ +N connected sockets that subscribed (gate 2 cleared) + │ + ▼ +Browser tab receives `socket.on(name, payload)` +``` + +### 2.4 The seam (client → server flow) + +``` +Browser tab: socket.emit(name, payload, ack) + │ + ▼ +SocketIORealtimeServer (gate 3) + │ + ├─ schema.safeParse(payload) + ├─ authorize(descriptor, params, socket.data.user) ← defense in depth + ▼ +handlerRegistry.invoke(name, validated, ctx) + │ + │ ctx = { userId, roles } (from socket.data.user) + ▼ +withSpan(withCapture(handlerFactory(deps)))(input, ctx) + │ + ▼ +Feature handler runs, throws or returns void + │ + ▼ +ack({ ok: true } | { ok: false, error }) +``` + +--- + +## 3. New package: `@repo/core-realtime` + +A new package, parallel in shape to `@repo/core-events`. Pure interfaces + symbols + the Socket.IO adapters. Tagged `core` (Turborepo boundary) and `core` (eslint-plugin-boundaries). + +### 3.1 Package shape + +``` +packages/core-realtime/ +├── package.json name: @repo/core-realtime +├── tsconfig.json +├── vitest.config.ts +├── eslint.config.js +├── turbo.json tags: ["core"] +├── AGENTS.md +└── src/ + ├── index.ts public barrel + ├── symbols.ts CORE_REALTIME_SYMBOLS + ├── realtime-channel.ts defineRealtimeChannel + descriptor + scope union + ├── realtime-broadcaster.interface.ts IRealtimeBroadcaster (server → client) + ├── realtime-handler.interface.ts IRealtimeHandler + IInboundDescriptor + ├── realtime-server.interface.ts IRealtimeServer (lifecycle, used at boot only) + ├── realtime-authenticator.interface.ts IRealtimeAuthenticator (cookie/header → user) + ├── realtime-handler-registry.ts in-memory map of channel → wrapped handler + ├── channel-template.ts matches "notifications.user.{userId}" → params + ├── authorize.ts pure function: descriptor + params + user → bool + ├── in-memory-realtime-broadcaster.ts test/dev (no socket.io dep) + ├── socket-io-realtime-server.ts production (depends on socket.io) + └── socket-io-realtime-broadcaster.ts production (wraps io.to(...).emit(...)) +``` + +### 3.2 Public exports (barrel) + +```ts +// packages/core-realtime/src/index.ts +export type { RealtimeChannelDescriptor, ChannelScope } from "./realtime-channel"; +export { defineRealtimeChannel } from "./realtime-channel"; +export type { IRealtimeBroadcaster } from "./realtime-broadcaster.interface"; +export type { IRealtimeHandler, IInboundDescriptor } 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 { SocketIORealtimeServer } from "./socket-io-realtime-server"; +export { SocketIORealtimeBroadcaster } from "./socket-io-realtime-broadcaster"; +``` + +### 3.3 Interfaces — exact shapes + +```ts +// packages/core-realtime/src/realtime-channel.ts +import type { z } from "zod"; + +export type ChannelScope = + | "public" + | "authenticated" + | { role: string } + | { userScoped: true; template: string }; + +export type RealtimeChannelDescriptor = { + readonly name: TName; + readonly schema: TSchema; + readonly scope: ChannelScope; +}; + +export function defineRealtimeChannel( + name: TName, + schema: TSchema, + options: { scope: ChannelScope }, +): RealtimeChannelDescriptor { + return { name, schema, scope: options.scope }; +} +``` + +```ts +// packages/core-realtime/src/realtime-broadcaster.interface.ts +import type { z } from "zod"; +import type { RealtimeChannelDescriptor } from "./realtime-channel"; + +export interface IRealtimeBroadcaster { + broadcast( + descriptor: RealtimeChannelDescriptor>, + payload: T, + ): Promise; +} +``` + +```ts +// packages/core-realtime/src/realtime-handler.interface.ts +import type { z } from "zod"; +import type { RealtimeChannelDescriptor } from "./realtime-channel"; + +export type RealtimeContext = { + userId: string | null; + roles: string[]; +}; + +export type IRealtimeHandler = (input: T, ctx: RealtimeContext) => Promise; + +export type IInboundDescriptor = { + readonly descriptor: RealtimeChannelDescriptor; + readonly handler: IRealtimeHandler>; +}; +``` + +```ts +// packages/core-realtime/src/realtime-authenticator.interface.ts +export interface IRealtimeAuthenticator { + authenticate(handshake: { + cookies: Record; + headers: Record; + }): Promise<{ userId: string; roles: string[] } | null>; +} +``` + +```ts +// packages/core-realtime/src/realtime-handler-registry.ts +import type { IInboundDescriptor } from "./realtime-handler.interface"; + +export interface IRealtimeHandlerRegistry { + register(entry: IInboundDescriptor>): void; + getInboundDescriptor(channelName: string): IInboundDescriptor | null; + list(): IInboundDescriptor[]; +} + +export class RealtimeHandlerRegistry implements IRealtimeHandlerRegistry { + // implementation: a Map keyed by descriptor.name; details in the impl plan +} +``` + +```ts +// packages/core-realtime/src/realtime-server.interface.ts +import type { Server as HttpServer } from "node:http"; +import type { IRealtimeAuthenticator } from "./realtime-authenticator.interface"; +import type { IRealtimeHandlerRegistry } from "./realtime-handler-registry"; + +export type IRealtimeServerOptions = { + httpServer: HttpServer; + authenticator: IRealtimeAuthenticator; + registry: IRealtimeHandlerRegistry; +}; + +export interface IRealtimeServer { + start(): Promise; + stop(): Promise; +} +``` + +The bus is generic over the schema's inferred type, so `realtime.broadcast(presenceChannel, payload)` infers `payload`'s type from `presenceChannel.schema` and rejects mismatches at compile time. + +--- + +## 4. Per-feature folder layout + +A feature that uses realtime has these additional directories (all optional; absent if unused): + +``` +packages//src/ +├── realtime/ +│ ├── .channel.ts ← channel descriptor (publisher-side public) +│ └── handlers/ +│ └── on-.handler.ts ← inbound handler (private, never re-exported) +└── ... +``` + +Top-level `realtime/`, **not** under `integrations/`. Rationale: + +- Channel descriptors are descriptor-shaped (`{ name, schema, scope }`) — same shape as event descriptors from ADR-015, which also live at top level (`events/`). +- tRPC's `integrations/api/` is router-shaped (procedures aggregated into a tree, mounted on a wire). Realtime has no router — channels are flat descriptors, the Socket.IO server lives once in `core-realtime`, no per-feature transport code. +- Consistency with `events/` and `jobs/` (ADR-015) outweighs consistency with tRPC's `integrations/api/`. + +The existing `integrations/api/` and `integrations/cms/` patterns are unchanged. + +--- + +## 5. File shapes + +### 5.1 Channel descriptor (publisher-side) + +```ts +// packages/blog/src/realtime/article-feed.channel.ts +import { z } from "zod"; +import { defineRealtimeChannel } from "@repo/core-realtime"; + +export const articleFeedSchema = z.object({ + id: z.string(), + slug: z.string(), + title: z.string(), + publishedAt: z.string().datetime(), +}).strict(); + +export type ArticleFeedPayload = z.infer; + +export const articleFeedChannel = defineRealtimeChannel( + "blog.article.feed", + articleFeedSchema, + { scope: "public" }, +); +``` + +Public re-export through the feature's root barrel (anchor-injected by the generator): + +```ts +// packages/blog/src/index.ts +// +export { + articleFeedChannel, + articleFeedSchema, + type ArticleFeedPayload, +} from "./realtime/article-feed.channel"; +``` + +### 5.2 Direct broadcast (publisher use case) + +```ts +// packages/blog/src/application/use-cases/publish-article.use-case.ts +import type { IRealtimeBroadcaster } from "@repo/core-realtime"; +import { articleFeedChannel } from "../../realtime/article-feed.channel"; + +export const publishArticleUseCase = + ( + articles: IArticlesRepository, + bus: IEventBus, + realtime: IRealtimeBroadcaster, + ) => + async (input: PublishArticleInput): Promise => { + const article = await articles.publish(input.id); + await bus.publish(articlePublishedEvent, article); // bus (durable consumers) + await realtime.broadcast(articleFeedChannel, article); // direct broadcaster + return publishArticleOutputSchema.parse(article); + }; +``` + +`bus` and `realtime` are both factory deps. Hybrid pattern: bus handles durable consumers (notification job, search index, audit log); broadcaster handles the live UI fan-out. + +### 5.3 Inbound handler (consumer-side) + +```ts +// packages/blog/src/realtime/handlers/on-presence-ping.handler.ts +import type { PresencePingPayload } from "../presence-ping.channel"; +import type { IPresenceService } from "../../application/services/presence.service.interface"; +import type { RealtimeContext } from "@repo/core-realtime"; + +export type IOnPresencePingHandler = ReturnType; + +export const onPresencePingHandler = + (presence: IPresenceService) => + async (input: PresencePingPayload, ctx: RealtimeContext): Promise => { + if (!ctx.userId) throw new Error("authenticated channel reached non-authed handler"); + await presence.markActive(ctx.userId, input.articleId, input.at); + }; +``` + +Handler factories follow the same shape as use cases (`(deps) => async (input, ctx) => result`). For `scope: "authenticated"` channels, `ctx.userId` is non-null (gate 2 + 3 ensure this); the runtime check is defensive, matching the `xOutputSchema.parse(...)` discipline use cases use. + +### 5.4 Handler binding (consumer's `bind-*`) + +```ts +// packages/blog/src/di/bind-production.ts (excerpt, anchor-injected by generator) +import { presencePingChannel } from "../realtime/presence-ping.channel"; +import { onPresencePingHandler } from "../realtime/handlers/on-presence-ping.handler"; + +export function bindProductionBlog( + config: SanitizedConfig, + tracer: ITracer, + logger: ILogger, + bus: IEventBus, + queue: IJobQueue, + realtime: IRealtimeBroadcaster, + realtimeRegistry: IRealtimeHandlerRegistry, +): void { + // ... existing bindings ... + + // + const wrappedOnPresencePing = withSpan( + tracer, + { name: "blog.onPresencePing", op: "realtime-handler" }, + withCapture( + logger, + { feature: "blog", layer: "realtime-handler", name: "blog.onPresencePing" }, + onPresencePingHandler(presenceService), + ), + ); + realtimeRegistry.register({ + descriptor: presencePingChannel, + handler: wrappedOnPresencePing, + }); +} +``` + +The handler is wrapped in the **same span+capture sandwich** as use cases (R41–R44 from ADR-014), with two new tag values: `op: "realtime-handler"`, `layer: "realtime-handler"`. The `// ` anchor marks where `gen realtime handler` inserts new blocks. + +**Note on binder signature growth:** `bindProductionX` now takes seven args `(config, tracer, logger, bus, queue, realtime, realtimeRegistry)`. This is consistent with how ADR-015 grew the signature from three to five. Future expansion may warrant collapsing into a single `BindContext` parameter; deferred. + +--- + +## 6. Topology: custom Node server + +`apps/web-next/server.ts` replaces `next dev` / `next start` as the boot entry. + +### 6.1 What runs in one process + +``` +apps/web-next/server.ts (Node http server, port 3000) + ├── Next.js handler (page routes, API routes, tRPC) + └── Socket.IO server (mounted on the same http server) + ├── Cookie-auth middleware (gate 1) + ├── Channel registry (subscribe → gate 2) + └── Handler dispatch (inbound message → gate 3) +``` + +Single process, single port. The Vercel adapter for Next.js serverless is no longer used for `web-next`; deployment is now any Node host (Render / Railway / Fly / self-hosted). `cms` and `web-tanstack` are unchanged — they can stay on their existing runtimes until they need realtime. + +### 6.2 Boot sequence + +```ts +// apps/web-next/server.ts (sketch) +import { createServer } from "node:http"; +import next from "next"; +import { Server as SocketIOServer } from "socket.io"; +import { SocketIORealtimeServer, SocketIORealtimeBroadcaster, RealtimeHandlerRegistry } from "@repo/core-realtime"; +import { bindAll } from "./src/server/bind-production"; + +const app = next({ dev: process.env.NODE_ENV !== "production" }); +await app.prepare(); + +const httpServer = createServer((req, res) => app.getRequestHandler()(req, res)); +const io = new SocketIOServer(httpServer, { /* CORS etc */ }); + +const broadcaster = new SocketIORealtimeBroadcaster(io); +const registry = new RealtimeHandlerRegistry(); + +await bindAll({ realtime: broadcaster, realtimeRegistry: registry }); +// ^ bindAll passes broadcaster + registry into every per-feature binder + +const realtimeServer = new SocketIORealtimeServer({ + io, + authenticator: realtimeAuthenticatorImpl(), // delegates to auth feature + registry, +}); +await realtimeServer.start(); + +httpServer.listen(3000); +``` + +`bindAll` gains two new resolution steps mirroring ADR-015's `resolveEventsAndJobs*`: + +- `resolveRealtime()` — picks broadcaster impl by env (in-memory in tests, Socket.IO otherwise). Idempotent + cached per process. +- `bindRealtimeBridge(bus, broadcaster, allowlist)` — explicit allowlist of bus events forwarded onto realtime channels. Empty in v1 (dashboard PR adds the first entries). + +--- + +## 7. Auth + authorization (the four checkpoints) + +### 7.1 Lifecycle gates + +| Gate | When | What it does | +|---|---|---| +| 1 | Connect | Read cookie, validate session via `IRealtimeAuthenticator`, attach `{ userId, roles } | null` to `socket.data.user`. | +| 2 | Subscribe | Match `requestedName` against registered descriptors (template-aware), apply `authorize(descriptor, params, user)`, on success `socket.join("ch:")`. | +| 3 | Inbound message | Validate via Zod, re-apply `authorize` (defense in depth), invoke wrapped handler with `ctx`. | +| 4 | Broadcast | **No gate.** `io.to("ch:").emit(...)` fans out to whoever cleared gate 2. Subscribe is the single source of truth. | + +### 7.2 Authenticator implementation lives at the app layer + +`core-realtime` defines `IRealtimeAuthenticator`. The implementation lives in `apps/web-next/server.ts` and delegates to the auth feature's `IAuthenticationService.validateSession()`: + +```ts +const authenticator: IRealtimeAuthenticator = { + authenticate: async ({ cookies }) => { + const sessionId = cookies[SESSION_COOKIE]; + if (!sessionId) return null; + const authService = authContainer.get( + AUTH_SYMBOLS.IAuthenticationService, + ); + const session = await authService.validateSession(sessionId); + return session + ? { userId: session.userId, roles: session.roles ?? [] } + : null; + }, +}; +``` + +`core-realtime` never imports the auth feature. The boundary stays clean (feature → core only); the app composes them. + +### 7.3 The `authorize` function + +```ts +// packages/core-realtime/src/authorize.ts +import type { ChannelScope, RealtimeChannelDescriptor } from "./realtime-channel"; + +export async function authorize( + descriptor: RealtimeChannelDescriptor, + params: Record, + user: { userId: string; roles: string[] } | null, +): Promise { + 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; +} +``` + +Pure function, no DB hit, no side effects. Each gate-2 / gate-3 invocation is constant-time. + +### 7.4 Channel-template matching + +Plain channels (`"blog.article.feed"`) match by exact equality. Templated channels (`"notifications.user.{userId}"`) match by parameterized lookup: + +- Client subscribes to `"notifications.user.user_42"`. +- Server iterates registered descriptors; finds the one whose template parses against the requested name. +- Yields `params = { userId: "user_42" }`. +- `authorize` reads `params.userId` for the `userScoped` scope check. + +--- + +## 8. Bus ↔ realtime — the hybrid pattern + +The repo already has `IEventBus` (ADR-015). Realtime is layered alongside it, not on top of it: + +- **Direct broadcast** (most cases): feature use case takes `realtime: IRealtimeBroadcaster` and calls `realtime.broadcast(channel, payload)`. The broadcaster has no knowledge of the bus. +- **Bridge**: a single `bindRealtimeBridge(bus, broadcaster, allowlist)` step subscribes to allowlisted bus events and forwards them onto realtime channels. Lives in `apps/web-next/src/server/bind-production.ts` next to the existing `bindAll` flow. + +### 8.1 Why both, not one + +| Use case | Bus only | Realtime only | Both (hybrid) | +|---|---|---|---| +| Welcome email after signup | ✓ — durable, retryable | ✗ — clients shouldn't trigger emails | | +| Article published → search index | ✓ | | | +| Live cursor positions in editor | ✗ — too high-frequency | ✓ | | +| Article published → live feed update | | | ✓ — durable consumers AND realtime fan-out | +| Comment posted → other readers see it | | | ✓ — moderation/notifications via bus, fan-out via bridge | + +A single API would force one of the wrong shapes onto the other use case. Two APIs, mutually independent, cover the full matrix. + +### 8.2 Bridge wiring + +```ts +// apps/web-next/src/server/bind-production.ts +function bindRealtimeBridge(bus: IEventBus, broadcaster: IRealtimeBroadcaster): void { + // v1 ships with empty allowlist. First entries land in the dashboard PR: + // bus.subscribe(userSignedUpEvent, "realtime-bridge", payload => + // broadcaster.broadcast(adminEventStreamChannel, { kind: "user.signed-up", payload })); +} +``` + +Each bridge entry is explicit, by design. There is no auto-bridge: bus events have no auth posture, realtime channels do; the mapping is where you assign it. + +--- + +## 9. Boundary rules and ESLint + +Three rule additions: + +- **`no-direct-socket-io`** — blocks `import "socket.io"` and `import "socket.io-client"` outside `packages/core-realtime/src/socket-io-*.ts` and `apps/*/server.ts`. Parallel to `no-direct-payload-jobs` (ADR-015). +- **`no-realtime-handler-reexport`** — blocks any export from `**/realtime/handlers/**` outside that file's own feature `bind-*`. Parallel to `no-handler-reexport` (ADR-015 R-E1). +- **Boundary tag for `core-realtime`**: `["core"]` in `packages/core-realtime/turbo.json`. The `feature` allowlist already includes `["core", "feature", "tooling"]` post-ADR-015; no further change. + +The `IRealtimeAuthenticator` impl in `apps/web-next/server.ts` imports from the `auth` feature. This crosses no boundary (apps are tagged `app` and may import from `core | core-composition | feature | tooling`). + +--- + +## 10. Testing + +Three layers, mirroring the existing patterns. + +### 10.1 Unit tests + +Feature handlers tested via direct factory injection — no realtime server, no Socket.IO, no sockets: + +```ts +const presence = new MockPresenceService(); +const handler = onPresencePingHandler(presence); +await handler( + { articleId: "a1", at: "2026-05-08T12:00:00.000Z" }, + { userId: "user_1", roles: [] }, +); +expect(presence.markActiveCalls).toHaveLength(1); +``` + +Use cases that broadcast inject `RecordingRealtimeBroadcaster` (§10.2) and assert against the recorded broadcasts. + +### 10.2 `RecordingRealtimeBroadcaster` (in `core-testing/instrumentation/`) + +Same shape as `RecordingEventBus`, same local-type-alias pattern (the cycle lesson from ADR-015 — core-testing must not import core-realtime as a runtime dep, or the typecheck graph cycles): + +```ts +// packages/core-testing/src/instrumentation/recording-realtime-broadcaster.ts +import type { z } from "zod"; + +// Local type aliases (mirror @repo/core-realtime's contracts). +type RealtimeChannelDescriptor = { + readonly name: TName; + readonly schema: TSchema; +}; + +interface IRealtimeBroadcaster { + broadcast( + descriptor: RealtimeChannelDescriptor>, + payload: T, + ): Promise; +} + +export class RecordingRealtimeBroadcaster implements IRealtimeBroadcaster { + readonly broadcasts: { channel: string; payload: unknown }[] = []; + async broadcast(descriptor, payload: T): Promise { + descriptor.schema.parse(payload); + this.broadcasts.push({ channel: descriptor.name, payload }); + } +} +``` + +### 10.3 Integration test: `realtime-ping` + +The proof-of-life. Brings up the Socket.IO server in-process, uses `socket.io-client` to connect with a seeded session cookie, exercises connect → subscribe → emit → broadcast end-to-end: + +```ts +// apps/web-next/src/__tests__/realtime-ping.test.ts (sketch) +describe("e2e: realtime-ping exercises all four checkpoints", () => { + it("authenticated client emits ping, receives pong on server-broadcast pong channel", async () => { + await bindAllDevSeed({ ... }); + const httpServer = createServer(); + const realtime = await startRealtimeServer({ httpServer, authenticator, registry }); + httpServer.listen(0); + const port = (httpServer.address() as AddressInfo).port; + + const sessionId = await seedTestSession(); // dev-seed Auth feature + const client = io(`http://localhost:${port}`, { + extraHeaders: { Cookie: `${SESSION_COOKIE}=${sessionId}` }, + }); + await once(client, "connect"); + + const pongs: unknown[] = []; + client.on("realtime.pong", (p) => pongs.push(p)); + await emitWithAck(client, "subscribe", "realtime.pong"); + + await emitWithAck(client, "realtime.ping", { at: new Date().toISOString() }); + await once(client, "realtime.pong"); + + expect(pongs).toHaveLength(1); + }); +}); +``` + +The `realtime-ping` channel pair lives inside `core-realtime` itself — not in any feature, not user-visible. Stripped or env-gated in production. + +### 10.4 Out of v1 scope + +- Load testing with N concurrent sockets +- Multi-instance fanout (sticky sessions, Redis adapter) +- Replay tests (pre-recorded socket sessions replayed against the server) + +--- + +## 11. Generators + +Two new generators, three new anchors per feature. Same protocol as ADR-015. + +### 11.1 New anchors (added to all five existing features + `feature` template) + +| File | Anchor | Used by | +|---|---|---| +| `src/index.ts` | `// ` | `gen realtime channel` | +| `src/di/symbols.ts` | `// ` | `gen realtime handler` | +| `src/di/bind-production.ts` | `// ` | `gen realtime handler` | +| `src/di/bind-dev-seed.ts` | `// ` | `gen realtime handler` | + +The CI guard at `packages/core-eslint/anchors.test.js` extends to assert these anchors stay present in every feature. + +### 11.2 `pnpm turbo gen realtime channel` + +```bash +pnpm turbo gen realtime --args channel +``` + +- Adds: `packages//src/realtime/.channel.ts` + test +- Modifies: `src/index.ts` at `` to re-export the descriptor + +The `` arg accepts: `public | authenticated | role: | user-scoped`. + +### 11.3 `pnpm turbo gen realtime handler` + +```bash +pnpm turbo gen realtime --args handler +``` + +- Adds: `packages//src/realtime/handlers/on-.handler.ts` + test +- Modifies: `src/di/symbols.ts` at `` (handler symbol) +- Modifies: both `bind-production.ts` and `bind-dev-seed.ts` at `` (registers wrapped handler with the registry) + +### 11.4 Notably absent + +- **No generator for outbound broadcasts.** A use case that wants to broadcast adds `realtime: IRealtimeBroadcaster` to its factory deps and calls `realtime.broadcast(channel, payload)`. Pattern is identical to adding `bus: IEventBus` — manual edit, no scaffold. +- **No generator for bridge entries.** Each bridge mapping is an explicit, hand-edited line in `apps/web-next/src/server/bind-production.ts`. There aren't enough of these for a generator to pay off. +- **No generator for new core packages** (the `core-realtime` package itself was hand-built). The user has flagged a future capability for this — see follow-up #2. + +--- + +## 12. Proof-of-life: `realtime-ping` + +A minimal channel-pair built into `core-realtime` itself, exercised by the integration test in §10.3. Not exposed through any feature; not user-visible. Validates that the four checkpoints + the cookie-auth path all integrate cleanly. + +```ts +// packages/core-realtime/src/realtime-ping.ts (or equivalent) +const pingSchema = z.object({ at: z.string().datetime() }).strict(); +const pongSchema = z.object({ at: z.string().datetime(), echo: z.string() }).strict(); + +export const realtimePingChannel = defineRealtimeChannel( + "realtime.ping", + pingSchema, + { scope: "authenticated" }, +); + +export const realtimePongChannel = defineRealtimeChannel( + "realtime.pong", + pongSchema, + { scope: "authenticated" }, +); + +export const realtimePingHandler = + (broadcaster: IRealtimeBroadcaster) => + async (input: PingPayload, ctx: RealtimeContext): Promise => { + await broadcaster.broadcast(realtimePongChannel, { + at: input.at, + echo: ctx.userId ?? "anonymous", + }); + }; +``` + +`SocketIORealtimeServer.start()` registers the ping handler unconditionally in v1 (env-gated `REALTIME_PING_DISABLED=true` to strip in production once the dashboard or another consumer ships). + +--- + +## 13. v1 scope vs deferred + +### In v1 + +- `@repo/core-realtime` package (interfaces, `SocketIORealtimeServer`, `SocketIORealtimeBroadcaster`, `InMemoryRealtimeBroadcaster`, `RealtimeHandlerRegistry`) +- Custom Node server `apps/web-next/server.ts` replacing `next start` / `next dev` +- `bindAll` extension: `resolveRealtime()` + `bindRealtimeBridge(bus, broadcaster, allowlist)` (allowlist empty) +- Cookie-session auth via `IRealtimeAuthenticator` +- Four scope kinds: `"public"`, `"authenticated"`, `{ role }`, `{ userScoped }` +- Two generators: `gen realtime channel`, `gen realtime handler` +- Three new anchors per feature + CI guard extension +- `RecordingRealtimeBroadcaster` test helper +- `realtime-ping` channel + `apps/web-next/src/__tests__/realtime-ping.test.ts` integration smoke +- ADR-016 documenting the design +- ESLint rules: `no-direct-socket-io`, `no-realtime-handler-reexport` + +### Deferred + +- Live observability dashboard + `admin-realtime` feature +- Concrete bridge consumers in feature packages (first one lands with the dashboard PR) +- Custom Node server for `cms` and `web-tanstack` apps +- DB-backed roles / permissions (see §14) +- `{ permission }` and `{ check: (user, params) => bool }` scope kinds (see §14) +- Production-mode e2e test with multiple connected sockets +- Multi-instance fanout (Redis adapter / sticky sessions) +- Generator for new core packages (separate brainstorm; see follow-up #2) + +--- + +## 14. Future evolution: DB-backed roles + permissions + +### 14.1 The migration is non-breaking + +The design is structured so that introducing a roles/permissions DB layer doesn't require any change to `core-realtime` or to the realtime layer itself. The seam is `IRealtimeAuthenticator`: + +- Today: `authenticate()` returns `{ userId, roles: [] }` (empty array, no role system yet) +- After the DB layer ships: `authenticate()` returns `{ userId, roles: ["admin", "editor"], permissions: [...] }` from a role + permissions lookup + +The realtime server reads whatever the auth feature decides to expose. Same pattern as `IUsersRepository` — the data source is irrelevant to the consumer. + +### 14.2 What lands when the DB layer ships + +In the `auth` feature: + +``` +packages/auth/src/ +├── application/ +│ ├── repositories/ +│ │ ├── users.repository.interface.ts existing +│ │ └── roles.repository.interface.ts NEW +│ └── services/ +│ ├── authentication.service.interface.ts existing +│ └── authorization.service.interface.ts NEW (hasPermission, getRoles) +├── entities/models/ +│ ├── user.ts existing +│ ├── role.ts NEW { name, permissions: string[] } +│ └── permission.ts NEW (typed string union) +├── infrastructure/repositories/ +│ ├── users.repository.ts existing +│ └── roles.repository.ts NEW (Payload-backed) +└── integrations/cms/collections/ + └── roles.ts NEW (Payload Roles collection) +``` + +The app-level `IRealtimeAuthenticator` impl widens to also load roles and permissions: + +```ts +authenticate: async ({ cookies }) => { + const session = await authService.validateSession(cookies[SESSION_COOKIE]); + if (!session) return null; + const roles = await authzService.getRolesForUser(session.userId); + return { + userId: session.userId, + roles: roles.map(r => r.name), + // future: also returning permissions for { permission } scope + }; +}; +``` + +Two DB hits at connect (validate session + load roles), then nothing per-message — the snapshot lives on `socket.data.user`. + +### 14.3 Scope union extends additively + +```ts +// future shape — additive, no breaking change to existing channels +type ChannelScope = + | "public" + | "authenticated" + | { role: string } // unchanged + | { permission: string } // NEW + | { permissions: string[]; mode: "all" | "any" } // NEW + | { userScoped: true; template: string } // unchanged + | { check: (user, params) => Promise }; // NEW (escape hatch) +``` + +`authorize()` gains one branch per new scope kind. Already-defined channels keep working unchanged. + +### 14.4 Staleness and the refresh path + +The connect-time snapshot is stale by definition: a user demoted while connected keeps the old roles until reconnect. For most apps this is acceptable. If it isn't, the existing event bus is the escape hatch: + +- Auth feature publishes `auth.user.role-changed` on the bus when roles mutate +- A small `RealtimeAuthRefresher` (lives next to `bindRealtimeBridge` in the app layer) finds connected sockets for that user and re-fetches their roles +- No realtime-protocol change, no DB hit per message — just an event-driven invalidation triggered by the bus + +This falls out for free from already having ADR-015's bus + the `userScoped` channel concept. + +### 14.5 What does NOT change + +- `core-realtime` package — zero edits +- Generators, anchors, CI guard — zero edits +- Existing channels with `{ role: "admin" }` — keep working + +--- + +## 15. Open follow-ups (out of v1 plan scope) + +Tracked here so they aren't forgotten: + +1. **Production-mode e2e test.** §10.3 runs in dev-seed mode (`InMemoryEventBus` + in-process Socket.IO + dev-seeded session). A parallel test against a real Postgres-backed Payload would prove the cookie-auth path more thoroughly. Out of scope for v1 because it requires a Payload test fixture. + +2. **Generic core-package generator.** The user has flagged this as a future capability: a `turbo gen core-package` with template variants (interface-and-adapters / utility / policy-only) so future core packages don't need to be hand-built. Saved as a project memory; brainstorm + spec separately after this lands. + +3. **Bridge wildcard subscription.** v1 requires explicit `bus.subscribe(eventX, ...)` per bridged event. A wildcard `bus.subscribe("*", ...)` API on the bus would let the bridge forward every event without enumeration. Useful for the dashboard. Out of v1 because the bus doesn't expose a wildcard today; the dashboard PR would add it if needed. + +4. **Multi-instance fanout.** When `apps/web-next` scales horizontally, broadcasts in one process don't reach sockets connected to another. Solved by Socket.IO's Redis adapter (sticky sessions for connection affinity, Redis pub/sub for cross-instance broadcast). Out of v1 because v1 ships single-instance. + +5. **Generator for outbound broadcasts.** Currently a manual edit. If the pattern proliferates, a `gen realtime broadcast ` could automate the use-case dep + call-site addition. Out of v1 because the broadcast call itself is one line. + +--- + +## 16. Self-review check + +This spec was self-reviewed against the brainstorming dialogue on the date of writing. Key points verified: + +- Every conversational decision (motivation, topology, framework choice, bus integration model, direction in v1, auth model, folder location, dashboard deferral, proof-of-life replacement) maps to a section above. +- All file paths are repo-relative; no "TBD" or placeholder text. +- The interface signatures in §3.3 are internally consistent with the file shapes in §5 and the `authorize` function in §7.3. +- Tests precede implementation per the TDD foundation (ADR-011); §10.3's integration test is the proof-of-life, written before the production code. +- v1 scope (§13) excludes everything the dialogue deferred; nothing shipped without a clear consumer or test exercise. +- The future-evolution section (§14) shows the DB-backed roles/permissions migration is non-breaking, addressing the user's question on 2026-05-08. +- ESLint boundary additions (§9) parallel ADR-015 patterns exactly.