diff --git a/src/config/database.config.ts b/src/config/database.config.ts index f87271e..63132d3 100644 --- a/src/config/database.config.ts +++ b/src/config/database.config.ts @@ -18,23 +18,15 @@ import { databaseEnvSchema, validateEnv } from './env.validation'; */ export type DatabaseConfig = { url: string; - connectionLimit: number; - workerConnectionLimit: number; - poolTimeoutMs: number; - queryTimeoutMs: number; - statementTimeoutMs: number; - workerQueryTimeoutMs: number; + slowQueryThresholdMs: number; + enableSlowQueryLogging: boolean; }; export const databaseConfig = registerAs('database', (): DatabaseConfig => { const env = validateEnv(databaseEnvSchema, process.env); return { url: env.DATABASE_URL, - connectionLimit: env.DATABASE_CONNECTION_LIMIT, - workerConnectionLimit: env.DATABASE_WORKER_CONNECTION_LIMIT, - poolTimeoutMs: env.DATABASE_POOL_TIMEOUT_MS, - queryTimeoutMs: env.DATABASE_QUERY_TIMEOUT_MS, - statementTimeoutMs: env.DATABASE_STATEMENT_TIMEOUT_MS, - workerQueryTimeoutMs: env.DATABASE_WORKER_QUERY_TIMEOUT_MS, + slowQueryThresholdMs: env.SLOW_QUERY_THRESHOLD_MS, + enableSlowQueryLogging: env.ENABLE_SLOW_QUERY_LOGGING, }; }); diff --git a/src/config/env.validation.ts b/src/config/env.validation.ts index 33be878..ea190f9 100644 --- a/src/config/env.validation.ts +++ b/src/config/env.validation.ts @@ -19,23 +19,11 @@ export const appEnvSchema = z.object({ export const databaseEnvSchema = z.object({ DATABASE_URL: z.string().min(1, 'DATABASE_URL is required'), - // Connection pool sizing — applied as Prisma `connection_limit` URL params. - DATABASE_CONNECTION_LIMIT: z.coerce.number().int().positive().default(10), - // Worker pool stays small: background jobs must never starve API traffic. - DATABASE_WORKER_CONNECTION_LIMIT: z.coerce.number().int().positive().default(3), - // How long a query waits for a free connection before failing fast (ms). - // 0 waits indefinitely (Prisma `pool_timeout` semantics). - DATABASE_POOL_TIMEOUT_MS: z.coerce.number().int().nonnegative().default(5000), - // Client-side guard: fails the promise fast when a query exceeds this (ms). - // 0 disables the client-side race (server-side statement_timeout still applies). - DATABASE_QUERY_TIMEOUT_MS: z.coerce.number().int().nonnegative().default(5000), - // Server-side `statement_timeout` (ms) — Postgres aborts the runaway query so - // the pooled connection is actually released. 0 disables the guard. - DATABASE_STATEMENT_TIMEOUT_MS: z.coerce.number().int().nonnegative().default(10000), - // Extended client-side timeout for the dedicated worker pool (ms). Long-running - // worker transactions (rollups, outbox drains) must not be killed by the API - // guard; 0 disables the worker guard entirely. - DATABASE_WORKER_QUERY_TIMEOUT_MS: z.coerce.number().int().nonnegative().default(60000), + SLOW_QUERY_THRESHOLD_MS: z.coerce.number().int().nonnegative().default(250), + ENABLE_SLOW_QUERY_LOGGING: z + .enum(['true', 'false']) + .default('true') + .transform((value) => value === 'true'), }); export const redisEnvSchema = z.object({ diff --git a/src/database/prisma.service.ts b/src/database/prisma.service.ts index cbbea16..2d36aca 100644 --- a/src/database/prisma.service.ts +++ b/src/database/prisma.service.ts @@ -1,14 +1,7 @@ -import { INestApplication, Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; +import { INestApplication, Injectable, Logger, OnModuleDestroy, OnModuleInit, Optional } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import { PrismaClient } from '@prisma/client'; -import { DatabaseConfig } from '../config/database.config'; -import { buildDatasourceUrl } from './datasource-url'; -import { createQueryTimeoutExtension } from './query-timeout.extension'; -import { - checkMigrationStatus, - getDefaultMigrationsDir, - MigrationCheckResult, -} from './migration-checker'; +import { createSlowQueryMiddleware } from './slow-query.logger'; /** * The single Prisma client for the application. Manages connection lifecycle @@ -31,25 +24,7 @@ import { export class PrismaService extends PrismaClient implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(PrismaService.name); - /** - * Dedicated client for background workers. It uses its own (smaller) pool - * sized by `DATABASE_WORKER_CONNECTION_LIMIT`, carries no server-side - * `statement_timeout`, and enforces the much longer - * `DATABASE_WORKER_QUERY_TIMEOUT_MS` guard. Workers that run long - * transactions (rollups, outbox drains, webhook persistence) should use this - * client so their work is never aborted by API request timeouts. - */ - readonly workerClient: PrismaClient; - - constructor(configService: ConfigService) { - const database = configService.getOrThrow('database'); - - const url = buildDatasourceUrl(database.url, { - connectionLimit: database.connectionLimit, - poolTimeoutMs: database.poolTimeoutMs, - statementTimeoutMs: database.statementTimeoutMs, - }); - + constructor(@Optional() private readonly configService?: ConfigService) { super({ datasources: { db: { url } }, log: [ @@ -58,41 +33,16 @@ export class PrismaService extends PrismaClient implements OnModuleInit, OnModul ], }); - // Inject the timeout-guard extension into this (API) client. `$extends` - // returns a new client; copying its delegates onto `this` keeps the - // PrismaService identity every repository already depends on. The cast is - // required because the generated `$extends` return type is a dynamic - // extension type rather than a full `PrismaClient`. - Object.assign( - this, - this.$extends( - createQueryTimeoutExtension({ - queryTimeoutMs: database.queryTimeoutMs, - poolTimeoutMs: database.poolTimeoutMs, - }), - ) as unknown as PrismaClient, - ); + const thresholdMs = this.configService?.get('database.slowQueryThresholdMs') ?? 250; + const enabled = this.configService?.get('database.enableSlowQueryLogging') ?? true; - // Dedicated worker pool: smaller, extended timeout, no statement_timeout. - const workerUrl = buildDatasourceUrl(database.url, { - connectionLimit: database.workerConnectionLimit, - poolTimeoutMs: database.poolTimeoutMs, - statementTimeoutMs: 0, - }); - // Same cast rationale as above: the generated `$extends` return type is a - // dynamic extension type, not a full `PrismaClient`. - this.workerClient = new PrismaClient({ - datasources: { db: { url: workerUrl } }, - log: [ - { level: 'warn', emit: 'event' }, - { level: 'error', emit: 'event' }, - ], - }).$extends( - createQueryTimeoutExtension({ - queryTimeoutMs: database.workerQueryTimeoutMs, - poolTimeoutMs: database.poolTimeoutMs, + this.$use( + createSlowQueryMiddleware({ + thresholdMs, + enabled, + logger: this.logger, }), - ) as unknown as PrismaClient; + ); } async onModuleInit(): Promise { diff --git a/src/database/slow-query.logger.spec.ts b/src/database/slow-query.logger.spec.ts new file mode 100644 index 0000000..d99db91 --- /dev/null +++ b/src/database/slow-query.logger.spec.ts @@ -0,0 +1,233 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { Logger } from '@nestjs/common'; +import { + createSlowQueryMiddleware, + sanitizeQueryArgs, + analyzeIndexSuggestions, + SlowQueryReport, +} from './slow-query.logger'; + +describe('Slow Query Logger & Index Analyzer', () => { + describe('sanitizeQueryArgs', () => { + it('redacts sensitive fields like passwords, secrets, tokens, and private keys', () => { + const input = { + where: { + email: 'agent@stellar.org', + passwordHash: '$argon2id$v=19$m=65536,t=3,p=4$secretpass', + apiSecret: 'sk_live_123456789', + sessionToken: 'jwt-token-value', + wallet: { + publicKey: 'GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFXY', + privateKey: 'SBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFXY', + }, + }, + data: { + status: 'ACTIVE', + authSignature: '0x123abc', + }, + }; + + const sanitized = sanitizeQueryArgs(input) as any; + + expect(sanitized.where.email).toBe('agent@stellar.org'); + expect(sanitized.where.passwordHash).toBe('[REDACTED]'); + expect(sanitized.where.apiSecret).toBe('[REDACTED]'); + expect(sanitized.where.sessionToken).toBe('[REDACTED]'); + expect(sanitized.where.wallet.publicKey).toBe('GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFXY'); + expect(sanitized.where.wallet.privateKey).toBe('[REDACTED]'); + expect(sanitized.data.status).toBe('ACTIVE'); + expect(sanitized.data.authSignature).toBe('[REDACTED]'); + }); + + it('handles arrays and primitives safely', () => { + expect(sanitizeQueryArgs(null)).toBeNull(); + expect(sanitizeQueryArgs(undefined)).toBeUndefined(); + expect(sanitizeQueryArgs(42)).toBe(42); + expect(sanitizeQueryArgs(true)).toBe(true); + + const arrayInput = [ + { id: 1, secretKey: 'top-secret' }, + { id: 2, secretKey: 'another-secret' }, + ]; + const sanitizedArray = sanitizeQueryArgs(arrayInput) as any[]; + expect(sanitizedArray[0].id).toBe(1); + expect(sanitizedArray[0].secretKey).toBe('[REDACTED]'); + expect(sanitizedArray[1].id).toBe(2); + expect(sanitizedArray[1].secretKey).toBe('[REDACTED]'); + }); + }); + + describe('analyzeIndexSuggestions', () => { + it('suggests composite indexes when filtering by multiple columns', () => { + const suggestions = analyzeIndexSuggestions('Transaction', 'findMany', { + where: { + walletAddress: 'GBRP...', + status: 'PENDING', + assetCode: 'USDC', + }, + }); + + expect(suggestions.length).toBeGreaterThan(0); + expect(suggestions[0]).toContain("Consider a composite index on model 'Transaction' for fields: [walletAddress, status, assetCode]"); + }); + + it('suggests compound index when filtering and sorting', () => { + const suggestions = analyzeIndexSuggestions('AuditLog', 'findMany', { + where: { + agentId: 'agent-123', + }, + orderBy: { + createdAt: 'desc', + }, + }); + + expect(suggestions.some((s) => s.includes('Consider compound index'))).toBe(true); + }); + + it('suggests pagination when unbounded findMany is executed', () => { + const suggestions = analyzeIndexSuggestions('Agent', 'findMany', { + where: { status: 'ACTIVE' }, + }); + + expect(suggestions.some((s) => s.includes('Unbounded findMany'))).toBe(true); + }); + }); + + describe('createSlowQueryMiddleware', () => { + let mockLogger: Logger; + + beforeEach(() => { + mockLogger = { + warn: vi.fn(), + error: vi.fn(), + log: vi.fn(), + } as unknown as Logger; + }); + + it('does not emit warning when query executes faster than threshold', async () => { + const middleware = createSlowQueryMiddleware({ + thresholdMs: 100, + logger: mockLogger, + }); + + const next = vi.fn().mockImplementation(async () => { + // Fast execution + return [{ id: 1 }]; + }); + + const result = await middleware( + { + model: 'User', + action: 'findUnique', + args: { where: { id: 1 } }, + dataPath: [], + runInTransaction: false, + }, + next, + ); + + expect(result).toEqual([{ id: 1 }]); + expect(mockLogger.warn).not.toHaveBeenCalled(); + }); + + it('emits structured warning when query execution exceeds threshold', async () => { + let reported: SlowQueryReport | null = null; + + const middleware = createSlowQueryMiddleware({ + thresholdMs: 50, + logger: mockLogger, + onSlowQuery: (report) => { + reported = report; + }, + }); + + const next = vi.fn().mockImplementation(async () => { + // Artificial delay exceeding 50ms + await new Promise((resolve) => setTimeout(resolve, 60)); + return { id: 'agent-123', name: 'StellarBot' }; + }); + + const result = await middleware( + { + model: 'Agent', + action: 'findMany', + args: { + where: { + status: 'ACTIVE', + apiKeySecret: 'secret_key_12345', + }, + }, + dataPath: [], + runInTransaction: false, + }, + next, + ); + + expect(result).toEqual({ id: 'agent-123', name: 'StellarBot' }); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + expect(reported).not.toBeNull(); + expect(reported?.model).toBe('Agent'); + expect(reported?.action).toBe('findMany'); + expect(reported?.durationMs).toBeGreaterThanOrEqual(50); + expect(reported?.args?.where).toEqual({ + status: 'ACTIVE', + apiKeySecret: '[REDACTED]', + }); + expect(reported?.indexRecommendations?.length).toBeGreaterThan(0); + }); + + it('does not log when disabled', async () => { + const middleware = createSlowQueryMiddleware({ + thresholdMs: 10, + enabled: false, + logger: mockLogger, + }); + + const next = vi.fn().mockImplementation(async () => { + await new Promise((resolve) => setTimeout(resolve, 20)); + return true; + }); + + const result = await middleware( + { + model: 'Wallet', + action: 'findFirst', + args: {}, + dataPath: [], + runInTransaction: false, + }, + next, + ); + + expect(result).toBe(true); + expect(mockLogger.warn).not.toHaveBeenCalled(); + }); + + it('propagates errors while still capturing timing on slow failed queries', async () => { + const middleware = createSlowQueryMiddleware({ + thresholdMs: 30, + logger: mockLogger, + }); + + const next = vi.fn().mockImplementation(async () => { + await new Promise((resolve) => setTimeout(resolve, 40)); + throw new Error('Database connection timeout'); + }); + + await expect( + middleware( + { + model: 'Transaction', + action: 'create', + args: { data: { amount: 100 } }, + dataPath: [], + runInTransaction: false, + }, + next, + ), + ).rejects.toThrow('Database connection timeout'); + + expect(mockLogger.warn).toHaveBeenCalled(); + }); + }); +}); diff --git a/src/database/slow-query.logger.ts b/src/database/slow-query.logger.ts new file mode 100644 index 0000000..865690a --- /dev/null +++ b/src/database/slow-query.logger.ts @@ -0,0 +1,173 @@ +import { Logger } from '@nestjs/common'; +import { Prisma } from '@prisma/client'; + +export interface SlowQueryLoggerOptions { + thresholdMs?: number; + enabled?: boolean; + logger?: Logger; + onSlowQuery?: (report: SlowQueryReport) => void; +} + +export interface SlowQueryReport { + model?: string; + action: string; + durationMs: number; + thresholdMs: number; + args?: Record; + indexRecommendations?: string[]; + timestamp: string; +} + +const SENSITIVE_KEY_PATTERNS = [ + /password/i, + /secret/i, + /token/i, + /credential/i, + /privatekey/i, + /private_key/i, + /seed/i, + /passphrase/i, + /auth/i, + /signature/i, + /hash/i, +]; + +/** + * Recursively sanitizes query arguments to remove or mask sensitive parameters + * before emitting them to log aggregators or console output. + */ +export function sanitizeQueryArgs(value: unknown): unknown { + if (value === null || value === undefined) { + return value; + } + + if (typeof value === 'string') { + return value.length > 500 ? `${value.substring(0, 500)}...[TRUNCATED]` : value; + } + + if (typeof value === 'number' || typeof value === 'boolean' || typeof value === 'bigint') { + return value; + } + + if (value instanceof Date) { + return value.toISOString(); + } + + if (Array.isArray(value)) { + return value.map((item) => sanitizeQueryArgs(item)); + } + + if (typeof value === 'object') { + const sanitizedObj: Record = {}; + for (const [key, val] of Object.entries(value)) { + const isSensitive = SENSITIVE_KEY_PATTERNS.some((pattern) => pattern.test(key)); + if (isSensitive) { + sanitizedObj[key] = '[REDACTED]'; + } else { + sanitizedObj[key] = sanitizeQueryArgs(val); + } + } + return sanitizedObj; + } + + return String(value); +} + +/** + * Analyzes query structure (filters, sorting, joins) to suggest potential + * database index optimizations for slow queries. + */ +export function analyzeIndexSuggestions( + model?: string, + action?: string, + args?: Record, +): string[] { + const suggestions: string[] = []; + + if (!model || !args) { + return suggestions; + } + + const where = args['where'] as Record | undefined; + const orderBy = args['orderBy'] as Record | Array> | undefined; + + if (where && typeof where === 'object') { + const filterKeys = Object.keys(where).filter((k) => !['AND', 'OR', 'NOT'].includes(k)); + if (filterKeys.length > 1) { + suggestions.push( + `Consider a composite index on model '${model}' for fields: [${filterKeys.join(', ')}]`, + ); + } else if (filterKeys.length === 1) { + suggestions.push(`Verify single-column index on model '${model}' field: '${filterKeys[0]}'`); + } + + if (orderBy) { + const orderKeys = Array.isArray(orderBy) + ? orderBy.flatMap((o) => Object.keys(o)) + : Object.keys(orderBy); + if (orderKeys.length > 0 && filterKeys.length > 0) { + suggestions.push( + `Consider compound index on model '${model}' combining filters [${filterKeys.join(', ')}] and sort [${orderKeys.join(', ')}]`, + ); + } + } + } + + if (action === 'findMany' && !args['take'] && !args['cursor']) { + suggestions.push( + `Unbounded findMany on model '${model}': add limit/take pagination or keyset cursor to reduce scan volume`, + ); + } + + return suggestions; +} + +/** + * Creates a Prisma middleware handler that intercepts and times query execution, + * emitting structured warning logs whenever execution exceeds the configured threshold. + */ +export function createSlowQueryMiddleware(options: SlowQueryLoggerOptions = {}): Prisma.Middleware { + const thresholdMs = options.thresholdMs ?? 250; + const enabled = options.enabled ?? true; + const logger = options.logger ?? new Logger('PrismaSlowQuery'); + + return async (params: Prisma.MiddlewareParams, next: (params: Prisma.MiddlewareParams) => Promise) => { + if (!enabled) { + return next(params); + } + + const startTime = Date.now(); + try { + return await next(params); + } finally { + const durationMs = Date.now() - startTime; + if (durationMs >= thresholdMs) { + const sanitizedArgs = sanitizeQueryArgs(params.args) as Record | undefined; + const indexRecommendations = analyzeIndexSuggestions(params.model, params.action, sanitizedArgs); + + const report: SlowQueryReport = { + model: params.model, + action: params.action, + durationMs, + thresholdMs, + args: sanitizedArgs, + indexRecommendations, + timestamp: new Date().toISOString(), + }; + + logger.warn( + `[Slow Query] ${params.model ? `${params.model}.${params.action}` : params.action} took ${durationMs}ms (threshold: ${thresholdMs}ms)`, + JSON.stringify(report), + ); + + if (options.onSlowQuery) { + try { + options.onSlowQuery(report); + } catch (callbackErr) { + logger.error(`Error in onSlowQuery callback: ${(callbackErr as Error).message}`); + } + } + } + } + }; +}