diff --git a/.env.example b/.env.example index 385ebeeb..90fafce4 100644 --- a/.env.example +++ b/.env.example @@ -24,3 +24,11 @@ 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= + +# Per-tenant leaky-bucket rate limiting for /api/v1/bookings/search. +# Sustained requests per second allowed per tenant (default 60). +BOOKINGS_SEARCH_RATE_PER_SECOND=60 +# Maximum burst capacity per tenant (default 120). +BOOKINGS_SEARCH_BURST=120 +# Hard timeout for the rate-limiter Redis call before failing open (default 250 ms). +BOOKINGS_SEARCH_REDIS_TIMEOUT_MS=250 diff --git a/docs/api/bookings-search-test-notes.md b/docs/api/bookings-search-test-notes.md new file mode 100644 index 00000000..1fa3be24 --- /dev/null +++ b/docs/api/bookings-search-test-notes.md @@ -0,0 +1,63 @@ +# Test notes — per-tenant leaky-bucket for /bookings/search + +Captured on branch `feat/tenant-leaky-bucket-search`. +Command per `package.json`: `npm test` (jest, `--runInBand`). + +## New suites (this change) + +``` +PASS src/middleware/__tests__/tenantLeakyBucket.redis.test.ts +PASS src/middleware/__tests__/tenantLeakyBucket.test.ts +PASS src/routes/__tests__/bookings.test.ts +Test Suites: 3 passed, 3 total +Tests: 62 passed, 62 total +``` + +Coverage (scoped to the two new modules, `--collectCoverageFrom`): + +| File | Stmts | Branch | Funcs | Lines | +| --- | --- | --- | --- | --- | +| src/middleware/tenantLeakyBucket.ts | 100% | 97.14% | 100% | 100% | +| src/routes/bookings.ts | 100% | 98.27% | 100% | 100% | +| **All files** | **100%** | **97.65%** | **100%** | **100%** | + +Residual uncovered branches are defensive by design: the production +`require("ioredis")` fallback (unreachable under test injection) and a +null-guard for `auth.userId` that the auth middleware makes unreachable over +HTTP. Both exceed the 95% requirement. + +## Edge cases from the issue — where each is covered + +| Required edge case | Test | +| --- | --- | +| Burst then sustained | `tenantLeakyBucket.test.ts` → “burst then sustained: 120-request burst passes, 121st is throttled, bucket refills at 60 rps” (also E2E in `bookings.test.ts`) | +| Tenant switch mid-connection | `tenantLeakyBucket.test.ts` → “tenant switch mid-connection charges the correct bucket each request” | +| Redis latency spike | `tenantLeakyBucket.test.ts` → “enforces the timeout on a latency spike instead of hanging” + “a slow store fails open within the timeout and the response completes” | +| Noisy tenant can't starve others | `tenantLeakyBucket.test.ts` → “noisy tenant cannot starve other tenants”; `bookings.test.ts` → E2E | +| Atomicity (Lua on Redis) | `tenantLeakyBucket.redis.test.ts` → Lua ↔ JS parity on a real Lua VM (ioredis-mock): burst, throttle, drain, clock skew, EVALSHA caching, NOSCRIPT fallback | +| 429 + Retry-After | middleware + route tests assert header and body contract | +| Metric | `rate_limit_bucket_burn{tenant}` set on every request; `rate_limit_redis_failures_total` on fail-open | + +## Full-suite regression check (`npm test`) + +The repository's `main` is already heavily broken upstream (654 failing tests +across 175 suites: bad upstream commits with syntax errors in +`marketplaceSearchSchema.ts` / `marketplaceSearchService.ts`, missing exports +like `FraudReasonCode`, undefined helpers like `resetSeniorPool`, etc.). To +compare like-for-like, all 257 suites were executed in chunks on both +`origin/main` and this branch in an otherwise memory-constrained sandbox (a +single monolithic `--runInBand` process OOMs after ~175 suites even on main). + +Result: + +- Failing suites on `main`: 175 +- Failing suites on this branch: 171 (all also failing on `main`) +- Suites failing **only** on this branch (regressions): **0** +- Two suites (`reranker`, `db-instrumentation`) failed on `main`'s chunk run + but pass when re-run individually on `main` → flaky pre-existing tests, + unrelated to this change. + +`tsc --noEmit`: 7 errors before and after the change — all pre-existing +syntax errors on `main`; this change adds zero compile errors. ESLint: clean +for every new/modified file (2 pre-existing unused-function errors in +`src/config/env.ts` remain on `main`, untouched). diff --git a/docs/api/bookings-search.md b/docs/api/bookings-search.md new file mode 100644 index 00000000..e358a7f7 --- /dev/null +++ b/docs/api/bookings-search.md @@ -0,0 +1,171 @@ +# Bookings Search API — `/api/v1/bookings/search` + +Search endpoint for booking records, protected by a **per-tenant leaky-bucket +rate limiter** (60 rps sustained, 120 burst) instead of the coarse global +fixed-window limiter used elsewhere. This document covers the endpoint itself +and the rate-limiting behavior clients and operators must understand. + +Implementation: + +| Piece | Location | +| --- | --- | +| Router | `src/routes/bookings.ts` | +| Per-tenant leaky-bucket middleware | `src/middleware/tenantLeakyBucket.ts` | +| Metrics | `src/metrics.ts` (`rate_limit_bucket_burn`, `rate_limit_redis_failures_total`) | +| Config | `src/config/env.ts` (`BOOKINGS_SEARCH_*`) | + +--- + +## Why per-tenant instead of the global limiter? + +The previous configuration shared one fixed-window budget across traffic. A +single misbehaving tenant (retry storm, broken sync loop, aggressive scraper) +could exhaust the window and starve every other tenant — the classic *noisy +neighbor* failure. The per-tenant leaky bucket gives **each tenant its own +independent budget**, so an abusive tenant can only ever throttle itself: + +``` +tenant A: [████████████████████████████] 429 — throttled, burns alone +tenant B: [█░░░░░░░░░░░░░░░░░░░░░░░░] 200 — unaffected +``` + +## Algorithm + +Leaky bucket, evaluated atomically inside Redis (Lua script): + +- Each admitted request adds **1 token** to the caller's bucket. +- The bucket **drains at a constant rate**: `BOOKINGS_SEARCH_RATE_PER_SECOND` + tokens/second (default **60**). +- Requests are admitted while `level + 1 <= BOOKINGS_SEARCH_BURST` (default + **120**). + +Consequences: + +- **Burst:** up to 120 requests may land simultaneously. +- **Sustained:** long-run throughput converges to exactly 60 rps. +- **Recovery:** after throttling, capacity returns continuously at 60 rps + (no hard window boundary, so no thundering-herd at window reset). + +The Lua script runs via `EVALSHA` with an automatic `EVAL` fallback on +`NOSCRIPT`, so the read-modify-write of bucket state is atomic across all app +instances — concurrent requests cannot double-spend capacity. + +### Redis state + +- Key: `rlb:bookings:search::` (hash `{level, ts}`). +- TTL: `2 × (burst / rate) + 1` seconds (5 s with defaults), refreshed on + every request. Buckets of quiet tenants expire automatically — no memory + leak from one-off tenants. + +## Tenant identity resolution + +**Tenant keys are derived from trusted auth context only.** Resolution order: + +1. `req.auth.tenantId` / `req.user.tenantId` — true tenant claims. +2. `req.auth.userId` / `req.user.sub || req.user.id` — the user is the tenant boundary. +3. `req.apiKeyId` — partner API keys map 1:1 to tenants (SHA-256 hashed). +4. Client IP (SHA-256 hashed) — anonymous traffic gets its own bucket and + cannot starve authenticated tenants. + +> [!IMPORTANT] +> The `x-tenant-id` **request header is never consulted**: trusting it would +> let any caller mint a fresh, empty bucket per request (limit evasion) and +> would flood Redis and Prometheus with unbounded keys. Identifiers are also +> canonicalized (strict charset + 128-char cap, otherwise replaced by a +> deterministic hash) before being embedded in keys, preventing Redis +> key-injection across logical namespaces. + +## Headers + +Every response (admitted or rejected) includes: + +| Header | Meaning | +| --- | --- | +| `X-RateLimit-Limit` | Burst capacity (120). | +| `X-RateLimit-Remaining` | Tokens remaining in the tenant's bucket right now. | +| `X-RateLimit-Reset` | Epoch seconds when the bucket will have fully drained. | +| `Retry-After` | **Only on 429.** Whole seconds until the request can be retried successfully. Computed from live bucket state, not a constant. | + +## Responses + +**Admitted** + +```json +{ "success": true, "data": { "results": [], "total": 0, "limit": 50, "offset": 0 } } +``` + +**Throttled — HTTP 429** + +```json +{ "success": false, "error": "Too many requests, please try again later.", "retryAfter": 1 } +``` + +Client guidance: respect `Retry-After`, add jitter, and never retry 429s in a +tight loop — a retry storm only re-fills your own bucket with condemned +requests. + +## Failure modes + +The limiter is **fail-open**: if Redis is unreachable or slower than +`BOOKINGS_SEARCH_REDIS_TIMEOUT_MS` (default 250 ms), the request is admitted +without accounting and `rate_limit_redis_failures_total` increments. + +Rationale: rate limiting exists to protect the endpoint; it must never take +the endpoint down with it. During a Redis outage, traffic degrades to the +pre-fix (unlimited) behavior and protection recovers automatically when Redis +returns. If `rate_limit_redis_failures_total` is sustained non-zero, treat it +as a production incident (Redis health), not as an application bug. + +A too-slow Redis additionally cannot hang request handling — every store call +is wrapped in the hard `BOOKINGS_SEARCH_REDIS_TIMEOUT_MS` budget. + +## Metrics + +| Metric | Type | Labels | Meaning | +| --- | --- | --- | --- | +| `rate_limit_bucket_burn` | Gauge | `tenant` | Live bucket fill level per tenant. Rising ⇒ about to be throttled; pinned at 120 ⇒ being throttled. Cardinality-budgeted (256) — excess tenants fold into the shared overflow label instead of harming the metrics pipeline. | +| `rate_limit_redis_failures_total` | Counter | — | Fail-open events (Redis error/timeout). Healthy steady state is 0. | + +## Configuration + +| Env var | Default | Purpose | +| --- | --- | --- | +| `BOOKINGS_SEARCH_RATE_PER_SECOND` | `60` | Sustained rps per tenant (leak rate). | +| `BOOKINGS_SEARCH_BURST` | `120` | Per-tenant burst capacity. | +| `BOOKINGS_SEARCH_REDIS_TIMEOUT_MS` | `250` | Hard budget for the limiter Redis call before failing open. | + +Tuning notes: keep `burst >= 2 × rate` so dashboards and batch syncs can +burst naturally; raise `rate` only together with capacity planning on the +search backing store. + +## Endpoint + +`GET /api/v1/bookings/search` + +Auth: `x-chronopay-user-id` + role header (`customer`, `professional`, +`admin`, `support`) — callers only ever see **their own** bookings. + +| Query param | Type | Description | +| --- | --- | --- | +| `q` | string ≤ 200 chars | Case-insensitive match over id, slotId, professional, note. | +| `status` | enum | `pending`, `confirmed`, `firm`, `cancelled`, `expired`, `hold_placed`, `hold_refunded`. | +| `slotId` | string | Exact slot filter. | +| `from`, `to` | ISO 8601 | Only bookings overlapping `[from, to]`. One-sided ranges allowed. `from` must not be after `to`. | +| `limit` | int 1–100 (default 50) | Page size. | +| `offset` | int ≥ 0 (default 0) | Page offset. | + +Validation failures return `400 { success: false, error }`. + +## Tests + +- `src/middleware/__tests__/tenantLeakyBucket.test.ts` — decision engine, + tenant resolution/security, middleware behavior, burst-then-sustained, + tenant switch mid-connection, fail-open, latency-spike timeout. +- `src/middleware/__tests__/tenantLeakyBucket.redis.test.ts` — proves the + Lua script on a real Lua VM (`ioredis-mock`) behaves identically to the JS + decision engine (burst, throttle, drain, clock skew), plus EVALSHA caching + and store lifecycle. +- `src/routes/__tests__/bookings.test.ts` — endpoint search/filter/pagination + contract and the end-to-end noisy-neighbor scenario. + +Coverage for both new modules: 100% lines / 100% functions / ≥97% branches. diff --git a/src/app.ts b/src/app.ts index 5107d86d..b6bd0db5 100644 --- a/src/app.ts +++ b/src/app.ts @@ -48,6 +48,7 @@ function isTruthyEnvValue(value: string | undefined): boolean { // Import routers import checkoutRouter from "./routes/checkout.js"; +import bookingsRouter from "./routes/bookings.js"; import buyerProfileRouter from "./buyer-profile/buyer-profile.routes.js"; import oauth2Router from "./routes/oauth2.js"; import adminRouter from "./routes/admin.js"; @@ -640,6 +641,9 @@ export function createApp(options: AppFactoryOptions = {}) { }, ); + // 4b. Bookings Search Routes (per-tenant leaky-bucket: 60 rps sustained / 120 burst) + app.use("/api/v1/bookings", bookingsRouter); + // 5. Webhooks Routes registerWebhookRoutes(app); app.use("/api/v1", webhookRoutes); diff --git a/src/config/env.ts b/src/config/env.ts index a23073be..2a0977d9 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -45,6 +45,12 @@ export interface EnvConfig { internalOverrideSecretPrev?: string; /** Acceptable clock skew (ms) for the bypass timestamp. Default 30 000. */ internalBypassToleranceMs: number; + /** Sustained requests per second per tenant on /api/v1/bookings/search. Default 60. */ + bookingsSearchRatePerSecond: number; + /** Burst capacity per tenant on /api/v1/bookings/search. Default 120. */ + bookingsSearchBurst: number; + /** Hard timeout (ms) for the leaky-bucket Redis call before failing open. Default 250. */ + bookingsSearchRedisTimeoutMs: number; } export class EnvValidationError extends Error { @@ -91,6 +97,25 @@ export function loadEnvConfig(env: NodeJS.ProcessEnv = process.env): EnvConfig { issues, ); + const bookingsSearchRatePerSecond = parsePositiveInteger( + env.BOOKINGS_SEARCH_RATE_PER_SECOND, + "BOOKINGS_SEARCH_RATE_PER_SECOND", + 60, + issues, + ); + const bookingsSearchBurst = parsePositiveInteger( + env.BOOKINGS_SEARCH_BURST, + "BOOKINGS_SEARCH_BURST", + 120, + issues, + ); + const bookingsSearchRedisTimeoutMs = parsePositiveInteger( + env.BOOKINGS_SEARCH_REDIS_TIMEOUT_MS, + "BOOKINGS_SEARCH_REDIS_TIMEOUT_MS", + 250, + issues, + ); + if (issues.length > 0) { throw new EnvValidationError(issues); } @@ -113,6 +138,9 @@ export function loadEnvConfig(env: NodeJS.ProcessEnv = process.env): EnvConfig { internalOverrideSecret, internalOverrideSecretPrev, internalBypassToleranceMs, + bookingsSearchRatePerSecond, + bookingsSearchBurst, + bookingsSearchRedisTimeoutMs, }; } diff --git a/src/metrics.ts b/src/metrics.ts index dc322811..849733d4 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -701,6 +701,38 @@ export const fairQueueBypassAttempts = createBudgetedCounter({ registers: [register], }); +// ─── Per-tenant leaky-bucket rate limiter metrics ──────────────────────────── + +/** + * Gauge of the current leaky-bucket fill level per tenant. Rising towards the + * configured burst capacity means the tenant is about to be throttled; + * a level pinned at capacity means the tenant is actively limited. + * Cardinality-budgeted so a flood of distinct tenants degrades to the + * shared overflow label rather than exhausting the metrics pipeline. + */ +export const rateLimitBucketBurn = createBudgetedGauge({ + name: "rate_limit_bucket_burn", + help: "Current leaky-bucket fill level (tokens charged) per tenant on rate-limited search endpoints", + labels: ["tenant"], + budget: 256, + buckets: [], + registers: [register], +}); + +/** + * Counter incremented each time the per-tenant leaky-bucket limiter fails + * open because Redis errored or exceeded the store timeout. Zero is the + * healthy steady state; a sustained non-zero rate means rate limiting is + * currently degraded (requests allowed without accounting). + */ +export const rateLimitRedisFailuresTotal = createBudgetedCounter({ + name: "rate_limit_redis_failures_total", + help: "Total number of times the tenant leaky-bucket limiter failed open due to Redis errors or timeouts", + labels: [], + budget: 0, + registers: [register], +}); + /** * Histogram tracking per-request SQL wall-clock time in milliseconds. * Labels: route, outcome (ok | breached). diff --git a/src/middleware/__tests__/tenantLeakyBucket.redis.test.ts b/src/middleware/__tests__/tenantLeakyBucket.redis.test.ts new file mode 100644 index 00000000..737a9a60 --- /dev/null +++ b/src/middleware/__tests__/tenantLeakyBucket.redis.test.ts @@ -0,0 +1,211 @@ +/** + * Integration test proving the production Lua script (executed on a real Lua + * VM via ioredis-mock) behaves *identically* to the pure JS decision engine + * powering the in-memory store. This guards the invariant that + * LEAKY_BUCKET_LUA and decideLeakyBucket must never drift apart, without + * needing a live Redis server in CI. + * + * `Date.now` is spied (not fake timers) so the mock's event-loop plumbing + * keeps working while the store sees a deterministic clock. + */ + +import { jest } from "@jest/globals"; +import RedisMock from "ioredis-mock"; + +const { + RedisLeakyBucketStore, + InMemoryLeakyBucketStore, + setTenantLeakyBucketStore, + resetTenantLeakyBucketStore, + getTenantLeakyBucketStore, + _setRedisCtorForTesting, + createTenantLeakyBucketRateLimiter, +} = await import("../tenantLeakyBucket.js"); + +const T0 = 1_700_000_000_000; +const PARAMS = { ratePerSecond: 60, capacity: 120 }; + +describe("Lua script ↔ JS decision engine parity (ioredis-mock Lua VM)", () => { + let now: number; + let nowSpy: ReturnType; + let redisStore: InstanceType; + let memoryStore: InstanceType; + + async function both(key: string) { + const fromRedis = await redisStore.consume(key, PARAMS); + const fromMemory = await memoryStore.consume(key, PARAMS); + return { fromRedis, fromMemory }; + } + + beforeEach(() => { + now = T0; + + nowSpy = jest.spyOn(Date, "now").mockImplementation(() => now) as any; + + redisStore = new RedisLeakyBucketStore(new RedisMock() as any, 1000); + memoryStore = new InMemoryLeakyBucketStore(() => now); + }); + + afterEach(() => { + nowSpy.mockRestore(); + }); + + it("120-request burst, throttle, drain and clock-skew behave identically on both stores", async () => { + const key = "rlb:bookings:search:user:parity"; + + // Burst: both stores admit exactly the 120-token burst… + for (let i = 0; i < 120; i++) { + const { fromRedis, fromMemory } = await both(key); + expect(fromRedis).toEqual(fromMemory); + expect(fromRedis.allowed).toBe(true); + } + + // …and throttle request 121 with the same retry-after. + const denied = await both(key); + expect(denied.fromRedis).toEqual(denied.fromMemory); + expect(denied.fromRedis).toEqual({ allowed: false, retryAfterMs: 17, level: 120 }); + + // Drain 60 tokens over 1 s on both. + now += 1000; + const refilled = await both(key); + expect(refilled.fromRedis).toEqual(refilled.fromMemory); + expect(refilled.fromRedis.allowed).toBe(true); + expect(refilled.fromRedis.level).toBeCloseTo(61); + + // Clock skew (now jumps backwards) must not inflate either bucket. + now -= 30_000; + const skewed = await both(key); + expect(skewed.fromRedis).toEqual(skewed.fromMemory); + expect(skewed.fromRedis.level).toBeGreaterThan(60); + expect(skewed.fromRedis.level).toBeLessThanOrEqual(120); + }); + + it("keeps tenant buckets independent on Redis too", async () => { + for (let i = 0; i < 121; i++) await redisStore.consume("rlb:bookings:search:user:a", PARAMS); + const a = await redisStore.consume("rlb:bookings:search:user:a", PARAMS); + expect(a.allowed).toBe(false); + const b = await redisStore.consume("rlb:bookings:search:user:b", PARAMS); + expect(b).toEqual({ allowed: true, retryAfterMs: 0, level: 1 }); + }); + + it("evalsha-first path works: first call triggers NOSCRIPT then caches, subsequent calls are cached", async () => { + const first = await redisStore.consume("rlb:bookings:search:user:script", PARAMS); + expect(first.allowed).toBe(true); + const second = await redisStore.consume("rlb:bookings:search:user:script", PARAMS); + expect(second).toEqual({ allowed: true, retryAfterMs: 0, level: 2 }); + }); +}); + +describe("store lifecycle", () => { + afterEach(async () => { + await resetTenantLeakyBucketStore(); + }); + + it("resolves an in-memory store under NODE_ENV=test and accepts injection", () => { + const resolved = getTenantLeakyBucketStore(); + expect(resolved).toBeInstanceOf(InMemoryLeakyBucketStore); + const injected = new InMemoryLeakyBucketStore(); + setTenantLeakyBucketStore(injected); + expect(getTenantLeakyBucketStore()).toBe(injected); + }); + + it("memoizes the default store", () => { + expect(getTenantLeakyBucketStore()).toBe(getTenantLeakyBucketStore()); + }); + + it("builds the production store over an injected Redis ctor (options, error handler, lifecycle)", async () => { + const handlers = new Map void>(); + const disconnect = jest.fn(); + const ctorCalls: Array<{ url: string; options: Record }> = []; + + class FakeRedis { + constructor(url: string, options: Record) { + ctorCalls.push({ url, options }); + } + on(event: string, handler: (...args: unknown[]) => void) { + handlers.set(event, handler); + return this; + } + evalsha() { + return Promise.resolve([1, 0, 1000]); + } + eval() { + return Promise.resolve([1, 0, 1000]); + } + script() { + return Promise.resolve("cached"); + } + quit() { + return Promise.resolve("OK"); + } + disconnect = disconnect; + status = "ready"; + } + + const prevEnv = process.env.NODE_ENV; + try { + process.env.NODE_ENV = "development"; + _setRedisCtorForTesting(FakeRedis as never); + await resetTenantLeakyBucketStore(); + + const store = getTenantLeakyBucketStore(); + expect(store).toBeInstanceOf(RedisLeakyBucketStore); + expect(ctorCalls).toHaveLength(1); + expect(ctorCalls[0].url).toMatch(/^redis:\/\//); + expect(ctorCalls[0].options.lazyConnect).toBe(true); + + // retryStrategy caps backoff and gives up after 10 attempts + const retry = ctorCalls[0].options.retryStrategy as (n: number) => number | null; + expect(retry(1)).toBe(100); + expect(retry(20)).toBeNull(); + + // error handler is registered and does not throw + expect(handlers.has("error")).toBe(true); + handlers.get("error")?.(new Error("link down")); + + // the store actually works through the fake client (evalsha path) + const decision = await store.consume("rlb:bookings:search:user:prod", { + ratePerSecond: 60, + capacity: 120, + }); + expect(decision).toEqual({ allowed: true, retryAfterMs: 0, level: 1 }); + + // reset tears the cached client down via disconnect() + await resetTenantLeakyBucketStore(); + expect(disconnect).toHaveBeenCalledTimes(1); + } finally { + _setRedisCtorForTesting(undefined); + process.env.NODE_ENV = prevEnv; + await resetTenantLeakyBucketStore(); + } + }); +}); + +describe("safety net", () => { + it("an unrecoverable middleware failure still fails open (never crashes the pipeline)", async () => { + const limiter = createTenantLeakyBucketRateLimiter({ + ratePerSecond: 60, + capacity: 120, + store: new InMemoryLeakyBucketStore(), + }); + // A request object whose every property access explodes — no store error, + // no res use; the failure escapes the inner try/catch entirely. + const boobyTrappedReq = new Proxy( + {}, + { + get() { + throw new Error("catastrophic request corruption"); + }, + }, + ); + const res = { setHeader: jest.fn(), status: jest.fn(), json: jest.fn() }; + const next = jest.fn(); + + + limiter(boobyTrappedReq as any, res as any, next as any); + await new Promise((resolve) => setTimeout(resolve, 10)); + + expect(next).toHaveBeenCalledTimes(1); + expect(res.status).not.toHaveBeenCalled(); + }); +}); diff --git a/src/middleware/__tests__/tenantLeakyBucket.test.ts b/src/middleware/__tests__/tenantLeakyBucket.test.ts new file mode 100644 index 00000000..71215f2e --- /dev/null +++ b/src/middleware/__tests__/tenantLeakyBucket.test.ts @@ -0,0 +1,478 @@ +/** + * Tests for the per-tenant leaky-bucket rate limiter. + * + * Edge cases required by the issue are covered explicitly: + * - burst then sustained traffic ("burst then sustained") + * - tenant identity changing between requests ("tenant switch mid-connection") + * - slow/failing Redis ("Redis latency spike") → bounded wait + fail-open + */ + +import { jest } from "@jest/globals"; +import request from "supertest"; +import express, { type Request, type Response } from "express"; + +const { + decideLeakyBucket, + bucketTtlSeconds, + sanitizeTenantIdentifier, + resolveTenantIdentity, + createTenantLeakyBucketRateLimiter, + InMemoryLeakyBucketStore, + RedisLeakyBucketStore, + LeakyBucketRedisTimeoutError, + LEAKY_BUCKET_LUA, +} = await import("../tenantLeakyBucket.js"); + +const T0 = 1_700_000_000_000; + +/** Build an express app with a fake auth layer + the limiter under test. */ +function buildApp( + limiter: express.RequestHandler, + handlerResponses?: (req: Request, res: Response) => void, +) { + const app = express(); + app.use(express.json()); + app.use((req: Request, _res: Response, next: express.NextFunction) => { + const userId = req.header("x-user-id"); + const tenantId = req.header("x-auth-tenant-id"); // trusted auth-context claim (set by upstream auth) + if (userId || tenantId) { + + (req as any).auth = { userId, tenantId, role: "customer", claims: {} }; + } + next(); + }); + app.get( + "/search", + limiter, + handlerResponses ?? ((_req: Request, res: Response) => res.status(200).json({ success: true })), + ); + return app; +} + +describe("decideLeakyBucket (pure transition)", () => { + const base = { ratePerSecond: 60, capacity: 120 }; + + it("admits the first request on a fresh bucket", () => { + const d = decideLeakyBucket(0, T0, { ...base, nowMs: T0 }); + expect(d).toEqual({ allowed: true, level: 1, retryAfterMs: 0 }); + }); + + it("admits at the exact capacity boundary", () => { + const d = decideLeakyBucket(119, T0, { ...base, nowMs: T0 }); + expect(d.allowed).toBe(true); + expect(d.level).toBe(120); + }); + + it("rejects when the bucket is full and reports exact retry-after", () => { + // level 120, 1 more unit needs 1 token to drain: 1/60rps = 16.67ms → 17ms + const d = decideLeakyBucket(120, T0, { ...base, nowMs: T0 }); + expect(d.allowed).toBe(false); + expect(d.retryAfterMs).toBe(17); + expect(d.level).toBeCloseTo(120); + }); + + it("drains at the leak rate over elapsed time", () => { + // 1000ms at 60 rps drains 60 tokens: 120 → 60, request admitted at 61. + const d = decideLeakyBucket(120, T0, { ...base, nowMs: T0 + 1000 }); + expect(d.allowed).toBe(true); + expect(d.level).toBeCloseTo(61); + }); + + it("clamps drain at zero for long-idle buckets", () => { + const d = decideLeakyBucket(120, T0, { ...base, nowMs: T0 + 60_000 }); + expect(d.allowed).toBe(true); + expect(d.level).toBe(1); + }); + + it("never grows the bucket on clock skew (negative elapsed)", () => { + // now goes backwards relative to the stored timestamp (cross-instance skew) + const d = decideLeakyBucket(120, T0 + 5000, { ...base, nowMs: T0 }); + expect(d.allowed).toBe(false); // NOT rejected-by-growth; level stays 120 + expect(d.level).toBe(120); + }); + + it("rejects non-positive rate/capacity", () => { + expect(() => decideLeakyBucket(0, T0, { ratePerSecond: 0, capacity: 10, nowMs: T0 })).toThrow(); + expect(() => decideLeakyBucket(0, T0, { ratePerSecond: 60, capacity: -1, nowMs: T0 })).toThrow(); + expect(() => + decideLeakyBucket(0, T0, { ratePerSecond: Number.NaN, capacity: 10, nowMs: T0 }), + ).toThrow(); + }); +}); + +describe("bucketTtlSeconds", () => { + it("covers two full drain periods plus slack", () => { + expect(bucketTtlSeconds(60, 120)).toBe(5); // ceil(120/60*2)+1 + }); + it("never returns less than 1", () => { + expect(bucketTtlSeconds(1000, 1)).toBeGreaterThanOrEqual(1); + }); +}); + +describe("sanitizeTenantIdentifier", () => { + it("passes safe identifiers through unchanged", () => { + expect(sanitizeTenantIdentifier("tenant-42_x.y")).toBe("tenant-42_x.y"); + }); + + it("hashes identifiers containing key-injection characters", () => { + const out = sanitizeTenantIdentifier("tenant:vip"); + expect(out.startsWith("h:")).toBe(true); + expect(out).toHaveLength(2 + 32); + expect(out).not.toContain(":vip"); + }); + + it("hashes overlong identifiers", () => { + const out = sanitizeTenantIdentifier("x".repeat(500)); + expect(out.startsWith("h:")).toBe(true); + }); + + it("hashes empty/whitespace identifiers", () => { + expect(sanitizeTenantIdentifier(" ")).toMatch(/^h:/); + }); +}); + +describe("resolveTenantIdentity", () => { + function reqWith(overrides: Record): express.Request { + return { + auth: overrides.auth, + user: overrides.user, + apiKeyId: overrides.apiKeyId, + headers: (overrides.headers ?? {}) as Record, + socket: { remoteAddress: overrides.ip ?? "10.0.0.1" }, + + } as any; + } + + it("prefers the trusted auth tenant claim over everything else", () => { + const identity = resolveTenantIdentity( + reqWith({ auth: { tenantId: "acme", userId: "u1" }, apiKeyId: "k9" }), + ); + expect(identity.key).toBe("rlb:bookings:search:tenant:acme"); + expect(identity.label).toBe("acme"); + }); + + it("falls back to the authenticated user id as the tenant boundary", () => { + const identity = resolveTenantIdentity(reqWith({ auth: { userId: "user-123" } })); + expect(identity.key).toBe("rlb:bookings:search:user:user-123"); + }); + + it("supports JWT user claims (sub / tenantId)", () => { + expect(resolveTenantIdentity(reqWith({ user: { sub: "u-sub" } })).key).toBe( + "rlb:bookings:search:user:u-sub", + ); + expect(resolveTenantIdentity(reqWith({ user: { tenantId: "t-1", sub: "u" } })).key).toBe( + "rlb:bookings:search:tenant:t-1", + ); + }); + + it("hashes API keys (never stores the raw key)", () => { + const identity = resolveTenantIdentity(reqWith({ apiKeyId: "super-secret-key" })); + expect(identity.key).toMatch(/^rlb:bookings:search:apiKey:[0-9a-f]{32}$/); + expect(identity.key).not.toContain("super-secret-key"); + }); + + it("never trusts a raw x-tenant-id request header (evasion prevention)", () => { + const identity = resolveTenantIdentity( + reqWith({ headers: { "x-tenant-id": "vip" }, ip: "10.0.0.9" }), + ); + // Without auth context the header is ignored → hashed-IP bucket. + expect(identity.key).toMatch(/^rlb:bookings:search:ip:/); + expect(identity.key).not.toContain("vip"); + }); + + it("hashes IPs so raw addresses never land in Redis keys", () => { + const identity = resolveTenantIdentity(reqWith({ ip: "203.0.113.7" })); + expect(identity.key).toMatch(/^rlb:bookings:search:ip:[0-9a-f]{32}$/); + expect(identity.key).not.toContain("203.0.113.7"); + }); + + it("falls back to a stable anonymous bucket when no IP is discoverable", () => { + + const identity = resolveTenantIdentity({ headers: {} } as any); + expect(identity.key).toMatch(/^rlb:bookings:search:ip:[0-9a-f]{32}$/); + expect(identity.label).toMatch(/^ip:[0-9a-f]{16}$/); + }); +}); + +describe("createTenantLeakyBucketRateLimiter (middleware, deterministic clock)", () => { + let now: number; + let store: InstanceType; + let limiter: express.RequestHandler; + let app: express.Express; + + beforeEach(() => { + now = T0; + store = new InMemoryLeakyBucketStore(() => now); + limiter = createTenantLeakyBucketRateLimiter({ ratePerSecond: 60, capacity: 120, store }); + app = buildApp(limiter); + }); + + it("constructor validates its parameters", () => { + expect(() => createTenantLeakyBucketRateLimiter({ ratePerSecond: 0 })).toThrow(); + expect(() => createTenantLeakyBucketRateLimiter({ capacity: 0 })).toThrow(); + expect(() => createTenantLeakyBucketRateLimiter({ amount: 0 })).toThrow(); + }); + + it("admits traffic under the burst ceiling and sets X-RateLimit headers", async () => { + const res = await request(app).get("/search").set("x-user-id", "alice").expect(200); + expect(res.headers["x-ratelimit-limit"]).toBe("120"); + expect(res.headers["x-ratelimit-remaining"]).toBe("119"); + expect(res.headers["x-ratelimit-reset"]).toBeDefined(); + expect(res.headers["retry-after"]).toBeUndefined(); + }); + + it("burst then sustained: 120-request burst passes, 121st is throttled, bucket refills at 60 rps", async () => { + const headers = { "x-user-id": "noisy" }; + + // 1) Full burst is absorbed (capacity = 120). + for (let i = 0; i < 120; i++) { + await request(app).get("/search").set(headers).expect(200); + } + + // 2) Burst exhausted → immediate retry is throttled with a precise Retry-After. + const denied = await request(app).get("/search").set(headers).expect(429); + expect(denied.headers["retry-after"]).toBe("1"); + expect(denied.headers["x-ratelimit-remaining"]).toBe("0"); + expect(denied.body).toMatchObject({ success: false, retryAfter: 1 }); + + // 3) 100ms later exactly 6 tokens have drained (60 rps): 6 pass, 7th throttled. + now += 100; + for (let i = 0; i < 6; i++) { + await request(app).get("/search").set(headers).expect(200); + } + await request(app).get("/search").set(headers).expect(429); + + // 4) After a full second of drain, exactly 60 succeed → sustained 60 rps. + now += 1000; + for (let i = 0; i < 60; i++) { + await request(app).get("/search").set(headers).expect(200); + } + await request(app).get("/search").set(headers).expect(429); + + // 5) Steady state: another second, exactly 60 more — never more than the leak rate. + now += 1000; + for (let i = 0; i < 60; i++) { + await request(app).get("/search").set(headers).expect(200); + } + await request(app).get("/search").set(headers).expect(429); + }); + + it("noisy tenant cannot starve other tenants (per-tenant isolation)", async () => { + // Tenant A exhausts its entire bucket. + for (let i = 0; i < 121; i++) { + await request(app).get("/search").set("x-user-id", "tenant-a"); + } + await request(app).get("/search").set("x-user-id", "tenant-a").expect(429); + + // Tenant B's bucket is untouched. + const res = await request(app).get("/search").set("x-user-id", "tenant-b").expect(200); + expect(res.headers["x-ratelimit-remaining"]).toBe("119"); + }); + + it("tenant switch mid-connection charges the correct bucket each request", async () => { + // Simulate a shared/proxied keep-alive connection alternating identities. + await request(app).get("/search").set("x-user-id", "first").expect(200); + await request(app).get("/search").set("x-user-id", "second").expect(200); + await request(app).get("/search").set("x-user-id", "first").expect(200); + + // Each identity owns an independent bucket with independent levels. + expect(store.size).toBe(2); + const r = await request(app).get("/search").set("x-user-id", "first").expect(200); + expect(r.headers["x-ratelimit-remaining"]).toBe("117"); // first: 3 requests total + const r2 = await request(app).get("/search").set("x-user-id", "second").expect(200); + expect(r2.headers["x-ratelimit-remaining"]).toBe("118"); // second: 2 requests total + }); + + it("trusted tenant claims override the per-user bucket", async () => { + // Same user, two different trusted tenant claims → two distinct buckets. + await request(app) + .get("/search") + .set("x-user-id", "shared-user") + .set("x-auth-tenant-id", "tenant-x") + .expect(200); + const res = await request(app) + .get("/search") + .set("x-user-id", "shared-user") + .set("x-auth-tenant-id", "tenant-y") + .expect(200); + expect(res.headers["x-ratelimit-remaining"]).toBe("119"); + }); + + it("anonymous callers fall back to a shared hashed-IP bucket which can be throttled", async () => { + const anon = createTenantLeakyBucketRateLimiter({ ratePerSecond: 60, capacity: 3, store }); + const anonApp = buildApp(anon); + await request(anonApp).get("/search").expect(200); + await request(anonApp).get("/search").expect(200); + await request(anonApp).get("/search").expect(200); + await request(anonApp).get("/search").expect(429); + // ... while an authenticated tenant is unaffected. + await request(anonApp).get("/search").set("x-user-id", "vip").expect(200); + }); + + it("works with zero options (all defaults from config)", async () => { + const defaultLimiter = createTenantLeakyBucketRateLimiter(); + const defaultApp = buildApp(defaultLimiter); + await request(defaultApp).get("/search").set("x-user-id", "cfg-user").expect(200); + }); + + it("fails open when the store throws", async () => { + const failingStore = { + consume: jest.fn().mockRejectedValue(new Error("redis connection lost")), + }; + const open = createTenantLeakyBucketRateLimiter({ + ratePerSecond: 60, + capacity: 120, + + store: failingStore as any, + }); + const openApp = buildApp(open); + const res = await request(openApp).get("/search").set("x-user-id", "u").expect(200); + expect(res.body.success).toBe(true); + expect(failingStore.consume).toHaveBeenCalled(); + }); + + it("fails open even when the store rejects with a non-Error value", async () => { + const weirdStore = { + consume: jest.fn().mockImplementation(() => Promise.reject("redis said no")), + }; + + const open = createTenantLeakyBucketRateLimiter({ ratePerSecond: 60, capacity: 120, store: weirdStore as any }); + const openApp = buildApp(open); + await request(openApp).get("/search").set("x-user-id", "u2").expect(200); + }); +}); + +describe("RedisLeakyBucketStore (unit, fake client)", () => { + + const makeClient = (overrides: Record = {}): any => ({ + evalsha: jest.fn().mockResolvedValue([1, 0, 1999]), + eval: jest.fn().mockResolvedValue([1, 0, 1999]), + script: jest.fn().mockResolvedValue(["cached"]), + quit: jest.fn().mockResolvedValue("OK"), + on: jest.fn(), + ...overrides, + }); + + it("uses EVALSHA with key, clock, rate, capacity, amount and TTL args", async () => { + const client = makeClient(); + const store = new RedisLeakyBucketStore(client, 250); + const decision = await store.consume("rlb:bookings:search:user:u1", { + ratePerSecond: 60, + capacity: 120, + }); + + expect(decision).toEqual({ allowed: true, retryAfterMs: 0, level: 1.999 }); + expect(client.evalsha).toHaveBeenCalledTimes(1); + const [sha, numKeys, key, nowMs, rate, capacity, amount, ttl] = client.evalsha.mock.calls[0] as [ + string, number, string, number, number, number, number, number, + ]; + expect(sha).toMatch(/^[0-9a-f]{40}$/); + expect(numKeys).toBe(1); + expect(key).toBe("rlb:bookings:search:user:u1"); + expect(Number.isFinite(nowMs)).toBe(true); + expect(rate).toBe(60); + expect(capacity).toBe(120); + expect(amount).toBe(1); + expect(ttl).toBe(5); // bucketTtlSeconds(60, 120) + expect(client.eval).not.toHaveBeenCalled(); + }); + + it("falls back to EVAL on NOSCRIPT and parses throttled responses", async () => { + const client = makeClient({ + evalsha: jest.fn().mockRejectedValue(new Error("NOSCRIPT No matching script. Please use EVAL.")), + eval: jest.fn().mockResolvedValue([0, 17, 120000]), + }); + const store = new RedisLeakyBucketStore(client, 250); + const decision = await store.consume("k", { ratePerSecond: 60, capacity: 120, amount: 1 }); + + expect(client.evalsha).toHaveBeenCalledTimes(1); + expect(client.eval).toHaveBeenCalledTimes(1); + const [scriptArg, numKeys, key] = client.eval.mock.calls[0] as [string, number, string]; + expect(scriptArg).toBe(LEAKY_BUCKET_LUA); + expect(numKeys).toBe(1); + expect(key).toBe("k"); + expect(decision).toEqual({ allowed: false, retryAfterMs: 17, level: 120 }); + }); + + it("propagates non-NOSCRIPT Redis errors", async () => { + const client = makeClient({ + evalsha: jest.fn().mockRejectedValue(new Error("READONLY replica")), + }); + const store = new RedisLeakyBucketStore(client, 250); + await expect(store.consume("k", { ratePerSecond: 60, capacity: 120 })).rejects.toThrow("READONLY"); + expect(client.eval).not.toHaveBeenCalled(); + }); + + it("enforces the timeout on a latency spike instead of hanging", async () => { + const client = makeClient({ + // Simulates a Redis that never answers within the budget. + evalsha: jest.fn().mockImplementation(() => new Promise(() => {})), + }); + const store = new RedisLeakyBucketStore(client, 40); + const started = Date.now(); + await expect(store.consume("k", { ratePerSecond: 60, capacity: 120 })).rejects.toBeInstanceOf( + LeakyBucketRedisTimeoutError, + ); + expect(Date.now() - started).toBeLessThan(1000); + }); + + it("tolerates degenerate Lua reply tuples", async () => { + const client = makeClient({ + evalsha: jest.fn().mockResolvedValue([1, null, undefined]), + }); + const store = new RedisLeakyBucketStore(client, 250); + const decision = await store.consume("k", { ratePerSecond: 60, capacity: 120 }); + expect(decision).toEqual({ allowed: true, retryAfterMs: 0, level: 0 }); + }); + + it("rejects an invalid timeout", () => { + expect(() => new RedisLeakyBucketStore(makeClient(), 0)).toThrow(); + expect(() => new RedisLeakyBucketStore(makeClient(), Number.NaN)).toThrow(); + }); +}); + +describe("middleware + RedisLeakyBucketStore: Redis latency spike end-to-end", () => { + it("a slow store fails open within the timeout and the response completes", async () => { + const slowClient = { + evalsha: jest.fn().mockImplementation(() => new Promise(() => {})), + eval: jest.fn().mockResolvedValue([1, 0, 1000]), + script: jest.fn(), + quit: jest.fn(), + on: jest.fn(), + }; + + const store = new RedisLeakyBucketStore(slowClient as any, 50); + const limiter = createTenantLeakyBucketRateLimiter({ ratePerSecond: 60, capacity: 120, store }); + const app = buildApp(limiter); + + const started = Date.now(); + const res = await request(app).get("/search").set("x-user-id", "spike-victim").expect(200); + expect(Date.now() - started).toBeLessThan(2000); + expect(res.body.success).toBe(true); + }); +}); + +describe("InMemoryLeakyBucketStore", () => { + it("tracks state per key and resets", async () => { + const store = new InMemoryLeakyBucketStore(() => T0); + await store.consume("a", { ratePerSecond: 1, capacity: 1 }); + expect(store.size).toBe(1); + await store.consume("b", { ratePerSecond: 1, capacity: 1 }); + expect(store.size).toBe(2); + store.reset(); + expect(store.size).toBe(0); + }); + + it("expires idle buckets after their TTL (mirrors Redis EXPIRE), keeps live ones", async () => { + let now = T0; + const store = new InMemoryLeakyBucketStore(() => now); + const params = { ratePerSecond: 1, capacity: 1 }; // ttl = 3s + await store.consume("stale", params); + await store.consume("live", params); + expect(store.size).toBe(2); + + now += 4_000; // past the 3s TTL of both entries + await store.consume("live", params); // refreshes 'live', prunes 'stale' + expect(store.size).toBe(1); + }); +}); diff --git a/src/middleware/tenantLeakyBucket.ts b/src/middleware/tenantLeakyBucket.ts new file mode 100644 index 00000000..53a18225 --- /dev/null +++ b/src/middleware/tenantLeakyBucket.ts @@ -0,0 +1,607 @@ +/** + * @file src/middleware/tenantLeakyBucket.ts + * + * Per-tenant leaky-bucket rate limiter for high-traffic search endpoints + * (initially `/api/v1/bookings/search`). + * + * Why this exists + * ─────────────── + * `createAuthAwareRateLimiter` (express-rate-limit) applies one coarse, + * fixed-window budget to every authenticated principal. On a search endpoint + * that is fine for fairness in the small, but a single noisy tenant could + * burn the shared window and starve every other tenant ("noisy neighbor"). + * This middleware instead maintains an *independent* leaky bucket per + * tenant, so an abusive tenant only ever degrades its own traffic. + * + * Algorithm + * ───────── + * Leaky bucket: every accepted request adds `amount` (1) work unit to the + * bucket; the bucket drains at a constant `ratePerSecond`. A request is + * admitted while `level + amount <= capacity`. Consequently: + * - burst traffic is absorbed up to `capacity` (default 120), and + * - sustained throughput converges to `ratePerSecond` (default 60 rps). + * Rejected requests are answered `429` with a strict `Retry-After` header + * computed from the exact drain time — never guessed. + * + * Atomicity + * ───────── + * The read-modify-write of the bucket state executes inside a single Redis + * Lua script (`LEAKY_BUCKET_LUA`), so concurrent requests from multiple app + * instances cannot race each other. The script is invoked via EVALSHA with + * an automatic EVAL fallback on NOSCRIPT (first call / after FLUSH). + * + * Failure policy (fail-open, availability-first) + * ────────────────────────────────────────────── + * If Redis is unreachable or slower than `redisTimeoutMs`, the limiter + * *allows* the request and increments `rate_limit_redis_failures_total`. + * A rate limiter must never take the protected endpoint down with it; the + * tradeoff is that during a Redis outage protection degrades to the + * pre-fix (unlimited) behavior. Documented in docs/api/bookings-search.md. + * + * Security notes + * ────────────── + * - Tenant identity is resolved from *trusted auth context only* + * (JWT/req.auth claims, API key). Client-supplied `x-tenant-id` headers + * are intentionally NOT trusted, preventing a caller from hopscotching + * across tenant buckets to evade the limit. + * - Identifiers are canonicalized before they are embedded in Redis keys + * (strict charset + length cap, otherwise SHA-256), which prevents Redis + * key-injection across logical namespaces and bounds Prometheus label + * cardinality. + * - Unauthenticated requests fall back to a SHA-256 hashed IP bucket so + * anonymous traffic cannot starve authenticated tenants either. + * - `Retry-After` is derived from live bucket state, not static. + * + * Testability + * ─────────── + * The store is dependency-injected (`setTenantLeakyBucketStore` / + * `resetTenantLeakyBucketStore`) and the clock is injectable, so tests can + * deterministically model bursts, leaks, tenant switches, and Redis + * latency spikes without sleeping. + */ + +import { createHash } from "node:crypto"; +import { createRequire } from "node:module"; +import type { NextFunction, Request, RequestHandler, Response } from "express"; +import { configService } from "../config/config.service.js"; +import { rateLimitBucketBurn, rateLimitRedisFailuresTotal } from "../metrics.js"; +import { logger } from "../utils/logger.js"; + +// ─── Decision engine (pure, shared semantics with the Lua script) ─────────── + +export interface LeakyBucketParams { + /** Tokens drained per second (sustained throughput). */ + ratePerSecond: number; + /** Maximum bucket fill level (burst allowance). */ + capacity: number; + /** Work units this request consumes (normally 1). */ + amount?: number; + /** Current time in epoch milliseconds (injectable for tests). */ + nowMs: number; +} + +export interface LeakyBucketDecision { + /** Whether the request is admitted. */ + allowed: boolean; + /** Bucket level after the decision (tokens currently "charged"). */ + level: number; + /** + * When rejected: milliseconds until enough capacity has drained for this + * request to succeed. 0 when allowed. + */ + retryAfterMs: number; +} + +/** + * Pure leaky-bucket transition. This function is the exact JavaScript mirror + * of `LEAKY_BUCKET_LUA`; both MUST evolve together so the in-memory store + * (tests / fallback) and the Redis store behave identically. + */ +export function decideLeakyBucket( + level: number, + lastTsMs: number, + params: LeakyBucketParams, +): LeakyBucketDecision { + const amount = params.amount ?? 1; + if ( + !Number.isFinite(params.ratePerSecond) || + params.ratePerSecond <= 0 || + !Number.isFinite(params.capacity) || + params.capacity <= 0 + ) { + throw new Error("ratePerSecond and capacity must be positive finite numbers"); + } + + // Guard against clock going backwards (cross-instance skew / NTP steps): + // a negative elapsed would *increase* the level and wrongly reject traffic. + const elapsedMs = Math.max(0, params.nowMs - lastTsMs); + const drained = (elapsedMs * params.ratePerSecond) / 1000; + let newLevel = Math.max(0, level - drained); + + if (newLevel + amount <= params.capacity) { + newLevel += amount; + return { allowed: true, level: newLevel, retryAfterMs: 0 }; + } + + const deficit = newLevel + amount - params.capacity; + const retryAfterMs = Math.ceil((deficit / params.ratePerSecond) * 1000); + return { allowed: false, level: newLevel, retryAfterMs }; +} + +/** + * Redis key TTL for an idle bucket. Once a bucket has not been touched for + * two full drain periods its level is guaranteed to be 0, so the key can be + * safely reclaimed — tenants that go quiet do not leak Redis memory. + */ +export function bucketTtlSeconds(ratePerSecond: number, capacity: number): number { + return Math.max(1, Math.ceil((capacity / ratePerSecond) * 2) + 1); +} + +// ─── Lua script (atomic read-modify-write on Redis) ───────────────────────── + +/** + * Atomic leaky-bucket update. + * + * KEYS[1] = bucket key (hash { level, ts }) + * ARGV[1] = now_ms + * ARGV[2] = rate (tokens/second) + * ARGV[3] = capacity + * ARGV[4] = amount + * ARGV[5] = ttl_seconds + * + * Returns { allowed(0|1), retry_after_ms, level_x1000 }. + * `level_x1000` is floor(level*1000) so fractional state crosses the + * Redis→JS boundary without float serialization drift (divide by 1000 in JS). + * + * NOTE: Redis forbids TIME inside scripts, so `now_ms` is passed by the + * caller. App instances share NTP; residual skew is clamped by the + * `math.max(0, ...)` guard and self-heals on the next write. + */ +export const LEAKY_BUCKET_LUA = ` +local level = tonumber(redis.call('HGET', KEYS[1], 'level') or '0') +local ts = tonumber(redis.call('HGET', KEYS[1], 'ts') or ARGV[1]) +local now_ms = tonumber(ARGV[1]) +local rate = tonumber(ARGV[2]) +local capacity = tonumber(ARGV[3]) +local amount = tonumber(ARGV[4]) +local ttl = tonumber(ARGV[5]) + +local elapsed = math.max(now_ms - ts, 0) +level = math.max(level - (elapsed * rate / 1000), 0) + +local allowed = 0 +local retry_after_ms = 0 +if level + amount <= capacity then + level = level + amount + allowed = 1 +else + retry_after_ms = math.ceil(((level + amount - capacity) / rate) * 1000) +end + +redis.call('HSET', KEYS[1], 'level', level, 'ts', now_ms) +redis.call('EXPIRE', KEYS[1], ttl) + +return { allowed, retry_after_ms, math.floor(level * 1000) } +`.trim(); + +const LEAKY_BUCKET_LUA_SHA = createHash("sha1").update(LEAKY_BUCKET_LUA, "utf8").digest("hex"); + +// ─── Store interface ───────────────────────────────────────────────────────── + +export interface LeakyBucketStoreConsume { + ratePerSecond: number; + capacity: number; + amount?: number; +} + +export interface LeakyBucketStore { + consume(key: string, params: LeakyBucketStoreConsume): Promise; +} + +/** Raised when the Redis round-trip exceeds the configured timeout. */ +export class LeakyBucketRedisTimeoutError extends Error { + constructor(timeoutMs: number) { + super(`leaky-bucket Redis eval exceeded ${timeoutMs}ms`); + this.name = "LeakyBucketRedisTimeoutError"; + } +} + +// ─── Redis store ───────────────────────────────────────────────────────────── + +/** + * The minimal ioredis surface used by the limiter. Declared as an interface + * so unit tests can inject fakes (latency spikes, NOSCRIPT, failures) + * without a live Redis. + */ +export interface RedisEvalLike { + eval(script: string, numKeys: number, ...args: Array): Promise; + evalsha(sha: string, numKeys: number, ...args: Array): Promise; + script( + subcommand: string, + ...args: Array + ): Promise; + quit(): Promise; + disconnect?(): void; + on(event: string, handler: (...args: unknown[]) => void): unknown; + status?: string; +} + +function isNoScriptError(err: unknown): boolean { + return err instanceof Error && err.message.toUpperCase().includes("NOSCRIPT"); +} + +/** + * Production store: executes {@link LEAKY_BUCKET_LUA} on Redis. + * + * Strategy: EVALSHA with the locally computed SHA-1. On NOSCRIPT (first use + * after boot / FLUSHALL) fall back to a full EVAL, which also re-caches the + * script server-side, so only one request per process lifetime pays the + * double round-trip. + * + * Every call is wrapped in a hard timeout (`timeoutMs`); a too-slow Redis + * surfaces as {@link LeakyBucketRedisTimeoutError} so the middleware can + * fail open instead of hanging the request (the "latency spike" case). + */ +export class RedisLeakyBucketStore implements LeakyBucketStore { + constructor( + private readonly client: RedisEvalLike, + private readonly timeoutMs: number = configService.bookingsSearchRedisTimeoutMs, + ) { + if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) { + throw new Error("timeoutMs must be a positive finite number"); + } + } + + async consume(key: string, params: LeakyBucketStoreConsume): Promise { + const nowMs = Date.now(); + const amount = params.amount ?? 1; + const ttl = bucketTtlSeconds(params.ratePerSecond, params.capacity); + const args: Array = [ + key, + nowMs, + params.ratePerSecond, + params.capacity, + amount, + ttl, + ]; + + let raw: unknown; + try { + raw = await this.withTimeout(this.client.evalsha(LEAKY_BUCKET_LUA_SHA, 1, ...args)); + } catch (err) { + if (!isNoScriptError(err)) throw err; + // Script not cached yet — EVAL re-registers it server-side. + raw = await this.withTimeout(this.client.eval(LEAKY_BUCKET_LUA, 1, ...args)); + } + + return parseLuaDecision(raw); + } + + private async withTimeout(promise: Promise): Promise { + let timer: NodeJS.Timeout | undefined; + const timeout = new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new LeakyBucketRedisTimeoutError(this.timeoutMs)), this.timeoutMs); + // Don't keep the event loop alive on behalf of an abandoned caller. + timer.unref?.(); + }); + try { + return await Promise.race([promise, timeout]); + } finally { + clearTimeout(timer); + } + } +} + +function parseLuaDecision(raw: unknown): LeakyBucketDecision { + const tuple = raw as [number | string, number | string, number | string]; + const allowed = Number(tuple[0]) === 1; + const retryAfterMs = Number(tuple[1]) || 0; + // level crosses the wire as floor(level * 1000) to avoid float drift. + const level = (Number(tuple[2]) || 0) / 1000; + return { allowed, retryAfterMs, level }; +} + +// ─── In-memory store (test environments / deterministic fallback) ─────────── + +/** + * Single-process store with semantics identical to the Lua script (it calls + * the same {@link decideLeakyBucket} transition). Used automatically in + * `NODE_ENV=test` so suites never need Redis, and injectable via + * `setTenantLeakyBucketStore` for deterministic clocks. + * + * In a multi-instance deployment this store must NOT be used (it cannot + * coordinate across processes) — hence its restriction to tests. + */ +interface BucketEntry { + level: number; + ts: number; + /** Mirrors the Redis EXPIRE set by the Lua script. Key is pruned once idle past it. */ + ttlMs: number; +} + +export class InMemoryLeakyBucketStore implements LeakyBucketStore { + private readonly buckets = new Map(); + + constructor(private readonly clock: () => number = () => Date.now()) {} + + async consume(key: string, params: LeakyBucketStoreConsume): Promise { + const ratePerSecond = params.ratePerSecond; + const capacity = params.capacity; + const nowMs = this.clock(); + const entry = this.buckets.get(key); + const decision = decideLeakyBucket(entry?.level ?? 0, entry?.ts ?? nowMs, { + ratePerSecond, + capacity, + amount: params.amount ?? 1, + nowMs, + }); + this.buckets.set(key, { + level: decision.level, + ts: nowMs, + ttlMs: bucketTtlSeconds(ratePerSecond, capacity) * 1000, + }); + this.pruneExpired(nowMs); + return decision; + } + + /** + * Reclaim keys idle beyond their TTL — the exact mirror of the Redis + * EXPIRE the Lua script applies — so quiet tenants don't leak memory. + */ + private pruneExpired(nowMs: number): void { + for (const [key, entry] of this.buckets) { + if (nowMs - entry.ts > entry.ttlMs) this.buckets.delete(key); + } + } + + /** Test helper — clears all bucket state. */ + reset(): void { + this.buckets.clear(); + } + + /** Test helper — number of live bucket keys. */ + get size(): number { + return this.buckets.size; + } +} + +// ─── Store lifecycle (lazy singleton with safe test injection) ─────────────── + +let cachedStore: LeakyBucketStore | undefined; +let cachedClient: RedisEvalLike | undefined; + +const require = createRequire(import.meta.url); + +type RedisCtor = new (url: string, options: Record) => RedisEvalLike; + +let redisCtorOverride: RedisCtor | undefined; + +/** + * @internal — test hook only. Lets tests run the production store-creation + * path (options, error handler, lifecycle) against a fake Redis constructor + * without touching ioredis' network stack. + */ +export function _setRedisCtorForTesting(ctor: RedisCtor | undefined): void { + redisCtorOverride = ctor; +} + +function createRedisStore(): LeakyBucketStore { + // Lazy require keeps ioredis (and its network client) out of test runs. + const Ctor: RedisCtor = + redisCtorOverride ?? + (require("ioredis") as { Redis: RedisCtor }).Redis; + cachedClient = new Ctor(process.env.REDIS_URL || "redis://localhost:6379", { + maxRetriesPerRequest: 3, + enableReadyCheck: true, + lazyConnect: true, + showFriendlyErrorStack: process.env.NODE_ENV !== "production", + retryStrategy: (times: number) => (times > 10 ? null : Math.min(times * 100, 2000)), + }); + cachedClient.on("error", (err: unknown) => { + logger.error({ err }, "tenantLeakyBucket redis error"); + }); + return new RedisLeakyBucketStore(cachedClient); +} + +/** + * Store resolution: + * 1. Test/injected store (via `setTenantLeakyBucketStore`), else + * 2. `NODE_ENV=test` → InMemoryLeakyBucketStore (no Redis dependency), else + * 3. Production → RedisLeakyBucketStore over the shared REDIS_URL client. + */ +export function getTenantLeakyBucketStore(): LeakyBucketStore { + if (!cachedStore) { + cachedStore = process.env.NODE_ENV === "test" ? new InMemoryLeakyBucketStore() : createRedisStore(); + } + return cachedStore; +} + +/** Inject a store (tests). Resets the production singleton. */ +export function setTenantLeakyBucketStore(store: LeakyBucketStore): void { + cachedStore = store; +} + +/** Restore default store resolution; closes any Redis client it created. */ +export async function resetTenantLeakyBucketStore(): Promise { + cachedStore = undefined; + if (cachedClient) { + + cachedClient.disconnect?.(); + cachedClient = undefined; + } +} + +// ─── Tenant identity resolution ────────────────────────────────────────────── + +const SAFE_ID = /^[A-Za-z0-9._-]{1,128}$/; + +export interface TenantIdentity { + /** namespaced Redis key, e.g. `rlb:bookings:search:user:alice` */ + key: string; + /** sanitized identifier used as the `tenant` metric label */ + label: string; +} + +function hashId(id: string): string { + return createHash("sha256").update(id, "utf8").digest("hex").slice(0, 32); +} + +/** + * Canonicalize a caller-controlled identifier before embedding it in a + * Redis key or a metric label. Safe identifiers pass through unchanged; + * anything else (overlong, exotic charset, attempted key injection like + * `tenant:vip`) is replaced by a deterministic hash so it can never collide + * with another logical namespace. + */ +export function sanitizeTenantIdentifier(id: string): string { + const trimmed = id.trim(); + if (SAFE_ID.test(trimmed)) return trimmed; + return `h:${hashId(trimmed)}`; +} + +function getClientIp(req: Request): string { + + const anyReq = req as any; + return anyReq.ip || anyReq.socket?.remoteAddress || "anonymous"; +} + +/** + * Resolve the *trusted* tenant identity for rate limiting. + * + * Priority (first match wins): + * 1. Tenant claim from trusted auth context (`req.auth.tenantId` or + * `req.user.tenantId`) — true multi-tenant deployments. + * 2. Authenticated user (`req.auth.userId` / `req.user.sub||id`) — the + * user is the tenant boundary. + * 3. API key (`req.apiKeyId`) — partner keys map 1:1 to tenants; hashed. + * 4. Hashed client IP — anonymous traffic cannot crowd out tenants. + * + * SECURITY: the `x-tenant-id` request header is deliberately NOT consulted. + * Trusting it would let any caller pick a fresh, empty bucket per request + * (limit evasion) and would create unbounded Redis keys / label cardinality. + */ +export function resolveTenantIdentity(req: Request, routeScope = "bookings:search"): TenantIdentity { + + const auth = (req as any).auth as Record | undefined; + const user = req.user as Record | undefined; + + const authTenant = firstString(auth?.tenantId, user?.tenantId); + if (authTenant) { + const id = sanitizeTenantIdentifier(authTenant); + return { key: `rlb:${routeScope}:tenant:${id}`, label: id }; + } + + const authUser = firstString(auth?.userId, user?.sub, user?.id); + if (authUser) { + const id = sanitizeTenantIdentifier(authUser); + return { key: `rlb:${routeScope}:user:${id}`, label: id }; + } + + + const apiKeyId = (req as any).apiKeyId as string | undefined; + if (typeof apiKeyId === "string" && apiKeyId.trim()) { + const id = hashId(apiKeyId.trim()); + return { key: `rlb:${routeScope}:apiKey:${id}`, label: `apiKey:${id.slice(0, 16)}` }; + } + + const id = hashId(getClientIp(req)); + return { key: `rlb:${routeScope}:ip:${id}`, label: `ip:${id.slice(0, 16)}` }; +} + +function firstString(...values: unknown[]): string | undefined { + for (const value of values) { + if (typeof value === "string" && value.trim()) return value; + } + return undefined; +} + +// ─── Express middleware ────────────────────────────────────────────────────── + +export interface TenantLeakyBucketOptions { + /** Sustained requests per second per tenant. Default: config / 60. */ + ratePerSecond?: number; + /** Burst capacity per tenant. Default: config / 120. */ + capacity?: number; + /** Work units per request (e.g. weigh expensive queries). Default: 1. */ + amount?: number; + /** Route scope embedded in Redis keys (namespaces the bucket). */ + routeScope?: string; + /** Store override (defaults to the lazily-resolved singleton). */ + store?: LeakyBucketStore; +} + +/** + * Per-tenant leaky-bucket rate-limit middleware. + * + * On admit: forwards the request and sets `X-RateLimit-Limit`, + * `X-RateLimit-Remaining`, `X-RateLimit-Reset`. + * On reject: `429` + `Retry-After` (whole seconds, derived from real bucket + * state) + error body `{ success: false, error, retryAfter }`. + * On Redis failure/timeout: fails open (admits) and counts + * `rate_limit_redis_failures_total`. + * + * Always emits the `rate_limit_bucket_burn{tenant}` gauge with the current + * bucket level so operators can watch per-tenant burn in real time. + */ +export function createTenantLeakyBucketRateLimiter( + options: TenantLeakyBucketOptions = {}, +): RequestHandler { + const ratePerSecond = options.ratePerSecond ?? configService.bookingsSearchRatePerSecond; + const capacity = options.capacity ?? configService.bookingsSearchBurst; + const amount = options.amount ?? 1; + const routeScope = options.routeScope ?? "bookings:search"; + + if (!(ratePerSecond > 0) || !(capacity > 0) || !(amount > 0)) { + throw new Error("tenant leaky-bucket: ratePerSecond, capacity and amount must be positive"); + } + + const handler = async (req: Request, res: Response, next: NextFunction): Promise => { + const identity = resolveTenantIdentity(req, routeScope); + const store = options.store ?? getTenantLeakyBucketStore(); + + let decision: LeakyBucketDecision; + try { + decision = await store.consume(identity.key, { ratePerSecond, capacity, amount }); + } catch (err) { + // Fail-open: availability first; never take the endpoint down with Redis. + rateLimitRedisFailuresTotal.inc(); + logger.warn( + { err: err instanceof Error ? err.message : String(err), tenant: identity.label }, + "tenant leaky-bucket store failure — failing open", + ); + next(); + return; + } + + rateLimitBucketBurn.labels(identity.label).set(decision.level); + + const remaining = Math.max(0, Math.floor(capacity - decision.level)); + const resetEpochSec = Math.ceil((Date.now() + (decision.level / ratePerSecond) * 1000) / 1000); + + res.setHeader("X-RateLimit-Limit", String(capacity)); + res.setHeader("X-RateLimit-Remaining", String(remaining)); + res.setHeader("X-RateLimit-Reset", String(resetEpochSec)); + + if (!decision.allowed) { + const retryAfterSec = Math.max(1, Math.ceil(decision.retryAfterMs / 1000)); + res.setHeader("Retry-After", String(retryAfterSec)); + res.status(429).json({ + success: false, + error: "Too many requests, please try again later.", + retryAfter: retryAfterSec, + }); + return; + } + + next(); + }; + + return (req: Request, res: Response, next: NextFunction): void => { + void handler(req, res, next).catch((err: unknown) => { + // Final safety net — still fail open, never crash the request pipeline. + rateLimitRedisFailuresTotal.inc(); + logger.error({ err }, "tenant leaky-bucket unexpected error — failing open"); + next(); + }); + }; +} diff --git a/src/routes/__tests__/bookings.test.ts b/src/routes/__tests__/bookings.test.ts new file mode 100644 index 00000000..464e592b --- /dev/null +++ b/src/routes/__tests__/bookings.test.ts @@ -0,0 +1,270 @@ +/** + * Route-level tests for /api/v1/bookings/search, including the end-to-end + * noisy-tenant scenario: one tenant burns its entire bucket while another + * tenant's traffic stays unaffected. + */ + +import request from "supertest"; +import express from "express"; + +const { createBookingsRouter } = await import("../bookings.js"); +const { InMemoryBookingIntentRepository } = await import( + "../../modules/booking-intents/booking-intent-repository.js" +); +const { createTenantLeakyBucketRateLimiter, InMemoryLeakyBucketStore } = await import( + "../../middleware/tenantLeakyBucket.js" +); + +const T0 = 1_700_000_000_000; +const AUTH = { "x-chronopay-user-id": "buyer-1", "x-chronopay-role": "customer" }; +const PRO = { "x-chronopay-user-id": "pro-1", "x-chronopay-role": "professional" }; + +function seed(repository: InstanceType) { + const base = { + professional: "dr-smith", + startTime: Date.parse("2026-08-01T10:00:00Z"), + endTime: Date.parse("2026-08-01T11:00:00Z"), + createdAt: "2026-07-01T00:00:00Z", + }; + return Promise.all([ + repository.create({ ...base, slotId: "slot-1", customerId: "buyer-1", status: "pending", note: "teeth cleaning" }), + repository.create({ ...base, slotId: "slot-2", customerId: "buyer-1", status: "confirmed", note: "root canal" }), + repository.create({ + ...base, + slotId: "slot-3", + customerId: "buyer-1", + status: "cancelled", + note: "checkup", + startTime: Date.parse("2026-09-01T10:00:00Z"), + endTime: Date.parse("2026-09-01T11:00:00Z"), + }), + repository.create({ ...base, slotId: "slot-9", customerId: "other-buyer", status: "confirmed", note: "not mine" }), + ]); +} + +function buildApp(options: { + repository?: InstanceType; + rateLimiter?: express.RequestHandler; +} = {}) { + const app = express(); + app.use(express.json()); + app.use("/api/v1/bookings", createBookingsRouter(options)); + return app; +} + +describe("GET /api/v1/bookings/search", () => { + let repository: InstanceType; + let app: express.Express; + + beforeEach(async () => { + repository = new InMemoryBookingIntentRepository(); + await seed(repository); + app = buildApp({ repository }); + }); + + it("rejects unauthenticated callers with 401", async () => { + const res = await request(app).get("/api/v1/bookings/search").expect(401); + expect(res.body.success).toBe(false); + }); + + it("rejects unknown roles with 400", async () => { + await request(app) + .get("/api/v1/bookings/search") + .set({ "x-chronopay-user-id": "u", "x-chronopay-role": "overlord" }) + .expect(400); + }); + + it("returns only the caller's bookings", async () => { + const res = await request(app).get("/api/v1/bookings/search").set(AUTH).expect(200); + expect(res.body.success).toBe(true); + expect(res.body.data.total).toBe(3); // "not mine" belongs to other-buyer + for (const record of res.body.data.results) { + expect(record.customerId).toBe("buyer-1"); + } + }); + + it("supports professional and support/admin roles", async () => { + await request(app).get("/api/v1/bookings/search").set(PRO).expect(200); + await request(app) + .get("/api/v1/bookings/search") + .set({ "x-chronopay-user-id": "s1", "x-chronopay-role": "admin" }) + .expect(200); + }); + + it("filters by status", async () => { + const res = await request(app) + .get("/api/v1/bookings/search?status=confirmed") + .set(AUTH) + .expect(200); + expect(res.body.data.total).toBe(1); + expect(res.body.data.results[0].slotId).toBe("slot-2"); + }); + + it("rejects invalid status values", async () => { + const res = await request(app) + .get("/api/v1/bookings/search?status=invented") + .set(AUTH) + .expect(400); + expect(res.body.error).toMatch(/status must be one of/); + }); + + it("filters by free-text q over note/slot/id (case-insensitive)", async () => { + const res = await request(app) + .get("/api/v1/bookings/search?q=ROOT") + .set(AUTH) + .expect(200); + expect(res.body.data.total).toBe(1); + expect(res.body.data.results[0].slotId).toBe("slot-2"); + }); + + it("filters by slotId", async () => { + const res = await request(app) + .get("/api/v1/bookings/search?slotId=slot-3") + .set(AUTH) + .expect(200); + expect(res.body.data.total).toBe(1); + }); + + it("filters by overlapping date range and validates the range", async () => { + const res = await request(app) + .get("/api/v1/bookings/search?from=2026-08-15T00:00:00Z&to=2026-09-15T00:00:00Z") + .set(AUTH) + .expect(200); + expect(res.body.data.total).toBe(1); + expect(res.body.data.results[0].slotId).toBe("slot-3"); + + await request(app) + .get("/api/v1/bookings/search?from=not-a-date") + .set(AUTH) + .expect(400); + await request(app) + .get("/api/v1/bookings/search?to=not-a-date") + .set(AUTH) + .expect(400); + await request(app) + .get("/api/v1/bookings/search?from=2026-09-02T00:00:00Z&to=2026-08-01T00:00:00Z") + .set(AUTH) + .expect(400); + }); + + it("free-text search matches note-less bookings by id/slot and tolerates empty search params", async () => { + repository = new InMemoryBookingIntentRepository(); + await repository.create({ + slotId: "slot-plain", + customerId: "buyer-1", + professional: "dr-no-notes", + startTime: Date.parse("2026-08-01T10:00:00Z"), + endTime: Date.parse("2026-08-01T11:00:00Z"), + status: "pending", + createdAt: "2026-07-01T00:00:00Z", + // deliberately no `note` + }); + app = buildApp({ repository }); + + const res = await request(app) + .get("/api/v1/bookings/search?q=no-notes") + .set(AUTH) + .expect(200); + expect(res.body.data.total).toBe(1); + + // q present but blank behaves like no filter; non-string query values are ignored + const blank = await request(app) + .get("/api/v1/bookings/search?q=%20&status=") + .set(AUTH) + .expect(200); + expect(blank.body.data.total).toBe(1); + }); + + it("supports one-sided ranges (only `to`, only `from`)", async () => { + const beforeMid = await request(app) + .get("/api/v1/bookings/search?to=2026-08-15T00:00:00Z") + .set(AUTH) + .expect(200); + expect(beforeMid.body.data.total).toBe(2); // both August bookings, slot-3 excluded + + const afterMid = await request(app) + .get("/api/v1/bookings/search?from=2026-08-15T00:00:00Z") + .set(AUTH) + .expect(200); + expect(afterMid.body.data.total).toBe(1); + expect(afterMid.body.data.results[0].slotId).toBe("slot-3"); + }); + + it("paginates with limit/offset and validates them", async () => { + const page1 = await request(app) + .get("/api/v1/bookings/search?limit=2&offset=0") + .set(AUTH) + .expect(200); + expect(page1.body.data.results).toHaveLength(2); + expect(page1.body.data.total).toBe(3); + + const page2 = await request(app) + .get("/api/v1/bookings/search?limit=2&offset=2") + .set(AUTH) + .expect(200); + expect(page2.body.data.results).toHaveLength(1); + + await request(app).get("/api/v1/bookings/search?limit=0").set(AUTH).expect(400); + await request(app).get("/api/v1/bookings/search?limit=101").set(AUTH).expect(400); + await request(app).get("/api/v1/bookings/search?offset=-1").set(AUTH).expect(400); + }); + + it("rejects overlong q", async () => { + await request(app) + .get(`/api/v1/bookings/search?q=${"a".repeat(201)}`) + .set(AUTH) + .expect(400); + }); + + it("surfaces repository failures as 500 without leaking internals", async () => { + const broken = new InMemoryBookingIntentRepository(); + broken.listByCustomer = async () => { + throw new Error("db exploded with sensitive detail"); + }; + const brokenApp = buildApp({ repository: broken }); + const res = await request(brokenApp).get("/api/v1/bookings/search").set(AUTH).expect(500); + expect(res.body).toEqual({ success: false, error: "Search failed" }); + }); +}); + +describe("GET /api/v1/bookings/search — per-tenant leaky bucket E2E", () => { + it("a noisy tenant is throttled at 120 burst + 60 rps while other tenants stay unaffected", async () => { + let now = T0; + const store = new InMemoryLeakyBucketStore(() => now); + const rateLimiter = createTenantLeakyBucketRateLimiter({ + ratePerSecond: 60, + capacity: 120, + routeScope: "bookings:search", + store, + }); + const app = buildApp({ repository: new InMemoryBookingIntentRepository(), rateLimiter }); + + const noisy = { "x-chronopay-user-id": "noisy-tenant", "x-chronopay-role": "customer" }; + const quiet = { "x-chronopay-user-id": "quiet-tenant", "x-chronopay-role": "customer" }; + + // Noisy tenant burns the full 120-token burst. + for (let i = 0; i < 120; i++) { + await request(app).get("/api/v1/bookings/search").set(noisy).expect(200); + } + const throttled = await request(app).get("/api/v1/bookings/search").set(noisy).expect(429); + expect(throttled.headers["retry-after"]).toBe("1"); + expect(throttled.body.retryAfter).toBe(1); + + // The quiet tenant still gets full service — no starvation. + const ok = await request(app).get("/api/v1/bookings/search").set(quiet).expect(200); + expect(ok.body.success).toBe(true); + + // One second of drain gives the noisy tenant its 60 rps sustained budget back. + now += 1000; + for (let i = 0; i < 60; i++) { + await request(app).get("/api/v1/bookings/search").set(noisy).expect(200); + } + await request(app).get("/api/v1/bookings/search").set(noisy).expect(429); + }); + + it("uses the default limiter (config-driven 60 rps / 120 burst) when none is injected", async () => { + const app = buildApp({ repository: new InMemoryBookingIntentRepository() }); + const res = await request(app).get("/api/v1/bookings/search").set(AUTH).expect(200); + expect(res.headers["x-ratelimit-limit"]).toBe("120"); + }); +}); diff --git a/src/routes/bookings.ts b/src/routes/bookings.ts new file mode 100644 index 00000000..621344dd --- /dev/null +++ b/src/routes/bookings.ts @@ -0,0 +1,197 @@ +/** + * @file src/routes/bookings.ts + * + * Express router for the /api/v1/bookings resource. + * + * GET /api/v1/bookings/search + * Searches booking records for the authenticated caller. This endpoint is + * the primary read path for dashboards and partner syncs, so it is the + * highest-traffic GET in the service — and therefore the first endpoint + * moved off the coarse global fixed-window limiter onto a per-tenant + * leaky bucket (60 rps sustained, 120 burst) keyed by trusted tenant + * identity. One noisy tenant can now only throttle itself, never the + * shared search path. See: + * - src/middleware/tenantLeakyBucket.ts (implementation) + * - docs/api/bookings-search.md (operator + client documentation) + * + * Query params: + * q - free-text match over id, slotId, professional, note (case-insensitive) + * status - exact booking status filter + * slotId - exact slot filter + * from/to - ISO-8601 datetimes; only bookings overlapping [from, to] are returned + * limit - page size, 1..100 (default 50) + * offset - page offset, >= 0 (default 0) + * + * Response: { success, data: { results, total, limit, offset } } + * Rate-limited responses: 429 + Retry-After header + { success: false, error, retryAfter }. + */ + +import { Router, type Request, type RequestHandler, type Response } from "express"; +import { requireAuthenticatedActor } from "../middleware/auth.js"; +import { createTenantLeakyBucketRateLimiter } from "../middleware/tenantLeakyBucket.js"; +import { + InMemoryBookingIntentRepository, + type BookingIntentRecord, + type BookingIntentStatus, +} from "../modules/booking-intents/booking-intent-repository.js"; +import { logger } from "../utils/logger.js"; + +const VALID_STATUSES: ReadonlySet = new Set([ + "pending", + "confirmed", + "firm", + "cancelled", + "expired", + "hold_placed", + "hold_refunded", +]); + +const MAX_LIMIT = 100; +const DEFAULT_LIMIT = 50; + +export interface BookingsRouterOptions { + /** Override the rate limiter (tests). Defaults to the per-tenant leaky bucket (60 rps / 120 burst). */ + rateLimiter?: RequestHandler; + /** Override the repository (tests / DB-backed wiring). */ + repository?: InMemoryBookingIntentRepository; +} + +interface ParsedSearchQuery { + q?: string; + status?: BookingIntentStatus; + slotId?: string; + fromMs?: number; + toMs?: number; + limit: number; + offset: number; +} + +type QueryParseResult = + | { ok: true; query: ParsedSearchQuery } + | { ok: false; status: number; error: string }; + +function parseSearchQuery(req: Request): QueryParseResult { + const raw = req.query; + + const q = typeof raw.q === "string" ? raw.q.trim() : undefined; + if (q !== undefined && q.length > 200) { + return { ok: false, status: 400, error: "q must be at most 200 characters" }; + } + + let status: BookingIntentStatus | undefined; + if (typeof raw.status === "string" && raw.status.trim()) { + const candidate = raw.status.trim().toLowerCase(); + if (!VALID_STATUSES.has(candidate)) { + return { + ok: false, + status: 400, + error: `status must be one of: ${[...VALID_STATUSES].join(", ")}`, + }; + } + status = candidate as BookingIntentStatus; + } + + const slotId = typeof raw.slotId === "string" && raw.slotId.trim() ? raw.slotId.trim() : undefined; + + let fromMs: number | undefined; + if (typeof raw.from === "string" && raw.from.trim()) { + fromMs = Date.parse(raw.from); + if (Number.isNaN(fromMs)) { + return { ok: false, status: 400, error: "from must be a valid ISO 8601 datetime" }; + } + } + + let toMs: number | undefined; + if (typeof raw.to === "string" && raw.to.trim()) { + toMs = Date.parse(raw.to); + if (Number.isNaN(toMs)) { + return { ok: false, status: 400, error: "to must be a valid ISO 8601 datetime" }; + } + } + + if (fromMs !== undefined && toMs !== undefined && fromMs > toMs) { + return { ok: false, status: 400, error: "from must not be after to" }; + } + + let limit = DEFAULT_LIMIT; + if (raw.limit !== undefined) { + limit = Number.parseInt(String(raw.limit), 10); + if (!Number.isFinite(limit) || limit < 1 || limit > MAX_LIMIT) { + return { ok: false, status: 400, error: `limit must be between 1 and ${MAX_LIMIT}` }; + } + } + + let offset = 0; + if (raw.offset !== undefined) { + offset = Number.parseInt(String(raw.offset), 10); + if (!Number.isFinite(offset) || offset < 0) { + return { ok: false, status: 400, error: "offset must be a non-negative integer" }; + } + } + + return { ok: true, query: { q, status, slotId, fromMs, toMs, limit, offset } }; +} + +function matchesQuery(record: BookingIntentRecord, query: ParsedSearchQuery): boolean { + if (query.status && record.status !== query.status) return false; + if (query.slotId && record.slotId !== query.slotId) return false; + if (query.fromMs !== undefined && record.endTime < query.fromMs) return false; + if (query.toMs !== undefined && record.startTime > query.toMs) return false; + if (query.q) { + const needle = query.q.toLowerCase(); + const haystack = [record.id, record.slotId, record.professional, record.note ?? ""] + .join("\n") + .toLowerCase(); + if (!haystack.includes(needle)) return false; + } + return true; +} + +export function createBookingsRouter(options: BookingsRouterOptions = {}): Router { + const router = Router(); + const repository = options.repository ?? new InMemoryBookingIntentRepository(); + const searchRateLimiter = + options.rateLimiter ?? + createTenantLeakyBucketRateLimiter({ routeScope: "bookings:search" }); + + router.get( + "/search", + requireAuthenticatedActor(["customer", "professional", "admin", "support"]), + searchRateLimiter, + async (req: Request, res: Response): Promise => { + try { + const parsed = parseSearchQuery(req); + if (!parsed.ok) { + res.status(parsed.status).json({ success: false, error: parsed.error }); + return; + } + + // Tenant data isolation: callers only ever see their own bookings. + + const callerId = String((req as any).auth?.userId ?? ""); + const records = await repository.listByCustomer(callerId); + + const matched = records.filter((record) => matchesQuery(record, parsed.query)); + const results = matched.slice(parsed.query.offset, parsed.query.offset + parsed.query.limit); + + res.status(200).json({ + success: true, + data: { + results, + total: matched.length, + limit: parsed.query.limit, + offset: parsed.query.offset, + }, + }); + } catch (err) { + logger.error({ err }, "bookings search failed"); + res.status(500).json({ success: false, error: "Search failed" }); + } + }, + ); + + return router; +} + +const bookingsRouter = createBookingsRouter(); +export default bookingsRouter;