feat(observability): full Prometheus metrics endpoint, Grafana dashboard, and indexer-lag alert (Closes #717)

backend/.env.example

@@ -21,6 +21,11 @@ HORIZON_URL=https://horizon-testnet.stellar.org
 # Generate with: openssl rand -hex 32
 ADMIN_API_KEY=your-admin-api-key-here-minimum-32-characters
 
+# Prometheus /metrics endpoint basic auth (format: "user:password")
+# Required in production — Prometheus must send the same credentials.
+# Example: METRICS_AUTH=metrics:super-secret
+METRICS_AUTH=metrics:change-me-in-production
+
 # Webhook configuration
 WEBHOOK_DESTINATION_URL=https://your-app.com/webhooks/stellar-stream
 WEBHOOK_SIGNING_SECRET=your-webhook-signing-secret-here

backend/src/cors.test.ts

@@ -14,6 +14,16 @@ vi.mock("./services/metrics", () => ({
   lastIndexedLedger: { set: vi.fn() },
   indexerErrorsTotal: { inc: vi.fn() },
   indexerCircuitState: { set: vi.fn() },
+  recordIndexerSuccess: vi.fn(),
+  httpRequestsTotal: { inc: vi.fn() },
+  httpRequestDurationMs: { observe: vi.fn() },
+  streamCountByStatus: { set: vi.fn() },
+  claimCount: { set: vi.fn() },
+  cancelCount: { set: vi.fn() },
+  indexerLagSeconds: { set: vi.fn() },
+  refreshPrometheusStreamMetrics: vi.fn(),
+  resetPrometheusStreamMetricsCache: vi.fn(),
+  resetIndexerLag: vi.fn(),
 }));
 
 vi.mock("@stellar/stellar-sdk", async (importOriginal) => {

backend/src/index.test.ts

@@ -11,13 +11,27 @@ const RECIPIENT_1 = Keypair.random().publicKey();
 const RECIPIENT_2 = Keypair.random().publicKey();
 
 const streamStoreMocks = vi.hoisted(() => ({
+  archiveOldStreams: vi.fn(),
   calculateProgress: vi.fn(),
   cancelStream: vi.fn(),
   createStream: vi.fn(),
+  deleteStreamById: vi.fn(),
+  estimateCreateStreamFee: vi.fn(),
+  getLatestLedgerTime: vi.fn(),
+  getOnChainClaimableAmount: vi.fn(),
+  getOnChainClaimableBatch: vi.fn(),
+  getOnChainStreamCount: vi.fn(),
   getStream: vi.fn(),
   initSoroban: vi.fn(),
   listStreams: vi.fn(),
+  listStreamsByRecipient: vi.fn(),
   listStreamsBySender: vi.fn(),
+  markStreamComplete: vi.fn(),
+  nowInSeconds: vi.fn(),
+  pauseStream: vi.fn(),
+  reconcileStream: vi.fn(),
+  refreshStreamStatuses: vi.fn(),
+  resumeStream: vi.fn(),
   syncStreams: vi.fn(),
   updateStreamStartAt: vi.fn(),
 }));
@@ -33,9 +47,15 @@ const eventHistoryMocks = vi.hoisted(() => ({
 }));
 
 vi.mock("./services/streamStore", () => streamStoreMocks);
+vi.mock("./services/db", () => ({
+  getAllowedAssets: vi.fn(() => ["USDC", "XLM"]),
+  searchStreamsFts: vi.fn(() => []),
+}));
 vi.mock("./services/eventHistory", () => eventHistoryMocks);
 vi.mock("./services/auth", () => ({
   authMiddleware: vi.fn((req: any, res: any, next: any) => next()),
+  adminJwtAuth: vi.fn((req: any, res: any, next: any) => next()),
+  getJwtSecret: vi.fn(() => "test_secret_for_integration"),
   generateChallenge: vi.fn(),
   refreshToken: vi.fn(),
   verifyChallengeAndIssueToken: vi.fn(),
@@ -163,7 +183,12 @@ function invokeListStreamsRoute(
     throw new Error("GET /api/streams route not found");
   }
 
-  const handler = layer.route.stack[0].handle as (req: any, res: any) => void;
+  // The route handler is the last entry in the route stack — middlewares such
+  // as readLimiter are registered before it.
+  const handler = layer.route.stack[layer.route.stack.length - 1].handle as (
+    req: any,
+    res: any,
+  ) => void;
 
   let statusCode = 200;
   let jsonBody: any;
@@ -178,6 +203,9 @@ function invokeListStreamsRoute(
       jsonBody = payload;
       return this;
     },
+    set() {
+      return this;
+    },
   };
 
   handler(req, res);
@@ -197,7 +225,12 @@ function invokeSenderStreamsRoute(
     throw new Error("GET /api/senders/:accountId/streams route not found");
   }
 
-  const handler = layer.route.stack[0].handle as (req: any, res: any) => void;
+  // The route handler is the last entry in the route stack — middlewares such
+  // as readLimiter are registered before it.
+  const handler = layer.route.stack[layer.route.stack.length - 1].handle as (
+    req: any,
+    res: any,
+  ) => void;
 
   let statusCode = 200;
   let jsonBody: any;
@@ -212,6 +245,9 @@ function invokeSenderStreamsRoute(
       jsonBody = payload;
       return this;
     },
+    set() {
+      return this;
+    },
   };
 
   handler(req, res);
@@ -645,7 +681,12 @@ function invokeGlobalEventsRoute(
     throw new Error("GET /api/events route not found");
   }
 
-  const handler = layer.route.stack[0].handle as (req: any, res: any) => void;
+  // The route handler is the last entry in the route stack — middlewares such
+  // as readLimiter are registered before it.
+  const handler = layer.route.stack[layer.route.stack.length - 1].handle as (
+    req: any,
+    res: any,
+  ) => void;
 
   let statusCode = 200;
   let jsonBody: any;
@@ -654,6 +695,7 @@ function invokeGlobalEventsRoute(
   const res = {
     status(code: number) { statusCode = code; return this; },
     json(payload: any) { jsonBody = payload; return this; },
+    set() { return this; },
   };
 
   handler(req, res);
@@ -672,7 +714,7 @@ describe("GET /api/events", () => {
     expect(status).toBe(200);
     expect(body.total).toBe(4);
     expect(body.page).toBe(1);
-    expect(body.limit).toBe(4);
+    expect(body.limit).toBe(20);
     expect(body.data).toHaveLength(4);
     expect(body.data[0].streamId).toBe("stream-1");
   });
@@ -686,7 +728,11 @@ describe("GET /api/events", () => {
 
     expect(status).toBe(200);
     expect(body.total).toBe(2);
-    expect(eventHistoryMocks.countAllEvents).toHaveBeenCalledWith("created");
+    expect(eventHistoryMocks.countAllEvents).toHaveBeenCalledWith(
+      "created",
+      undefined,
+      undefined,
+    );
 
   });
 

backend/src/index.ts

@@ -95,7 +95,7 @@ import {
 } from "./validation/schemas";
 import { validateEnv } from "./config/validateEnv";
 import { getMetricsHistory } from "./services/metricsHistory";
-import { register } from "./services/metrics";
+import { register, refreshPrometheusStreamMetrics } from "./services/metrics";
 import { initCache } from "./services/cache";
 import { getGlobalStats } from "./services/stats";
 import { logger } from "./logger";
@@ -411,6 +411,15 @@ app.get("/metrics", async (_req: Request, res: Response) => {
     }
   }
 
+  // Refresh DB-backed gauges (stream_count, claim_count, cancel_count) before
+  // serialising. If the database is unavailable, still serve the in-memory
+  // metrics so a scrape never fails outright.
+  try {
+    refreshPrometheusStreamMetrics();
+  } catch (err) {
+    logger.warn({ err }, "failed to refresh DB-backed Prometheus metrics");
+  }
+
   const output = await register.metrics();
   res.setHeader("Content-Type", "text/plain; version=0.0.4");
   res.send(output);

backend/src/metrics.route.test.ts

@@ -0,0 +1,58 @@
+import { describe, it, expect, vi } from "vitest";
+import request from "supertest";
+
+// METRICS_AUTH is read from the environment at module load time, so it must
+// be set before the app module is imported.
+vi.hoisted(() => {
+  process.env.METRICS_AUTH = "metrics:test-secret";
+});
+
+import { app } from "./index";
+
+const AUTH_HEADER = "Basic " + Buffer.from("metrics:test-secret").toString("base64");
+const WRONG_AUTH_HEADER = "Basic " + Buffer.from("metrics:wrong-password").toString("base64");
+
+describe("GET /metrics", () => {
+  it("returns 401 for unauthenticated scrapes when METRICS_AUTH is set", async () => {
+    const res = await request(app).get("/metrics");
+    expect(res.status).toBe(401);
+    expect(res.headers["www-authenticate"]).toContain("Basic");
+  });
+
+  it("returns 401 for scrapes with wrong credentials", async () => {
+    const res = await request(app).get("/metrics").set("Authorization", WRONG_AUTH_HEADER);
+    expect(res.status).toBe(401);
+  });
+
+  it("serves Prometheus text format with all required metric families", async () => {
+    const res = await request(app).get("/metrics").set("Authorization", AUTH_HEADER);
+    expect(res.status).toBe(200);
+    expect(res.headers["content-type"]).toContain("text/plain");
+    expect(res.headers["content-type"]).toContain("version=0.0.4");
+
+    for (const name of [
+      "request_count",
+      "request_duration_ms",
+      "stream_count",
+      "claim_count",
+      "cancel_count",
+      "indexer_lag_seconds",
+      "events_indexed_total",
+    ]) {
+      expect(res.text).toContain(name);
+    }
+  });
+
+  it("records request metrics observed by the request logger", async () => {
+    await request(app).get("/api/health");
+    await request(app).get("/api/health");
+
+    const res = await request(app).get("/metrics").set("Authorization", AUTH_HEADER);
+    expect(res.status).toBe(200);
+    expect(res.text).toContain(
+      'request_count{method="GET",route="/api/health",status_code="200"} 2',
+    );
+    expect(res.text).toContain("request_duration_ms_count");
+    expect(res.text).toContain("request_duration_ms_bucket");
+  });
+});

backend/src/middleware/requestLogger.ts

@@ -2,6 +2,7 @@ import { Request, Response, NextFunction } from "express";
 import crypto from "crypto";
 import { logger } from "../logger";
 import { runWithCorrelation } from "../correlationContext";
+import { httpRequestsTotal, httpRequestDurationMs } from "../services/metrics";
 
 declare global {
   namespace Express {
@@ -56,6 +57,13 @@ export function requestLogger(req: Request, res: Response, next: NextFunction) {
     res.on("finish", () => {
       const durationMs = Date.now() - start;
 
+      // Record Prometheus request metrics. Prefer the matched route pattern
+      // (e.g. /api/streams/:id) to keep label cardinality bounded; fall back
+      // to the request path for unmatched routes such as 404s.
+      const route = req.route?.path ?? req.originalUrl.split("?")[0];
+      httpRequestsTotal.inc({ method: req.method, route, status_code: String(res.statusCode) });
+      httpRequestDurationMs.observe({ method: req.method, route }, durationMs);
+
       const logEntry = {
         correlation_id: correlationId,
         method: req.method,

backend/src/services/indexer.circuitbreaker.test.ts

@@ -11,6 +11,16 @@ vi.mock("./metrics", () => ({
   lastIndexedLedger: { set: vi.fn() },
   indexerErrorsTotal: { inc: vi.fn() },
   indexerCircuitState: { set: vi.fn() },
+  recordIndexerSuccess: vi.fn(),
+  httpRequestsTotal: { inc: vi.fn() },
+  httpRequestDurationMs: { observe: vi.fn() },
+  streamCountByStatus: { set: vi.fn() },
+  claimCount: { set: vi.fn() },
+  cancelCount: { set: vi.fn() },
+  indexerLagSeconds: { set: vi.fn() },
+  refreshPrometheusStreamMetrics: vi.fn(),
+  resetPrometheusStreamMetricsCache: vi.fn(),
+  resetIndexerLag: vi.fn(),
 }));
 
 import { CircuitBreaker } from "./indexer";

backend/src/services/indexer.gap.test.ts

@@ -31,6 +31,16 @@ vi.mock("./metrics", () => ({
   lastIndexedLedger: { set: vi.fn() },
   indexerErrorsTotal: { inc: vi.fn() },
   indexerCircuitState: { set: vi.fn() },
+  recordIndexerSuccess: vi.fn(),
+  httpRequestsTotal: { inc: vi.fn() },
+  httpRequestDurationMs: { observe: vi.fn() },
+  streamCountByStatus: { set: vi.fn() },
+  claimCount: { set: vi.fn() },
+  cancelCount: { set: vi.fn() },
+  indexerLagSeconds: { set: vi.fn() },
+  refreshPrometheusStreamMetrics: vi.fn(),
+  resetPrometheusStreamMetricsCache: vi.fn(),
+  resetIndexerLag: vi.fn(),
 }));
 
 // ── In-memory DB (replaced per-test via setupDb) ──────────────────────────────

backend/src/services/indexer.test.ts

@@ -10,13 +10,24 @@ const mockLedgersScannedTotal = vi.hoisted(() => ({ inc: vi.fn() }));
 const mockLastIndexedLedger = vi.hoisted(() => ({ set: vi.fn() }));
 const mockIndexerErrorsTotal = vi.hoisted(() => ({ inc: vi.fn() }));
 const mockIndexerCircuitState = vi.hoisted(() => ({ set: vi.fn() }));
+const mockRecordIndexerSuccess = vi.hoisted(() => vi.fn());
 
 vi.mock("./metrics", () => ({
   eventsIndexedTotal: mockEventsIndexedTotal,
   ledgersScannedTotal: mockLedgersScannedTotal,
   lastIndexedLedger: mockLastIndexedLedger,
   indexerErrorsTotal: mockIndexerErrorsTotal,
   indexerCircuitState: mockIndexerCircuitState,
+  recordIndexerSuccess: mockRecordIndexerSuccess,
+  httpRequestsTotal: { inc: vi.fn() },
+  httpRequestDurationMs: { observe: vi.fn() },
+  streamCountByStatus: { set: vi.fn() },
+  claimCount: { set: vi.fn() },
+  cancelCount: { set: vi.fn() },
+  indexerLagSeconds: { set: vi.fn() },
+  refreshPrometheusStreamMetrics: vi.fn(),
+  resetPrometheusStreamMetricsCache: vi.fn(),
+  resetIndexerLag: vi.fn(),
 }));
 
 let db: InstanceType<typeof Database>;

backend/src/services/indexer.ts

@@ -13,6 +13,7 @@ import {
   lastIndexedLedger,
   indexerErrorsTotal,
   indexerCircuitState,
+  recordIndexerSuccess,
 } from "./metrics";
 import { logger } from "../logger";
 
@@ -269,6 +270,7 @@ async function indexEvents(): Promise<void> {
 
     if (currentLedger <= lastProcessedLedger) {
       circuitBreaker.onSuccess();
+      recordIndexerSuccess();
       return;
     }
 
@@ -279,6 +281,7 @@ async function indexEvents(): Promise<void> {
     }
 
     circuitBreaker.onSuccess();
+    recordIndexerSuccess();
   } catch (err) {
     circuitBreaker.onFailure();
     indexerErrorsTotal.inc();