server.ts · Source

server.tsapps/worker/src/health/server.tsraw ← health7,338 B← Back to the app
1/**
2 * apps/worker — minimal HTTP health server (no NestJS), port 4003.
3 *
4 * Contract (TASK-0009, docs/development/development-guide.md port table):
5 *   GET /health          -> 200 { status:"ok", app, version, uptimeSeconds, timestamp }
6 *   GET /health/ready    -> 200 { status:"ready", checks:{redis,queue} }
7 *                           | 503 { status:"not_ready", checks:{...} } naming the failing dep
8 *   GET /health/search-index:lag  -> 200 SearchIndexLagResult (TASK-0100)
9 *                           | 503 { available:false, ... } when the probe fails
10 *   GET /metrics                  -> 200 text/plain Prometheus exposition
11 *                                    (TASK-0135; 204 when OTel is disabled
12 *                                    or the render fails — the scrape target
13 *                                    exists either way)
14 *
15 * Unauthenticated by design (documented exception to the Phase 1 global guard);
16 * it exposes no data beyond app name / version / dependency state / metrics.
17 *
18 * The worker is otherwise headless (BullMQ + stream consumers, no business HTTP
19 * surface). This is the ONLY port the worker listens on.
20 */
21import { createServer, type Server } from "node:http";
22 
23export const APP_NAME = "factory-worker";
24export const APP_VERSION = "0.1.0";
25 
26export interface WorkerHealthDeps {
27  /** Redis PING — resolves true when the shared Redis is reachable. */
28  pingRedis: () => Promise<boolean>;
29  /** BullMQ queue probe — true when the queue backend is reachable. */
30  checkQueue?: () => Promise<boolean>;
31  /**
32   * Search-index consumer lag probe (TASK-0100 req 1: the lag suite asserts
33   * the worker consumes a 1k-event burst without backlog growth > 30 s via
34   * the consumer-lag metric `search-index:lag`).
35   *
36   * Resolves a snapshot of the `worker-search-index` consumer group on
37   * `factory:events`: `pending` = entries delivered but not yet ACKed
38   * (in-flight or awaiting redelivery); `lagEntries` = entries in the stream
39   * not yet delivered to THIS group (group last-delivered id → stream tail).
40   * Both are "the backlog" in different senses; the E2E lag suite reads
41   * `pending + lagEntries` as the composite backlog number.
42   *
43   * Optional: when absent (unit tests / older wiring) the endpoint reports
44   * `available: false` and never fails readiness.
45   */
46  checkSearchIndexLag?: () => Promise<{ pending: number; lagEntries: number }>;
47  /**
48   * TASK-0135 — render the OTel Prometheus exposition (the worker's
49   * queue_depth / queue_job_duration_seconds series + target_info). Returns
50   * "" when OTel is disabled; the route answers 204 in that case so the
51   * Prometheus scrape target is always present on the documented port.
52   */
53  metricsBody?: () => Promise<string>;
54}
55 
56export interface WorkerHealthResult {
57  status: "ready" | "not_ready";
58  checks: {
59    redis: boolean;
60    queue: boolean;
61  };
62}
63 
64export interface SearchIndexLagResult {
65  metric: "search-index:lag";
66  stream: "factory:events";
67  group: "worker-search-index";
68  /** False when no probe is wired (or the Redis query failed). */
69  available: boolean;
70  /** Delivered to the group but not ACKed (in-flight / redelivery). */
71  pending: number;
72  /** Stream entries not yet delivered to the group (tail − last-delivered). */
73  lagEntries: number;
74  /** pending + lagEntries — the number the lag E2E asserts (the backlog). */
75  backlog: number;
76  timestamp: string;
77}
78 
79/**
80 * Build the health server. `readyFn` is called per readiness request (1 s
81 * timeout enforced inside) so a hung dependency answers < 2 s. Liveness is
82 * always 200 — it never probes dependencies.
83 */
84export function createHealthServer(deps: WorkerHealthDeps): Server {
85  const startedAt = Date.now();
86 
87  const timeout = (ms: number) =>
88    new Promise<never>((_, reject) =>
89      setTimeout(() => reject(new Error("health check timeout")), ms).unref()
90    );
91 
92  const ready = async (): Promise<WorkerHealthResult> => {
93    const [redis, queue] = await Promise.all([
94      withTimeout(deps.pingRedis(), timeout(1000)).catch(() => false),
95      deps.checkQueue ? withTimeout(deps.checkQueue(), timeout(1000)).catch(() => false) : true
96    ]);
97    return {
98      status: redis && queue ? "ready" : "not_ready",
99      checks: { redis, queue }
100    };
101  };
102 
103  /**
104   * `GET /health/search-index:lag` (TASK-0100 req 1, Backend Changes) — the
105   * consumer-lag metric the lag E2E polls during a 1k-event burst.
106   *
107   * The probe is OBSERVABILITY, not readiness: it never gates
108   * /health/ready. When the probe is not wired (deps.checkSearchIndexLag
109   * absent) or Redis fails, the response is `available: false` with zeroed
110   * counts — callers (the E2E) must treat that as "cannot measure" and skip
111   * the assertion rather than fail the run.
112   */
113  const lag = async (): Promise<SearchIndexLagResult> => {
114    const base: SearchIndexLagResult = {
115      metric: "search-index:lag",
116      stream: "factory:events",
117      group: "worker-search-index",
118      available: false,
119      pending: 0,
120      lagEntries: 0,
121      backlog: 0,
122      timestamp: new Date().toISOString()
123    };
124    if (!deps.checkSearchIndexLag) return base;
125    try {
126      const { pending, lagEntries } = await withTimeout(deps.checkSearchIndexLag(), timeout(1500));
127      return { ...base, available: true, pending, lagEntries, backlog: pending + lagEntries };
128    } catch {
129      return base; // probe failure → "cannot measure", never 5xx (liveness stays 200)
130    }
131  };
132 
133  return createServer(async (req, res) => {
134    const url = (req.url ?? "").split("?")[0];
135    if (req.method !== "GET") {
136      res.writeHead(405, { "content-type": "application/json" });
137      res.end(JSON.stringify({ status: "method_not_allowed" }));
138      return;
139    }
140    // Liveness at /api/v1/health (TASK-0009 contract); /health kept as an
141    // alias for the documented manual curl.
142    if (url === "/api/v1/health" || url === "/api/v1/health/" || url === "/health" || url === "/health/") {
143      res.writeHead(200, { "content-type": "application/json" });
144      res.end(
145        JSON.stringify({
146          status: "ok",
147          app: APP_NAME,
148          version: APP_VERSION,
149          uptimeSeconds: Math.round((Date.now() - startedAt) / 1000),
150          timestamp: new Date().toISOString()
151        })
152      );
153      return;
154    }
155    if (url === "/health/ready") {
156      const result = await ready();
157      res.writeHead(result.status === "ready" ? 200 : 503, {
158        "content-type": "application/json"
159      });
160      res.end(JSON.stringify(result));
161      return;
162    }
163    if (url === "/health/search-index:lag") {
164      res.writeHead(200, { "content-type": "application/json" });
165      res.end(JSON.stringify(await lag()));
166      return;
167    }
168    if (url === "/metrics") {
169      const body = deps.metricsBody ? await deps.metricsBody().catch(() => "") : "";
170      if (!body) {
171        res.writeHead(204);
172        res.end();
173        return;
174      }
175      res.writeHead(200, { "content-type": "text/plain; version=0.0.4; charset=utf-8" });
176      res.end(body);
177      return;
178    }
179    res.writeHead(404, { "content-type": "application/json" });
180    res.end(JSON.stringify({ status: "not_found" }));
181  });
182}
183 
184async function withTimeout<T>(p: Promise<T>, t: Promise<never>): Promise<T> {
185  return Promise.race([p, t]);
186}
187