feat(indexer): add Prometheus /metrics endpoint with indexer and transfer counters

README.md

@@ -343,6 +343,7 @@ curl http://localhost:3000/status
 {
   "ok": true,
   "lastIndexedLedger": 5842100,
+  "last_indexed_ledger": 5842100,
   "latestLedger": 5842102,
   "lagLedgers": 2,
   "startedAt": "2025-10-01T10:00:00.000Z",
@@ -351,6 +352,43 @@ curl http://localhost:3000/status
 }
 ```
 
+`last_indexed_ledger` is a snake_case alias of `lastIndexedLedger` — the same
+name the Prometheus gauge is exported under. Both always carry the same value.
+
+***
+
+### `GET /metrics`
+
+Indexer and process metrics in Prometheus text exposition format, for scraping,
+dashboards, and alerting.
+
+```bash
+curl http://localhost:3000/metrics
+```
+
+```
+# HELP ledgers_indexed_total Ledgers advanced through by the indexer, per network
+# TYPE ledgers_indexed_total counter
+ledgers_indexed_total{network="mainnet"} 48213
+# HELP last_indexed_ledger Highest ledger sequence committed by the indexer, per network
+# TYPE last_indexed_ledger gauge
+last_indexed_ledger{network="mainnet"} 5842100
+```
+
+| Metric | Type | Labels | What it says |
+| ------ | ---- | ------ | ------------ |
+| `ledgers_indexed_total` | counter | `network` | Ledgers the indexer has advanced through. A flat `rate()` on a network whose loop should be running means it has stalled. |
+| `transfers_stored_total` | counter | `network`, `type` | Rows persisted, split `fungible` / `nft` — one parse path can break while the other keeps working. |
+| `rpc_errors_total` | counter | `outcome` | Failed RPC attempts. `retry` counts attempts `withRetry` absorbed, `exhausted` counts calls that gave up — a degrading endpoint shows up in `retry` long before it fails a call. |
+| `last_indexed_ledger` | gauge | `network` | Highest committed ledger. Against the chain tip, this is lag. |
+| `db_query_duration_seconds` | histogram | `operation` | Duration of instrumented DB operations, failures included. |
+
+Standard `process_*` and `nodejs_*` metrics are exported alongside these.
+
+The endpoint reads in-process counters only — no DB, no RPC — so it keeps
+answering while the subsystems it reports on are down, and it is exempt from the
+API rate limit so scrapes do not go dark under load.
+
 ***
 
 ### `GET /transfers/incoming/:address`
@@ -531,6 +569,8 @@ curl -H "Accept: application/vnd.api+json" http://localhost:3000/summary/GABC123
 | `GET /healthz` | `health` |
 | `GET /readyz` | `readiness` |
 
+`GET /metrics` is not a JSON:API resource — it serves Prometheus text format.
+
 ***
 
 ## Event Types Indexed

openapi.json

@@ -257,6 +257,24 @@
         }
       }
     },
+    "/metrics": {
+      "get": {
+        "summary": "Prometheus metrics",
+        "description": "Indexer and process metrics in Prometheus text exposition format. Served from in-process counters only — no database or RPC call — so it keeps answering while the subsystems it reports on are down.",
+        "responses": {
+          "200": {
+            "description": "Prometheus text exposition format",
+            "content": {
+              "text/plain": {
+                "schema": {
+                  "type": "string"
+                }
+              }
+            }
+          }
+        }
+      }
+    },
     "/status": {
       "get": {
         "summary": "Indexer status",

package.json

@@ -75,6 +75,7 @@
     "graphql": "^16.11.0",
     "ioredis": "^5.11.1",
     "parquetjs-lite": "^0.8.7",
+    "prom-client": "^15.1.3",
     "ws": "^8.20.0",
     "zod": "^4.4.3"
   },

src/__tests__/metrics.test.ts

@@ -0,0 +1,142 @@
+import request from "supertest";
+import { createApp } from "../api";
+import {
+  ledgersIndexedTotal,
+  transfersStoredTotal,
+  lastIndexedLedger,
+  observeDbQuery,
+  recordRpcError,
+  registry,
+  _resetMetrics,
+} from "../metrics";
+
+jest.mock("../db", () => ({
+  getLastIndexedLedger: jest.fn().mockResolvedValue(1000),
+  getLastIndexedState: jest.fn(),
+  queryTransfers: jest.fn(),
+  queryAllTransfers: jest.fn(),
+  queryByTxHash: jest.fn(),
+  querySummary: jest.fn(),
+  queryNftTransfers: jest.fn(),
+  getNftOwner: jest.fn(),
+  getNftMetadata: jest.fn(),
+  prisma: { $queryRaw: jest.fn().mockResolvedValue([{ 1: 1 }]) },
+}));
+
+jest.mock("../rpc", () => ({
+  getLatestLedger: jest.fn().mockResolvedValue(1050),
+}));
+
+jest.mock("../indexer", () => ({
+  getAllIndexerStats: jest.fn().mockReturnValue({}),
+  runningNetworks: jest.fn().mockReturnValue([]),
+  getIndexerStats: jest
+    .fn()
+    .mockReturnValue({ startedAt: "2024-01-01T00:00:00.000Z", uptimeSeconds: 100, totalIndexed: 50 }),
+}));
+
+describe("Prometheus metrics (#39)", () => {
+  const app = createApp();
+
+  beforeEach(() => {
+    _resetMetrics();
+  });
+
+  describe("GET /metrics", () => {
+    it("serves Prometheus text exposition format", async () => {
+      const res = await request(app).get("/metrics");
+
+      expect(res.status).toBe(200);
+      expect(res.headers["content-type"]).toContain("text/plain");
+      // Every series in the exposition format is preceded by its HELP and TYPE
+      // lines; a body without them is not something Prometheus will scrape.
+      expect(res.text).toMatch(/^# HELP /m);
+      expect(res.text).toMatch(/^# TYPE /m);
+    });
+
+    it("exports every custom metric, declared with the right type", async () => {
+      const res = await request(app).get("/metrics");
+
+      expect(res.text).toContain("# TYPE ledgers_indexed_total counter");
+      expect(res.text).toContain("# TYPE transfers_stored_total counter");
+      expect(res.text).toContain("# TYPE rpc_errors_total counter");
+      expect(res.text).toContain("# TYPE last_indexed_ledger gauge");
+      expect(res.text).toContain("# TYPE db_query_duration_seconds histogram");
+    });
+
+    it("reports recorded samples with their labels", async () => {
+      ledgersIndexedTotal.inc({ network: "testnet" }, 12);
+      transfersStoredTotal.inc({ network: "testnet", type: "fungible" }, 5);
+      transfersStoredTotal.inc({ network: "testnet", type: "nft" }, 2);
+      lastIndexedLedger.set({ network: "testnet" }, 987_654);
+      recordRpcError("retry");
+
+      const res = await request(app).get("/metrics");
+
+      expect(res.text).toContain('ledgers_indexed_total{network="testnet"} 12');
+      expect(res.text).toContain('transfers_stored_total{network="testnet",type="fungible"} 5');
+      expect(res.text).toContain('transfers_stored_total{network="testnet",type="nft"} 2');
+      expect(res.text).toContain('last_indexed_ledger{network="testnet"} 987654');
+      expect(res.text).toContain('rpc_errors_total{outcome="retry"} 1');
+    });
+
+    it("does not depend on the database or RPC being up", async () => {
+      // The scrape must still answer when the things it reports on are broken —
+      // otherwise monitoring goes dark exactly when it is needed.
+      const { prisma } = jest.requireMock("../db");
+      const { getLatestLedger } = jest.requireMock("../rpc");
+      prisma.$queryRaw.mockRejectedValueOnce(new Error("db down"));
+      getLatestLedger.mockRejectedValueOnce(new Error("rpc down"));
+
+      const res = await request(app).get("/metrics");
+
+      expect(res.status).toBe(200);
+      expect(res.text).toContain("# TYPE ledgers_indexed_total counter");
+    });
+  });
+
+  describe("observeDbQuery", () => {
+    it("times a successful query under its operation label", async () => {
+      const value = await observeDbQuery("someQuery", async () => "result");
+
+      expect(value).toBe("result");
+      const text = await registry.metrics();
+      expect(text).toContain('db_query_duration_seconds_count{operation="someQuery"} 1');
+    });
+
+    it("still times a query that throws, and rethrows it", async () => {
+      // A query that takes eight seconds and then fails is the one worth
+      // seeing; dropping it would make the histogram describe only good runs.
+      await expect(
+        observeDbQuery("failingQuery", async () => {
+          throw new Error("boom");
+        })
+      ).rejects.toThrow("boom");
+
+      const text = await registry.metrics();
+      expect(text).toContain('db_query_duration_seconds_count{operation="failingQuery"} 1');
+    });
+  });
+
+  describe("recordRpcError", () => {
+    it("separates a retried attempt from an exhausted one", async () => {
+      recordRpcError("retry");
+      recordRpcError("retry");
+      recordRpcError("exhausted");
+
+      const res = await request(app).get("/metrics");
+      expect(res.text).toContain('rpc_errors_total{outcome="retry"} 2');
+      expect(res.text).toContain('rpc_errors_total{outcome="exhausted"} 1');
+    });
+  });
+
+  describe("GET /status", () => {
+    it("reports last_indexed_ledger alongside the camelCase field", async () => {
+      const res = await request(app).get("/status");
+
+      expect(res.status).toBe(200);
+      expect(res.body.last_indexed_ledger).toBe(1000);
+      expect(res.body.lastIndexedLedger).toBe(1000);
+    });
+  });
+});

src/api.ts

@@ -24,6 +24,7 @@ import {
   transferQuerySchema,
 } from "./openapi/schemas";
 import { parseOr400 } from "./openapi/validation";
+import { renderMetrics, metricsContentType } from "./metrics";
 
 // ─── RPC Health Check Cache ───────────────────────────────────────────────
 let cachedRpcHealth: { healthy: boolean; timestamp: number } | null = null;
@@ -58,7 +59,9 @@ const limiter = rateLimit({
   standardHeaders: true,
   legacyHeaders: false,
   message: { error: "Too many requests, please try again later." },
-  skip: () => process.env.NODE_ENV === "test",
+  // /metrics is exempt: a scrape endpoint that starts 429ing goes blind exactly
+  // when load is high enough to matter, which is when you need the graphs.
+  skip: (req) => process.env.NODE_ENV === "test" || req.path === "/metrics",
 });
 
 // ─── Response cache (opt-in via CACHE_ENABLED) ─────────────────────────────
@@ -128,7 +131,9 @@ export function createApp(): express.Application {
   // Attaches X-Data-Stale and X-As-Of-Ledger headers plus `stale` / `as_of_ledger`
   // body parameters when serving DB data during an RPC outage.
   app.use(async (req: Request, res: Response, next: NextFunction) => {
-    if (req.method !== "GET" || req.path === "/healthz") {
+    // /metrics is served from in-process counters, so the RPC health probe
+    // below would add a network round-trip to every scrape and report nothing.
+    if (req.method !== "GET" || req.path === "/healthz" || req.path === "/metrics") {
       return next();
     }
 
@@ -224,6 +229,23 @@ export function createApp(): express.Application {
     res.json({ ok: true, uptime: process.uptime() });
   });
 
+  // ─── GET /metrics — Prometheus scrape endpoint ──────────────────────────
+  /**
+   * Indexer and process metrics in Prometheus text exposition format.
+   *
+   * Reads only in-process counters — no DB, no RPC — so it stays answerable
+   * while the things it is reporting on are down, which is the point.
+   */
+  app.get("/metrics", async (_req: Request, res: Response, next: NextFunction) => {
+    try {
+      const body = await renderMetrics();
+      res.setHeader("Content-Type", metricsContentType);
+      res.send(body);
+    } catch (err) {
+      next(err);
+    }
+  });
+
   // ─── GET /readyz — K8s/Render readiness probe ───────────────────────────
   /**
    * Returns 200 when database connection is alive.
@@ -343,6 +365,9 @@ export function createApp(): express.Application {
           ok: true,
           status: "healthy",
           lastIndexedLedger,
+          // snake_case alias alongside the camelCase field, for consumers that
+          // read the same name the metric is exported under. Both always agree.
+          last_indexed_ledger: lastIndexedLedger,
           latestLedger,
           lagLedgers: latestLedger - (lastIndexedLedger ?? latestLedger),
           ...stats,
@@ -359,6 +384,7 @@ export function createApp(): express.Application {
           stale: true,
           as_of_ledger: lastIndexedLedger ?? undefined,
           lastIndexedLedger,
+          last_indexed_ledger: lastIndexedLedger,
           latestLedger: null,
           lagLedgers: null,
           ...stats,