feat(web): observable read-replica and slot replication health

apps/web/src/app/api/cron/db-replication-health/route.test.ts

@@ -0,0 +1,121 @@
+import { captureException } from '@sentry/nextjs';
+
+import { collectReplicationHealth, type ReplicationHealthReport } from '@/lib/replication-health';
+
+jest.mock('@/lib/config.server', () => ({
+  CRON_SECRET: 'cron-secret',
+}));
+
+jest.mock('@/lib/replication-health', () => ({
+  collectReplicationHealth: jest.fn(),
+}));
+
+jest.mock('@sentry/nextjs', () => ({
+  captureException: jest.fn(),
+}));
+
+import { GET } from './route';
+
+const mockCollect = jest.mocked(collectReplicationHealth);
+const mockCaptureException = jest.mocked(captureException);
+
+function report(overrides: Partial<ReplicationHealthReport> = {}): ReplicationHealthReport {
+  return {
+    healthy: true,
+    timestamp: '2026-07-30T09:39:06.000Z',
+    replicas: [],
+    walSenders: [],
+    slots: [],
+    errors: [],
+    ...overrides,
+  };
+}
+
+function createRequest(headers: Record<string, string> = {}) {
+  return new Request('http://localhost:3000/api/cron/db-replication-health', {
+    method: 'GET',
+    headers,
+  });
+}
+
+describe('GET /api/cron/db-replication-health', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+    jest.spyOn(console, 'log').mockImplementation(() => {});
+    jest.spyOn(console, 'error').mockImplementation(() => {});
+    mockCollect.mockResolvedValue(report());
+  });
+
+  afterEach(() => {
+    jest.restoreAllMocks();
+  });
+
+  it('returns 401 with the wrong cron secret', async () => {
+    const response = await GET(createRequest({ authorization: 'Bearer wrong' }));
+
+    expect(response.status).toBe(401);
+    expect(mockCollect).not.toHaveBeenCalled();
+  });
+
+  it('does not alert when replication is healthy', async () => {
+    const response = await GET(createRequest({ authorization: 'Bearer cron-secret' }));
+
+    expect(response.status).toBe(200);
+    await expect(response.json()).resolves.toEqual(
+      expect.objectContaining({ healthy: true, problems: [] })
+    );
+    expect(mockCaptureException).not.toHaveBeenCalled();
+  });
+
+  it('alerts when a replica is lagging', async () => {
+    mockCollect.mockResolvedValue(
+      report({
+        healthy: false,
+        replicas: [
+          {
+            name: 'us-west',
+            status: 'lagging',
+            in_recovery: true,
+            replay_lsn: '1EE6/65035140',
+            last_xact_replay_timestamp: '2026-07-22 10:41:00+00',
+            replay_delay_seconds: 691200,
+            error: null,
+          },
+        ],
+      })
+    );
+
+    const response = await GET(createRequest({ authorization: 'Bearer cron-secret' }));
+
+    const body = await response.json();
+    expect(body.healthy).toBe(false);
+    expect(body.problems).toEqual([expect.stringContaining('replica us-west: lagging')]);
+    expect(mockCaptureException).toHaveBeenCalledTimes(1);
+  });
+
+  it('alerts when a slot is at risk', async () => {
+    mockCollect.mockResolvedValue(
+      report({
+        healthy: false,
+        slots: [
+          {
+            slot_name: 'snowflake_connector_gfqyzuertw',
+            slot_type: 'logical',
+            active: false,
+            wal_status: 'lost',
+            retained_wal_bytes: '0',
+            at_risk: true,
+          },
+        ],
+      })
+    );
+
+    const response = await GET(createRequest({ authorization: 'Bearer cron-secret' }));
+
+    const body = await response.json();
+    expect(body.problems).toEqual([
+      expect.stringContaining('slot snowflake_connector_gfqyzuertw: wal_status=lost'),
+    ]);
+    expect(mockCaptureException).toHaveBeenCalledTimes(1);
+  });
+});

apps/web/src/app/api/cron/db-replication-health/route.ts

