diff --git a/backend/src/lib/indexer-state.ts b/backend/src/lib/indexer-state.ts index fb0218e7..e6568647 100644 --- a/backend/src/lib/indexer-state.ts +++ b/backend/src/lib/indexer-state.ts @@ -1 +1,59 @@ +import { prisma } from './prisma.js'; +import logger from '../logger.js'; + export const INDEXER_STATE_ID = 'singleton'; + +export interface IndexerStateRow { + id: string; + lastLedger: number; + lastCursor: string | null; + createdAt: Date; + updatedAt: Date; +} + +/** + * Ensure the singleton indexer_state row exists. + * Uses a catch-and-retry pattern to handle the race condition where two + * concurrent callers attempt the first insert simultaneously. If the + * unique-constraint violation fires, we treat it as success and re-read. + */ +export async function ensureIndexerState( + startLedger: number, +): Promise { + const existing = await prisma.indexerState.findUnique({ + where: { id: INDEXER_STATE_ID }, + }); + if (existing) return existing; + + try { + const created = await prisma.indexerState.create({ + data: { + id: INDEXER_STATE_ID, + lastLedger: startLedger, + lastCursor: null, + }, + }); + return created; + } catch (err: unknown) { + // P2002 = Prisma unique-constraint violation (code "P2002") + if ( + err instanceof Error && + 'code' in err && + (err as { code: string }).code === 'P2002' + ) { + logger.warn( + '[IndexerState] Concurrent first-insert detected; re-reading existing row.', + ); + const existingAfterRace = await prisma.indexerState.findUnique({ + where: { id: INDEXER_STATE_ID }, + }); + if (!existingAfterRace) { + throw new Error( + '[IndexerState] Unique-constraint violation but row not found after race.', + ); + } + return existingAfterRace; + } + throw err; + } +} diff --git a/backend/src/lib/pg-pool.ts b/backend/src/lib/pg-pool.ts index fcb098f2..31337dea 100644 --- a/backend/src/lib/pg-pool.ts +++ b/backend/src/lib/pg-pool.ts @@ -19,3 +19,17 @@ export const createPgPoolConfig = (): pg.PoolConfig => ({ }); export const createPgPool = () => new pg.Pool(createPgPoolConfig()); + +export interface PoolMetrics { + totalCount: number; + idleCount: number; + waitingCount: number; +} + +export function getPoolMetrics(pool: pg.Pool): PoolMetrics { + return { + totalCount: pool.totalCount, + idleCount: pool.idleCount, + waitingCount: pool.waitingCount, + }; +} diff --git a/backend/src/lib/prisma.ts b/backend/src/lib/prisma.ts index fd4ddfe2..2c001c30 100644 --- a/backend/src/lib/prisma.ts +++ b/backend/src/lib/prisma.ts @@ -23,4 +23,7 @@ export const prisma = if (process.env.NODE_ENV !== 'production') globalForPrisma.prisma = prisma; +export { getPoolMetrics } from './pg-pool.js'; +export const pool = globalForPrisma.pool!; + export default prisma; diff --git a/backend/src/middleware/admin-rate-limiter.middleware.ts b/backend/src/middleware/admin-rate-limiter.middleware.ts new file mode 100644 index 00000000..1f665ef7 --- /dev/null +++ b/backend/src/middleware/admin-rate-limiter.middleware.ts @@ -0,0 +1,29 @@ +import { rateLimit } from 'express-rate-limit'; +import type { Request, Response } from 'express'; + +export const adminRateLimiter = rateLimit({ + windowMs: 1 * 60 * 1000, // 1 minute + max: 30, // Stricter limit: 30 requests per minute for admin endpoints + standardHeaders: true, + legacyHeaders: false, + message: { + error: 'Too many admin requests', + message: 'You have exceeded the admin rate limit. Please try again later.', + status: 429, + }, + keyGenerator: (req: Request): string => { + // Use x-forwarded-for or remote address as key + const forwarded = req.headers['x-forwarded-for']; + if (typeof forwarded === 'string') { + return forwarded.split(',')[0].trim(); + } + return req.ip ?? 'unknown'; + }, + skip: (req: Request): boolean => { + // Skip rate limiting in test environment + return process.env.NODE_ENV === 'test'; + }, + handler: (req: Request, res: Response, _next, options): void => { + res.status(options.statusCode).json(options.message); + }, +}); diff --git a/backend/src/routes/v1/admin.routes.ts b/backend/src/routes/v1/admin.routes.ts index f6427c74..c6a7b681 100644 --- a/backend/src/routes/v1/admin.routes.ts +++ b/backend/src/routes/v1/admin.routes.ts @@ -1,13 +1,15 @@ import { Router } from 'express'; import type { Request, Response } from 'express'; import { requireAdmin } from '../../middleware/auth.js'; +import { adminRateLimiter } from '../../middleware/admin-rate-limiter.middleware.js'; import { getIndexerStatus, resetIndexer, replayFromLedger, } from '../../services/indexerService.js'; -import { prisma } from '../../lib/prisma.js'; +import { prisma, pool } from '../../lib/prisma.js'; +import { getPoolMetrics } from '../../lib/pg-pool.js'; import { INDEXER_STATE_ID } from '../../lib/indexer-state.js'; import { sseService } from '../../services/sse.service.js'; import { cache } from '../../lib/redis.js'; @@ -17,6 +19,7 @@ const router = Router(); // All admin routes require admin JWT router.use(requireAdmin); +router.use(adminRateLimiter); /** * @openapi @@ -131,6 +134,7 @@ async function buildAdminMetrics() { }, sse: { activeConnections: sseService.getClientCount() }, cache: cache.getStats(), + pgPool: getPoolMetrics(pool), indexer: { lastLedger: indexerState?.lastLedger ?? 0, lagSeconds, diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index c2ac91ab..6e1f186c 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/backend/src/workers/soroban-event-worker.ts @@ -1,6 +1,6 @@ import { rpc, xdr, StrKey } from "@stellar/stellar-sdk"; import { prisma } from "../lib/prisma.js"; -import { INDEXER_STATE_ID } from "../lib/indexer-state.js"; +import { INDEXER_STATE_ID, ensureIndexerState } from "../lib/indexer-state.js"; import { sseService } from "../services/sse.service.js"; import logger from "../logger.js"; import { Prisma } from "../generated/prisma/index.js"; @@ -189,15 +189,7 @@ export class SorobanEventWorker { */ private async fetchAndProcessEvents(): Promise { // Ensure an IndexerState row exists on first run. - const state = await prisma.indexerState.upsert({ - where: { id: INDEXER_STATE_ID }, - create: { - id: INDEXER_STATE_ID, - lastLedger: this.startLedger, - lastCursor: null, - }, - update: {}, - }); + const state = await ensureIndexerState(this.startLedger); const baseFilter = { filters: [ diff --git a/backend/tests/admin-rate-limiter.test.ts b/backend/tests/admin-rate-limiter.test.ts new file mode 100644 index 00000000..c7c858bc --- /dev/null +++ b/backend/tests/admin-rate-limiter.test.ts @@ -0,0 +1,21 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; + +vi.mock('../src/lib/prisma.js', () => ({ + prisma: {}, +})); + +vi.mock('../src/logger.js', () => ({ + default: { info: vi.fn(), error: vi.fn(), warn: vi.fn() }, +})); + +import { adminRateLimiter } from '../src/middleware/admin-rate-limiter.middleware.js'; + +describe('adminRateLimiter', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('is a function', () => { + expect(typeof adminRateLimiter).toBe('function'); + }); +}); diff --git a/backend/tests/indexer-state.test.ts b/backend/tests/indexer-state.test.ts new file mode 100644 index 00000000..05a92a38 --- /dev/null +++ b/backend/tests/indexer-state.test.ts @@ -0,0 +1,98 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; + +const mockFindUnique = vi.fn(); +const mockCreate = vi.fn(); + +vi.mock('../src/lib/prisma.js', () => ({ + prisma: { + indexerState: { + findUnique: mockFindUnique, + create: mockCreate, + }, + }, +})); + +vi.mock('../src/logger.js', () => ({ + default: { + warn: vi.fn(), + info: vi.fn(), + error: vi.fn(), + }, +})); + +import { ensureIndexerState } from '../src/lib/indexer-state.js'; + +describe('ensureIndexerState', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('returns existing row if it already exists', async () => { + const existing = { + id: 'singleton', + lastLedger: 42, + lastCursor: 'cur', + createdAt: new Date(), + updatedAt: new Date(), + }; + mockFindUnique.mockResolvedValueOnce(existing); + + const result = await ensureIndexerState(0); + + expect(result).toEqual(existing); + expect(mockCreate).not.toHaveBeenCalled(); + }); + + it('creates and returns a new row when none exists', async () => { + const created = { + id: 'singleton', + lastLedger: 10, + lastCursor: null, + createdAt: new Date(), + updatedAt: new Date(), + }; + mockFindUnique.mockResolvedValueOnce(null); + mockCreate.mockResolvedValueOnce(created); + + const result = await ensureIndexerState(10); + + expect(result).toEqual(created); + expect(mockCreate).toHaveBeenCalledWith({ + data: { id: 'singleton', lastLedger: 10, lastCursor: null }, + }); + }); + + it('re-reads the row on unique-constraint violation (P2002) without throwing', async () => { + const existingRow = { + id: 'singleton', + lastLedger: 5, + lastCursor: null, + createdAt: new Date(), + updatedAt: new Date(), + }; + + // First findUnique returns null (no row yet) + mockFindUnique.mockResolvedValueOnce(null); + // Create throws P2002 (race condition duplicate insert) + const p2002Error = Object.assign(new Error('Unique constraint failed'), { + code: 'P2002', + }); + mockCreate.mockRejectedValueOnce(p2002Error); + // Second findUnique returns the existing row created by the concurrent caller + mockFindUnique.mockResolvedValueOnce(existingRow); + + const result = await ensureIndexerState(0); + + expect(result).toEqual(existingRow); + expect(mockCreate).toHaveBeenCalledTimes(1); + expect(mockFindUnique).toHaveBeenCalledTimes(2); + }); + + it('re-throws non-P2002 errors', async () => { + mockFindUnique.mockResolvedValueOnce(null); + const genericError = new Error('connection refused'); + mockCreate.mockRejectedValueOnce(genericError); + + await expect(ensureIndexerState(0)).rejects.toThrow('connection refused'); + }); +}); diff --git a/backend/tests/pg-pool.test.ts b/backend/tests/pg-pool.test.ts index 2b16b7e3..d5aa7cfd 100644 --- a/backend/tests/pg-pool.test.ts +++ b/backend/tests/pg-pool.test.ts @@ -45,4 +45,22 @@ describe('pg-pool', () => { }), ); }); + + it('getPoolMetrics returns totalCount, idleCount, and waitingCount from the pool', async () => { + const { getPoolMetrics } = await import('../src/lib/pg-pool.js'); + + const mockPool = { + totalCount: 10, + idleCount: 5, + waitingCount: 2, + } as any; + + const metrics = getPoolMetrics(mockPool); + + expect(metrics).toEqual({ + totalCount: 10, + idleCount: 5, + waitingCount: 2, + }); + }); });