feat: expire stale pending booking-intents

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}`);

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({

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([]);
+    });
+  });
 });

src/modules/booking-intents/booking-intent-repository.ts

@@ -92,6 +92,13 @@ export interface BookingIntentRepository {
   update(id: string, updates: Partial<Omit<BookingIntentRecord, "id">>): BookingIntentRecord | Promise<BookingIntentRecord>;
   updateTokenInfo?(id: string, tokenAsset: string, mintTxHash: string): Promise<void> | void;
   findExpiredHolds(nowMs: number): BookingIntentRecord[] | Promise<BookingIntentRecord[]>;
+  /**
+   * 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<BookingIntentRecord[]>;
 }
 
 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 }));
+  }
 }

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<BookingIntentRecord[]> {
+    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.