diff --git a/.env.example b/.env.example index 385ebeeb..d90f3aad 100644 --- a/.env.example +++ b/.env.example @@ -17,10 +17,20 @@ PORT=3001 # Example: CORS_ALLOWED_ORIGINS=https://app.chronopay.com,https://*.chronopay.com CORS_ALLOWED_ORIGINS=http://localhost:3000,http://localhost:3001 +# Redis connection URL (used by the slot cache, idempotency store and the +# scheduler pause kill-switch). Defaults to redis://localhost:6379. +REDIS_URL=redis://localhost:6379 + +# Shared secret for admin-only control-plane endpoints (sent as the +# `x-chronopay-admin-token` header). Required for POST /api/v1/admin/scheduler/*. +# Generate one with: +# node -e "console.log(require('crypto').randomBytes(32).toString('hex'))" +CHRONOPAY_ADMIN_TOKEN=change-me-admin-token + # Optional: encrypt idempotency payloads stored in Redis with AES-256-GCM. # Leave disabled unless you have generated and securely provisioned 32-byte base64 keys. IDEMPOTENCY_REDIS_ENCRYPTION_ENABLED=false IDEMPOTENCY_REDIS_ENCRYPTION_ACTIVE_KEY_ID= IDEMPOTENCY_REDIS_ENCRYPTION_ACTIVE_KEY= # Comma-separated key-id:base64-key entries kept temporarily for decryption during rotation. -IDEMPOTENCY_REDIS_ENCRYPTION_PREVIOUS_KEYS= +IDEMPOTENCY_REDIS_ENCRYPTION_PREVIOUS_KEYS= \ No newline at end of file diff --git a/README.md b/README.md index 3436189d..40df8f1d 100644 --- a/README.md +++ b/README.md @@ -186,6 +186,7 @@ Detailed API contracts and endpoint documentation: - **[Booking Intent API](docs/booking-intent-api.md)** — POST /api/v1/booking-intents endpoint with request/response schemas, validation rules, and error codes - **[Slots API](docs/api/slots.md)** — Complete slots API documentation covering GET/POST/PATCH/DELETE endpoints, conflict detection, caching, and security considerations +- **[Scheduler Pause / Resume](docs/scheduler-pause.md)** — Admin-only incident kill-switch (POST /api/v1/admin/scheduler/pause|resume) that freezes new booking-intent creation platform-wide via a Redis flag while leaving read paths intact ## API (slot listing) diff --git a/docs/scheduler-pause.md b/docs/scheduler-pause.md new file mode 100644 index 00000000..da990608 --- /dev/null +++ b/docs/scheduler-pause.md @@ -0,0 +1,167 @@ +# Scheduler Pause / Resume Kill-Switch + +An admin-only, platform-wide kill-switch that **freezes new booking-intent +creation** during an incident (e.g. a downstream outage, a fraud spike, a bad +deploy) while leaving **read paths intact** — customers can still view existing +bookings, cancel previews, hold status, etc. + +The switch is a single Redis flag (`scheduler:paused`) checked by a lightweight +guard middleware on the booking-intent *create* route. It is intentionally +cheap to read (one `GET`) and safe to fail. + +- Relevant code: [`src/redis.ts`](../src/redis.ts), + [`src/middleware/schedulerGate.ts`](../src/middleware/schedulerGate.ts), + [`src/routes/admin/scheduler.ts`](../src/routes/admin/scheduler.ts), + [`src/services/schedulerStatusBus.ts`](../src/services/schedulerStatusBus.ts) + +--- + +## Endpoints + +All three require the shared admin token in the `x-chronopay-admin-token` +header (see [`requireAdminToken`](../src/middleware/authorization.ts)). They are +mounted under `/api/v1/admin/scheduler`. + +### `POST /api/v1/admin/scheduler/pause` + +Freeze new booking-intent creation platform-wide. + +Body: + +| Field | Type | Required | Notes | +| -------------- | ------ | -------- | ------------------------------------------ | +| `reason` | string | yes | Human-readable incident reason. | +| `initiated_by` | string | yes | Operator identity (also accepts `initiatedBy`). | + +```bash +curl -X POST https://api.chronopay.com/api/v1/admin/scheduler/pause \ + -H "x-chronopay-admin-token: $CHRONOPAY_ADMIN_TOKEN" \ + -H "content-type: application/json" \ + -d '{ "reason": "payments provider degraded", "initiated_by": "oncall-jane" }' +``` + +```json +200 OK +{ + "success": true, + "scheduler": { + "paused": true, + "reason": "payments provider degraded", + "initiatedBy": "oncall-jane", + "pausedAt": "2026-07-31T09:15:04.512Z" + } +} +``` + +### `POST /api/v1/admin/scheduler/resume` + +Lift the freeze. + +| Field | Type | Required | +| -------------- | ------ | -------- | +| `initiated_by` | string | yes | + +```json +200 OK +{ "success": true, "scheduler": { "paused": false, "initiatedBy": "oncall-jane" } } +``` + +### `GET /api/v1/admin/scheduler/status` + +Read the current state (a read path — safe to call during a freeze). + +```json +200 OK +{ "success": true, "scheduler": { "paused": true, "reason": "...", "initiatedBy": "...", "pausedAt": "..." } } +``` + +### Error responses + +| Status | `code` | When | +| ------ | --------------------- | -------------------------------------------------------- | +| 401 | (auth) | Missing `x-chronopay-admin-token`. | +| 403 | (auth) | Wrong admin token. | +| 400 | `INVALID_REASON` | `reason` missing/blank on pause. | +| 400 | `INVALID_INITIATED_BY`| `initiated_by` missing/blank. | +| 503 | `REDIS_UNAVAILABLE` | Redis could not be reached to read/update the flag. | +| 500 | `INTERNAL_ERROR` | Unexpected failure. | + +--- + +## What callers see while paused + +The guard ([`schedulerGate`](../src/middleware/schedulerGate.ts)) is attached to +the booking-intent **create** route only. While paused it returns: + +```json +503 Service Unavailable +Retry-After: 120 +{ + "success": false, + "error": "Booking creation is temporarily paused by an operator.", + "code": "SCHEDULER_PAUSED", + "reason": "payments provider degraded", + "initiatedBy": "oncall-jane", + "pausedAt": "2026-07-31T09:15:04.512Z" +} +``` + +Read routes (`GET /:id/hold-status`, `GET /:id/cancel-preview`, listings, …) are +**not** guarded and keep working. + +--- + +## The Redis flag + +- **Key:** `scheduler:paused` +- **Value:** a small JSON payload — `{"paused":1,"reason":"…","initiated_by":"…","paused_at":"…"}`. + The `paused` field is `1`, satisfying the "`scheduler:paused=1`" contract while + still carrying the audit metadata. A bare legacy `"1"` value is also honoured. +- **Resume** deletes the key, so "not paused" is simply the *absence* of the key — + the safest default if the value is ever evicted. + +## Fail-open contract (important) + +The pause flag is a **safety** mechanism, not a correctness one. If Redis is +unreachable **at guard time**, the guard **fails open** (allows the request) and +logs a warning. A kill-switch must never turn an unrelated Redis outage into a +total booking outage. + +- Guard, Redis down → `next()` + `warn` log (traffic flows). +- Guard, flag set → `503 SCHEDULER_PAUSED` (traffic blocked). +- Control-plane write, Redis down → `503 REDIS_UNAVAILABLE` (the operator is told + the pause/resume could not be persisted, rather than silently succeeding). + +## Realtime broadcast + +Every pause/resume is broadcast on the in-process status bus +([`schedulerStatusBus`](../src/services/schedulerStatusBus.ts), channel +`scheduler:status`). The WebSocket layer subscribes via `onSchedulerStatus(...)` +and relays the event to connected dashboards so a freeze is visible immediately +instead of by polling. Broadcasting is fire-and-forget and never fails the +underlying operation. + +## Metrics + +Two Prometheus counters (registered in [`src/metrics.ts`](../src/metrics.ts) and +exposed on `/metrics`): + +- `scheduler_pause_total` — incremented on every successful pause. +- `scheduler_resume_total` — incremented on every successful resume. + +## Configuration + +| Env var | Purpose | Default | +| ----------------------- | ------------------------------------------------ | -------------------------- | +| `REDIS_URL` | Redis connection string for the flag. | `redis://localhost:6379` | +| `CHRONOPAY_ADMIN_TOKEN` | Shared secret for the admin control-plane. | *(required)* | + +## Runbook + +1. **Pause:** `POST /pause` with a clear `reason` and your handle in `initiated_by`. +2. Confirm with `GET /status` and watch `scheduler_pause_total` tick up. +3. Mitigate the incident. Booking creation returns `503 SCHEDULER_PAUSED`. +4. **Resume:** `POST /resume` with `initiated_by`. Confirm `paused:false` and + `scheduler_resume_total`. +5. If `/pause` or `/resume` returns `503 REDIS_UNAVAILABLE`, Redis itself is the + problem — bookings are already flowing (guard fails open); fix Redis first. diff --git a/src/__tests__/schedulerRedisFlag.test.ts b/src/__tests__/schedulerRedisFlag.test.ts new file mode 100644 index 00000000..71737d91 --- /dev/null +++ b/src/__tests__/schedulerRedisFlag.test.ts @@ -0,0 +1,229 @@ +/** + * Unit tests for the scheduler pause flag helpers in src/redis.ts. + * + * Runs under NODE_ENV=test, so `getRedisClient()` returns whatever fake we + * inject with `setRedisClient()`. Injecting `null` simulates "Redis down". + */ +import { jest } from "@jest/globals"; +import { + SCHEDULER_PAUSED_KEY, + RedisUnavailableError, + pauseScheduler, + resumeScheduler, + readSchedulerPauseState, + setRedisClient, + isRedisReady, + closeRedisClient, + type RedisLike, +} from "../redis.js"; + +function makeFakeRedis(overrides: Partial = {}) { + const store = new Map(); + const client: RedisLike & { store: Map } = { + store, + get: async (key: string) => (store.has(key) ? store.get(key)! : null), + set: async (key: string, value: string) => { + store.set(key, value); + return "OK"; + }, + del: async (key: string) => { + const existed = store.has(key); + store.delete(key); + return existed ? 1 : 0; + }, + ping: async () => "PONG", + quit: async () => "OK", + ...overrides, + }; + return client; +} + +describe("scheduler redis flag", () => { + afterEach(() => { + setRedisClient(null); + jest.restoreAllMocks(); + }); + + it("exposes the canonical redis key", () => { + expect(SCHEDULER_PAUSED_KEY).toBe("scheduler:paused"); + }); + + it("pauseScheduler stores a structured paused=1 payload and returns state", async () => { + const redis = makeFakeRedis(); + setRedisClient(redis); + + const state = await pauseScheduler({ reason: "db incident", initiatedBy: "alice" }); + + expect(state).toMatchObject({ + paused: true, + reason: "db incident", + initiatedBy: "alice", + }); + expect(typeof state.pausedAt).toBe("string"); + + const stored = JSON.parse(redis.store.get(SCHEDULER_PAUSED_KEY)!); + expect(stored).toMatchObject({ + paused: 1, + reason: "db incident", + initiated_by: "alice", + }); + expect(typeof stored.paused_at).toBe("string"); + }); + + it("readSchedulerPauseState reflects a pause and its metadata", async () => { + const redis = makeFakeRedis(); + setRedisClient(redis); + + await pauseScheduler({ reason: "spike", initiatedBy: "bob" }); + const state = await readSchedulerPauseState(); + + expect(state).toMatchObject({ + paused: true, + reason: "spike", + initiatedBy: "bob", + }); + }); + + it("resumeScheduler clears the flag (pause then immediate resume)", async () => { + const redis = makeFakeRedis(); + setRedisClient(redis); + + await pauseScheduler({ reason: "spike", initiatedBy: "bob" }); + expect((await readSchedulerPauseState()).paused).toBe(true); + + const resumed = await resumeScheduler({ initiatedBy: "bob" }); + expect(resumed).toEqual({ paused: false, initiatedBy: "bob" }); + expect(redis.store.has(SCHEDULER_PAUSED_KEY)).toBe(false); + expect(await readSchedulerPauseState()).toEqual({ paused: false }); + }); + + it("readSchedulerPauseState returns { paused: false } when the key is absent", async () => { + setRedisClient(makeFakeRedis()); + expect(await readSchedulerPauseState()).toEqual({ paused: false }); + }); + + it("tolerates a bare '1' legacy value", async () => { + const redis = makeFakeRedis(); + redis.store.set(SCHEDULER_PAUSED_KEY, "1"); + setRedisClient(redis); + expect(await readSchedulerPauseState()).toEqual({ paused: true }); + }); + + it("tolerates a JSON-quoted '1' primitive value", async () => { + const redis = makeFakeRedis(); + redis.store.set(SCHEDULER_PAUSED_KEY, JSON.stringify("1")); // '"1"' + setRedisClient(redis); + expect(await readSchedulerPauseState()).toEqual({ paused: true }); + }); + + it("treats a boolean-true paused field as paused", async () => { + const redis = makeFakeRedis(); + redis.store.set(SCHEDULER_PAUSED_KEY, JSON.stringify({ paused: true, reason: "r" })); + setRedisClient(redis); + expect(await readSchedulerPauseState()).toMatchObject({ paused: true, reason: "r" }); + }); + + it("tolerates a bare non-'1' legacy value as not paused", async () => { + const redis = makeFakeRedis(); + redis.store.set(SCHEDULER_PAUSED_KEY, "nope"); + setRedisClient(redis); + expect(await readSchedulerPauseState()).toEqual({ paused: false }); + }); + + it("treats a JSON payload with paused=0 as not paused", async () => { + const redis = makeFakeRedis(); + redis.store.set(SCHEDULER_PAUSED_KEY, JSON.stringify({ paused: 0 })); + setRedisClient(redis); + expect(await readSchedulerPauseState()).toEqual({ paused: false }); + }); + + it("accepts a string '1' paused field and camelCase metadata keys", async () => { + const redis = makeFakeRedis(); + redis.store.set( + SCHEDULER_PAUSED_KEY, + JSON.stringify({ paused: "1", reason: "r", initiatedBy: "carol", pausedAt: "t" }), + ); + setRedisClient(redis); + expect(await readSchedulerPauseState()).toEqual({ + paused: true, + reason: "r", + initiatedBy: "carol", + pausedAt: "t", + }); + }); + + describe("Redis unavailable (fail-open contract source)", () => { + it("pauseScheduler throws RedisUnavailableError when client is null", async () => { + setRedisClient(null); + await expect(pauseScheduler({ reason: "x", initiatedBy: "y" })).rejects.toBeInstanceOf( + RedisUnavailableError, + ); + }); + + it("resumeScheduler throws RedisUnavailableError when client is null", async () => { + setRedisClient(null); + await expect(resumeScheduler({ initiatedBy: "y" })).rejects.toBeInstanceOf( + RedisUnavailableError, + ); + }); + + it("readSchedulerPauseState throws RedisUnavailableError when client is null", async () => { + setRedisClient(null); + await expect(readSchedulerPauseState()).rejects.toBeInstanceOf(RedisUnavailableError); + }); + + it("wraps a set() failure as RedisUnavailableError", async () => { + setRedisClient( + makeFakeRedis({ + set: async () => { + throw new Error("connection reset"); + }, + }), + ); + await expect(pauseScheduler({ reason: "x", initiatedBy: "y" })).rejects.toBeInstanceOf( + RedisUnavailableError, + ); + }); + + it("wraps a del() failure as RedisUnavailableError", async () => { + setRedisClient( + makeFakeRedis({ + del: async () => { + throw new Error("connection reset"); + }, + }), + ); + await expect(resumeScheduler({ initiatedBy: "y" })).rejects.toBeInstanceOf( + RedisUnavailableError, + ); + }); + + it("wraps a get() failure as RedisUnavailableError", async () => { + setRedisClient( + makeFakeRedis({ + get: async () => { + throw new Error("connection reset"); + }, + }), + ); + await expect(readSchedulerPauseState()).rejects.toBeInstanceOf(RedisUnavailableError); + }); + }); + + it("isRedisReady tracks the injected client", () => { + setRedisClient(makeFakeRedis()); + expect(isRedisReady()).toBe(true); + setRedisClient(null); + expect(isRedisReady()).toBe(false); + }); + + it("closeRedisClient quits and resets readiness (idempotent)", async () => { + const quit = jest.fn<() => Promise>().mockResolvedValue("OK"); + setRedisClient(makeFakeRedis({ quit })); + await closeRedisClient(); + expect(quit).toHaveBeenCalledTimes(1); + expect(isRedisReady()).toBe(false); + // Idempotent: second call is a no-op. + await expect(closeRedisClient()).resolves.toBeUndefined(); + }); +}); diff --git a/src/app.ts b/src/app.ts index 5107d86d..69ab6f3b 100644 --- a/src/app.ts +++ b/src/app.ts @@ -51,6 +51,8 @@ import checkoutRouter from "./routes/checkout.js"; import buyerProfileRouter from "./buyer-profile/buyer-profile.routes.js"; import oauth2Router from "./routes/oauth2.js"; import adminRouter from "./routes/admin.js"; +import schedulerAdminRouter from "./routes/admin/scheduler.js"; +import { schedulerGate } from "./middleware/schedulerGate.js"; import graceWindowRouter from "./routes/graceWindow.js"; import { legalHoldRouter } from "./routes/legalHold.js"; import webhookRoutes, { registerWebhookRoutes } from "./routes/webhooks.js"; @@ -516,6 +518,9 @@ export function createApp(options: AppFactoryOptions = {}) { // 3b-i. Fraud model admin routes (#455 rollback hotkey) app.use("/api/v1/admin/fraud-models", fraudModelsRouter); + // 3b-i-scheduler. Incident scheduler kill-switch (pause/resume) + app.use("/api/v1/admin/scheduler", schedulerAdminRouter); + // 3b-i-a. Scheduled feature-flag rollout admin routes (#570) app.use("/api/v1/admin/flag-rollouts", flagRolloutsRouter); @@ -620,6 +625,7 @@ export function createApp(options: AppFactoryOptions = {}) { app.post( "/api/v1/booking-intents", requireAuth(["customer"]), + schedulerGate, async (req: any, res: Response) => { try { const { slotId, note } = req.body; diff --git a/src/metrics.ts b/src/metrics.ts index dc322811..62777142 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -742,3 +742,27 @@ export const treasuryBalance = createBudgetedGauge({ budget: 50, }); +/** + * Counter incremented each time an operator PAUSES the scheduler + * (freezes platform-wide booking-intent creation during an incident). + */ +export const schedulerPauseTotal = createBudgetedCounter({ + name: "scheduler_pause_total", + help: "Total number of times the scheduler (booking-intent creation) was paused by an operator", + labels: [], + budget: 0, + registers: [register], +}); + +/** + * Counter incremented each time an operator RESUMES the scheduler + * (lifts a platform-wide booking-intent creation freeze). + */ +export const schedulerResumeTotal = createBudgetedCounter({ + name: "scheduler_resume_total", + help: "Total number of times the scheduler (booking-intent creation) was resumed by an operator", + labels: [], + budget: 0, + registers: [register], +}); + diff --git a/src/middleware/__tests__/schedulerGate.test.ts b/src/middleware/__tests__/schedulerGate.test.ts new file mode 100644 index 00000000..3c075023 --- /dev/null +++ b/src/middleware/__tests__/schedulerGate.test.ts @@ -0,0 +1,105 @@ +/** + * Tests for the schedulerGate middleware. + * + * Verifies the three-way contract: + * - not paused → next() + * - paused → 503 SCHEDULER_PAUSED (fail closed) + * - Redis unavailable → next() + warning (fail OPEN) + */ +import { jest } from "@jest/globals"; +import express from "express"; +import request from "supertest"; +import { schedulerGate } from "../schedulerGate.js"; +import { setRedisClient, pauseScheduler, type RedisLike } from "../../redis.js"; + +function makeFakeRedis(overrides: Partial = {}): RedisLike { + const store = new Map(); + return { + get: async (key: string) => (store.has(key) ? store.get(key)! : null), + set: async (key: string, value: string) => { + store.set(key, value); + return "OK"; + }, + del: async (key: string) => { + store.delete(key); + return 1; + }, + ping: async () => "PONG", + quit: async () => "OK", + ...overrides, + }; +} + +function makeApp() { + const app = express(); + app.use(express.json()); + app.post("/create", schedulerGate, (_req, res) => { + res.status(201).json({ success: true, created: true }); + }); + return app; +} + +describe("schedulerGate", () => { + afterEach(() => { + setRedisClient(null); + jest.restoreAllMocks(); + }); + + it("allows the request when the scheduler is not paused", async () => { + setRedisClient(makeFakeRedis()); + const res = await request(makeApp()).post("/create").send({}); + expect(res.status).toBe(201); + expect(res.body).toEqual({ success: true, created: true }); + }); + + it("blocks with 503 SCHEDULER_PAUSED when paused, exposing metadata + Retry-After", async () => { + setRedisClient(makeFakeRedis()); + await pauseScheduler({ reason: "db incident", initiatedBy: "alice" }); + + const res = await request(makeApp()).post("/create").send({}); + + expect(res.status).toBe(503); + expect(res.headers["retry-after"]).toBe("120"); + expect(res.body).toMatchObject({ + success: false, + code: "SCHEDULER_PAUSED", + reason: "db incident", + initiatedBy: "alice", + }); + expect(res.body.pausedAt).toBeTruthy(); + }); + + it("fails OPEN (allows traffic) when Redis is unavailable", async () => { + setRedisClient(null); // getRedisClient() → null → RedisUnavailableError + const res = await request(makeApp()).post("/create").send({}); + expect(res.status).toBe(201); + }); + + it("fails OPEN when the redis read throws mid-flight", async () => { + setRedisClient( + makeFakeRedis({ + get: async () => { + throw new Error("connection reset"); + }, + }), + ); + const res = await request(makeApp()).post("/create").send({}); + expect(res.status).toBe(201); + }); + + it("nulls out missing metadata fields in the paused response", async () => { + // Simulate a bare "1" legacy flag with no metadata. + const store = new Map([["scheduler:paused", "1"]]); + setRedisClient( + makeFakeRedis({ + get: async (key: string) => store.get(key) ?? null, + }), + ); + + const res = await request(makeApp()).post("/create").send({}); + expect(res.status).toBe(503); + expect(res.body.reason).toBeNull(); + expect(res.body.initiatedBy).toBeNull(); + expect(res.body.pausedAt).toBeNull(); + }); +}); diff --git a/src/middleware/schedulerGate.ts b/src/middleware/schedulerGate.ts new file mode 100644 index 00000000..5c429579 --- /dev/null +++ b/src/middleware/schedulerGate.ts @@ -0,0 +1,65 @@ +/** + * @file src/middleware/schedulerGate.ts + * + * Express guard that freezes NEW booking-intent creation platform-wide while an + * operator has paused the scheduler during an incident. + * + * Design contract + * ─────────────── + * - Attach ONLY to mutating booking-intent create routes. Read paths (status, + * previews, listings) are intentionally left untouched so customers can still + * inspect existing bookings during a freeze. + * - Fail OPEN: if the pause flag cannot be read (Redis outage), allow the + * request through and emit a warning. A kill-switch must never amplify an + * unrelated Redis incident into a full booking outage. + * - Fail CLOSED only when the flag is explicitly set: respond `503` with a + * machine-readable `SCHEDULER_PAUSED` code and a `Retry-After` hint. + */ + +import type { Request, Response, NextFunction } from "express"; +import { readSchedulerPauseState } from "../redis.js"; +import { logger } from "../utils/logger.js"; + +/** Seconds clients should wait before retrying while paused. */ +const RETRY_AFTER_SECONDS = 120; + +export async function schedulerGate( + req: Request, + res: Response, + next: NextFunction, +): Promise { + let state; + try { + state = await readSchedulerPauseState(); + } catch (err) { + // Fail-open: cannot determine pause state → do not block traffic. + logger.warn( + { err, path: req.originalUrl }, + "schedulerGate: unable to read scheduler pause flag — failing open", + ); + next(); + return; + } + + if (!state.paused) { + next(); + return; + } + + logger.info( + { path: req.originalUrl, reason: state.reason, initiatedBy: state.initiatedBy }, + "schedulerGate: rejecting booking-intent create — scheduler is paused", + ); + + res.setHeader("Retry-After", String(RETRY_AFTER_SECONDS)); + res.status(503).json({ + success: false, + error: "Booking creation is temporarily paused by an operator.", + code: "SCHEDULER_PAUSED", + reason: state.reason ?? null, + initiatedBy: state.initiatedBy ?? null, + pausedAt: state.pausedAt ?? null, + }); +} + +export default schedulerGate; diff --git a/src/redis.ts b/src/redis.ts new file mode 100644 index 00000000..cdc322d8 --- /dev/null +++ b/src/redis.ts @@ -0,0 +1,280 @@ +/** + * @file src/redis.ts + * + * Shared Redis access plus the platform "scheduler pause" flag used to freeze + * new booking-intent creation during an incident. + * + * ──────────────────────────────────────────────────────────────────────────── + * Why this lives here + * ──────────────────────────────────────────────────────────────────────────── + * The incident kill-switch has to be reachable from two independent call sites: + * + * 1. The admin control-plane route (`src/routes/admin/scheduler.ts`) that + * *writes* the flag (pause / resume). + * 2. The data-plane guard middleware (`src/middleware/schedulerGate.ts`) that + * *reads* the flag on every booking-intent create request. + * + * Keeping the flag semantics in one module guarantees both sides agree on the + * Redis key, the value encoding, and the fail-open contract. + * + * ──────────────────────────────────────────────────────────────────────────── + * Fail-open contract + * ──────────────────────────────────────────────────────────────────────────── + * The pause flag is a *safety* mechanism, not a correctness one. If Redis is + * unreachable we must NOT wedge the whole booking funnel shut on top of an + * unrelated Redis outage. Reads therefore surface a distinguishable + * `RedisUnavailableError` so the guard can fail *open* (allow traffic) while + * logging a warning, whereas an explicit paused flag fails *closed* (503). + */ + +import { createRequire } from "module"; +import { logger } from "./utils/logger.js"; + +const require = createRequire(import.meta.url); +/* istanbul ignore next -- deployment-env fallback; the test env always sets REDIS_URL */ +const REDIS_URL = process.env.REDIS_URL ?? "redis://localhost:6379"; + +/** + * Minimal Redis surface this module depends on. Declared as an interface so + * tests can inject a fake without pulling in ioredis. + */ +export interface RedisLike { + get(key: string): Promise; + set(key: string, value: string, ...args: unknown[]): Promise; + del(key: string): Promise; + ping(): Promise; + quit(): Promise; + on?(event: string, handler: (...args: unknown[]) => void): unknown; +} + +let _client: RedisLike | null = null; +let _ready = false; + +/** Returns true once the client has emitted the "ready" event. */ +export function isRedisReady(): boolean { + return _ready; +} + +/** + * Replace the active client. Used by tests to inject a fake (or `null` to + * simulate "Redis unavailable"). + */ +export function setRedisClient(client: RedisLike | null): void { + _client = client; + _ready = client !== null; +} + +/** + * Returns the shared Redis client, creating it lazily on first use. + * + * In `NODE_ENV=test` the singleton starts as `null`; tests inject a fake via + * `setRedisClient()`. Returning `null` (rather than throwing) lets callers + * decide how to degrade. + */ +export function getRedisClient(): RedisLike | null { + if (process.env.NODE_ENV === "test") { + return _client; + } + + /* istanbul ignore next -- real ioredis network construction; not exercisable + under the jest ESM runner, which maps `ioredis` to an ESM-only mock. Mirrors + the established pattern in src/cache/redisClient.ts. */ + if (!_client) { + const { Redis } = require("ioredis") as { + Redis: new (url: string, options: Record) => RedisLike; + }; + + const redis = new Redis(REDIS_URL, { + // Exponential back-off capped at 2s; give up after a few attempts so a + // dead Redis doesn't hold requests hostage — the guard fails open anyway. + retryStrategy: (times: number) => Math.min(times * 100, 2000), + maxRetriesPerRequest: 3, + enableReadyCheck: true, + }); + + redis.on?.("connect", () => logger.info({ url: REDIS_URL }, "redis connected")); + redis.on?.("ready", () => { + _ready = true; + logger.info({ url: REDIS_URL }, "redis ready"); + }); + redis.on?.("error", (...args: unknown[]) => logger.error({ err: args[0] }, "redis error")); + redis.on?.("close", () => { + _ready = false; + }); + + _client = redis; + } + + return _client; +} + +/** + * Gracefully close the connection. Idempotent — safe to call multiple times. + */ +export async function closeRedisClient(): Promise { + if (_client) { + const closing = _client; + _client = null; + _ready = false; + await closing.quit(); + logger.info("redis connection closed gracefully"); + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// Scheduler pause flag +// ──────────────────────────────────────────────────────────────────────────── + +/** Redis key holding the platform-wide scheduler pause flag. */ +export const SCHEDULER_PAUSED_KEY = "scheduler:paused"; + +/** + * Thrown when the pause flag cannot be read/written because Redis is + * unreachable. Callers use this to decide between fail-open (guard) and a + * 503 (control-plane write). + */ +export class RedisUnavailableError extends Error { + public readonly code = "REDIS_UNAVAILABLE"; + constructor(message = "Redis is unavailable") { + super(message); + this.name = "RedisUnavailableError"; + } +} + +/** In-memory view of the pause flag returned to callers. */ +export interface SchedulerPauseState { + paused: boolean; + reason?: string; + initiatedBy?: string; + pausedAt?: string; +} + +interface StoredPausePayload { + paused: 1; + reason: string; + initiated_by: string; + paused_at: string; +} + +/** + * Pause the scheduler platform-wide. + * + * Stores `scheduler:paused` as a JSON payload carrying the incident `reason` + * and the operator who `initiated_by` the pause, so the flag doubles as an + * audit breadcrumb. The `paused` field is `1`, satisfying the "scheduler:paused=1" + * contract while still leaving room for structured metadata. + * + * @throws {RedisUnavailableError} when Redis cannot be reached. + */ +export async function pauseScheduler(input: { + reason: string; + initiatedBy: string; +}): Promise { + const client = getRedisClient(); + if (!client) { + throw new RedisUnavailableError(); + } + + const pausedAt = new Date().toISOString(); + const payload: StoredPausePayload = { + paused: 1, + reason: input.reason, + initiated_by: input.initiatedBy, + paused_at: pausedAt, + }; + + try { + await client.set(SCHEDULER_PAUSED_KEY, JSON.stringify(payload)); + } catch (err) { + throw new RedisUnavailableError((err as Error)?.message); + } + + return { + paused: true, + reason: input.reason, + initiatedBy: input.initiatedBy, + pausedAt, + }; +} + +/** + * Resume the scheduler by clearing the pause flag. + * + * Deleting the key (rather than setting `paused=0`) keeps "not paused" as the + * absence of the key, which is the safest default should the value ever be + * evicted or lost. + * + * @throws {RedisUnavailableError} when Redis cannot be reached. + */ +export async function resumeScheduler(input: { + initiatedBy: string; +}): Promise { + const client = getRedisClient(); + if (!client) { + throw new RedisUnavailableError(); + } + + try { + await client.del(SCHEDULER_PAUSED_KEY); + } catch (err) { + throw new RedisUnavailableError((err as Error)?.message); + } + + return { paused: false, initiatedBy: input.initiatedBy }; +} + +/** + * Read the current pause state. + * + * Returns `{ paused: false }` when the key is absent. Tolerates both the + * structured JSON payload and a bare `"1"` legacy value. + * + * @throws {RedisUnavailableError} when Redis cannot be reached. + */ +export async function readSchedulerPauseState(): Promise { + const client = getRedisClient(); + if (!client) { + throw new RedisUnavailableError(); + } + + let raw: string | null; + try { + raw = await client.get(SCHEDULER_PAUSED_KEY); + } catch (err) { + throw new RedisUnavailableError((err as Error)?.message); + } + + if (!raw) { + return { paused: false }; + } + + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + // Non-JSON legacy encoding: a bare "1" means paused. + return { paused: raw === "1" }; + } + + // Structured payload written by pauseScheduler(). + if (parsed !== null && typeof parsed === "object") { + const obj = parsed as Partial & { + paused?: unknown; + initiatedBy?: string; + pausedAt?: string; + }; + const paused = obj.paused === 1 || obj.paused === "1" || obj.paused === true; + if (!paused) { + return { paused: false }; + } + return { + paused: true, + reason: obj.reason, + initiatedBy: obj.initiated_by ?? obj.initiatedBy, + pausedAt: obj.paused_at ?? obj.pausedAt, + }; + } + + // Primitive legacy encoding: JSON.parse("1") === 1, JSON.parse('"1"') === "1". + return { paused: parsed === 1 || parsed === "1" }; +} diff --git a/src/routes/admin/__tests__/scheduler.errors.test.ts b/src/routes/admin/__tests__/scheduler.errors.test.ts new file mode 100644 index 00000000..1e4c4652 --- /dev/null +++ b/src/routes/admin/__tests__/scheduler.errors.test.ts @@ -0,0 +1,72 @@ +/** + * Covers the scheduler control-plane's defensive 500 path: when the underlying + * flag store throws something OTHER than RedisUnavailableError, the async + * handlers must translate it to a clean 500 rather than leaking an unhandled + * rejection. The redis module is mocked so we can inject a generic failure. + */ +import { jest } from "@jest/globals"; +import express from "express"; +import request from "supertest"; + +class FakeRedisUnavailableError extends Error { + code = "REDIS_UNAVAILABLE"; +} + +jest.unstable_mockModule("../../../redis.js", () => ({ + RedisUnavailableError: FakeRedisUnavailableError, + SCHEDULER_PAUSED_KEY: "scheduler:paused", + pauseScheduler: jest.fn(async () => { + throw new Error("boom"); + }), + resumeScheduler: jest.fn(async () => { + throw new Error("boom"); + }), + readSchedulerPauseState: jest.fn(async () => { + throw new Error("boom"); + }), + setRedisClient: jest.fn(), +})); + +const ADMIN_TOKEN = "test-admin-token-scheduler-errors"; +process.env.CHRONOPAY_ADMIN_TOKEN = ADMIN_TOKEN; + +let schedulerRouter: express.Router; + +beforeAll(async () => { + schedulerRouter = (await import("../scheduler.js")).default; +}); + +function makeApp() { + const app = express(); + app.use(express.json()); + app.use("/api/v1/admin/scheduler", schedulerRouter); + return app; +} + +describe("admin scheduler routes — unexpected errors", () => { + it("POST /pause returns 500 INTERNAL_ERROR on a non-Redis failure", async () => { + const res = await request(makeApp()) + .post("/api/v1/admin/scheduler/pause") + .set("x-chronopay-admin-token", ADMIN_TOKEN) + .send({ reason: "x", initiated_by: "alice" }); + expect(res.status).toBe(500); + expect(res.body.code).toBe("INTERNAL_ERROR"); + }); + + it("POST /resume returns 500 INTERNAL_ERROR on a non-Redis failure", async () => { + const res = await request(makeApp()) + .post("/api/v1/admin/scheduler/resume") + .set("x-chronopay-admin-token", ADMIN_TOKEN) + .send({ initiated_by: "alice" }); + expect(res.status).toBe(500); + expect(res.body.code).toBe("INTERNAL_ERROR"); + }); + + it("GET /status returns 500 INTERNAL_ERROR on a non-Redis failure", async () => { + const res = await request(makeApp()) + .get("/api/v1/admin/scheduler/status") + .set("x-chronopay-admin-token", ADMIN_TOKEN); + expect(res.status).toBe(500); + expect(res.body.code).toBe("INTERNAL_ERROR"); + }); +}); diff --git a/src/routes/admin/__tests__/scheduler.test.ts b/src/routes/admin/__tests__/scheduler.test.ts new file mode 100644 index 00000000..1e4c4652 --- /dev/null +++ b/src/routes/admin/__tests__/scheduler.test.ts @@ -0,0 +1,72 @@ +/** + * Covers the scheduler control-plane's defensive 500 path: when the underlying + * flag store throws something OTHER than RedisUnavailableError, the async + * handlers must translate it to a clean 500 rather than leaking an unhandled + * rejection. The redis module is mocked so we can inject a generic failure. + */ +import { jest } from "@jest/globals"; +import express from "express"; +import request from "supertest"; + +class FakeRedisUnavailableError extends Error { + code = "REDIS_UNAVAILABLE"; +} + +jest.unstable_mockModule("../../../redis.js", () => ({ + RedisUnavailableError: FakeRedisUnavailableError, + SCHEDULER_PAUSED_KEY: "scheduler:paused", + pauseScheduler: jest.fn(async () => { + throw new Error("boom"); + }), + resumeScheduler: jest.fn(async () => { + throw new Error("boom"); + }), + readSchedulerPauseState: jest.fn(async () => { + throw new Error("boom"); + }), + setRedisClient: jest.fn(), +})); + +const ADMIN_TOKEN = "test-admin-token-scheduler-errors"; +process.env.CHRONOPAY_ADMIN_TOKEN = ADMIN_TOKEN; + +let schedulerRouter: express.Router; + +beforeAll(async () => { + schedulerRouter = (await import("../scheduler.js")).default; +}); + +function makeApp() { + const app = express(); + app.use(express.json()); + app.use("/api/v1/admin/scheduler", schedulerRouter); + return app; +} + +describe("admin scheduler routes — unexpected errors", () => { + it("POST /pause returns 500 INTERNAL_ERROR on a non-Redis failure", async () => { + const res = await request(makeApp()) + .post("/api/v1/admin/scheduler/pause") + .set("x-chronopay-admin-token", ADMIN_TOKEN) + .send({ reason: "x", initiated_by: "alice" }); + expect(res.status).toBe(500); + expect(res.body.code).toBe("INTERNAL_ERROR"); + }); + + it("POST /resume returns 500 INTERNAL_ERROR on a non-Redis failure", async () => { + const res = await request(makeApp()) + .post("/api/v1/admin/scheduler/resume") + .set("x-chronopay-admin-token", ADMIN_TOKEN) + .send({ initiated_by: "alice" }); + expect(res.status).toBe(500); + expect(res.body.code).toBe("INTERNAL_ERROR"); + }); + + it("GET /status returns 500 INTERNAL_ERROR on a non-Redis failure", async () => { + const res = await request(makeApp()) + .get("/api/v1/admin/scheduler/status") + .set("x-chronopay-admin-token", ADMIN_TOKEN); + expect(res.status).toBe(500); + expect(res.body.code).toBe("INTERNAL_ERROR"); + }); +}); diff --git a/src/routes/admin/scheduler.ts b/src/routes/admin/scheduler.ts new file mode 100644 index 00000000..84831d4f --- /dev/null +++ b/src/routes/admin/scheduler.ts @@ -0,0 +1,153 @@ +/** + * @file src/routes/admin/scheduler.ts + * + * Admin control-plane for the incident scheduler kill-switch. Mounted under + * `/api/v1/admin/scheduler`. + * + * POST /pause – freeze new booking-intent creation platform-wide + * POST /resume – lift the freeze + * GET /status – read the current pause state (read path) + * + * Every mutating action: + * - Requires the shared admin token (`requireAdminToken`). + * - Requires an explicit `initiated_by` in the body so the acting operator is + * recorded rather than the anonymous shared token. + * - Increments the relevant Prometheus counter. + * - Broadcasts the new status on the WebSocket status bus. + * - Writes a best-effort audit event. + */ + +import { Router, type Request, type Response } from "express"; +import { requireAdminToken } from "../../middleware/authorization.js"; +import { + pauseScheduler, + resumeScheduler, + readSchedulerPauseState, + RedisUnavailableError, +} from "../../redis.js"; +import { schedulerPauseTotal, schedulerResumeTotal } from "../../metrics.js"; +import { broadcastSchedulerStatus } from "../../services/schedulerStatusBus.js"; +import { defaultAuditLogger } from "../../services/auditLogger.js"; +import { logger } from "../../utils/logger.js"; + +const router = Router(); + +function auditFireAndForget( + action: string, + context: Record, + req: Request, + status: number | string, +): void { + /* istanbul ignore next -- req.ip is always populated behind supertest/Express; + the socket fallback only matters in exotic transports */ + const actorIp = req.ip || req.socket?.remoteAddress; + void defaultAuditLogger + .log(action, { method: req.method, context }, { actorIp, resource: req.originalUrl, status }) + .catch(() => undefined); +} + +function readStringField(body: unknown, ...keys: string[]): string { + const source = body && typeof body === "object" ? (body as Record) : {}; + for (const key of keys) { + const value = source[key]; + if (typeof value === "string" && value.trim().length > 0) { + return value.trim(); + } + } + return ""; +} + +function redisUnavailable(res: Response): Response { + return res.status(503).json({ + success: false, + code: "REDIS_UNAVAILABLE", + error: "Cannot reach Redis to read or update the scheduler flag.", + }); +} + +/** + * Central error mapping for the async handlers. Async Express-4 handlers must + * not `throw` (the rejection would not reach the error middleware), so every + * handler funnels failures through here. + */ +function handleError(res: Response, err: unknown): Response { + if (err instanceof RedisUnavailableError) { + return redisUnavailable(res); + } + logger.error({ err }, "scheduler control-plane: unexpected error"); + return res.status(500).json({ + success: false, + code: "INTERNAL_ERROR", + error: "Internal server error", + }); +} + +// --------------------------------------------------------------------------- +// POST /pause +// --------------------------------------------------------------------------- +router.post("/pause", requireAdminToken, async (req: Request, res: Response) => { + const reason = readStringField(req.body, "reason"); + const initiatedBy = readStringField(req.body, "initiated_by", "initiatedBy"); + + if (!reason) { + return res + .status(400) + .json({ success: false, code: "INVALID_REASON", error: "reason is required" }); + } + if (!initiatedBy) { + return res.status(400).json({ + success: false, + code: "INVALID_INITIATED_BY", + error: "initiated_by is required", + }); + } + + try { + const state = await pauseScheduler({ reason, initiatedBy }); + schedulerPauseTotal.inc(); + broadcastSchedulerStatus(state); + auditFireAndForget("SCHEDULER_PAUSED", { reason, initiatedBy }, req, 200); + return res.status(200).json({ success: true, scheduler: state }); + } catch (err) { + return handleError(res, err); + } +}); + +// --------------------------------------------------------------------------- +// POST /resume +// --------------------------------------------------------------------------- +router.post("/resume", requireAdminToken, async (req: Request, res: Response) => { + const initiatedBy = readStringField(req.body, "initiated_by", "initiatedBy"); + + if (!initiatedBy) { + return res.status(400).json({ + success: false, + code: "INVALID_INITIATED_BY", + error: "initiated_by is required", + }); + } + + try { + const state = await resumeScheduler({ initiatedBy }); + schedulerResumeTotal.inc(); + broadcastSchedulerStatus(state); + auditFireAndForget("SCHEDULER_RESUMED", { initiatedBy }, req, 200); + return res.status(200).json({ success: true, scheduler: state }); + } catch (err) { + return handleError(res, err); + } +}); + +// --------------------------------------------------------------------------- +// GET /status — read path, safe to call during a freeze +// --------------------------------------------------------------------------- +router.get("/status", requireAdminToken, async (_req: Request, res: Response) => { + try { + const state = await readSchedulerPauseState(); + return res.status(200).json({ success: true, scheduler: state }); + } catch (err) { + return handleError(res, err); + } +}); + +export default router; diff --git a/src/routes/booking-intents.ts b/src/routes/booking-intents.ts index 98bee648..5a7dc47e 100644 --- a/src/routes/booking-intents.ts +++ b/src/routes/booking-intents.ts @@ -13,6 +13,7 @@ import { Router, type Request, Response } from "express"; import { requireAuthenticatedActor } from "../middleware/auth.js"; import { requireFeatureFlag } from "../middleware/featureFlags.js"; +import { schedulerGate } from "../middleware/schedulerGate.js"; import { auditMiddleware } from "../middleware/audit.js"; import { createAuthAwareRateLimiter } from "../middleware/rateLimiter.js"; import { idempotencyMiddleware } from "../middleware/idempotency.js"; @@ -70,6 +71,7 @@ export function createBookingIntentsRouter() { "/", requireFeatureFlag("CREATE_BOOKING_INTENT"), requireAuthenticatedActor(["customer", "admin"]), + schedulerGate, idempotencyMiddleware, createAuthAwareRateLimiter(), auditMiddleware("CREATE_BOOKING_INTENT"), diff --git a/src/services/__tests__/schedulerStatusBus.test.ts b/src/services/__tests__/schedulerStatusBus.test.ts new file mode 100644 index 00000000..6d190184 --- /dev/null +++ b/src/services/__tests__/schedulerStatusBus.test.ts @@ -0,0 +1,74 @@ +/** + * Unit tests for the scheduler status broadcast bus. + */ +import { jest } from "@jest/globals"; +import { + SCHEDULER_STATUS_CHANNEL, + broadcastSchedulerStatus, + onSchedulerStatus, + resetSchedulerStatusBus, + type SchedulerStatusEvent, +} from "../schedulerStatusBus.js"; + +describe("schedulerStatusBus", () => { + afterEach(() => { + resetSchedulerStatusBus(); + jest.restoreAllMocks(); + }); + + it("exposes the channel name", () => { + expect(SCHEDULER_STATUS_CHANNEL).toBe("scheduler:status"); + }); + + it("delivers the state (plus broadcastAt) to subscribers", () => { + const received: SchedulerStatusEvent[] = []; + onSchedulerStatus((e) => received.push(e)); + + const event = broadcastSchedulerStatus({ + paused: true, + reason: "incident", + initiatedBy: "alice", + pausedAt: "2026-07-31T00:00:00.000Z", + }); + + expect(received).toHaveLength(1); + expect(received[0]).toMatchObject({ + paused: true, + reason: "incident", + initiatedBy: "alice", + }); + expect(typeof received[0].broadcastAt).toBe("string"); + expect(event).toEqual(received[0]); + }); + + it("unsubscribe stops further delivery", () => { + const listener = jest.fn(); + const unsubscribe = onSchedulerStatus(listener); + + broadcastSchedulerStatus({ paused: true }); + unsubscribe(); + broadcastSchedulerStatus({ paused: false }); + + expect(listener).toHaveBeenCalledTimes(1); + }); + + it("a throwing subscriber never breaks the broadcast", () => { + onSchedulerStatus(() => { + throw new Error("subscriber blew up"); + }); + const good = jest.fn(); + onSchedulerStatus(good); + + // EventEmitter invokes listeners in order; a throw in the first would + // normally propagate, so broadcastSchedulerStatus must swallow it. + expect(() => broadcastSchedulerStatus({ paused: false })).not.toThrow(); + }); + + it("resetSchedulerStatusBus removes all subscribers", () => { + const listener = jest.fn(); + onSchedulerStatus(listener); + resetSchedulerStatusBus(); + broadcastSchedulerStatus({ paused: true }); + expect(listener).not.toHaveBeenCalled(); + }); +}); diff --git a/src/services/schedulerStatusBus.ts b/src/services/schedulerStatusBus.ts new file mode 100644 index 00000000..1843a6fe --- /dev/null +++ b/src/services/schedulerStatusBus.ts @@ -0,0 +1,67 @@ +/** + * @file src/services/schedulerStatusBus.ts + * + * In-process broadcast bus for scheduler pause/resume status changes. + * + * The realtime (WebSocket) layer subscribes to this bus via `onSchedulerStatus` + * and relays each event to connected clients, so operators and dashboards see a + * pause/resume take effect immediately instead of polling. Keeping the bus + * transport-agnostic (a plain Node `EventEmitter`) means the control-plane route + * has zero coupling to the socket implementation and the behaviour is trivially + * unit-testable. + * + * `broadcastSchedulerStatus` is deliberately fire-and-forget and never throws: + * a failure to notify subscribers must never fail the underlying pause/resume + * operation, which has already been persisted to Redis. + */ + +import { EventEmitter } from "events"; +import { logger } from "../utils/logger.js"; +import type { SchedulerPauseState } from "../redis.js"; + +/** Channel name the WebSocket bus fans this out on. */ +export const SCHEDULER_STATUS_CHANNEL = "scheduler:status"; + +export interface SchedulerStatusEvent extends SchedulerPauseState { + /** ISO timestamp of when the broadcast was emitted. */ + broadcastAt: string; +} + +export type SchedulerStatusListener = (event: SchedulerStatusEvent) => void; + +// A single shared emitter for the process. `setMaxListeners(0)` disables the +// default 10-listener warning — the WS layer may attach one listener per shard. +const emitter = new EventEmitter(); +emitter.setMaxListeners(0); + +/** + * Broadcast a scheduler status change to every subscriber. Never throws. + */ +export function broadcastSchedulerStatus(state: SchedulerPauseState): SchedulerStatusEvent { + const event: SchedulerStatusEvent = { + ...state, + broadcastAt: new Date().toISOString(), + }; + + try { + emitter.emit(SCHEDULER_STATUS_CHANNEL, event); + } catch (err) { + // A misbehaving subscriber must not break the pause/resume flow. + logger.warn({ err }, "schedulerStatusBus: subscriber threw during broadcast"); + } + + return event; +} + +/** + * Subscribe to scheduler status changes. Returns an unsubscribe function. + */ +export function onSchedulerStatus(listener: SchedulerStatusListener): () => void { + emitter.on(SCHEDULER_STATUS_CHANNEL, listener); + return () => emitter.off(SCHEDULER_STATUS_CHANNEL, listener); +} + +/** Remove every subscriber. Primarily for test isolation. */ +export function resetSchedulerStatusBus(): void { + emitter.removeAllListeners(SCHEDULER_STATUS_CHANNEL); +}