diff --git a/src/index.ts b/src/index.ts index 24934bc6..29318006 100644 --- a/src/index.ts +++ b/src/index.ts @@ -122,6 +122,65 @@ const _shutdownHooks: Array<() => void> = []; logger.info("outbox-relay worker started"); })(); +// ─── Expire Pending Booking-Intents Worker (#588) ─────────────────────────── +// Scans for booking intents stuck in `pending` (awaiting payment/confirmation) +// beyond the TTL (default 30 min), cancels them, releases the reserved slot +// inventory, and emits a durable `booking_intent_expired` outbox event. +// Enabled by default; set EXPIRE_BOOKING_INTENTS_DISABLED=true to skip. +(async () => { + if (process.env.EXPIRE_BOOKING_INTENTS_DISABLED === "true") { + logger.info("expire-booking-intents worker disabled via EXPIRE_BOOKING_INTENTS_DISABLED"); + return; + } + + const { createExpireBookingIntentsWorker } = await import( + "./scheduler/expireBookingIntents.js" + ); + const { BookingIntentService } = await import( + "./modules/booking-intents/booking-intent-service.js" + ); + const { InMemoryBookingIntentRepository } = await import( + "./modules/booking-intents/booking-intent-repository.js" + ); + const { InMemorySlotRepository } = await import("./modules/slots/slot-repository.js"); + const { insertIntoOutbox } = await import("./services/outboxRelay.js"); + const { withTransaction } = await import("./db/connection.js"); + + const slotRepo = new InMemorySlotRepository(); + const bookingIntentRepo = new InMemoryBookingIntentRepository(); + const bookingIntentService = new BookingIntentService(bookingIntentRepo, slotRepo); + + const controller = new AbortController(); + _shutdownHooks.push(() => controller.abort()); + + createExpireBookingIntentsWorker( + { + bookingIntentRepository: bookingIntentRepo, + bookingIntentService, + emitExpired: async (event) => { + // Durable event emission through the transactional outbox; the relay + // worker publishes it to downstream consumers (at-least-once). + await withTransaction(async (client) => { + await insertIntoOutbox(client, "booking_intent_expired", event.intentId, { + intentId: event.intentId, + slotId: event.slotId, + customerId: event.customerId, + expiredAtMs: event.expiredAtMs, + }); + }); + }, + }, + { + ttlMs: Number(process.env.EXPIRE_BOOKING_INTENTS_TTL_MS) || undefined, + batchSize: Number(process.env.EXPIRE_BOOKING_INTENTS_BATCH_SIZE) || undefined, + safetyThreshold: Number(process.env.EXPIRE_BOOKING_INTENTS_SAFETY_THRESHOLD) || undefined, + intervalMs: Number(process.env.EXPIRE_BOOKING_INTENTS_INTERVAL_MS) || undefined, + }, + ).start(); + + logger.info("expire-booking-intents worker started"); +})(); + const PORT = config.port || 3001; const server = app.listen(PORT, () => { logger.info({ port: PORT }, `ChronoPay API listening on http://localhost:${PORT}`); diff --git a/src/metrics.ts b/src/metrics.ts index dc322811..b038d02f 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -376,6 +376,33 @@ export const expiryCleanupSafetyBrakeTriggers = createBudgetedCounter({ registers: [register], }); +// ─── Expire pending booking-intents worker metrics ─────────────────────────── + +/** + * Counter incremented each time the expire-booking-intents worker cancels a + * booking intent stuck in `pending` beyond the configured TTL. Equivalent to + * the `booking_intents_expired_total` gauge requested in issue #588. + */ +export const bookingIntentsExpiredTotal = createBudgetedCounter({ + name: "booking_intents_expired_total", + help: "Total number of stale pending booking intents expired by the expire-booking-intents worker", + labels: [], + budget: 0, + registers: [register], +}); + +/** + * Counter incremented each time the expire-booking-intents worker skips a + * sweep because the candidate count exceeded the safety threshold. + */ +export const expireBookingIntentsSafetyBrakeTriggers = createBudgetedCounter({ + name: "expire_booking_intents_safety_brake_triggers_total", + help: "Total number of expire-booking-intents sweeps skipped because the candidate count exceeded the safety threshold", + labels: [], + budget: 0, + registers: [register], +}); + // ─── Refundable hold expiry sweeper metrics ──────────────────────────────────── export const refundableHoldReleaseLagSeconds = createBudgetedHistogram({ diff --git a/src/modules/booking-intents/__tests__/pg-booking-intent-repository.test.ts b/src/modules/booking-intents/__tests__/pg-booking-intent-repository.test.ts index 7a80fef6..b23475e8 100644 --- a/src/modules/booking-intents/__tests__/pg-booking-intent-repository.test.ts +++ b/src/modules/booking-intents/__tests__/pg-booking-intent-repository.test.ts @@ -212,4 +212,37 @@ describe("PgBookingIntentRepository", () => { expect(result).toEqual(expectedRecord); }); }); + + describe("findStalePendingIntents", () => { + it("claims stale pending intents with FOR UPDATE SKIP LOCKED", async () => { + const cutoffMs = new Date("2026-01-02T00:00:00.000Z").getTime(); + mockQuery.mockResolvedValueOnce({ rows: [dbRow], rowCount: 1 } as any); + + const result = await repository.findStalePendingIntents(cutoffMs, 50); + + expect(mockQuery).toHaveBeenCalledWith( + expect.stringContaining("SELECT * FROM booking_intents"), + expect.arrayContaining([new Date(cutoffMs), 50]), + ); + + const sql = mockQuery.mock.calls[0][0] as string; + expect(sql).toContain("WHERE status = 'pending' AND created_at <= $1"); + expect(sql).toContain("ORDER BY created_at ASC"); + expect(sql).toContain("FOR UPDATE SKIP LOCKED"); + + expect(result).toEqual([expectedRecord]); + }); + + it("returns an empty array when no stale intents exist", async () => { + mockQuery.mockResolvedValueOnce({ rows: [], rowCount: 0 } as any); + + const result = await repository.findStalePendingIntents(1000, 10); + + expect(mockQuery).toHaveBeenCalledWith( + expect.stringContaining("SELECT * FROM booking_intents"), + [new Date(1000), 10], + ); + expect(result).toEqual([]); + }); + }); }); diff --git a/src/modules/booking-intents/booking-intent-repository.ts b/src/modules/booking-intents/booking-intent-repository.ts index 36f0babd..70106272 100644 --- a/src/modules/booking-intents/booking-intent-repository.ts +++ b/src/modules/booking-intents/booking-intent-repository.ts @@ -92,6 +92,13 @@ export interface BookingIntentRepository { update(id: string, updates: Partial>): BookingIntentRecord | Promise; updateTokenInfo?(id: string, tokenAsset: string, mintTxHash: string): Promise | void; findExpiredHolds(nowMs: number): BookingIntentRecord[] | Promise; + /** + * Returns up to `limit` intents stuck in `pending` whose `createdAt` is at or + * before `cutoffMs`, oldest first. Used by the expire-booking-intents worker. + * PostgreSQL implementations should claim rows with `FOR UPDATE SKIP LOCKED` + * so concurrent worker instances never process the same intent twice. + */ + findStalePendingIntents(cutoffMs: number, limit: number): BookingIntentRecord[] | Promise; } const ACTIVE_HOLD_STATUSES: BookingIntentStatus[] = ["pending", "hold_placed"]; @@ -188,4 +195,18 @@ export class InMemoryBookingIntentRepository implements BookingIntentRepository ) .map((i) => ({ ...i })); } + + findStalePendingIntents(cutoffMs: number, limit: number): BookingIntentRecord[] { + return this.intents + .filter( + (entry) => + entry.status === "pending" && + new Date(entry.createdAt).getTime() <= cutoffMs, + ) + .sort( + (a, b) => new Date(a.createdAt).getTime() - new Date(b.createdAt).getTime(), + ) + .slice(0, limit) + .map((i) => ({ ...i })); + } } diff --git a/src/modules/booking-intents/pg-booking-intent-repository.ts b/src/modules/booking-intents/pg-booking-intent-repository.ts index 7bc924e7..9d927713 100644 --- a/src/modules/booking-intents/pg-booking-intent-repository.ts +++ b/src/modules/booking-intents/pg-booking-intent-repository.ts @@ -86,6 +86,27 @@ export class PgBookingIntentRepository implements BookingIntentRepository { await this.dbQuery(sql, [id, tokenAsset, mintTxHash]); } + /** + * Claims up to `limit` stale `pending` intents older than `cutoffMs`. + * + * `FOR UPDATE SKIP LOCKED` lets multiple worker instances sweep safely: rows + * locked by a concurrent transaction (a competing claim) are skipped instead + * of blocking, so an intent can never be claimed by two workers at once. + * Rows are returned oldest-first so retries/backfills naturally drain the + * longest-stuck intents first. + */ + async findStalePendingIntents(cutoffMs: number, limit: number): Promise { + const sql = ` + SELECT * FROM booking_intents + WHERE status = 'pending' AND created_at <= $1 + ORDER BY created_at ASC + LIMIT $2 + FOR UPDATE SKIP LOCKED + `; + const res = await this.dbQuery(sql, [new Date(cutoffMs), limit]); + return res.rows.map((row) => this.mapRowToRecord(row)); + } + /** * Maps a database row to a BookingIntentRecord domain object. * Converts database TIMESTAMPTZ to milliseconds for the domain record. diff --git a/src/scheduler/__tests__/expireBookingIntents.test.ts b/src/scheduler/__tests__/expireBookingIntents.test.ts new file mode 100644 index 00000000..25b46b9a --- /dev/null +++ b/src/scheduler/__tests__/expireBookingIntents.test.ts @@ -0,0 +1,623 @@ +// @ts-nocheck +import { jest } from "@jest/globals"; +import { + expireBookingIntentsOnce, + runExpireBookingIntentsWorker, + createExpireBookingIntentsWorker, + bookingIntentExpiredEvents, + BookingIntentExpiredEvent, +} from "../expireBookingIntents.js"; +import { BookingIntentService } from "../../modules/booking-intents/booking-intent-service.js"; +import { + InMemoryBookingIntentRepository, + BookingIntentRepository, +} from "../../modules/booking-intents/booking-intent-repository.js"; +import { InMemorySlotRepository } from "../../modules/slots/slot-repository.js"; +import { register } from "../../metrics.js"; + +const SLOT_1 = "slot-11111111-1111-4111-8111-111111111111"; +const SLOT_2 = "slot-22222222-2222-4222-8222-222222222222"; +const SLOT_3 = "slot-33333333-3333-4333-8333-333333333333"; + +const TTL_30_MIN = 30 * 60 * 1000; + +async function metricValue(metric: string): Promise { + const metrics = await register.metrics(); + const line = metrics.split("\n").find((entry) => entry.startsWith(metric)); + if (!line) return 0; + return Number(line.trim().split(/\s+/).at(-1)); +} + +describe("expireBookingIntents worker", () => { + const baseTime = 1_700_000_000_000; + let bookingIntentRepo: InMemoryBookingIntentRepository; + let slotRepo: InMemorySlotRepository; + let bookingIntentService: BookingIntentService; + + beforeEach(() => { + jest.useFakeTimers().setSystemTime(baseTime); + bookingIntentRepo = new InMemoryBookingIntentRepository(); + slotRepo = new InMemorySlotRepository(); + bookingIntentService = new BookingIntentService( + bookingIntentRepo, + slotRepo, + () => new Date(baseTime).toISOString(), + ); + register.resetMetrics(); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + async function createPendingIntent(slotId: string, customerId = "customer-1") { + return bookingIntentService.createIntent( + { slotId }, + { userId: customerId, role: "customer" }, + ); + } + + it("expires stale pending booking intents older than the TTL and releases the slot", async () => { + const intent = await createPendingIntent(SLOT_1); + + const emitted: BookingIntentExpiredEvent[] = []; + const listener = (event: BookingIntentExpiredEvent) => emitted.push(event); + bookingIntentExpiredEvents.on("booking_intent_expired", listener); + + try { + const result = await expireBookingIntentsOnce( + { bookingIntentRepository: bookingIntentRepo, bookingIntentService }, + { ttlMs: TTL_30_MIN, batchSize: 10 }, + baseTime + TTL_30_MIN, // createdAt (baseTime) is exactly at the cutoff + ); + + expect(result.expiredCount).toBe(1); + expect(result.candidatesCount).toBe(1); + expect(result.failures).toEqual([]); + + const updated = bookingIntentRepo.findById(intent.id); + expect(updated?.status).toBe("expired"); + + // Slot inventory released back to bookable. + expect(slotRepo.findById(SLOT_1)?.bookable).toBe(true); + + expect(emitted).toHaveLength(1); + expect(emitted[0].intentId).toBe(intent.id); + expect(emitted[0].slotId).toBe(SLOT_1); + expect(emitted[0].customerId).toBe("customer-1"); + expect(emitted[0].expiredAtMs).toBe(baseTime + TTL_30_MIN); + + expect(await metricValue("booking_intents_expired_total")).toBe(1); + } finally { + bookingIntentExpiredEvents.off("booking_intent_expired", listener); + } + }); + + it("does not expire intents still within the TTL", async () => { + const intent = await createPendingIntent(SLOT_1); + + const result = await expireBookingIntentsOnce( + { bookingIntentRepository: bookingIntentRepo, bookingIntentService }, + { ttlMs: TTL_30_MIN }, + baseTime + TTL_30_MIN - 1, // 1 ms before the 30-minute cutoff + ); + + expect(result.expiredCount).toBe(0); + expect(bookingIntentRepo.findById(intent.id)?.status).toBe("pending"); + // Slot must remain reserved. + expect(slotRepo.findById(SLOT_1)?.bookable).toBe(false); + expect(await metricValue("booking_intents_expired_total")).toBe(0); + }); + + it("expires an intent whose createdAt is exactly at the 30-minute boundary", async () => { + const intent = await createPendingIntent(SLOT_1); + + const result = await expireBookingIntentsOnce( + { bookingIntentRepository: bookingIntentRepo, bookingIntentService }, + { ttlMs: TTL_30_MIN }, + baseTime + TTL_30_MIN, // createdAt === cutoff → stale + ); + + expect(result.expiredCount).toBe(1); + expect(bookingIntentRepo.findById(intent.id)?.status).toBe("expired"); + }); + + it("never double-cancels when two workers sweep the same intents", async () => { + const intent = await createPendingIntent(SLOT_1); + const deps = { bookingIntentRepository: bookingIntentRepo, bookingIntentService }; + + const first = await expireBookingIntentsOnce(deps, { ttlMs: TTL_30_MIN }, baseTime + TTL_30_MIN); + const second = await expireBookingIntentsOnce(deps, { ttlMs: TTL_30_MIN }, baseTime + TTL_30_MIN); + + expect(first.expiredCount).toBe(1); + // Second sweep finds nothing to do: status is already expired and the + // candidate query only returns `pending` intents. + expect(second.expiredCount).toBe(0); + expect(bookingIntentRepo.findById(intent.id)?.status).toBe("expired"); + expect(slotRepo.findById(SLOT_1)?.bookable).toBe(true); + expect(await metricValue("booking_intents_expired_total")).toBe(1); + }); + + it("skips intents that are no longer pending (e.g. slot already completed)", async () => { + const intent = await createPendingIntent(SLOT_1); + // Simulate a concurrent transition (e.g. buyer confirmed before the worker). + const confirmed = bookingIntentRepo.updateStatus(intent.id, "confirmed"); + + // Force the stale-candidate query to return the now-confirmed intent, as + // could happen between the candidate fetch and the re-verification. + const raceyRepo: BookingIntentRepository = { + ...bookingIntentRepo, + findStalePendingIntents: () => [confirmed], + }; + + const result = await expireBookingIntentsOnce( + { bookingIntentRepository: raceyRepo, bookingIntentService }, + { ttlMs: TTL_30_MIN }, + baseTime + TTL_30_MIN, + ); + + expect(result.expiredCount).toBe(0); + // Intent untouched, slot not released. + expect(bookingIntentRepo.findById(intent.id)?.status).toBe("confirmed"); + expect(slotRepo.findById(SLOT_1)?.bookable).toBe(false); + expect(await metricValue("booking_intents_expired_total")).toBe(0); + }); + + it("expires multiple stale intents and leaves fresh ones alone", async () => { + const stale1 = await createPendingIntent(SLOT_1); + const stale2 = await createPendingIntent(SLOT_2); + + // SLOT_3 is seeded non-bookable; make it bookable so a fresh intent can exist. + slotRepo.updateBookable(SLOT_3, true); + const fresh = await createPendingIntent(SLOT_3, "customer-3"); + // Push the fresh intent's creation time inside the TTL window. + bookingIntentRepo.update(fresh.id, { + createdAt: new Date(baseTime + TTL_30_MIN + 1000).toISOString(), + }); + + const result = await expireBookingIntentsOnce( + { bookingIntentRepository: bookingIntentRepo, bookingIntentService }, + { ttlMs: TTL_30_MIN }, + baseTime + TTL_30_MIN, + ); + + expect(result.expiredCount).toBe(2); + expect(bookingIntentRepo.findById(stale1.id)?.status).toBe("expired"); + expect(bookingIntentRepo.findById(stale2.id)?.status).toBe("expired"); + expect(bookingIntentRepo.findById(fresh.id)?.status).toBe("pending"); + }); + + it("respects the batch size when more intents are stale than the batch limit", async () => { + const stale1 = await createPendingIntent(SLOT_1); + const stale2 = await createPendingIntent(SLOT_2); + slotRepo.updateBookable(SLOT_3, true); + const stale3 = await createPendingIntent(SLOT_3, "customer-3"); + + const result = await expireBookingIntentsOnce( + { bookingIntentRepository: bookingIntentRepo, bookingIntentService }, + { ttlMs: TTL_30_MIN, batchSize: 2 }, + baseTime + TTL_30_MIN, + ); + + expect(result.expiredCount).toBe(2); + expect(result.candidatesCount).toBe(2); + // Oldest two intents are processed in this sweep; the third waits for the + // next sweep. Only one slot release for SLOT_1 remains for the next run. + expect(bookingIntentRepo.findById(stale1.id)?.status).toBe("expired"); + expect(bookingIntentRepo.findById(stale2.id)?.status).toBe("expired"); + expect(bookingIntentRepo.findById(stale3.id)?.status).toBe("pending"); + }); + + it("trips the safety brake when candidates exceed the threshold", async () => { + await createPendingIntent(SLOT_1); + await createPendingIntent(SLOT_2); + + const result = await expireBookingIntentsOnce( + { bookingIntentRepository: bookingIntentRepo, bookingIntentService }, + { ttlMs: TTL_30_MIN, safetyThreshold: 1 }, + baseTime + TTL_30_MIN, + ); + + expect(result.skippedBecauseThreshold).toBe(true); + expect(result.expiredCount).toBe(0); + expect(result.candidatesCount).toBe(2); + expect(await metricValue("expire_booking_intents_safety_brake_triggers_total")).toBe(1); + }); + + it("records per-intent failures without aborting the sweep", async () => { + await createPendingIntent(SLOT_1); + + const emitExpired = jest.fn(async () => { + throw new Error("outbox write failed"); + }); + + const result = await expireBookingIntentsOnce( + { + bookingIntentRepository: bookingIntentRepo, + bookingIntentService, + emitExpired, + }, + { ttlMs: TTL_30_MIN }, + baseTime + TTL_30_MIN, + ); + + // The domain transition succeeds; only the event delivery fails. + expect(result.expiredCount).toBe(1); + expect(result.failures).toHaveLength(1); + expect(result.failures[0].error).toMatch(/event emission failed: outbox write failed/); + expect(bookingIntentRepo.listAll()[0].status).toBe("expired"); + }); + + it("records a string (non-Error) event emission failure", async () => { + await createPendingIntent(SLOT_1); + + const emitExpired = jest.fn(async () => { + throw "outbox is down"; + }); + + const result = await expireBookingIntentsOnce( + { + bookingIntentRepository: bookingIntentRepo, + bookingIntentService, + emitExpired, + }, + { ttlMs: TTL_30_MIN }, + baseTime + TTL_30_MIN, + ); + + expect(result.failures).toHaveLength(1); + expect(result.failures[0].error).toMatch(/event emission failed: outbox is down/); + }); + + it("skips candidates whose intent can no longer be found", async () => { + class MissingFindRepo extends InMemoryBookingIntentRepository { + findById() { + return undefined; + } + } + + const repo = new MissingFindRepo(); + const service = new BookingIntentService( + repo, + slotRepo, + () => new Date(baseTime).toISOString(), + ); + const intent = await service.createIntent( + { slotId: SLOT_1 }, + { userId: "customer-1", role: "customer" }, + ); + + const result = await expireBookingIntentsOnce( + { bookingIntentRepository: repo, bookingIntentService: service }, + { ttlMs: TTL_30_MIN }, + baseTime + TTL_30_MIN, + ); + + expect(result.expiredCount).toBe(0); + expect(repo.listAll()[0].status).toBe("pending"); + expect(intent.id).toBeDefined(); + }); + + it("survives non-Error failures thrown by the repository", async () => { + class ThrowingFindRepo extends InMemoryBookingIntentRepository { + findById() { + throw "repository exploded"; + } + } + + const repo = new ThrowingFindRepo(); + const service = new BookingIntentService( + repo, + slotRepo, + () => new Date(baseTime).toISOString(), + ); + const intent = await service.createIntent( + { slotId: SLOT_1 }, + { userId: "customer-1", role: "customer" }, + ); + + const result = await expireBookingIntentsOnce( + { bookingIntentRepository: repo, bookingIntentService: service }, + { ttlMs: TTL_30_MIN }, + baseTime + TTL_30_MIN, + ); + + expect(result.expiredCount).toBe(0); + expect(result.failures).toHaveLength(1); + expect(result.failures[0].intentId).toBe(intent.id); + expect(result.failures[0].error).toBe("repository exploded"); + }); + + it("honors the emitExpired dependency for durable outbox wiring", async () => { + const intent = await createPendingIntent(SLOT_1); + const emitExpired = jest.fn(async (event: BookingIntentExpiredEvent) => { + expect(event.intentId).toBe(intent.id); + }); + + await expireBookingIntentsOnce( + { + bookingIntentRepository: bookingIntentRepo, + bookingIntentService, + emitExpired, + }, + { ttlMs: TTL_30_MIN }, + baseTime + TTL_30_MIN, + ); + + expect(emitExpired).toHaveBeenCalledTimes(1); + }); + + it("resolves the TTL from the environment when no override is provided", async () => { + const intent = await createPendingIntent(SLOT_1); + const previous = process.env.EXPIRE_BOOKING_INTENTS_TTL_MS; + process.env.EXPIRE_BOOKING_INTENTS_TTL_MS = String(60 * 1000); + + try { + // 30 min elapses → way past the 1-minute env TTL. + const result = await expireBookingIntentsOnce( + { bookingIntentRepository: bookingIntentRepo, bookingIntentService }, + {}, + baseTime + TTL_30_MIN, + ); + expect(result.expiredCount).toBe(1); + expect(bookingIntentRepo.findById(intent.id)?.status).toBe("expired"); + } finally { + if (previous === undefined) { + delete process.env.EXPIRE_BOOKING_INTENTS_TTL_MS; + } else { + process.env.EXPIRE_BOOKING_INTENTS_TTL_MS = previous; + } + } + }); + + it("falls back to defaults when env values are empty or invalid", async () => { + const intent = await createPendingIntent(SLOT_1); + const previousTtl = process.env.EXPIRE_BOOKING_INTENTS_TTL_MS; + const previousBatch = process.env.EXPIRE_BOOKING_INTENTS_BATCH_SIZE; + const previousSafety = process.env.EXPIRE_BOOKING_INTENTS_SAFETY_THRESHOLD; + const previousInterval = process.env.EXPIRE_BOOKING_INTENTS_INTERVAL_MS; + + process.env.EXPIRE_BOOKING_INTENTS_TTL_MS = "not-a-number"; + process.env.EXPIRE_BOOKING_INTENTS_BATCH_SIZE = ""; + process.env.EXPIRE_BOOKING_INTENTS_SAFETY_THRESHOLD = "0"; + process.env.EXPIRE_BOOKING_INTENTS_INTERVAL_MS = "-5"; + + try { + // Invalid TTL falls back to the 30-minute default: createdAt === cutoff, + // so the intent is stale at exactly the boundary. + const result = await expireBookingIntentsOnce( + { bookingIntentRepository: bookingIntentRepo, bookingIntentService }, + {}, + baseTime + TTL_30_MIN, + ); + expect(result.expiredCount).toBe(1); + expect(bookingIntentRepo.findById(intent.id)?.status).toBe("expired"); + } finally { + if (previousTtl === undefined) { + delete process.env.EXPIRE_BOOKING_INTENTS_TTL_MS; + } else { + process.env.EXPIRE_BOOKING_INTENTS_TTL_MS = previousTtl; + } + if (previousBatch === undefined) { + delete process.env.EXPIRE_BOOKING_INTENTS_BATCH_SIZE; + } else { + process.env.EXPIRE_BOOKING_INTENTS_BATCH_SIZE = previousBatch; + } + if (previousSafety === undefined) { + delete process.env.EXPIRE_BOOKING_INTENTS_SAFETY_THRESHOLD; + } else { + process.env.EXPIRE_BOOKING_INTENTS_SAFETY_THRESHOLD = previousSafety; + } + if (previousInterval === undefined) { + delete process.env.EXPIRE_BOOKING_INTENTS_INTERVAL_MS; + } else { + process.env.EXPIRE_BOOKING_INTENTS_INTERVAL_MS = previousInterval; + } + } + }); + + it("reads all worker config values from the environment", async () => { + await createPendingIntent(SLOT_1); + const previous = { + ttl: process.env.EXPIRE_BOOKING_INTENTS_TTL_MS, + batch: process.env.EXPIRE_BOOKING_INTENTS_BATCH_SIZE, + safety: process.env.EXPIRE_BOOKING_INTENTS_SAFETY_THRESHOLD, + interval: process.env.EXPIRE_BOOKING_INTENTS_INTERVAL_MS, + }; + + process.env.EXPIRE_BOOKING_INTENTS_TTL_MS = "60000"; + process.env.EXPIRE_BOOKING_INTENTS_BATCH_SIZE = "50"; + process.env.EXPIRE_BOOKING_INTENTS_SAFETY_THRESHOLD = "500"; + process.env.EXPIRE_BOOKING_INTENTS_INTERVAL_MS = "5000"; + + try { + const result = await expireBookingIntentsOnce( + { bookingIntentRepository: bookingIntentRepo, bookingIntentService }, + {}, + baseTime + TTL_30_MIN, + ); + // TTL of 60s makes the intent (created at baseTime) deeply stale. + expect(result.expiredCount).toBe(1); + } finally { + if (previous.ttl === undefined) { + delete process.env.EXPIRE_BOOKING_INTENTS_TTL_MS; + } else { + process.env.EXPIRE_BOOKING_INTENTS_TTL_MS = previous.ttl; + } + if (previous.batch === undefined) { + delete process.env.EXPIRE_BOOKING_INTENTS_BATCH_SIZE; + } else { + process.env.EXPIRE_BOOKING_INTENTS_BATCH_SIZE = previous.batch; + } + if (previous.safety === undefined) { + delete process.env.EXPIRE_BOOKING_INTENTS_SAFETY_THRESHOLD; + } else { + process.env.EXPIRE_BOOKING_INTENTS_SAFETY_THRESHOLD = previous.safety; + } + if (previous.interval === undefined) { + delete process.env.EXPIRE_BOOKING_INTENTS_INTERVAL_MS; + } else { + process.env.EXPIRE_BOOKING_INTENTS_INTERVAL_MS = previous.interval; + } + } + }); + + it("supports repositories whose findStalePendingIntents returns a Promise", async () => { + class AsyncFindRepo extends InMemoryBookingIntentRepository { + async findStalePendingIntents(cutoffMs: number, limit: number) { + return super.findStalePendingIntents(cutoffMs, limit); + } + } + + const asyncRepo = new AsyncFindRepo(); + const service = new BookingIntentService( + asyncRepo, + slotRepo, + () => new Date(baseTime).toISOString(), + ); + const intent = await service.createIntent( + { slotId: SLOT_1 }, + { userId: "customer-1", role: "customer" }, + ); + + const result = await expireBookingIntentsOnce( + { bookingIntentRepository: asyncRepo, bookingIntentService: service }, + { ttlMs: TTL_30_MIN }, + baseTime + TTL_30_MIN, + ); + + expect(result.expiredCount).toBe(1); + expect(asyncRepo.findById(intent.id)?.status).toBe("expired"); + }); + + it("managed worker ignores duplicate start calls", async () => { + const intent = await createPendingIntent(SLOT_1); + jest.useRealTimers(); + + const worker = createExpireBookingIntentsWorker( + { bookingIntentRepository: bookingIntentRepo, bookingIntentService }, + { ttlMs: 1, intervalMs: 1000 }, + ); + + worker.start(); + worker.start(); // second start is a no-op + let result = worker.getLastResult(); + for (let i = 0; i < 20 && result === null; i += 1) { + await new Promise((resolve) => setTimeout(resolve, 5)); + result = worker.getLastResult(); + } + worker.stop(); + + expect(result?.expiredCount).toBe(1); + expect(bookingIntentRepo.findById(intent.id)?.status).toBe("expired"); + }); + + it("stops gracefully when aborted mid-run", async () => { + const intent = await createPendingIntent(SLOT_1); + jest.setSystemTime(baseTime + 1_000_000); + + const abortController = new AbortController(); + const workerPromise = runExpireBookingIntentsWorker( + abortController.signal, + { bookingIntentRepository: bookingIntentRepo, bookingIntentService }, + { intervalMs: 1000, ttlMs: 1 }, + ); + + await Promise.resolve(); + abortController.abort(); + await expect(workerPromise).resolves.toBeUndefined(); + expect(bookingIntentRepo.findById(intent.id)?.status).toBe("expired"); + }); + + it("aborts the worker loop while it is sleeping between sweeps", async () => { + const intent = await createPendingIntent(SLOT_1); + jest.setSystemTime(baseTime + 1_000_000); + + const abortController = new AbortController(); + const workerPromise = runExpireBookingIntentsWorker( + abortController.signal, + { bookingIntentRepository: bookingIntentRepo, bookingIntentService }, + { intervalMs: 1000, ttlMs: 1 }, + ); + + // Flush the microtask queue so the first sweep completes and the loop + // reaches its sleep between sweeps, then abort mid-sleep. + for (let i = 0; i < 10; i += 1) { + await Promise.resolve(); + } + abortController.abort(); + await expect(workerPromise).resolves.toBeUndefined(); + expect(bookingIntentRepo.findById(intent.id)?.status).toBe("expired"); + }); + + it("managed worker records the last sweep result and can be stopped", async () => { + const intent = await createPendingIntent(SLOT_1); + // Use real timers for the loop/poll so Date.now() is far past the intent's + // creation time (making it stale) and setTimeout callbacks actually fire. + jest.useRealTimers(); + + const worker = createExpireBookingIntentsWorker( + { bookingIntentRepository: bookingIntentRepo, bookingIntentService }, + { ttlMs: 1, intervalMs: 1000 }, + ); + + worker.start(); + let result = worker.getLastResult(); + for (let i = 0; i < 20 && result === null; i += 1) { + await new Promise((resolve) => setTimeout(resolve, 5)); + result = worker.getLastResult(); + } + worker.stop(); + + expect(result?.expiredCount).toBe(1); + expect(bookingIntentRepo.findById(intent.id)?.status).toBe("expired"); + }); + + it("falls back to default config when no overrides or clock are supplied", async () => { + const intent = await createPendingIntent(SLOT_1); + // Fake clock is frozen at baseTime, so the default 30-minute TTL means the + // intent (created at baseTime) is still fresh and must not be expired. + const result = await expireBookingIntentsOnce({ + bookingIntentRepository: bookingIntentRepo, + bookingIntentService, + }); + + expect(result.expiredCount).toBe(0); + expect(bookingIntentRepo.findById(intent.id)?.status).toBe("pending"); + }); + + it("worker loop uses default config when none is supplied", async () => { + const abortController = new AbortController(); + abortController.abort(); + + // Already-aborted signal: the loop never sweeps and resolves immediately. + const workerPromise = runExpireBookingIntentsWorker(abortController.signal, { + bookingIntentRepository: bookingIntentRepo, + bookingIntentService, + }); + + await expect(workerPromise).resolves.toBeUndefined(); + }); + + it("managed worker uses default config and stops cleanly when aborted mid-sweep", async () => { + const intent = await createPendingIntent(SLOT_1); + // Default config (30-minute TTL): the intent is fresh at the frozen clock, + // so a completed sweep reports zero expirations. + const worker = createExpireBookingIntentsWorker({ + bookingIntentRepository: bookingIntentRepo, + bookingIntentService, + }); + + worker.start(); + worker.stop(); // abort lands while the first sweep is still in flight + + // Flush the microtasks so the in-flight sweep and loop teardown complete. + for (let i = 0; i < 5; i += 1) { + await Promise.resolve(); + } + + const result = worker.getLastResult(); + expect(result?.expiredCount).toBe(0); + expect(bookingIntentRepo.findById(intent.id)?.status).toBe("pending"); + }); +}); diff --git a/src/scheduler/expireBookingIntents.ts b/src/scheduler/expireBookingIntents.ts new file mode 100644 index 00000000..7b015c14 --- /dev/null +++ b/src/scheduler/expireBookingIntents.ts @@ -0,0 +1,297 @@ +// @ts-nocheck +/** + * expireBookingIntents.ts + * + * Background job that scans for booking intents stuck in `pending` (i.e. + * awaiting payment/confirmation) for longer than the configured TTL and + * cancels them: + * + * - Marks the intent `expired` (the domain's equivalent of `cancelled_expired`) + * - Releases the reserved slot inventory back to the marketplace + * - Emits a `booking_intent_expired` outbox event + * - Records a `booking_intents_expired_total` Prometheus metric + * + * Multi-instance safety + * --------------------- + * The Postgres repository claims stale rows with `FOR UPDATE SKIP LOCKED` + * (see PgBookingIntentRepository.findStalePendingIntents), so two worker + * instances never process the same intent concurrently. On top of that, this + * worker re-verifies the intent is still `pending` immediately before + * expiring it, so even a re-queued batch can never double-cancel an intent + * that was confirmed or completed in the meantime. + * + * Config (all optional, env-overridable): + * - ttlMs: how long an intent may stay `pending` (default 30 min) + * - batchSize: max intents claimed per sweep (default 100) + * - safetyThreshold: skip the sweep if candidates exceed this (default 10,000) + * - intervalMs: cron interval between sweeps (default 60 s) + */ + +import { EventEmitter } from "node:events"; +import type { + BookingIntentRepository, +} from "../modules/booking-intents/booking-intent-repository.js"; +import type { BookingIntentService } from "../modules/booking-intents/booking-intent-service.js"; +import { + bookingIntentsExpiredTotal, + expireBookingIntentsSafetyBrakeTriggers, +} from "../metrics.js"; +import { logger } from "../utils/logger.js"; + +// ─── Types ──────────────────────────────────────────────────────────────────── + +export interface ExpireBookingIntentsConfig { + /** Max age (ms) of a `pending` intent before it is expired. Default: 30 min. */ + ttlMs?: number; + /** Maximum number of intents claimed in a single sweep. Default: 100. */ + batchSize?: number; + /** Sweeps are skipped when candidate count exceeds this. Default: 10,000. */ + safetyThreshold?: number; + /** Interval (ms) between sweeps. Default: 60 s. */ + intervalMs?: number; +} + +export interface BookingIntentExpiredEvent { + intentId: string; + slotId: string; + customerId: string; + expiredAtMs: number; +} + +export interface ExpireBookingIntentsResult { + /** Number of intents successfully expired in this sweep. */ + expiredCount: number; + /** Number of stale candidates found before the safety brake. */ + candidatesCount: number; + /** True when the sweep was skipped because candidates exceeded the safety threshold. */ + skippedBecauseThreshold?: boolean; + /** Per-intent errors encountered while expiring. */ + failures: { intentId: string; error: string }[]; +} + +export interface ExpireBookingIntentsDependencies { + bookingIntentRepository: BookingIntentRepository; + bookingIntentService: BookingIntentService; + /** + * Emits the `booking_intent_expired` event. Defaults to an in-process + * EventEmitter emit; production wiring (src/index.ts) replaces this with an + * outbox insert so the event is durably relayed downstream. + */ + emitExpired?: (event: BookingIntentExpiredEvent) => void | Promise; +} + +// ─── Defaults & config resolution ───────────────────────────────────────────── + +const DEFAULT_CONFIG: Required = { + ttlMs: 30 * 60 * 1000, // 30 minutes + batchSize: 100, + safetyThreshold: 10_000, + intervalMs: 60 * 1000, // every 60 seconds +}; + +function parsePositiveInteger(value: string | undefined, defaultValue: number): number { + if (!value) return defaultValue; + const parsed = Number(value); + if (!Number.isInteger(parsed) || parsed < 1) return defaultValue; + return parsed; +} + +function resolveConfig( + overrides: ExpireBookingIntentsConfig = {}, +): Required { + return { + ttlMs: + overrides.ttlMs ?? + parsePositiveInteger(process.env.EXPIRE_BOOKING_INTENTS_TTL_MS, DEFAULT_CONFIG.ttlMs), + batchSize: + overrides.batchSize ?? + parsePositiveInteger(process.env.EXPIRE_BOOKING_INTENTS_BATCH_SIZE, DEFAULT_CONFIG.batchSize), + safetyThreshold: + overrides.safetyThreshold ?? + parsePositiveInteger( + process.env.EXPIRE_BOOKING_INTENTS_SAFETY_THRESHOLD, + DEFAULT_CONFIG.safetyThreshold, + ), + intervalMs: + overrides.intervalMs ?? + parsePositiveInteger(process.env.EXPIRE_BOOKING_INTENTS_INTERVAL_MS, DEFAULT_CONFIG.intervalMs), + }; +} + +// ─── Event emitter ──────────────────────────────────────────────────────────── + +/** + * In-process `booking_intent_expired` event bus. Production deployments should + * prefer the outbox wiring in src/index.ts so events survive restarts. + */ +export const bookingIntentExpiredEvents = new EventEmitter(); + +function defaultEmitExpired(event: BookingIntentExpiredEvent): void { + bookingIntentExpiredEvents.emit("booking_intent_expired", event); +} + +// ─── Sweep execution ────────────────────────────────────────────────────────── + +/** + * Runs a single scan-and-expire sweep. + * + * 1. Claims up to `batchSize` stale `pending` intents (oldest first). + * 2. Trips the safety brake when the candidate count exceeds the threshold. + * 3. Re-verifies each intent is still `pending` before expiring it, so a + * confirmed/completed intent is never cancelled and two concurrent workers + * can never double-cancel. + * 4. Expires the intent (status -> `expired`) and releases the slot inventory. + * 5. Emits the `booking_intent_expired` event and records the metric. + * + * @param nowMs Override clock for deterministic tests. + */ +export async function expireBookingIntentsOnce( + dependencies: ExpireBookingIntentsDependencies, + configOverrides: ExpireBookingIntentsConfig = {}, + nowMs?: number, +): Promise { + const config = resolveConfig(configOverrides); + const now = nowMs ?? Date.now(); + const cutoff = now - config.ttlMs; + + const candidates = dependencies.bookingIntentRepository.findStalePendingIntents( + cutoff, + config.batchSize, + ); + const staleIntents = Array.isArray(candidates) ? candidates : await candidates; + + if (staleIntents.length > config.safetyThreshold) { + expireBookingIntentsSafetyBrakeTriggers.inc(); + return { + expiredCount: 0, + candidatesCount: staleIntents.length, + skippedBecauseThreshold: true, + failures: [], + }; + } + + const emitExpired = dependencies.emitExpired ?? defaultEmitExpired; + const failures: { intentId: string; error: string }[] = []; + let expiredCount = 0; + + for (const candidate of staleIntents) { + try { + // Re-verify current state before touching anything. This is the guard + // that prevents double-cancels when two workers sweep concurrently or a + // buyer confirmed/completed the intent in the meantime. + const current = await dependencies.bookingIntentRepository.findById(candidate.id); + if (!current || current.status !== "pending") { + continue; + } + + // Expires the intent (status -> 'expired') and releases its slot. + dependencies.bookingIntentService.expireIntent(candidate.id); + bookingIntentsExpiredTotal.inc(); + expiredCount += 1; + + const event: BookingIntentExpiredEvent = { + intentId: candidate.id, + slotId: candidate.slotId, + customerId: candidate.customerId, + expiredAtMs: now, + }; + try { + await emitExpired(event); + } catch (emitErr) { + // The domain transition already happened; surface the delivery + // failure separately so operators can re-emit if needed. + failures.push({ + intentId: candidate.id, + error: `event emission failed: ${emitErr instanceof Error ? emitErr.message : String(emitErr)}`, + }); + } + } catch (err) { + failures.push({ + intentId: candidate.id, + error: err instanceof Error ? err.message : String(err), + }); + logger.warn( + { intentId: candidate.id, error: err instanceof Error ? err.message : String(err) }, + "expire-booking-intents: failed to expire intent", + ); + } + } + + return { expiredCount, candidatesCount: staleIntents.length, failures }; +} + +// ─── Background worker loop ─────────────────────────────────────────────────── + +function sleep(ms: number, signal: AbortSignal): Promise { + return new Promise((resolve) => { + if (signal.aborted) { + return resolve(); + } + const timer = setTimeout(resolve, ms); + signal.addEventListener( + "abort", + () => { + clearTimeout(timer); + resolve(); + }, + { once: true }, + ); + }); +} + +/** + * Long-running cron loop. Sweeps immediately, then sleeps `intervalMs` + * (default 60 s) between sweeps until the signal is aborted. + */ +export async function runExpireBookingIntentsWorker( + signal: AbortSignal, + dependencies: ExpireBookingIntentsDependencies, + configOverrides: ExpireBookingIntentsConfig = {}, +): Promise { + const config = resolveConfig(configOverrides); + + while (!signal.aborted) { + await expireBookingIntentsOnce(dependencies, configOverrides); + if (signal.aborted) break; + await sleep(config.intervalMs, signal); + } +} + +/** + * Managed worker handle with start/stop semantics for use in index.ts or + * tests. `getLastResult()` returns the most recent sweep result. + */ +export function createExpireBookingIntentsWorker( + dependencies: ExpireBookingIntentsDependencies, + configOverrides: ExpireBookingIntentsConfig = {}, +): { start: () => void; stop: () => void; getLastResult: () => ExpireBookingIntentsResult | null } { + const controller = new AbortController(); + let running = false; + let lastResult: ExpireBookingIntentsResult | null = null; + + async function loop(): Promise { + const config = resolveConfig(configOverrides); + while (!controller.signal.aborted) { + lastResult = await expireBookingIntentsOnce(dependencies, configOverrides); + if (controller.signal.aborted) break; + await sleep(config.intervalMs, controller.signal); + } + } + + return { + start() { + if (running) return; + running = true; + loop().finally(() => { + running = false; + }); + }, + stop() { + controller.abort(); + running = false; + }, + getLastResult() { + return lastResult; + }, + }; +}