agentic-dev/docs/guides/realtime.md

# Realtime

> **Prerequisite:** This guide assumes `@repo/core-realtime` is present. If you started from the slim template, run `pnpm turbo gen core-package realtime` first.

Walkthrough for adding Socket.IO realtime channels, broadcasts, and inbound handlers to a feature. For the architectural rationale, see [ADR-016](../decisions/adr-016-realtime-layer.md).

The three rules to keep in mind:

- **R0** — Realtime is for state delivery, not for replacing tRPC. Persistent operations with request/response semantics belong on tRPC procedures.
- **R1** — Channel descriptors are exported; handlers are private (never re-exported, ESLint-enforced via `no-realtime-handler-reexport`).
- **R2** — `socket.io` lives in one package only. Feature packages MUST NOT `import "socket.io"` or `import "socket.io-client"`. Only `packages/core-realtime/src/socket-io-*.ts` and `apps/*/server.ts` are allowed.

Two generators do the boilerplate. Each inserts at fixed `// <gen:*>` anchor comments that are present in every feature.

```bash
pnpm turbo gen realtime channel    # channel descriptor
pnpm turbo gen realtime handler    # inbound realtime handler
```

The bus-bridge pattern from [ADR-015](../decisions/adr-015-events-and-jobs.md) is also available: `bindRealtimeBridge(bus, broadcaster, allowlist)` in `apps/web-next/src/server/bind-production.ts` forwards allowlisted bus events onto realtime channels. The allowlist ships empty in v1 — see the spec §8 for the full hybrid pattern.

---

## 1. Declare a channel

Run from the repo root:

```bash
pnpm turbo gen realtime --args channel blog article-feed public
```

The fourth argument is the channel scope. Valid values:

| Scope arg       | What it means                                                                    |
| --------------- | -------------------------------------------------------------------------------- |
| `public`        | Any connected socket may subscribe (no auth required)                            |
| `authenticated` | Socket must have a valid session (gate 1 cleared)                                |
| `role:<name>`   | Socket must have the named role in `roles[]`                                     |
| `user-scoped`   | Socket must match `params.userId === socket.data.user.userId` (template channel) |

This scaffolds:

- `packages/blog/src/realtime/article-feed.channel.ts` (the descriptor — `defineRealtimeChannel` + Zod schema)
- `packages/blog/src/realtime/article-feed.channel.test.ts`
- A re-export at the `// <gen:realtime-channels>` anchor in `packages/blog/src/index.ts`

Then **fill in the schema** with the fields the channel's payload carries:

```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" },
);
```

The generator wires the re-export automatically — verify:

```bash
pnpm --filter @repo/blog lint typecheck
```

### User-scoped channels

For per-user delivery, use `user-scoped` scope and a template channel name:

```ts
export const userNotificationsChannel = defineRealtimeChannel(
  "notifications.user.{userId}",
  notificationsSchema,
  { scope: { userScoped: true, template: "notifications.user.{userId}" } },
);
```

Clients subscribe to `"notifications.user.user_42"`. The server matches the template, extracts `params = { userId: "user_42" }`, and `authorize` checks `params.userId === socket.data.user.userId`. Only the matching user's socket clears gate 2.

---

## 2. Broadcast from a use case

Direct broadcast is the primary path — a use case adds `realtime: IRealtimeBroadcaster` to its factory signature and calls `realtime.broadcast(channel, payload)` after the success path.

```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, realtime: IRealtimeBroadcaster) =>
  async (input: PublishArticleInput): Promise<PublishArticleOutput> => {
    const article = await articles.publish(input.id);
    await realtime.broadcast(articleFeedChannel, {
      id: article.id,
      slug: article.slug,
      title: article.title,
      publishedAt: article.publishedAt,
    });
    return publishArticleOutputSchema.parse(article);
  };
```

`realtime.broadcast` is type-safe — TypeScript infers the payload type from `articleFeedChannel.schema` and rejects a mismatched object at compile time.

**Update DI.** Both `bind-production.ts` and `bind-dev-seed.ts` receive `realtime` as a parameter. Thread it into the factory call:

```ts
// packages/blog/src/di/bind-production.ts
export function bindProductionBlog(
  config: SanitizedConfig,
  tracer: ITracer,
  logger: ILogger,
  bus: IEventBus,
  queue: IJobQueue,
  realtime: IRealtimeBroadcaster,
  realtimeRegistry: IRealtimeHandlerRegistry,
): void {
  // ... existing bindings ...

  const wrappedPublishArticle = withSpan(
    tracer,
    { name: "blog.publishArticle", op: "use-case" },
    withCapture(
      logger,
      { feature: "blog", layer: "use-case", name: "blog.publishArticle" },
      publishArticleUseCase(articlesRepo, realtime),
    ),
  );
  blogContainer
    .bind<IPublishArticleUseCase>(BLOG_SYMBOLS.IPublishArticleUseCase)
    .toConstantValue(wrappedPublishArticle);
}
```

**Testing.** Inject `RecordingRealtimeBroadcaster` from `@repo/core-testing/instrumentation` and assert against its `broadcasts` array:

