feat: add synthetic journey monitor for critical Utix workflow

.github/workflows/synthetic-monitor.yml

@@ -19,6 +19,11 @@ jobs:
       fail-fast: false
       matrix:
         network: [testnet, mainnet]
+        include:
+          - network: testnet
+            horizon: https://horizon-testnet.stellar.org
+          - network: mainnet
+            horizon: https://horizon.stellar.org
 
     steps:
       - name: Checkout
@@ -33,4 +38,5 @@ jobs:
       - name: Run synthetic monitor
         env:
           UTIX_USER_ROLE: service
+          HORIZON_URL: ${{ matrix.horizon }}
         run: npm run synthetic:monitor -- --network ${{ matrix.network }}

core/health/monitor.ts

@@ -7,6 +7,45 @@
 
 import { HealthIndicator, HealthStatus, HealthSummary, UnresolvedException, StaleRecord } from './types';
 
+export interface JourneyStepResult {
+  name: string;
+  status: 'passed' | 'failed' | 'skipped';
+  latencyMs: number;
+  message?: string;
+  dependency?: string;
+}
+
+export interface JourneyMonitorResult {
+  journey: string;
+  status: 'passed' | 'failed';
+  startedAt: Date;
+  finishedAt: Date;
+  totalLatencyMs: number;
+  steps: JourneyStepResult[];
+  failures: JourneyStepResult[];
+  dependencyBlame: string[];
+}
+
+export interface JourneyMonitorOptions {
+  horizonUrl?: string;
+  timeoutMs?: number;
+  fetchImpl?: typeof fetch;
+}
+
+const DEFAULT_JOURNEY_TIMEOUT_MS = 10_000;
+
+async function withTimeout<T>(promise: Promise<T>, timeoutMs: number, label: string): Promise<T> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  const timeout = new Promise<never>((_, reject) => {
+    timer = setTimeout(() => reject(new Error(`${label} timed out after ${timeoutMs}ms`)), timeoutMs);
+  });
+  try {
+    return await Promise.race([promise, timeout]);
+  } finally {
+    if (timer) clearTimeout(timer);
+  }
+}
+
 class HealthMonitor {
   private indicators: HealthIndicator[] = [];
   private exceptions: UnresolvedException[] = [];
@@ -183,10 +222,126 @@ class HealthMonitor {
   getStaleRecords(maxAgeMs: number): StaleRecord[] {
     return this.staleRecords.filter(r => r.age > maxAgeMs);
   }
+
+  /**
+   * Runs the synthetic end-to-end journey monitor for the most valuable
+   * contributor workflow: registry load -> network reachability -> ledger
+   * fetch. Reports per-step latency, failures, and dependency blame.
+   */
+  async runJourneyMonitor(options: JourneyMonitorOptions = {}): Promise<JourneyMonitorResult> {
+    const startedAt = new Date();
+    const steps: JourneyStepResult[] = [];
+    const timeoutMs = options.timeoutMs ?? DEFAULT_JOURNEY_TIMEOUT_MS;
+    const fetchImpl = options.fetchImpl ?? (typeof fetch !== 'undefined' ? fetch : undefined);
+
+    const runStep = async (
+      name: string,
+      dependency: string,
+      fn: () => Promise<string | void>,
+    ): Promise<void> => {
+      const stepStart = Date.now();
+      try {
+        const message = await withTimeout(Promise.resolve().then(fn), timeoutMs, name);
+        steps.push({
+          name,
+          status: 'passed',
+          latencyMs: Date.now() - stepStart,
+          message: typeof message === 'string' ? message : undefined,
+          dependency,
+        });
+      } catch (error) {
+        const message = error instanceof Error ? error.message : String(error);
+        steps.push({
+          name,
+          status: 'failed',
+          latencyMs: Date.now() - stepStart,
+          message,
+          dependency,
+        });
+        this.recordException({
+          type: 'JourneyMonitorFailure',
+          message: `Journey step "${name}" failed: ${message}`,
+          error,
+          context: { dependency },
+        });
+      }
+    };
+
+    await runStep('Load feature registry', 'registry', async () => {
+      const { registry } = await import('../registry/registry');
+      const count = Object.keys(registry).length;
+      if (count === 0) throw new Error('Feature registry is empty');
+      return `${count} features loaded`;
+    });
+
+    const horizonUrl = options.horizonUrl ?? process.env.NEXT_PUBLIC_HORIZON_TESTNET_URL;
+
+    await runStep('Resolve Horizon endpoint', 'configuration', async () => {
+      if (!horizonUrl) throw new Error('Horizon endpoint not configured');
+      new URL(horizonUrl);
+      return horizonUrl;
+    });
+
+    await runStep('Fetch latest ledger', 'horizon', async () => {
+      if (!horizonUrl) throw new Error('Horizon endpoint not configured');
+      if (!fetchImpl) throw new Error('fetch is not available in this runtime');
+      const url = `${horizonUrl.replace(/\/$/, '')}/ledgers?order=desc&limit=1`;
+      const response = await fetchImpl(url, { headers: { Accept: 'application/json' } });
+      if (!response.ok) {
+        throw new Error(`Horizon responded with HTTP ${response.status}`);
+      }
+      const body = (await response.json()) as { _embedded?: { records?: unknown[] } };
+      const records = body?._embedded?.records;
+      if (!Array.isArray(records) || records.length === 0) {
+        throw new Error('Horizon returned no ledger records');
+      }
+      return 'Latest ledger retrieved';
+    });
+
+    const finishedAt = new Date();
+    const failures = steps.filter(s => s.status === 'failed');
+    const dependencyBlame = Array.from(
+      new Set(failures.map(f => f.dependency).filter((d): d is string => Boolean(d))),
+    );
+
+    return {
+      journey: 'contributor-happy-path',
+      status: failures.length === 0 ? 'passed' : 'failed',
+      startedAt,
+      finishedAt,
+      totalLatencyMs: finishedAt.getTime() - startedAt.getTime(),
+      steps,
+      failures,
+      dependencyBlame,
+    };
+  }
 }
 
 export const healthMonitor = new HealthMonitor();
 
+export function formatJourneyReport(result: JourneyMonitorResult): string {
+  const lines: string[] = [];
+  lines.push('='.repeat(70));
+  lines.push(`SYNTHETIC JOURNEY MONITOR: ${result.journey}`);
+  lines.push('='.repeat(70));
+  lines.push(`Status: ${result.status.toUpperCase()}`);
+  lines.push(`Started: ${result.startedAt.toISOString()}`);
+  lines.push(`Finished: ${result.finishedAt.toISOString()}`);
+  lines.push(`Total Latency: ${result.totalLatencyMs}ms`);
+  lines.push('');
+  lines.push('STEPS:');
+  for (const step of result.steps) {
+    const symbol = step.status === 'passed' ? '✓' : step.status === 'failed' ? '✗' : '-';
+    lines.push(`  ${symbol} ${step.name} (${step.latencyMs}ms)${step.message ? `: ${step.message}` : ''}`);
+  }
+  if (result.dependencyBlame.length > 0) {
+    lines.push('');
+    lines.push(`DEPENDENCY BLAME: ${result.dependencyBlame.join(', ')}`);
+  }
+  lines.push('='.repeat(70));
+  return lines.join('\n');
+}
+
 export function formatHealthReport(summary: HealthSummary): string {
   const lines: string[] = [];
 

core/health/synthetic.ts

@@ -1,3 +1,4 @@
+
 /**
  * Synthetic monitoring for the critical end-to-end journey (#269)
  *
@@ -91,6 +92,8 @@ export interface SyntheticMonitorOptions {
   readonly latencyBudgetMs?: number;
   /** Ingestion lag above this is `degraded`. */
   readonly maxIngestLagLedgers?: number;
+  /** Injectable clock for deterministic tests of `startedAt`/`durationMs`. */
+  readonly clock?: () => number;
 }
 
 /** Mirrors the operation browser's page size, so page two is exercised exactly as users reach it. */
@@ -151,6 +154,16 @@ function isDescendingPage(page: Record<string, unknown>[], cursor?: string): boo
   return true;
 }
 
