feat: add Prometheus metrics endpoint

README.md

@@ -16,6 +16,12 @@ Every HTTP API route is mounted under `/v1`, including health and admin routes.
 For example: `GET /v1/health`, `GET /v1/ngos`, `GET /v1/streams`, and
 `GET /v1/indexer/status`.
 
+## Prometheus metrics
+
+`GET /metrics` returns Prometheus text format (`text/plain; version=0.0.4`).
+It exposes HTTP request totals and duration, indexer checkpoint and lag, and webhook
+delivery totals. Configure Prometheus to scrape it on the API host and port.
+
 ## NGO Verification Model
 
 An NGO's verified status consists of two distinct steps kept deliberately separate:

src/indexer/worker.ts

@@ -1,3 +1,4 @@
+import { recordIndexerPosition } from '../metrics.js';
 import { getLatestLedgerSequence, rpcServer } from '../stellar/rpc.js';
 import { getCheckpointState, saveCheckpoint } from './checkpoint.js';
 import { WATCHED_CONTRACT_IDS } from './contracts.js';
@@ -94,6 +95,7 @@ export async function pollOnce(handleEvent: EventHandler): Promise<void> {
       // otherwise start from "now" to avoid replaying all history.
       lastProcessedLedger = START_LEDGER ?? (await getLatestLedgerSequence());
       await saveCheckpoint(lastProcessedLedger);
+      recordIndexerPosition(lastProcessedLedger, lastProcessedLedger);
       return;
     }
   }
@@ -108,6 +110,7 @@ export async function pollOnce(handleEvent: EventHandler): Promise<void> {
   //     Skip ahead to the current ledger; the alternative is an indexer that
   //     never recovers. Events in the gap are lost, so say so loudly.
   const latestLedger = await getLatestLedgerSequence();
+  recordIndexerPosition(lastProcessedLedger, latestLedger);
 
   if (lastProcessedLedger >= latestLedger && lastProcessedEventId === undefined) {
     return;

src/metrics.ts

@@ -0,0 +1,80 @@
+type HttpKey = `${string}\u0000${string}\u0000${number}`;
+
+const httpRequests = new Map<HttpKey, number>();
+const httpDuration = new Map<HttpKey, { count: number; sum: number }>();
+const notificationDeliveries = new Map<'success' | 'failure', number>();
+
+let indexerCheckpointLedger = 0;
+let indexerLatestLedger = 0;
+
+function labels(method: string, route: string, statusCode: number): string {
+  return `method="${method}",route="${route.replaceAll('"', '\\"')}",status_code="${statusCode}"`;
+}
+
+export function recordHttpRequest(
+  method: string,
+  route: string,
+  statusCode: number,
+  durationSeconds: number,
+): void {
+  const key: HttpKey = `${method}\u0000${route}\u0000${statusCode}`;
+  httpRequests.set(key, (httpRequests.get(key) ?? 0) + 1);
+  const current = httpDuration.get(key) ?? { count: 0, sum: 0 };
+  httpDuration.set(key, { count: current.count + 1, sum: current.sum + durationSeconds });
+}
+
+export function recordIndexerPosition(checkpointLedger: number, latestLedger: number): void {
+  indexerCheckpointLedger = checkpointLedger;
+  indexerLatestLedger = latestLedger;
+}
+
+export function recordNotificationDelivery(outcome: 'success' | 'failure'): void {
+  notificationDeliveries.set(outcome, (notificationDeliveries.get(outcome) ?? 0) + 1);
+}
+
+export function renderMetrics(): string {
+  const lines = [
+    '# HELP streamgive_http_requests_total Total HTTP requests handled.',
+    '# TYPE streamgive_http_requests_total counter',
+  ];
+
+  for (const [key, count] of httpRequests) {
+    const [method, route, statusCode] = key.split('\u0000');
+    lines.push(
+      `streamgive_http_requests_total{${labels(method, route, Number(statusCode))}} ${count}`,
+    );
+  }
+
+  lines.push(
+    '# HELP streamgive_http_request_duration_seconds HTTP request duration in seconds.',
+    '# TYPE streamgive_http_request_duration_seconds summary',
+  );
+  for (const [key, value] of httpDuration) {
+    const [method, route, statusCode] = key.split('\u0000');
+    const metricLabels = labels(method, route, Number(statusCode));
+    lines.push(`streamgive_http_request_duration_seconds_sum{${metricLabels}} ${value.sum}`);
+    lines.push(`streamgive_http_request_duration_seconds_count{${metricLabels}} ${value.count}`);
+  }
+
+  lines.push(
+    '# HELP streamgive_indexer_checkpoint_ledger Last ledger checkpointed by the indexer.',
+    '# TYPE streamgive_indexer_checkpoint_ledger gauge',
+    `streamgive_indexer_checkpoint_ledger ${indexerCheckpointLedger}`,
+    '# HELP streamgive_indexer_latest_ledger Latest ledger observed from Stellar RPC.',
+    '# TYPE streamgive_indexer_latest_ledger gauge',
+    `streamgive_indexer_latest_ledger ${indexerLatestLedger}`,
+    '# HELP streamgive_indexer_lag_ledgers Difference between the latest and checkpoint ledgers.',
+    '# TYPE streamgive_indexer_lag_ledgers gauge',
+    `streamgive_indexer_lag_ledgers ${Math.max(0, indexerLatestLedger - indexerCheckpointLedger)}`,
+    '# HELP streamgive_notification_deliveries_total Webhook notification delivery attempts.',
+    '# TYPE streamgive_notification_deliveries_total counter',
+  );
+
+  for (const outcome of ['success', 'failure'] as const) {
+    lines.push(
+      `streamgive_notification_deliveries_total{outcome="${outcome}"} ${notificationDeliveries.get(outcome) ?? 0}`,
+    );
+  }
+
+  return `${lines.join('\n')}\n`;
+}

src/notifications/service.ts

@@ -1,6 +1,7 @@
 import { createHmac } from 'node:crypto';
 
 import { prisma } from '../db.js';
+import { recordNotificationDelivery } from '../metrics.js';
 import type { NotificationEvent } from './types.js';
 
 const deliveredEventIds = new Set<string>();
@@ -64,15 +65,14 @@ async function notifyWebhook(event: NotificationEvent): Promise<void> {
       success = true;
     } else {
       error = `status ${res.status}`;
-      console.error(
-        `webhook notification failed with status ${res.status} for ${webhookUrl}`,
-      );
+      console.error(`webhook notification failed with status ${res.status} for ${webhookUrl}`);
     }
   } catch (err) {
     error = err instanceof Error ? err.message : String(err);
     console.error('webhook notification failed', err);
   }
 
+  recordNotificationDelivery(success ? 'success' : 'failure');
   await prisma.notificationLog.create({
     data: { eventType: event.type, channel: 'webhook', success, error },
   });

src/server.ts

@@ -7,6 +7,8 @@ import Fastify from 'fastify';
 import type { FastifyError } from 'fastify';
 
 import { prisma } from './db.js';
+import { recordHttpRequest, renderMetrics } from './metrics.js';
+import { getCheckpoint } from './indexer/checkpoint.js';
 import { donorRoutes } from './routes/donors.js';
 import { impactRoutes } from './routes/impact.js';
 import { indexerStatusRoutes } from './routes/indexerStatus.js';
@@ -72,6 +74,7 @@ export interface BuildServerOptions {
 }
 
 export function buildServer(options?: BuildServerOptions) {
+  const requestStartedAt = new WeakMap<object, bigint>();
   const trustProxySetting =
     options?.trustProxy !== undefined
       ? options.trustProxy
@@ -165,6 +168,26 @@ export function buildServer(options?: BuildServerOptions) {
     },
   });
 