```ts
import { RecordingRealtimeBroadcaster } from "@repo/core-testing/instrumentation";

it("broadcasts the article after publish", async () => {
  const articles = new MockArticlesRepository([mockArticle]);
  const realtime = new RecordingRealtimeBroadcaster();

  const useCase = publishArticleUseCase(articles, realtime);
  await useCase({ id: "article-1" });

  expect(realtime.broadcasts).toHaveLength(1);
  expect(realtime.broadcasts[0]).toMatchObject({
    channel: "blog.article.feed",
    payload: { id: "article-1" },
  });
});
```

**Verify:**

```bash
pnpm --filter @repo/blog lint typecheck test
```

---

## 3. Receive client messages

Use an inbound handler when a connected client emits a message that the server should react to — presence pings, cursor positions, votes, ephemeral state that shouldn't go through tRPC.

Run from the repo root:

```bash
pnpm turbo gen realtime --args handler blog article-feed
```

This scaffolds:

- `packages/blog/src/realtime/handlers/on-article-feed.handler.ts` + test
- A symbol at `// <gen:realtime-handler-symbols>` in `packages/blog/src/di/symbols.ts`
- A wrapped registration block at `// <gen:realtime-handlers>` in both `bind-production.ts` and `bind-dev-seed.ts`

The generator prints two manual steps.

**1. Add the imports** to the top of both bind files (the modify-block can't add imports):

```ts
import { articleFeedChannel } from "../realtime/article-feed.channel";
import { onArticleFeedHandler } from "../realtime/handlers/on-article-feed.handler";
```

**2. Fill in the handler body.** The generated handler factory is `(deps) => async (input, ctx) => Promise<void>`. Inject what the handler needs and implement the body:

```ts
// packages/blog/src/realtime/handlers/on-article-feed.handler.ts
import type { ArticleFeedPayload } from "../article-feed.channel";
import type { IPresenceService } from "../../application/services/presence.service.interface";
import type { RealtimeContext } from "@repo/core-realtime";

export type IOnArticleFeedHandler = ReturnType<typeof onArticleFeedHandler>;

export const onArticleFeedHandler =
  (presence: IPresenceService) =>
  async (input: ArticleFeedPayload, ctx: RealtimeContext): Promise<void> => {
    if (!ctx.userId) return;
    await presence.markViewing(ctx.userId, input.id);
  };
```

**3. Wire the dependency.** The generated bind-block emits `onArticleFeedHandler()` with no args. Edit it to pass the service:

```ts
// packages/blog/src/di/bind-production.ts (excerpt at <gen:realtime-handlers>)
const wrappedOnArticleFeed = withSpan(
  tracer,
  { name: "blog.onArticleFeed", op: "realtime-handler" },
  withCapture(
    logger,
    { feature: "blog", layer: "realtime-handler", name: "blog.onArticleFeed" },
    onArticleFeedHandler(presenceService),
  ),
);
realtimeRegistry.register({
  descriptor: articleFeedChannel,
  handler: wrappedOnArticleFeed,
});
```

The same block is generated in `bind-dev-seed.ts`. Edit both.

**Testing.** Unit-test the handler factory by injecting mocks directly — no server, no sockets:

```ts
it("marks the user as viewing the article", async () => {
  const presence = new MockPresenceService();
  const handler = onArticleFeedHandler(presence);

  await handler(
    {
      id: "article-1",
      slug: "hello",
      title: "Hello",
      publishedAt: "2026-05-08T00:00:00.000Z",
    },
    { userId: "user_1", roles: [] },
  );

  expect(presence.markViewingCalls).toHaveLength(1);
  expect(presence.markViewingCalls[0]).toMatchObject({
    userId: "user_1",
    articleId: "article-1",
  });
});
```

Handlers MUST NOT be re-exported from the feature's public surface (enforced by `core-eslint`'s `no-realtime-handler-reexport` rule). The bind files wire handlers internally; `src/index.ts` must never re-export from `realtime/handlers/`.

**Verify:**

```bash
pnpm --filter @repo/blog lint typecheck test
```

---

## Anchor protocol

Three fixed anchor comments live in every feature for realtime:

| 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` asserts these stay present in every feature. Remove one and CI fails — restore it and the test goes green.

The six ADR-015 anchors (`<gen:events>`, `<gen:event-handler-symbols>`, `<gen:job-symbols>`, `<gen:event-handlers>` in both binders, `<gen:jobs>` in both binders) continue to exist alongside them. Each anchor is independent.

## Integration test reference

`apps/web-next/src/__tests__/realtime-ping.test.ts` exercises gates 1 + 2 over a real Socket.IO connection: build broadcaster + registry + stub authenticator → start Socket.IO server in-process → connect with a stub session cookie → subscribe → emit ping → receive pong. Gates 3 + 4 are covered in `packages/core-realtime/src/socket-io-realtime-server.test.ts` (inbound rejection of unknown channels and forbidden scope). Use this test as a template when adding cross-feature realtime flows.

## Related

- [ADR-016](../decisions/adr-016-realtime-layer.md) — design decision record (full design including topology, auth gates, and v1 scope)
- [ADR-015](../decisions/adr-015-events-and-jobs.md) — cross-feature events and background jobs (bus-bridge pattern)