+/**
+ * A step that never ran (because an earlier step failed) is recorded as a
+ * skipped failure with `errorCode: "skipped"` so the report shows the whole
+ * journey, not just the prefix that executed. `attempts: 0` distinguishes it
+ * from a real failed attempt.
+ */
+function skippedStep(name: SyntheticStepName): SyntheticStepResult {
+  return { step: name, ok: false, attempts: 0, latencyMs: 0, errorCode: "skipped" };
+}
+
 export async function runSyntheticMonitor(
   auth: AuthContext,
   options: SyntheticMonitorOptions
@@ -181,6 +194,7 @@ export async function runSyntheticMonitor(
   const fetchImpl = options.fetch ?? fetch;
   const sleep = options.sleep ?? ((ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms)));
   const now = options.now ?? (() => Date.now());
+  const clock = options.clock ?? (() => Date.now());
   const base = `${(options.horizonUrl ?? HORIZON_URLS[options.network]).replace(/\/+$/, "")}/`;
   const steps: SyntheticStepResult[] = [];
   const startedAt = now();
@@ -262,6 +276,7 @@ export async function runSyntheticMonitor(
       const source = page[0].source_account;
       return typeof source === "string" && PUBLIC_KEY.test(source) ? ok(source) : err("malformed_payload");
     }));
