index.ts · Source

index.tsapps/realtime/src/index.tsraw ← src11,643 B← Back to the app
1/**
2 * Entry point for the realtime server (apps/realtime) — TASK-0091.
3 *
4 * Wires configuration, logging, and the production seams (Prisma
5 * persistence + permission/meta resolvers, Redis update log) into the
6 * server built in `server.ts`, and manages the process lifecycle:
7 * startup, SIGTERM/SIGINT → the R6 drain (4010 + retryAfterMs to all
8 * sockets, flush all dirty rooms, ≤ 10 s, exit 0).
9 *
10 * Dev mode without the sidecars: `REALTIME_REDIS=0` / `REALTIME_DB=0`
11 * (config.ts) fall back to the in-memory update log + doc store.
12 *
13 * The production seam modules (`@prisma/client`, `ioredis`) load lazily
14 * via DYNAMIC imports — this is an ESM package, and a top-level
15 * `require()` crashed the process on boot in every mode (recorded
16 * TASK-0091 fix). A missing/unbuildable client degrades to the
17 * in-memory seam (dev mode) with a warning; any other seam error fails
18 * the boot loudly.
19 */
20 
21import pino from "pino";
22import { loadConfig, type RealtimeConfig } from "./config.js";
23import {
24  createRealtimeServer,
25  type AuditSink,
26  type EventSink,
27  type PermissionResolver,
28  type RoomMetaResolver,
29} from "./server.js";
30import { InMemoryDocStore, type DocStore } from "./rooms.js";
31import {
32  InMemoryUpdateLog,
33  RedisUpdateLog,
34  type UpdateLog,
35  type UpdateLogBackend,
36} from "./update-log.js";
37import {
38  PrismaDocStore,
39  createPrismaMetaResolver,
40  createPrismaPermissionResolver,
41} from "./prisma-store.js";
42import { HttpDocStore } from "./http-store.js";
43import { createDomainEventSource, type DomainEventSource } from "./event-source.js";
44import { initOtel, shutdownOtel } from "@factory/observability";
45 
46async function main(): Promise<void> {
47  const config = loadConfig();
48  const logger = pino({ level: config.logLevel });
49 
50  // TASK-0135: OTel FIRST (same contract as the API/worker — the lazy metric
51  // instruments bind to the SDK provider at first touch; disabled → no-op).
52  // OTEL_* come straight from process.env (NOT in config.ts's schema).
53  const otelEndpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
54  await initOtel({
55    serviceName: "factory-realtime",
56    environment: process.env.NODE_ENV,
57    ...(process.env.OTEL_ENABLED === "true" && otelEndpoint ? { endpoint: otelEndpoint } : {})
58  });
59 
60  const store = await buildStore(config, logger);
61  const log = await buildUpdateLog(config, logger);
62  const permissions = await buildPermissions(config, logger);
63  const meta = await buildMeta(config, logger);
64 
65  const audit: AuditSink = {
66    // The audit seam publishes the abuse event; the api's audit
67    // consumer (the event bus) writes the row — the Phase-0/1 audit
68    // path, no new table (task Backend Changes).
69    publish: (event, payload) => {
70      logger.warn({ event, ...payload }, "audit: realtime abuse event");
71    },
72  };
73 
74  // The domain-event seam (TASK-0092 R3): the flush emits `page.updated`
75  // (the Phase-3 shape, the `blockDelta` additive field). The realtime app
76  // has no Redis dependency of its own, so the seam is logged here (the
77  // event is reconstructable from the structured log); the event-bus
78  // wiring (XADD `factory:events`) is the Phase-8/production seam —
79  // recorded in the task file (see the `EventSink` docs in server.ts).
80  const events: EventSink = {
81    publish: (event) => {
82      logger.info(
83        { name: event.name, pageId: (event.payload as { pageId?: string }).pageId },
84        "event: domain event emitted",
85      );
86    },
87  };
88 
89  // The (b) CONTROL producer (TASK-0092 6b, generalized in TASK-0094): poll
90  // the API's `factory:events` stream ONCE and route every event —
91  // `page.updated` → `block:changed` refetch-hint, `comment.*` →
92  // `comment.event` room frames, `notification.created` → `notification.push`
93  // control-room push (the server's routing sink does the per-event work).
94  // Lazy ioredis, same degradation pattern as the update log — no Redis → no
95  // control fan-out (the web overlay still works via its REST path; recorded
96  // in the task file).
97  const domainEventSourceFactory = await buildDomainEventSource(config, logger);
98 
99  const rt = createRealtimeServer(config, {
100    logger,
101    store,
102    log,
103    permissions,
104    meta,
105    audit,
106    events,
107    ...(domainEventSourceFactory !== null ? { domainEventSource: domainEventSourceFactory } : {}),
108  });
109 
110  await rt.start();
111  logger.info(
112    { port: config.port, healthPort: config.healthPort },
113    "realtime server started: ws + health listening",
114  );
115 
116  let shuttingDown = false;
117  const shutdown = (signal: string): void => {
118    if (shuttingDown) return;
119    shuttingDown = true;
120    logger.info({ signal }, "shutting down realtime server (drain)");
121    rt.drain()
122      .catch((err: unknown) => {
123        logger.error({ err: String(err), signal }, "error during realtime drain");
124      })
125      .finally(async () => {
126        // TASK-0135: flush the OTel SDK (pending spans + the final metric
127        // collection) before the process dies.
128        await shutdownOtel().catch(() => undefined);
129        process.exit(0);
130      });
131  };
132  process.on("SIGINT", () => shutdown("SIGINT"));
133  process.on("SIGTERM", () => shutdown("SIGTERM"));
134}
135 
136/** True when a dynamic import failed because `@prisma/client` is not
137 *  installed/generated yet (the dev-mode fallback case). */
138function isMissingPrismaClient(err: unknown): boolean {
139  if (err === null || typeof err !== "object") return false;
140  const code = (err as { code?: unknown }).code;
141  const msg = err instanceof Error ? err.message : "";
142  return (
143    code === "ERR_MODULE_NOT_FOUND" || /@prisma\/client|prisma.*not (yet )?generated/i.test(msg)
144  );
145}
146 
147/** Load the Prisma-backed seams (doc store + permission + meta). The
148 *  modules are imported lazily so the process boots even when the
149 *  generated client is absent (the `REALTIME_DB=0` dev contract). */
150async function loadPrismaSeams(): Promise<{
151  PrismaClient: new (cfg?: unknown) => import("@prisma/client").PrismaClient;
152} | null> {
153  try {
154    // Both imports must land: prisma-store.js statically imports
155    // @prisma/client, so they succeed or fail together.
156    const [client] = await Promise.all([import("@prisma/client"), import("./prisma-store.js")]);
157    return {
158      PrismaClient: client.PrismaClient as new (
159        cfg?: unknown,
160      ) => import("@prisma/client").PrismaClient,
161    };
162  } catch (err) {
163    if (isMissingPrismaClient(err)) return null;
164    throw err; // unexpected seam error — fail the boot, do not degrade silently
165  }
166}
167 
168async function buildStore(config: RealtimeConfig, logger: pino.Logger): Promise<DocStore> {
169  if (!config.useDatabase) {
170    logger.warn({}, "REALTIME_DB=0 — using the in-memory doc store (dev mode)");
171    return new InMemoryDocStore();
172  }
173  if (config.internalServiceToken.length > 0) {
174    // Production seam (TASK-0092 slice 2): the doc store persists through
175    // the API's internal endpoints (one Postgres write funnel — the API's
176    // audit + event layers). Fail loud when the token is rejected on boot
177    // (a misconfigured token would otherwise surface as a flush 401 at
178    // first edit, mid-session).
179    const store = new HttpDocStore({
180      apiUrl: config.apiUrl,
181      serviceToken: config.internalServiceToken,
182    });
183    if (!(await store.ping())) {
184      throw new Error(
185        "HttpDocStore boot check failed: the API's internal endpoints are unreachable " +
186          `(${config.apiUrl}/internal/* — check API_URL + INTERNAL_SERVICE_TOKEN on BOTH sides)`,
187      );
188    }
189    logger.info({ apiUrl: config.apiUrl }, "wired the HTTP doc store (API internal endpoints)");
190    return store;
191  }
192  const seams = await loadPrismaSeams();
193  if (seams === null) {
194    logger.warn({}, "@prisma/client unavailable — in-memory doc store (dev mode)");
195    return new InMemoryDocStore();
196  }
197  logger.info({}, "wired the Prisma doc store");
198  return new PrismaDocStore(new seams.PrismaClient());
199}
200 
201async function buildPermissions(
202  config: RealtimeConfig,
203  logger: pino.Logger,
204): Promise<PermissionResolver> {
205  if (!config.useDatabase) {
206    logger.warn({}, "no DB — fail-closed permission resolver (view-only joins)");
207  }
208  const seams = await loadPrismaSeams();
209  if (seams === null) {
210    logger.warn({}, "no DB — fail-closed permission resolver (view-only joins)");
211    return { resolveLevel: async () => "view" as const };
212  }
213  return createPrismaPermissionResolver(new seams.PrismaClient());
214}
215 
216async function buildMeta(config: RealtimeConfig, logger: pino.Logger): Promise<RoomMetaResolver> {
217  const seams = await loadPrismaSeams();
218  if (seams === null) {
219    logger.warn({}, "no DB — fail-closed meta resolver (all rooms 4002)");
220    return { resolve: async () => null };
221  }
222  return createPrismaMetaResolver(new seams.PrismaClient());
223}
224 
225function buildUpdateLog(config: RealtimeConfig, logger: pino.Logger): Promise<UpdateLog> {
226  return (async () => {
227    if (!config.useRedis) {
228      logger.warn({}, "REALTIME_REDIS=0 — using the in-memory update log (dev mode)");
229      return new InMemoryUpdateLog();
230    }
231    try {
232      const { Redis } = await import("ioredis");
233      const redis = new Redis(config.redisUrl, { maxRetriesPerRequest: 2 });
234      redis.on("error", (err: Error) => {
235        logger.warn({ err: err.message }, "redis connection error");
236      });
237      logger.info({}, "wired the Redis update log");
238      const log = new RedisUpdateLog({
239        redis: redis as unknown as UpdateLogBackend,
240      });
241      return {
242        append: (k: string, u: string, v: number) => log.append(k, u, v),
243        entriesAfter: (k: string, s: number) => log.entriesAfter(k, s),
244        maxSeq: (k: string) => log.maxSeq(k),
245        trimAfter: (k: string, s: number) => log.trimAfter(k, s),
246        clear: (k: string) => log.clear(k),
247        ping: async () => redis.status === "ready" || redis.status === "connecting",
248      };
249    } catch (err) {
250      logger.warn({ err: String(err) }, "ioredis unavailable — in-memory update log (dev mode)");
251      return new InMemoryUpdateLog();
252    }
253  })();
254}
255 
256/** The (b) control producer (TASK-0092 6b, generalized in TASK-0094): a
257 *  domain-event source that polls the API's `factory:events` stream (lazy
258 *  ioredis, same degradation pattern as the update log). Returns NULL when
259 *  there is no Redis (dev mode) — the server then runs no control fan-out
260 *  (the web overlay still works via its REST path). */
261async function buildDomainEventSource(
262  config: RealtimeConfig,
263  logger: pino.Logger,
264): Promise<
265  ((on: (event: import("@factory/shared").DomainEvent) => void) => DomainEventSource) | null
266> {
267  if (!config.useRedis) {
268    logger.warn({}, "REALTIME_REDIS=0 — no domain-event control source (dev mode)");
269    return null;
270  }
271  try {
272    const { Redis } = await import("ioredis");
273    const redis = new Redis(config.redisUrl, { maxRetriesPerRequest: 2 });
274    redis.on("error", (err: Error) => {
275      logger.warn({ err: err.message }, "redis connection error (domain-event source)");
276    });
277    logger.info({}, "wired the domain-event control source (factory:events stream)");
278    return (on) =>
279      createDomainEventSource(on, {
280        client: redis as unknown as import("./event-source.js").DomainEventStreamClient,
281        onError: (err) => logger.warn({ err: String(err) }, "domain-event source error"),
282      });
283  } catch (err) {
284    logger.warn(
285      { err: String(err) },
286      "ioredis unavailable — no domain-event control source (dev mode)",
287    );
288    return null;
289  }
290}
291 
292main().catch((err: unknown) => {
293  console.error("realtime server failed to start", err);
294  process.exit(1);
295});
296