@@ -0,0 +1,71 @@
+import { captureException } from '@sentry/nextjs';
+import { NextResponse } from 'next/server';
+
+import { CRON_SECRET } from '@/lib/config.server';
+import { collectReplicationHealth } from '@/lib/replication-health';
+
+/**
+ * Emits replication health to Axiom (via the Vercel log drain) and alerts Sentry
+ * when a replica is lagging/unreachable or a slot is at risk of losing WAL.
+ *
+ * This intentionally uses the replica-side SQL probe from `collectReplicationHealth`
+ * rather than the Supabase Prometheus `physical_replication_lag_*` metric: that
+ * metric has a documented history of returning no data, and — like the primary's
+ * `pg_stat_replication` — cannot see a replica whose walreceiver has died. The
+ * per-replica probe is the authoritative signal.
+ */
+export async function GET(request: Request) {
+  const authHeader = request.headers.get('authorization');
+  if (!CRON_SECRET || authHeader !== `Bearer ${CRON_SECRET}`) {
+    return NextResponse.json({ error: 'Unauthorized' }, { status: 401 });
+  }
+
+  const report = await collectReplicationHealth();
+
+  // Structured JSON → Vercel log drain → Axiom, one line per series.
+  for (const replica of report.replicas) {
+    console.log(
+      JSON.stringify({ type: 'db_replication_health', ...replica, timestamp: report.timestamp })
+    );
+  }
+  for (const slot of report.slots) {
+    console.log(
+      JSON.stringify({ type: 'db_replication_slot', ...slot, timestamp: report.timestamp })
+    );
+  }
+
+  const problems: string[] = [];
+  for (const replica of report.replicas) {
+    if (replica.status !== 'ok') {
+      problems.push(
+        `replica ${replica.name}: ${replica.status}` +
+          (replica.replay_delay_seconds !== null
+            ? ` (${Math.round(replica.replay_delay_seconds)}s behind)`
+            : replica.error
+              ? ` (${replica.error})`
+              : '')
+      );
+    }
+  }
+  for (const slot of report.slots) {
+    if (slot.at_risk) {
+      problems.push(`slot ${slot.slot_name}: wal_status=${slot.wal_status}, active=${slot.active}`);
+    }
+  }
+  for (const error of report.errors) {
+    problems.push(`primary query failed: ${error}`);
+  }
+
+  if (problems.length > 0) {
+    console.error(JSON.stringify({ type: 'db_replication_health_alert', problems }));
+    captureException(new Error(`Replication health degraded: ${problems.join('; ')}`));
+  }
+
+  return NextResponse.json({
+    healthy: report.healthy,
+    problems,
+    replicas: report.replicas.length,
+    slots: report.slots.length,
+    timestamp: report.timestamp,
+  });
+}

apps/web/src/app/api/internal/db/replication-lag/route.test.ts

@@ -1,27 +1,28 @@
 import { NextRequest } from 'next/server';
-import { db } from '@/lib/drizzle';
+
+import { collectReplicationHealth, type ReplicationHealthReport } from '@/lib/replication-health';
 
 jest.mock('@/lib/config.server', () => ({
   INTERNAL_API_SECRET: 'internal-secret',
 }));
 
