From 832f043fc580955a0ae811c9b5766b021f980252 Mon Sep 17 00:00:00 2001 From: Tim Oyelabi Date: Mon, 29 Jun 2026 15:53:03 +0100 Subject: [PATCH] feat(indexer): add event statistics endpoint with type breakdown --- pnpm-lock.yaml | 29 ++++++++++++ services/indexer/package.json | 6 ++- services/indexer/src/index.test.ts | 56 +++++++++++++++++++++++ services/indexer/src/index.ts | 73 ++++++++++++++++++------------ 4 files changed, 135 insertions(+), 29 deletions(-) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 2cddaef..24a7690 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -4,6 +4,9 @@ settings: autoInstallPeers: true excludeLinksFromLockfile: false +overrides: + axios: ^1.7.4 + importers: .: @@ -119,6 +122,12 @@ importers: '@fastify/cors': specifier: '*' version: 11.2.0 + '@fastify/rate-limit': + specifier: ^11.1.0 + version: 11.1.0 + '@prisma/adapter-pg': + specifier: ^7.8.0 + version: 7.8.0 '@prisma/client': specifier: '*' version: 7.8.0(prisma@7.8.0(@types/react@19.2.17)(react-dom@19.2.7(react@19.2.7))(react@19.2.7)(typescript@5.9.3))(typescript@5.9.3) @@ -134,6 +143,9 @@ importers: ioredis: specifier: '*' version: 5.11.1 + pg: + specifier: ^8.22.0 + version: 8.22.0 zod: specifier: ^3.22.4 version: 3.25.76 @@ -141,6 +153,9 @@ importers: '@types/node': specifier: ^20.10.0 version: 20.19.43 + '@types/pg': + specifier: ^8.20.0 + version: 8.20.0 '@types/tape': specifier: ^5.6.4 version: 5.8.1 @@ -277,6 +292,9 @@ packages: '@fastify/rate-limit@11.0.0': resolution: {integrity: sha512-kCs+G59SitZw9TL/ekFe+MrzXk20dEp6zPAM8WEZjFl5Ubvv5ksTbEXYr4jGlBwWAKn78q+NFsj5CN75zXLjaw==} + '@fastify/rate-limit@11.1.0': + resolution: {integrity: sha512-BeJ9tizLvmTXGD7deYU5G04OtHhwk5uHxbpEPVp09gKvUBIXmau/4Bshxhu9ci54MvVWfGjCEx4RzvsTntojwA==} + '@hono/node-server@1.19.11': resolution: {integrity: sha512-dr8/3zEaB+p0D2n/IUrlPF1HZm586qgJNXK1a9fhg/PzdtkK7Ksd5l312tJX2yBuALqDYBlG20QEbayqPyxn+g==} engines: {node: '>=18.14.1'} @@ -878,6 +896,9 @@ packages: fastify-plugin@5.1.0: resolution: {integrity: sha512-FAIDA8eovSt5qcDgcBvDuX/v0Cjz0ohGhENZ/wpc3y+oZCY2afZ9Baqql3g/lC+OHRnciQol4ww7tuthOb9idw==} + fastify-plugin@6.0.0: + resolution: {integrity: sha512-fZOty7z3O7vOliF6d8bHE3wiEh1KcNnKEQensSgTk9C1DvN6nRLS++XVd86v33Hw/8u9Un8A1zDrQ8ujcQDHEg==} + fastify@5.8.5: resolution: {integrity: sha512-Yqptv59pQzPgQUSIm87hMqHJmdkb1+GPxdE6vW6FRyVE9G86mt7rOghitiU4JHRaTyDUk9pfeKmDeu70lAwM4Q==} @@ -1804,6 +1825,12 @@ snapshots: fastify-plugin: 5.1.0 toad-cache: 3.7.1 + '@fastify/rate-limit@11.1.0': + dependencies: + '@lukeed/ms': 2.0.2 + fastify-plugin: 6.0.0 + toad-cache: 3.7.1 + '@hono/node-server@1.19.11(hono@4.12.25)': dependencies: hono: 4.12.25 @@ -2507,6 +2534,8 @@ snapshots: fastify-plugin@5.1.0: {} + fastify-plugin@6.0.0: {} + fastify@5.8.5: dependencies: '@fastify/ajv-compiler': 4.0.5 diff --git a/services/indexer/package.json b/services/indexer/package.json index b86f3c8..46f69fd 100644 --- a/services/indexer/package.json +++ b/services/indexer/package.json @@ -13,16 +13,20 @@ }, "dependencies": { "@bettapay/validation": "workspace:*", + "@fastify/cors": "*", + "@fastify/rate-limit": "^11.1.0", + "@prisma/adapter-pg": "^7.8.0", "@prisma/client": "*", "@stellar/stellar-sdk": "*", "bullmq": "*", "fastify": "*", - "@fastify/cors": "*", "ioredis": "*", + "pg": "^8.22.0", "zod": "^3.22.4" }, "devDependencies": { "@types/node": "^20.10.0", + "@types/pg": "^8.20.0", "@types/tape": "^5.6.4", "cross-env": "^7.0.3", "tape": "^5.9.0", diff --git a/services/indexer/src/index.test.ts b/services/indexer/src/index.test.ts index 382ea09..dbaf2f9 100644 --- a/services/indexer/src/index.test.ts +++ b/services/indexer/src/index.test.ts @@ -1,6 +1,62 @@ import test from 'tape'; import { fastify } from './index.js'; +test('GET /api/events/stats - rejects requests without service token', async (t) => { + await fastify.ready(); + + try { + const res = await fastify.inject({ + method: 'GET', + url: '/api/events/stats', + }); + t.equal(res.statusCode, 401, 'should return 401 without x-service-token'); + const body = JSON.parse(res.body); + t.equal(body.error?.code, 'UNAUTHORIZED', 'error code should be UNAUTHORIZED'); + } catch (err: any) { + t.fail(err); + } finally { + t.end(); + } +}); + +test('GET /api/events/stats - rejects invalid date strings with 400', async (t) => { + await fastify.ready(); + + try { + const res = await fastify.inject({ + method: 'GET', + url: '/api/events/stats?from=not-a-date', + headers: { 'x-service-token': process.env.INTER_SERVICE_SECRET || 'test-secret-that-is-at-least-16-chars' }, + }); + t.equal(res.statusCode, 400, 'should return 400 for invalid from date'); + const body = JSON.parse(res.body); + t.equal(body.error?.code, 'VALIDATION_ERROR', 'error code should be VALIDATION_ERROR'); + } catch (err: any) { + t.fail(err); + } finally { + t.end(); + } +}); + +test('GET /api/events/stats - rejects from > to with 400', async (t) => { + await fastify.ready(); + + try { + const res = await fastify.inject({ + method: 'GET', + url: '/api/events/stats?from=2025-06-01T00:00:00Z&to=2024-06-01T00:00:00Z', + headers: { 'x-service-token': process.env.INTER_SERVICE_SECRET || 'test-secret-that-is-at-least-16-chars' }, + }); + t.equal(res.statusCode, 400, 'should return 400 when from is after to'); + const body = JSON.parse(res.body); + t.equal(body.error?.code, 'VALIDATION_ERROR', 'error code should be VALIDATION_ERROR'); + } catch (err: any) { + t.fail(err); + } finally { + t.end(); + } +}); + test('Indexer rate limiting - requests below the limit succeed', async (t) => { await fastify.ready(); diff --git a/services/indexer/src/index.ts b/services/indexer/src/index.ts index d9fea28..f4c9a27 100644 --- a/services/indexer/src/index.ts +++ b/services/indexer/src/index.ts @@ -21,6 +21,8 @@ import { Redis } from 'ioredis'; import { Queue, Worker } from 'bullmq'; import { PrismaClient } from '@prisma/client'; import { rpc, scValToNative, xdr } from '@stellar/stellar-sdk'; +import pg from 'pg'; +import { PrismaPg } from '@prisma/adapter-pg'; import { z } from 'zod'; import { validateEnv, @@ -28,9 +30,12 @@ import { registerRequestId, registerServiceAuth, PaginationQuery, + DateRangeQuery, EVENT_TYPES, connectWithRetry, createLoggerOptions, + getPrismaLogLevels, + setupPrismaQueryLogging, registerTracing, } from '@bettapay/validation'; import type { EventType } from '@bettapay/validation'; @@ -40,7 +45,10 @@ const PORT = Number(process.env.PORT ?? '3003'); const fastify = Fastify({ logger: createLoggerOptions({ level: env.LOG_LEVEL }) }); registerRequestId(fastify); -const prisma = new PrismaClient(); +const pool = new pg.Pool({ connectionString: env.DATABASE_URL }); +const prismaAdapter = new PrismaPg(pool); +const prisma = new PrismaClient({ adapter: prismaAdapter, log: getPrismaLogLevels() }); +setupPrismaQueryLogging(prisma, fastify.log); fastify.register(cors, { origin: env.ALLOWED_ORIGINS }); registerErrorHandler(fastify); @@ -49,17 +57,12 @@ registerTracing(fastify); // Inter-service auth: internal endpoints require a valid x-service-token (#117). registerServiceAuth(fastify, env.INTER_SERVICE_SECRET); -<<<<<<< HEAD -// Polling state -======= fastify.register(rateLimit, { max: 500, timeWindow: '1 minute' }); -// In-memory event ring buffer (50 events max) -const events: any[] = []; ->>>>>>> 35d765e (.) +// Polling state let latestLedgerCursor: number | undefined = undefined; let latestLedgerSequence: number | undefined = undefined; const BASE_BACKOFF = 1000; @@ -215,7 +218,41 @@ fastify.get('/api/events', { preValidation: [fastify.serviceAuth] }, async (requ return { events: dbEvents, total, limit, offset, hasMore, latestLedgerCursor }; }); -<<<<<<< HEAD +// Issue #79 — event counts by type within a configurable time window. +fastify.get('/api/events/stats', { preValidation: [fastify.serviceAuth] }, async (request) => { + const { from: fromStr, to: toStr } = DateRangeQuery.parse(request.query ?? {}); + + const from = fromStr ? new Date(fromStr) : undefined; + const to = new Date(toStr); + + const grouped = await prisma.indexedEvent.groupBy({ + by: ['type'], + _count: true, + where: { + indexedAt: { + ...(from ? { gte: from } : {}), + lte: to, + }, + }, + }); + + const byType: Record = {}; + let total = 0; + for (const entry of grouped) { + byType[entry.type] = entry._count; + total += entry._count; + } + + return { + total, + byType, + timeRange: { + from: from?.toISOString() ?? toStr, + to: to.toISOString(), + }, + }; +}); + // Issue #68 — replay historical events for a ledger range const ReplayBody = z.object({ fromLedger: z.number().int().min(1), @@ -322,22 +359,6 @@ fastify.delete<{ Params: { id: string } }>('/api/webhooks/:id', async (request, // ── Stellar RPC polling loop ────────────────────────────────────────────────── -======= -fastify.route({ - method: ['GET', 'POST'], - url: '/api/events/replay', - config: { - rateLimit: { - max: 60, - timeWindow: '1 minute' - } - }, - handler: async (request, reply) => { - return { status: 'ok', replayed: true }; - } -}); - ->>>>>>> 35d765e (.) const server = new rpc.Server(env.STELLAR_RPC_URL, { allowHttp: true }); async function pollEvents() { @@ -414,7 +435,6 @@ const start = async () => { } }; -<<<<<<< HEAD process.on('SIGTERM', async () => { await prisma.$disconnect(); await webhookQueue.close(); @@ -423,11 +443,8 @@ process.on('SIGTERM', async () => { process.exit(0); }); -start(); -======= if (process.env.NODE_ENV !== 'test') { start(); } export { fastify }; ->>>>>>> 35d765e (.)