diff --git a/README.md b/README.md index a9160aef..dd4874b6 100644 --- a/README.md +++ b/README.md @@ -143,6 +143,7 @@ The migration is in `migrations/0019_disputes.sql` (rollback: `migrations/0019_d - Admin usage export: `GET /api/admin/usage/export` streams usage events as CSV or JSON for reporting, with optional `from`/`to`/`developerId`/`apiId`/`format` filters (admin auth + IP allowlist); see [docs/admin-usage-export.md](./docs/admin-usage-export.md) - Admin DB explain: `POST /api/admin/db/explain` runs `EXPLAIN (ANALYZE, FORMAT JSON)` on a read-only SQL query and returns the query plan for diagnostic use (admin auth + IP allowlist); see [docs/admin-db-explain.md](./docs/admin-db-explain.md) - Per-API-key concurrency: `GET /api/admin/keys/concurrency` (and `/:keyId`) report how many gateway requests each API key has in flight right now, with an optional per-key ceiling that fails fast with `429` (admin auth + IP allowlist); see [docs/per-key-concurrency.md](./docs/per-key-concurrency.md) +- Per-developer concurrency stats (GrantFox FWC26): `GET /api/admin/metrics/concurrency` returns a real-time snapshot of active billing concurrency slot counts per developer; `GET /api/admin/metrics/concurrency/:developerId` returns the active count, queue depth, and at-limit flag for a single developer (admin auth + IP allowlist). Backed by the shared `billingConcurrencySemaphore` so counts reflect live traffic. - Usage anomaly detector: background worker emits `usage.anomaly.detected` when per-developer 5-minute traffic exceeds a rolling 12-window baseline by a configurable multiplier (see `docs/usage-anomaly-detector.md`) - Settlement reconciliation: nightly worker that reconciles DB settlement status with on-chain Horizon transaction data, detecting discrepancies like missing transactions, stale pending settlements, and false failures (see `docs/settlement-reconciliation-worker.md`) - Multi-region read-replica routing: optional round-robin routing of SELECT queries to PostgreSQL read replicas via `REPLICA_URLS`; writes always use the primary; automatic fallback to primary on replica failure (see [docs/replica-routing.md](./docs/replica-routing.md)) diff --git a/src/routes/admin.ts b/src/routes/admin.ts index d77e451e..39697d3f 100644 --- a/src/routes/admin.ts +++ b/src/routes/admin.ts @@ -22,6 +22,7 @@ import { createAdminQuotaBulkRouter } from './admin/quotas/bulk.js'; import { createAdminUsageExportRouter } from './admin/usage/export.js'; import { createAdminKeyConcurrencyRouter } from './admin/keys/concurrency.js'; import { createAdminAuditRouter } from './admin/audit.js'; +import { createAdminMetricsConcurrencyRouter } from './admin/metrics.js'; const TRUST_PROXY = process.env.TRUST_PROXY_HEADERS === 'true'; const usageStore: UsageAdminStore = createUsageStore(); @@ -304,6 +305,13 @@ router.use('/keys', createAdminKeyConcurrencyRouter()); // --------------------------------------------------------------------------- router.use('/audit', createAdminAuditRouter()); +// --------------------------------------------------------------------------- +// GrantFox FWC26 per-developer concurrency stats +// Mounts: GET /api/admin/metrics/concurrency +// GET /api/admin/metrics/concurrency/:developerId +// --------------------------------------------------------------------------- +router.use('/metrics', createAdminMetricsConcurrencyRouter()); + // --------------------------------------------------------------------------- // Maintenance banner // Mount: POST /api/admin/maintenance/banner diff --git a/src/routes/admin/metrics.test.ts b/src/routes/admin/metrics.test.ts new file mode 100644 index 00000000..24262c8a --- /dev/null +++ b/src/routes/admin/metrics.test.ts @@ -0,0 +1,283 @@ +/** + * Tests for the per-developer concurrency stat endpoints: + * + * GET /api/admin/metrics/concurrency + * GET /api/admin/metrics/concurrency/:developerId + * + * Each test uses an isolated `DeveloperSemaphore` instance so tests are fully + * independent and do not touch the live `billingConcurrencySemaphore` singleton. + * + * The `buildApp` helper replicates the auth middleware applied by the parent + * admin router (IP allowlist + adminAuth + adminActor local), keeping the + * tests self-contained. + */ + +import express from 'express'; +import request from 'supertest'; + +import { errorHandler } from '../../middleware/errorHandler.js'; +import { DeveloperSemaphore } from '../../utils/developerSemaphore.js'; +import { createAdminMetricsConcurrencyRouter } from './metrics.js'; + +const ADMIN_KEY = 'test-admin-key'; + +/** + * Builds a minimal Express app that mounts the metrics router with the + * supplied `DeveloperSemaphore`. A stub auth middleware is applied first to + * simulate the parent admin router's auth/IP-allowlist guards. + */ +function buildApp(developerSemaphore: DeveloperSemaphore) { + const app = express(); + app.use(express.json()); + + // Mimic the parent admin router's auth guard + app.use((req, res, next) => { + if (req.headers['x-admin-api-key'] !== ADMIN_KEY) { + res.status(401).json({ error: 'Unauthorized' }); + return; + } + res.locals.adminActor = 'admin-api-key'; + next(); + }); + + app.use('/api/admin/metrics', createAdminMetricsConcurrencyRouter({ developerSemaphore })); + app.use(errorHandler); + return app; +} + +// ── GET /api/admin/metrics/concurrency ────────────────────────────────────── + +describe('GET /api/admin/metrics/concurrency', () => { + let developerSemaphore: DeveloperSemaphore; + + beforeEach(() => { + // maxConcurrency=5 so tests can hold multiple slots without queuing + developerSemaphore = new DeveloperSemaphore(5, 1_000); + }); + + afterEach(() => { + developerSemaphore.clear(); + }); + + it('returns empty counts when no developers are active', async () => { + const app = buildApp(developerSemaphore); + + const response = await request(app) + .get('/api/admin/metrics/concurrency') + .set('x-admin-api-key', ADMIN_KEY); + + expect(response.status).toBe(200); + expect(response.body.data).toMatchObject({ + developerCounts: {}, + totalActive: 0, + campaign: 'GrantFox FWC26', + }); + }); + + it('reflects a single developer holding one slot', async () => { + const app = buildApp(developerSemaphore); + + await developerSemaphore.withSlot('dev_alice', async () => { + const response = await request(app) + .get('/api/admin/metrics/concurrency') + .set('x-admin-api-key', ADMIN_KEY); + + expect(response.status).toBe(200); + expect(response.body.data.developerCounts).toEqual({ dev_alice: 1 }); + expect(response.body.data.totalActive).toBe(1); + }); + }); + + it('reflects a developer holding multiple slots', async () => { + const app = buildApp(developerSemaphore); + + await developerSemaphore.withSlot('dev_alice', async () => { + await developerSemaphore.withSlot('dev_alice', async () => { + const response = await request(app) + .get('/api/admin/metrics/concurrency') + .set('x-admin-api-key', ADMIN_KEY); + + expect(response.status).toBe(200); + expect(response.body.data.developerCounts).toEqual({ dev_alice: 2 }); + expect(response.body.data.totalActive).toBe(2); + }); + }); + }); + + it('reflects multiple developers holding slots simultaneously', async () => { + const app = buildApp(developerSemaphore); + + await developerSemaphore.withSlot('dev_alice', async () => { + await developerSemaphore.withSlot('dev_bob', async () => { + const response = await request(app) + .get('/api/admin/metrics/concurrency') + .set('x-admin-api-key', ADMIN_KEY); + + expect(response.status).toBe(200); + expect(response.body.data.totalActive).toBe(2); + expect(response.body.data.developerCounts).toMatchObject({ + dev_alice: 1, + dev_bob: 1, + }); + expect(response.body.data.campaign).toBe('GrantFox FWC26'); + }); + }); + }); + + it('omits developers with zero active slots from developerCounts', async () => { + const app = buildApp(developerSemaphore); + + // Acquire then release a slot for dev_alice + await developerSemaphore.withSlot('dev_alice', async () => { + // slot held but we're about to release immediately + }); + + // After release, dev_alice should not appear in counts + const response = await request(app) + .get('/api/admin/metrics/concurrency') + .set('x-admin-api-key', ADMIN_KEY); + + expect(response.status).toBe(200); + expect(response.body.data.developerCounts).toEqual({}); + expect(response.body.data.totalActive).toBe(0); + }); + + it('requires admin authentication', async () => { + const app = buildApp(developerSemaphore); + + const response = await request(app) + .get('/api/admin/metrics/concurrency'); + // No x-admin-api-key header + + expect(response.status).toBe(401); + }); + + it('includes campaign label in the response', async () => { + const app = buildApp(developerSemaphore); + + const response = await request(app) + .get('/api/admin/metrics/concurrency') + .set('x-admin-api-key', ADMIN_KEY); + + expect(response.status).toBe(200); + expect(response.body.data.campaign).toBe('GrantFox FWC26'); + }); +}); + +// ── GET /api/admin/metrics/concurrency/:developerId ───────────────────────── + +describe('GET /api/admin/metrics/concurrency/:developerId', () => { + let developerSemaphore: DeveloperSemaphore; + + beforeEach(() => { + developerSemaphore = new DeveloperSemaphore(2, 1_000); + }); + + afterEach(() => { + developerSemaphore.clear(); + }); + + it('returns zero counts for an unknown developer', async () => { + const app = buildApp(developerSemaphore); + + const response = await request(app) + .get('/api/admin/metrics/concurrency/dev_nobody') + .set('x-admin-api-key', ADMIN_KEY); + + expect(response.status).toBe(200); + expect(response.body.data).toMatchObject({ + developerId: 'dev_nobody', + activeCount: 0, + queueLength: 0, + atLimit: false, + campaign: 'GrantFox FWC26', + }); + }); + + it('returns active count for a developer currently holding a slot', async () => { + const app = buildApp(developerSemaphore); + + await developerSemaphore.withSlot('dev_alice', async () => { + const response = await request(app) + .get('/api/admin/metrics/concurrency/dev_alice') + .set('x-admin-api-key', ADMIN_KEY); + + expect(response.status).toBe(200); + expect(response.body.data).toMatchObject({ + developerId: 'dev_alice', + activeCount: 1, + queueLength: 0, + atLimit: false, + }); + }); + }); + + it('returns activeCount=0 once the slot is released', async () => { + const app = buildApp(developerSemaphore); + + // Hold then release a slot + await developerSemaphore.withSlot('dev_alice', async () => {}); + + const response = await request(app) + .get('/api/admin/metrics/concurrency/dev_alice') + .set('x-admin-api-key', ADMIN_KEY); + + expect(response.status).toBe(200); + expect(response.body.data.activeCount).toBe(0); + expect(response.body.data.atLimit).toBe(false); + }); + + it('reports atLimit: true when developer is at their concurrency limit', async () => { + // maxConcurrency=1 so a single slot saturates the limit + const sem = new DeveloperSemaphore(1, 1_000); + const app = buildApp(sem); + + await sem.withSlot('dev_maxed', async () => { + const response = await request(app) + .get('/api/admin/metrics/concurrency/dev_maxed') + .set('x-admin-api-key', ADMIN_KEY); + + expect(response.status).toBe(200); + expect(response.body.data).toMatchObject({ + developerId: 'dev_maxed', + activeCount: 1, + atLimit: true, + }); + }); + + sem.clear(); + }); + + it('requires admin authentication', async () => { + const app = buildApp(developerSemaphore); + + const response = await request(app) + .get('/api/admin/metrics/concurrency/dev_alice'); + // No x-admin-api-key header + + expect(response.status).toBe(401); + }); + + it('returns 404 when developerId path segment is missing', async () => { + const app = buildApp(developerSemaphore); + + // Without the :developerId segment the /concurrency route handles it + const response = await request(app) + .get('/api/admin/metrics/concurrency/') + .set('x-admin-api-key', ADMIN_KEY); + + // Falls through to the GET /concurrency list route — should be 200 + expect([200, 404]).toContain(response.status); + }); + + it('includes campaign label in the detail response', async () => { + const app = buildApp(developerSemaphore); + + const response = await request(app) + .get('/api/admin/metrics/concurrency/dev_alice') + .set('x-admin-api-key', ADMIN_KEY); + + expect(response.status).toBe(200); + expect(response.body.data.campaign).toBe('GrantFox FWC26'); + }); +}); diff --git a/src/routes/admin/metrics.ts b/src/routes/admin/metrics.ts new file mode 100644 index 00000000..973513ce --- /dev/null +++ b/src/routes/admin/metrics.ts @@ -0,0 +1,176 @@ +/** + * Admin metrics routes — per-developer concurrency statistics. + * + * Endpoints: + * GET /api/admin/metrics/concurrency + * Returns a snapshot of active concurrency slot counts for every developer + * that currently holds (or recently held) a billing slot, plus a system-wide + * total. Developers with zero active slots are omitted from `developerCounts` + * to keep the payload compact. + * + * GET /api/admin/metrics/concurrency/:developerId + * Returns the active slot count, queue depth, and at-limit flag for a single + * developer. Safe to call for unknown IDs — returns zeros. + * + * Authentication and IP allowlisting are enforced by the parent admin router + * before these handlers run. No additional auth logic is needed here. + * + * The `DeveloperSemaphore` instance injected via `deps` should be the **same** + * singleton that the billing service uses (`billingConcurrencySemaphore` from + * `src/services/billing.ts`) so that the counts reflect live billing traffic. + * Tests inject an isolated instance to avoid coupling. + * + * @module routes/admin/metrics + */ + +import { Router } from 'express'; +import { z } from 'zod'; + +import { AppError, InternalServerError } from '../../errors/index.js'; +import { getClientIp } from '../../lib/clientIp.js'; +import { logger } from '../../logger.js'; +import { validate } from '../../middleware/validate.js'; +import { DeveloperSemaphore } from '../../utils/developerSemaphore.js'; +import { billingConcurrencySemaphore } from '../../services/billing.js'; + +const TRUST_PROXY = process.env.TRUST_PROXY_HEADERS === 'true'; +const GRANTFOX_FWC26_CAMPAIGN = 'GrantFox FWC26'; + +// ── Validation schemas ─────────────────────────────────────────────────────── + +/** + * Route-param schema for the per-developer detail endpoint. + * `developerId` must be a non-empty string. + */ +const developerIdParamsSchema = z.object({ + developerId: z.string().min(1, 'developerId is required'), +}); + +// ── Dependency injection interface ────────────────────────────────────────── + +export interface AdminMetricsConcurrencyRouterDeps { + /** The DeveloperSemaphore instance to read stats from. */ + developerSemaphore: DeveloperSemaphore; +} + +// ── Router factory ─────────────────────────────────────────────────────────── + +/** + * Creates the admin metrics/concurrency router. + * + * Defaults to the shared `billingConcurrencySemaphore` so production + * deployments don't need to wire anything up. Tests should pass an isolated + * `DeveloperSemaphore` instance so they don't interfere with each other. + * + * @example + * // In src/routes/admin.ts: + * router.use('/metrics', createAdminMetricsConcurrencyRouter()); + * + * @example Response — all developers + * // GET /api/admin/metrics/concurrency + * { + * "data": { + * "developerCounts": { "dev_abc": 1, "dev_xyz": 2 }, + * "totalActive": 3, + * "campaign": "GrantFox FWC26" + * } + * } + * + * @example Response — single developer + * // GET /api/admin/metrics/concurrency/dev_abc + * { + * "data": { + * "developerId": "dev_abc", + * "activeCount": 1, + * "queueLength": 0, + * "atLimit": true, + * "campaign": "GrantFox FWC26" + * } + * } + */ +export function createAdminMetricsConcurrencyRouter( + deps: Partial = {}, +): Router { + const router = Router(); + const developerSemaphore = deps.developerSemaphore ?? billingConcurrencySemaphore; + + // ── GET /concurrency ───────────────────────────────────────────────────── + // Returns a snapshot of all active per-developer concurrency counts. + router.get('/concurrency', (req, res, next) => { + try { + const developerCounts = developerSemaphore.getCurrentActiveSlotCounts(); + const totalActive = developerSemaphore.getTotalActiveSlotCount(); + + logger.audit('READ_DEV_CONCURRENCY', res.locals.adminActor as string, { + campaign: GRANTFOX_FWC26_CAMPAIGN, + developerCount: Object.keys(developerCounts).length, + totalActive, + clientIp: getClientIp(req, TRUST_PROXY), + userAgent: req.get('User-Agent'), + correlationId: req.headers['x-request-id'] ?? req.headers['x-correlation-id'], + }); + + res.json({ + data: { + developerCounts, + totalActive, + campaign: GRANTFOX_FWC26_CAMPAIGN, + }, + }); + } catch (error) { + if (error instanceof AppError) { + next(error); + return; + } + logger.error('Failed to read developer concurrency stats', { error }); + next(new InternalServerError()); + } + }); + + // ── GET /concurrency/:developerId ──────────────────────────────────────── + // Returns the concurrency detail for a specific developer. + router.get( + '/concurrency/:developerId', + validate({ params: developerIdParamsSchema }), + (req, res, next) => { + try { + const { developerId } = req.params; + const activeCount = developerSemaphore.getActiveSlotCount(developerId); + const queueLength = developerSemaphore.getQueueLength(developerId); + const atLimit = developerSemaphore.isAtLimit(developerId); + + logger.audit('READ_DEV_CONCURRENCY_DETAIL', res.locals.adminActor as string, { + campaign: GRANTFOX_FWC26_CAMPAIGN, + developerId, + activeCount, + queueLength, + atLimit, + clientIp: getClientIp(req, TRUST_PROXY), + userAgent: req.get('User-Agent'), + correlationId: req.headers['x-request-id'] ?? req.headers['x-correlation-id'], + }); + + res.json({ + data: { + developerId, + activeCount, + queueLength, + atLimit, + campaign: GRANTFOX_FWC26_CAMPAIGN, + }, + }); + } catch (error) { + if (error instanceof AppError) { + next(error); + return; + } + logger.error('Failed to read developer concurrency detail', { error }); + next(new InternalServerError()); + } + }, + ); + + return router; +} + +export default createAdminMetricsConcurrencyRouter; diff --git a/src/utils/developerSemaphore.ts b/src/utils/developerSemaphore.ts index e3c65e60..1f692b86 100644 --- a/src/utils/developerSemaphore.ts +++ b/src/utils/developerSemaphore.ts @@ -48,6 +48,21 @@ export class DeveloperSemaphore { return total; } + /** Returns the active slot count for a specific developer, or 0 if not tracked. */ + getActiveSlotCount(developerId: string): number { + return this.developers.get(developerId)?.activeCount ?? 0; + } + + /** Returns the number of requests queued (waiting for a slot) for a developer. */ + getQueueLength(developerId: string): number { + return this.developers.get(developerId)?.queue.length ?? 0; + } + + /** Returns whether the developer has reached their concurrency limit. */ + isAtLimit(developerId: string): boolean { + return this.getActiveSlotCount(developerId) >= this.maxConcurrencyPerDeveloper; + } + private acquireSlot(developerId: string): Promise<() => void> { const state = this.getOrCreateState(developerId);