-jest.mock('@/lib/drizzle', () => ({
-  db: {
-    execute: jest.fn(),
-  },
+jest.mock('@/lib/replication-health', () => ({
+  collectReplicationHealth: jest.fn(),
 }));
 
 import { GET } from './route';
 
-const mockExecute = jest.mocked(db.execute);
+const mockCollect = jest.mocked(collectReplicationHealth);
 
-function queryResult(rows: Record<string, unknown>[]) {
+function report(overrides: Partial<ReplicationHealthReport> = {}): ReplicationHealthReport {
   return {
-    command: 'SELECT',
-    rowCount: rows.length,
-    oid: 0,
-    fields: [],
-    rows,
+    healthy: true,
+    timestamp: '2026-07-30T09:39:06.000Z',
+    replicas: [],
+    walSenders: [],
+    slots: [],
+    errors: [],
+    ...overrides,
   };
 }
 
@@ -35,56 +36,43 @@ function createRequest(headers: Record<string, string> = {}) {
 describe('GET /api/internal/db/replication-lag', () => {
   beforeEach(() => {
     jest.clearAllMocks();
-    mockExecute.mockResolvedValue(queryResult([]));
+    mockCollect.mockResolvedValue(report());
   });
 
   it('returns 401 without the internal secret', async () => {
     const response = await GET(createRequest());
 
     expect(response.status).toBe(401);
-    expect(mockExecute).not.toHaveBeenCalled();
+    expect(mockCollect).not.toHaveBeenCalled();
   });
 
-  it('returns replication lag for every target reported by Postgres', async () => {
-    mockExecute.mockResolvedValue(
-      queryResult([
-        {
-          pid: 123,
-          application_name: 'walreceiver',
-          client_addr: '10.0.0.10',
-          client_hostname: null,
-          client_port: 5432,
-          state: 'streaming',
-          sync_state: 'async',
-          sent_lsn: '0/5000000',
-          write_lsn: '0/4FFFFF0',
-          flush_lsn: '0/4FFFFE0',
-          replay_lsn: '0/4FFFFD0',
-          sent_lag_bytes: '0',
-          write_lag_bytes: '16',
-          flush_lag_bytes: '32',
-          replay_lag_bytes: '48',
-          write_lag_seconds: 0.1,
-          flush_lag_seconds: 0.2,
-          replay_lag_seconds: 0.3,
-        },
-      ])
+  it('returns the replication health report with the secret', async () => {
+    mockCollect.mockResolvedValue(
+      report({
+        healthy: false,
+        replicas: [
+          {
+            name: 'us-west',
+            status: 'unreachable',
+            in_recovery: null,
+            replay_lsn: null,
+            last_xact_replay_timestamp: null,
+            replay_delay_seconds: null,
+            error: 'connection timeout',
+          },
+        ],
+      })
     );
 
     const response = await GET(createRequest({ 'X-Internal-Secret': 'internal-secret' }));
 
     expect(response.status).toBe(200);
-    expect(mockExecute).toHaveBeenCalledTimes(1);
-    await expect(response.json()).resolves.toEqual({
-      targets: [
-        expect.objectContaining({
-          application_name: 'walreceiver',
-          client_addr: '10.0.0.10',
-          replay_lag_bytes: '48',
-          replay_lag_seconds: 0.3,
-        }),
-      ],
-      timestamp: expect.any(String),
-    });
+    expect(mockCollect).toHaveBeenCalledTimes(1);
+    await expect(response.json()).resolves.toEqual(
+      expect.objectContaining({
+        healthy: false,
+        replicas: [expect.objectContaining({ name: 'us-west', status: 'unreachable' })],
+      })
+    );
   });
 });

apps/web/src/app/api/internal/db/replication-lag/route.ts

@@ -1,30 +1,8 @@
 import { timingSafeEqual } from 'crypto';
-import { sql } from 'drizzle-orm';
 import { NextResponse } from 'next/server';
 
 import { INTERNAL_API_SECRET } from '@/lib/config.server';
-import { db } from '@/lib/drizzle';
-
-type ReplicationLagRow = {
-  pid: number;
-  application_name: string;
-  client_addr: string | null;
-  client_hostname: string | null;
-  client_port: number | null;
-  state: string | null;
-  sync_state: string | null;
-  sent_lsn: string | null;
-  write_lsn: string | null;
-  flush_lsn: string | null;
-  replay_lsn: string | null;
-  sent_lag_bytes: string;
-  write_lag_bytes: string;
-  flush_lag_bytes: string;
-  replay_lag_bytes: string;
-  write_lag_seconds: number | null;
-  flush_lag_seconds: number | null;
-  replay_lag_seconds: number | null;
-};
+import { collectReplicationHealth } from '@/lib/replication-health';
 
 function secretMatches(provided: string | null, expected: string): boolean {
   if (!provided) return false;
@@ -42,29 +20,7 @@ export async function GET(request: Request) {
     return NextResponse.json({ error: 'Unauthorized' }, { status: 401 });
   }
 
-  const { rows } = await db.execute<ReplicationLagRow>(sql`
-    SELECT
-      pid,
-      application_name,
-      client_addr::text,
-      client_hostname,
-      client_port,
-      state,
-      sync_state,
-      sent_lsn::text,
-      write_lsn::text,
-      flush_lsn::text,
-      replay_lsn::text,
-      COALESCE(pg_wal_lsn_diff(pg_current_wal_lsn(), sent_lsn), 0)::text AS sent_lag_bytes,
-      COALESCE(pg_wal_lsn_diff(pg_current_wal_lsn(), write_lsn), 0)::text AS write_lag_bytes,
-      COALESCE(pg_wal_lsn_diff(pg_current_wal_lsn(), flush_lsn), 0)::text AS flush_lag_bytes,
-      COALESCE(pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn), 0)::text AS replay_lag_bytes,
-      EXTRACT(EPOCH FROM write_lag)::double precision AS write_lag_seconds,
-      EXTRACT(EPOCH FROM flush_lag)::double precision AS flush_lag_seconds,
-      EXTRACT(EPOCH FROM replay_lag)::double precision AS replay_lag_seconds
-    FROM pg_stat_replication
-    ORDER BY application_name, client_addr, client_port
-  `);
+  const report = await collectReplicationHealth();
 
-  return NextResponse.json({ targets: rows, timestamp: new Date().toISOString() });
+  return NextResponse.json(report);
 }