Danijel Martinek
500d835f0e
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.
880 lines
38 KiB
Markdown
880 lines
38 KiB
Markdown
# 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/<name>.channel.ts` is re-exported from the package root barrel. A feature's `realtime/handlers/<name>.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:<name>").emit(name, payload)
|
||
▼
|
||
Socket.IO room "ch:<name>"
|
||
│
|
||
▼
|
||
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<T> + 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<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 };
|
||
}
|
||
```
|
||
|
||
```ts
|
||
// packages/core-realtime/src/realtime-broadcaster.interface.ts
|
||
import type { z } from "zod";
|
||
import type { RealtimeChannelDescriptor } from "./realtime-channel";
|
||
|
||
export interface IRealtimeBroadcaster {
|
||
broadcast<T>(
|
||
descriptor: RealtimeChannelDescriptor<string, z.ZodType<T>>,
|
||
payload: T,
|
||
): Promise<void>;
|
||
}
|
||
```
|
||
|
||
```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<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>>;
|
||
};
|
||
```
|
||
|
||
```ts
|
||
// packages/core-realtime/src/realtime-authenticator.interface.ts
|
||
export interface IRealtimeAuthenticator {
|
||
authenticate(handshake: {
|
||
cookies: Record<string, string>;
|
||
headers: Record<string, string>;
|
||
}): 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<T>(entry: IInboundDescriptor<string, z.ZodType<T>>): void;
|
||
getInboundDescriptor(channelName: string): IInboundDescriptor<string, z.ZodType> | null;
|
||
list(): IInboundDescriptor<string, z.ZodType>[];
|
||
}
|
||
|
||
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<void>;
|
||
stop(): Promise<void>;
|
||
}
|
||
```
|
||
|
||
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/<feature>/src/
|
||
├── realtime/
|
||
│ ├── <name>.channel.ts ← channel descriptor (publisher-side public)
|
||
│ └── handlers/
|
||
│ └── on-<name>.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<typeof articleFeedSchema>;
|
||
|
||
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
|
||
// <gen:realtime-channels>
|
||
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<PublishArticleOutput> => {
|
||
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<typeof onPresencePingHandler>;
|
||
|
||
export const onPresencePingHandler =
|
||
(presence: IPresenceService) =>
|
||
async (input: PresencePingPayload, ctx: RealtimeContext): Promise<void> => {
|
||
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 ...
|
||
|
||
// <gen:realtime-handlers>
|
||
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 `// <gen:realtime-handlers>` 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:<name>")`. |
|
||
| 3 | Inbound message | Validate via Zod, re-apply `authorize` (defense in depth), invoke wrapped handler with `ctx`. |
|
||
| 4 | Broadcast | **No gate.** `io.to("ch:<name>").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<IAuthenticationService>(
|
||
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<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;
|
||
}
|
||
```
|
||
|
||
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<TName extends string, TSchema extends z.ZodType> = {
|
||
readonly name: TName;
|
||
readonly schema: TSchema;
|
||
};
|
||
|
||
interface IRealtimeBroadcaster {
|
||
broadcast<T>(
|
||
descriptor: RealtimeChannelDescriptor<string, z.ZodType<T>>,
|
||
payload: T,
|
||
): Promise<void>;
|
||
}
|
||
|
||
export class RecordingRealtimeBroadcaster implements IRealtimeBroadcaster {
|
||
readonly broadcasts: { channel: string; payload: unknown }[] = [];
|
||
async broadcast<T>(descriptor, payload: T): Promise<void> {
|
||
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-channels>` | `gen realtime channel` |
|
||
| `src/di/symbols.ts` | `// <gen:realtime-handler-symbols>` | `gen realtime handler` |
|
||
| `src/di/bind-production.ts` | `// <gen:realtime-handlers>` | `gen realtime handler` |
|
||
| `src/di/bind-dev-seed.ts` | `// <gen:realtime-handlers>` | `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 <feature> <slug> <scope>
|
||
```
|
||
|
||
- Adds: `packages/<feature>/src/realtime/<slug>.channel.ts` + test
|
||
- Modifies: `src/index.ts` at `<gen:realtime-channels>` to re-export the descriptor
|
||
|
||
The `<scope>` arg accepts: `public | authenticated | role:<name> | user-scoped`.
|
||
|
||
### 11.3 `pnpm turbo gen realtime handler`
|
||
|
||
```bash
|
||
pnpm turbo gen realtime --args handler <feature> <channel-slug>
|
||
```
|
||
|
||
- Adds: `packages/<feature>/src/realtime/handlers/on-<channel>.handler.ts` + test
|
||
- Modifies: `src/di/symbols.ts` at `<gen:realtime-handler-symbols>` (handler symbol)
|
||
- Modifies: both `bind-production.ts` and `bind-dev-seed.ts` at `<gen:realtime-handlers>` (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<void> => {
|
||
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<boolean> }; // 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 <feature> <channel-slug>` 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.
|