+  if (!root) steps.push(skippedStep("operations.discover"));
 
   const operationsPath = accountId ? `accounts/${encodeURIComponent(accountId)}/operations` : "";
   const query = { order: "desc", limit: String(SYNTHETIC_PAGE_SIZE) };
@@ -274,6 +289,7 @@ export async function runSyntheticMonitor(
       if (page.length === 0 || !isDescendingPage(page)) return err("pagination_broken");
       return ok(page);
     }));
+  if (!accountId) steps.push(skippedStep("account.operations"));
 
   if (firstPage && firstPage.length === SYNTHETIC_PAGE_SIZE) {
     const cursor = firstPage[firstPage.length - 1].paging_token as string;
@@ -283,6 +299,7 @@ export async function runSyntheticMonitor(
       return isDescendingPage(page, cursor) ? ok(page) : err("pagination_broken");
     });
   }
+  if (firstPage && firstPage.length < SYNTHETIC_PAGE_SIZE) steps.push(skippedStep("account.operations.older"));
 
   const durationMs = now() - startedAt;
   const failed = steps.find((result) => !result.ok);
@@ -309,7 +326,7 @@ export async function runSyntheticMonitor(
     network: options.network,
     status,
     correlationId,
-    startedAt: new Date(startedAt).toISOString(),
+    startedAt: new Date(clock()).toISOString(),
     durationMs,
     steps,
     ...(ingestLagLedgers === undefined ? {} : { ingestLagLedgers })

core/health/types.ts

@@ -30,6 +30,43 @@ export interface StaleRecord {
   details: Record<string, unknown>;
 }
 
+export interface JourneyStepResult {
+  name: string;
+  status: HealthStatus;
+  latencyMs: number;
+  error?: string;
+  dependency?: string;
+}
+
+export interface JourneyRun {
+  id: string;
+  journey: string;
+  startedAt: Date;
+  finishedAt: Date;
+  status: HealthStatus;
+  totalLatencyMs: number;
+  steps: JourneyStepResult[];
+  failedStep?: string;
+  blamedDependency?: string;
+}
+
+export interface JourneyMonitorSummary {
+  timestamp: Date;
+  journey: string;
+  status: HealthStatus;
+  lastRun?: JourneyRun;
+  recentRuns: JourneyRun[];
+  metrics: {
+    runCount: number;
+    successCount: number;
+    failureCount: number;
+    successRate: number;
+    p50LatencyMs: number;
+    p95LatencyMs: number;
+    maxLatencyMs: number;
+  };
+}
+
 export interface HealthSummary {
   timestamp: Date;
   overallStatus: HealthStatus;