+  app.addHook('onRequest', async (request) => {
+    requestStartedAt.set(request, process.hrtime.bigint());
+  });
+
+  app.addHook('onResponse', async (request, reply) => {
+    const startedAt = requestStartedAt.get(request);
+    if (startedAt === undefined) return;
+    const durationSeconds = Number(process.hrtime.bigint() - startedAt) / 1_000_000_000;
+    recordHttpRequest(
+      request.method,
+      request.routeOptions.url ?? 'unmatched',
+      reply.statusCode,
+      durationSeconds,
+    );
+  });
+
+  app.get('/metrics', async (_request, reply) => {
+    return reply.type('text/plain; version=0.0.4; charset=utf-8').send(renderMetrics());
+  });
+
   app.register(swagger, {
     openapi: {
       info: {

test/metrics.test.ts

@@ -0,0 +1,29 @@
+import { describe, expect, it, vi } from 'vitest';
+
+vi.mock('../src/routes/ngoApplications.js', () => ({
+  ngoApplicationRoutes: async () => {},
+}));
+
+import { buildServer } from '../src/server.js';
+
+describe('GET /metrics', () => {
+  it('returns Prometheus text with HTTP and indexer metrics', async () => {
+    const app = buildServer();
+
+    await app.inject({ method: 'GET', url: '/not-a-route' });
+    const response = await app.inject({ method: 'GET', url: '/metrics' });
+
+    expect(response.statusCode).toBe(200);
+    expect(response.headers['content-type']).toContain('text/plain');
+    expect(response.body).toContain('# TYPE streamgive_http_requests_total counter');
+    expect(response.body).toContain(
+      'streamgive_http_requests_total{method="GET",route="unmatched",status_code="404"}',
+    );
+    expect(response.body).toContain('# TYPE streamgive_http_request_duration_seconds summary');
+    expect(response.body).toContain('# TYPE streamgive_indexer_lag_ledgers gauge');
+    expect(response.body).toContain('streamgive_indexer_checkpoint_ledger 0');
+    expect(response.body).toContain('# TYPE streamgive_notification_deliveries_total counter');
+
+    await app.close();
+  });
+});