Instrument pending external-event queue health

packages/daemon/src/lib/external-events/external-event-store.ts

@@ -31,6 +31,7 @@ import {
   TERMINAL_DELIVERY_STATES,
   TERMINAL_EVENT_STATES,
 } from './types';
+import type { DeliveryTerminalEvent, QueueAgeStats } from './queue-health-metrics';
 import { validateLiteralTopic, validateSource } from './topic-validator';
 
 interface ExternalEventRow {
@@ -90,6 +91,20 @@ export class ExternalEventValidationError extends Error {
 export class ExternalEventStore {
   constructor(private readonly db: BunDatabase) {}
 
+  /**
+   * Optional hook fired when a delivery row transitions to a terminal state
+   * (`delivered`, or `failed` via `failure.terminal=true`). Set by the space
+   * runtime so queue-health metrics can count every delivery outcome from a
+   * single observation point, regardless of which call path reached the
+   * transition. Only fired on an actual transition (`changes > 0`).
+   */
+  private deliveryTerminalHook?: (event: DeliveryTerminalEvent) => void;
+
+  /** Install the delivery-terminal observation hook. */
+  setDeliveryTerminalHook(hook: (event: DeliveryTerminalEvent) => void): void {
+    this.deliveryTerminalHook = hook;
+  }
+
   // ---------------------------------------------------------------------------
   // Source event lifecycle
   // ---------------------------------------------------------------------------
@@ -430,14 +445,17 @@ export class ExternalEventStore {
   /** Mark the delivery row terminal `delivered`. No-op if already terminal. */
   markDeliveryDelivered(eventId: string, deliveryKey: string): void {
     const now = Date.now();
-    this.db
+    const result = this.db
       .prepare(
         `UPDATE space_external_event_deliveries
 				 SET state = 'delivered', failure_reason = NULL, delivered_at = ?, updated_at = ?
 				 WHERE event_id = ? AND delivery_key = ?
 				 AND state NOT IN ('delivered', 'failed')`
       )
       .run(now, now, eventId, deliveryKey);
+    if (result.changes > 0 && this.deliveryTerminalHook) {
+      this.deliveryTerminalHook({ eventId, deliveryKey, outcome: 'delivered', reason: null });
+    }
   }
 
   /**
@@ -452,14 +470,22 @@ export class ExternalEventStore {
   markDeliveryFailed(eventId: string, deliveryKey: string, failure: DeliveryFailure): void {
     const now = Date.now();
     const newState: ExternalEventDeliveryState = failure.terminal ? 'failed' : 'pending';
-    this.db
+    const result = this.db
       .prepare(
         `UPDATE space_external_event_deliveries
 				 SET state = ?, failure_reason = ?, updated_at = ?
 				 WHERE event_id = ? AND delivery_key = ?
 				 AND state NOT IN ('delivered', 'failed')`
       )
       .run(newState, failure.reason, now, eventId, deliveryKey);
+    if (failure.terminal && result.changes > 0 && this.deliveryTerminalHook) {
+      this.deliveryTerminalHook({
+        eventId,
+        deliveryKey,
+        outcome: 'failed',
+        reason: failure.reason,
+      });
+    }
   }
 
   /** List delivery rows for an event (for diagnostics and tests). */
@@ -555,6 +581,56 @@ export class ExternalEventStore {
     return rows.map(deliveryRowToRecord);
   }
 
+  /**
+   * Summarize DB-persisted `pending` deliveries for the queue-health snapshot
+   * without materializing every row: count + min/max/avg via SQL aggregates, and
+   * p95 via a single `LIMIT 1 OFFSET k` lookup over a sorted scan (no per-row JS
+   * allocation). The anchor is the source event's ingestion time
+   * (`space_external_events.created_at`), matching the runtime's event-age TTL
+   * semantics — see `EXTERNAL_EVENT_QUEUE_TTL_MS`. Returns `null` when there are
+   * no pending deliveries.
+   */
+  summarizePendingDeliveries(now: number = Date.now()): QueueAgeStats | null {
+    const agg = this.db
+      .prepare(
+        `SELECT
+           COUNT(*) AS count,
+           MIN(? - e.created_at) AS minMs,
+           MAX(? - e.created_at) AS maxMs,
+           AVG(? - e.created_at) AS avgMs
+         FROM space_external_event_deliveries d
+         INNER JOIN space_external_events e ON e.id = d.event_id
+         WHERE d.state = 'pending'`
+      )
+      .get(now, now, now) as {
+      count: number;
+      minMs: number | null;
+      maxMs: number | null;
+      avgMs: number | null;
+    };
+    const count = agg?.count ?? 0;
+    if (count === 0) return null;
+    // Nearest-rank p95 (no interpolation): the ceil(0.95 * n)th value, clamped.
+    const p95Offset = Math.min(count - 1, Math.ceil(count * 0.95) - 1);
+    const p95 = this.db
+      .prepare(
+        `SELECT (? - e.created_at) AS age
+           FROM space_external_event_deliveries d
+           INNER JOIN space_external_events e ON e.id = d.event_id
+           WHERE d.state = 'pending'
+           ORDER BY age
+           LIMIT 1 OFFSET ?`
+      )
+      .get(now, p95Offset) as { age: number } | undefined;
+    return {
+      count,
+      minMs: agg.minMs ?? 0,
+      maxMs: agg.maxMs ?? 0,
+      avgMs: Math.round(agg.avgMs ?? 0),
+      p95Ms: p95?.age ?? agg.maxMs ?? 0,
+    };
+  }
+
   getDelivery(eventId: string, deliveryKey: string): ExternalEventDeliveryRecord | null {
     const row = this.db
       .prepare(