From 2871bf4cfd972efec1168e0788263c4124abe3e5 Mon Sep 17 00:00:00 2001 From: wandooadzer-cmyk <“wandooadzer@gmail.com”> Date: Sun, 18 Oct 2026 10:51:02 +0100 Subject: [PATCH] fix: rotate database credentials from Vault without restart Database credentials were loaded from Vault once at startup and baked into the PrismaClient, so rotating them in Vault left the app on stale (and eventually revoked) credentials until a restart. - PrismaService now wraps a swappable PrismaClient (interface merge + Proxy keep the existing API for all callers) and exposes rotateCredentials(): new clients are connected before being swapped in, a rejected credential keeps the current connection, and replaced clients are drained for DATABASE_ROTATION_DRAIN_MS (default 30s) before disconnecting. - New DatabaseCredentialRotationService polls Vault every VAULT_REFRESH_INTERVAL_MS (default 5m, 0 disables) and applies changes to DATABASE_URL / DATABASE_REPLICA_URL; Vault or connect failures are logged and retried on the next tick. - Split fetchVaultSecrets() out of loadVaultSecrets() for runtime reuse. - Validate the new env vars, document rotation in the README, and add tests covering rotation, failure fallback, draining, and polling. Co-Authored-By: Claude Opus 5.5 --- README.md | 10 ++ src/app.module.ts | 17 ++ src/common/secrets/vault-secrets.loader.ts | 73 +++++--- src/config/env.d.ts | 5 + ...tabase-credential-rotation.service.spec.ts | 128 ++++++++++++++ .../database-credential-rotation.service.ts | 107 ++++++++++++ src/prisma/prisma.module.ts | 5 +- src/prisma/prisma.service.spec.ts | 150 ++++++++++++++++ src/prisma/prisma.service.ts | 161 ++++++++++++++++-- 9 files changed, 623 insertions(+), 33 deletions(-) create mode 100644 src/prisma/database-credential-rotation.service.spec.ts create mode 100644 src/prisma/database-credential-rotation.service.ts diff --git a/README.md b/README.md index 2d1bd5b..5f30e65 100644 --- a/README.md +++ b/README.md @@ -320,6 +320,16 @@ If none of the three are set, Vault is skipped. Once any of them is set, the ser On success the loaded key names (never values) are logged. +#### Database credential rotation + +When Vault is configured, the server re-reads the Vault secret every `VAULT_REFRESH_INTERVAL_MS` (default `300000`, i.e. 5 minutes; `0` disables polling). If `DATABASE_URL` or `DATABASE_REPLICA_URL` changed, it reconnects without a restart: + +1. A new Prisma client is built from the new URL and connected **before** anything is swapped. If Vault is unreachable or the database rejects the new credentials, the error is logged, the existing connections stay in use, and the next refresh retries. +2. Once connected, the new client takes over all new queries. +3. The previous client is disconnected after `DATABASE_ROTATION_DRAIN_MS` (default `30000`) so in-flight queries and transactions can finish. Keep the old credentials valid in the database for at least that long after rotating them in Vault. + +Only the database URLs are applied at runtime; other secrets are still read once at startup. + ### OpenTelemetry tracing Tracing is enabled when the OpenTelemetry packages are installed and the following variables are set: diff --git a/src/app.module.ts b/src/app.module.ts index 1a7af93..5d3aee8 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -94,6 +94,23 @@ function validateConfig(config: Record) { } } + // Database credential rotation tuning (see src/prisma/database-credential-rotation.service.ts) + const refreshInterval = config['VAULT_REFRESH_INTERVAL_MS'] as string | undefined; + if (refreshInterval !== undefined && refreshInterval !== '') { + const parsed = Number(refreshInterval); + if (!Number.isInteger(parsed) || parsed < 0 || (parsed > 0 && parsed < 10_000)) { + errors.push('VAULT_REFRESH_INTERVAL_MS must be 0 (disabled) or an integer of at least 10000'); + } + } + + const rotationDrain = config['DATABASE_ROTATION_DRAIN_MS'] as string | undefined; + if (rotationDrain !== undefined && rotationDrain !== '') { + const parsed = Number(rotationDrain); + if (!Number.isInteger(parsed) || parsed < 0) { + errors.push('DATABASE_ROTATION_DRAIN_MS must be a non-negative integer'); + } + } + if (!config['STELLAR_RPC_URL']) { errors.push('STELLAR_RPC_URL is required'); diff --git a/src/common/secrets/vault-secrets.loader.ts b/src/common/secrets/vault-secrets.loader.ts index a5fbbab..02ea940 100644 --- a/src/common/secrets/vault-secrets.loader.ts +++ b/src/common/secrets/vault-secrets.loader.ts @@ -49,9 +49,58 @@ export class VaultSecretsError extends Error { export async function loadVaultSecrets(): Promise { const logger = new Logger('VaultSecrets'); + const secrets = await fetchVaultSecrets(); + if (!secrets) { + return; + } + const vaultPath = process.env['VAULT_KV_PATH']!; + + const loaded: string[] = []; + for (const [key, value] of Object.entries(secrets)) { + process.env[key] = value; + loaded.push(key); + } + + const missingRequired = parseCsv(process.env['VAULT_REQUIRED_KEYS']) + .filter((key) => !isNonEmptyString(process.env[key])); + if (missingRequired.length > 0) { + throw new VaultSecretsError( + `Required secret(s) missing after loading Vault path ${vaultPath}: ${missingRequired.join(', ')}. ` + + 'Add them to the Vault secret or set them as environment variables.', + ); + } + + if (loaded.length === 0) { + logger.warn(`Vault path ${vaultPath} returned no secrets`); + } else { + // Log key names only — never values. + logger.log(`Loaded ${loaded.length} secret(s) from Vault path ${vaultPath}: ${loaded.join(', ')}`); + } +} + +/** + * Whether Vault is configured at all (any of VAULT_ADDR / VAULT_TOKEN / + * VAULT_KV_PATH set). Partial configuration counts as configured so that + * fetchVaultSecrets() reports it. + */ +export function isVaultConfigured(): boolean { + return VAULT_CONFIG_VARS.some((name) => isNonEmptyString(process.env[name])); +} + +/** + * Fetch the KV v2 secret from Vault without touching process.env. Returns + * null when Vault is not configured; otherwise the secret's key/value pairs + * (null/undefined values dropped, everything else stringified). Throws a + * VaultSecretsError describing why they could not be read. + * + * Used both at startup (via loadVaultSecrets) and at runtime by + * DatabaseCredentialRotationService to pick up rotated credentials without a + * restart. + */ +export async function fetchVaultSecrets(): Promise | null> { const configured = VAULT_CONFIG_VARS.filter((name) => isNonEmptyString(process.env[name])); if (configured.length === 0) { - return; + return null; } if (configured.length < VAULT_CONFIG_VARS.length) { const missing = VAULT_CONFIG_VARS.filter((name) => !configured.includes(name)); @@ -117,28 +166,12 @@ export async function loadVaultSecrets(): Promise { ); } - const loaded: string[] = []; + const result: Record = {}; for (const [key, value] of Object.entries(secrets)) { if (value === undefined || value === null) { continue; } - process.env[key] = String(value); - loaded.push(key); - } - - const missingRequired = parseCsv(process.env['VAULT_REQUIRED_KEYS']) - .filter((key) => !isNonEmptyString(process.env[key])); - if (missingRequired.length > 0) { - throw new VaultSecretsError( - `Required secret(s) missing after loading Vault path ${vaultPath}: ${missingRequired.join(', ')}. ` + - 'Add them to the Vault secret or set them as environment variables.', - ); - } - - if (loaded.length === 0) { - logger.warn(`Vault path ${vaultPath} returned no secrets`); - } else { - // Log key names only — never values. - logger.log(`Loaded ${loaded.length} secret(s) from Vault path ${vaultPath}: ${loaded.join(', ')}`); + result[key] = String(value); } + return result; } diff --git a/src/config/env.d.ts b/src/config/env.d.ts index 0e249e1..79b8ee1 100644 --- a/src/config/env.d.ts +++ b/src/config/env.d.ts @@ -46,6 +46,11 @@ export interface EnvironmentVariables { VAULT_ADDR?: string; VAULT_TOKEN?: string; VAULT_KV_PATH?: string; + VAULT_TIMEOUT_MS?: string; + VAULT_REQUIRED_KEYS?: string; + // Database credential rotation (see src/prisma/database-credential-rotation.service.ts) + VAULT_REFRESH_INTERVAL_MS?: string; + DATABASE_ROTATION_DRAIN_MS?: string; OTEL_SERVICE_NAME?: string; OTEL_EXPORTER_OTLP_ENDPOINT?: string; OTEL_EXPORTER_OTLP_HEADERS?: string; diff --git a/src/prisma/database-credential-rotation.service.spec.ts b/src/prisma/database-credential-rotation.service.spec.ts new file mode 100644 index 0000000..3c620ce --- /dev/null +++ b/src/prisma/database-credential-rotation.service.spec.ts @@ -0,0 +1,128 @@ +import { Logger } from '@nestjs/common'; +import { DatabaseCredentialRotationService } from './database-credential-rotation.service'; +import { PrismaService } from './prisma.service'; + +describe('DatabaseCredentialRotationService', () => { + const originalEnv = process.env; + const OLD_URL = 'postgresql://old:old-pass@localhost:5432/db'; + const NEW_URL = 'postgresql://new:new-pass@localhost:5432/db'; + + let fetchMock: jest.Mock; + let prisma: { rotateCredentials: jest.Mock }; + let service: DatabaseCredentialRotationService; + + function vaultReturns(data: Record) { + fetchMock.mockResolvedValue({ + ok: true, + status: 200, + statusText: 'OK', + json: jest.fn().mockResolvedValue({ data: { data } }), + }); + } + + beforeEach(() => { + jest.useFakeTimers(); + process.env = { + ...originalEnv, + VAULT_ADDR: 'https://vault.example.com', + VAULT_TOKEN: 's.token', + VAULT_KV_PATH: 'secret/data/parashield', + DATABASE_URL: OLD_URL, + JWT_SECRET: 'startup-jwt', + }; + delete process.env.VAULT_REFRESH_INTERVAL_MS; + delete process.env.DATABASE_REPLICA_URL; + fetchMock = jest.fn(); + global.fetch = fetchMock as unknown as typeof fetch; + prisma = { rotateCredentials: jest.fn().mockResolvedValue(true) }; + service = new DatabaseCredentialRotationService(prisma as unknown as PrismaService); + jest.spyOn(Logger.prototype, 'log').mockImplementation(() => undefined); + jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined); + }); + + afterEach(() => { + service.onModuleDestroy(); + process.env = originalEnv; + jest.useRealTimers(); + jest.restoreAllMocks(); + }); + + it('applies rotated database credentials from Vault and reconnects', async () => { + vaultReturns({ DATABASE_URL: NEW_URL, JWT_SECRET: 'rotated-jwt' }); + + await expect(service.refresh()).resolves.toBe(true); + + expect(process.env.DATABASE_URL).toBe(NEW_URL); + expect(prisma.rotateCredentials).toHaveBeenCalledTimes(1); + // Only database credentials are applied at runtime. + expect(process.env.JWT_SECRET).toBe('startup-jwt'); + }); + + it('does not reconnect when the credentials are unchanged', async () => { + vaultReturns({ DATABASE_URL: OLD_URL }); + + await expect(service.refresh()).resolves.toBe(false); + expect(prisma.rotateCredentials).not.toHaveBeenCalled(); + }); + + it('keeps current credentials when Vault is unreachable', async () => { + fetchMock.mockRejectedValue(new TypeError('fetch failed')); + + await expect(service.refresh()).resolves.toBe(false); + expect(process.env.DATABASE_URL).toBe(OLD_URL); + expect(prisma.rotateCredentials).not.toHaveBeenCalled(); + expect(Logger.prototype.warn).toHaveBeenCalledWith( + expect.stringContaining('keeping current credentials'), + ); + }); + + it('restores the previous credentials when reconnecting fails, so the next refresh retries', async () => { + vaultReturns({ DATABASE_URL: NEW_URL }); + prisma.rotateCredentials.mockRejectedValueOnce(new Error('P1000: Authentication failed')); + + await expect(service.refresh()).resolves.toBe(false); + expect(process.env.DATABASE_URL).toBe(OLD_URL); + + await expect(service.refresh()).resolves.toBe(true); + expect(process.env.DATABASE_URL).toBe(NEW_URL); + expect(prisma.rotateCredentials).toHaveBeenCalledTimes(2); + }); + + it('polls Vault on the configured interval', async () => { + process.env.VAULT_REFRESH_INTERVAL_MS = '60000'; + vaultReturns({ DATABASE_URL: NEW_URL }); + service.onModuleInit(); + + await jest.advanceTimersByTimeAsync(59_999); + expect(fetchMock).not.toHaveBeenCalled(); + await jest.advanceTimersByTimeAsync(1); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(prisma.rotateCredentials).toHaveBeenCalledTimes(1); + }); + + it('does not poll when Vault is not configured or polling is disabled', async () => { + service.onModuleInit(); + expect(jest.getTimerCount()).toBe(1); + service.onModuleDestroy(); + + process.env.VAULT_REFRESH_INTERVAL_MS = '0'; + service.onModuleInit(); + expect(jest.getTimerCount()).toBe(0); + + delete process.env.VAULT_REFRESH_INTERVAL_MS; + delete process.env.VAULT_ADDR; + delete process.env.VAULT_TOKEN; + delete process.env.VAULT_KV_PATH; + service.onModuleInit(); + expect(jest.getTimerCount()).toBe(0); + }); + + it('shares one refresh between overlapping calls', async () => { + vaultReturns({ DATABASE_URL: NEW_URL }); + + await Promise.all([service.refresh(), service.refresh()]); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(prisma.rotateCredentials).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/prisma/database-credential-rotation.service.ts b/src/prisma/database-credential-rotation.service.ts new file mode 100644 index 0000000..bb5f440 --- /dev/null +++ b/src/prisma/database-credential-rotation.service.ts @@ -0,0 +1,107 @@ +import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; +import { fetchVaultSecrets, isVaultConfigured } from '../common/secrets/vault-secrets.loader'; +import { PrismaService } from './prisma.service'; + +// Keys whose rotation PrismaService knows how to apply without a restart. +// Other secrets are read once at startup by their consumers, so they are +// deliberately not overwritten in process.env at runtime. +const DATABASE_CREDENTIAL_KEYS = ['DATABASE_URL', 'DATABASE_REPLICA_URL'] as const; + +const DEFAULT_REFRESH_INTERVAL_MS = 300_000; + +function getRefreshIntervalMs(): number { + const raw = process.env.VAULT_REFRESH_INTERVAL_MS; + if (raw === undefined || raw.trim() === '') return DEFAULT_REFRESH_INTERVAL_MS; + const parsed = Number(raw); + return Number.isInteger(parsed) && parsed >= 0 ? parsed : DEFAULT_REFRESH_INTERVAL_MS; +} + +/** + * Periodically re-reads the Vault secret loaded at startup and, when the + * database credentials in it have changed, applies them to process.env and + * asks PrismaService to reconnect with them. + * + * Only active when Vault is configured. Runs every VAULT_REFRESH_INTERVAL_MS + * (default 5 minutes, 0 disables polling; refresh() can still be called). + * Failures — Vault unreachable, or the new credentials not yet accepted by the + * database — are logged and the current connections are kept, so a rotation + * hiccup never takes the API down; the next tick retries. + */ +@Injectable() +export class DatabaseCredentialRotationService implements OnModuleInit, OnModuleDestroy { + private readonly logger = new Logger(DatabaseCredentialRotationService.name); + private timer: NodeJS.Timeout | null = null; + private inFlight: Promise | null = null; + + constructor(private readonly prisma: PrismaService) {} + + onModuleInit(): void { + if (!isVaultConfigured()) return; + + const intervalMs = getRefreshIntervalMs(); + if (intervalMs === 0) { + this.logger.log('Vault credential refresh polling disabled (VAULT_REFRESH_INTERVAL_MS=0)'); + return; + } + this.timer = setInterval(() => void this.refresh(), intervalMs); + this.timer.unref?.(); + this.logger.log(`Polling Vault for database credential rotation every ${intervalMs / 1000}s`); + } + + onModuleDestroy(): void { + if (this.timer) { + clearInterval(this.timer); + this.timer = null; + } + } + + /** + * Fetch the current secret from Vault and rotate database connections if + * the credentials changed. Overlapping calls share one refresh. Resolves + * true when connections were swapped; never rejects. + */ + refresh(): Promise { + this.inFlight ??= this.doRefresh().finally(() => { + this.inFlight = null; + }); + return this.inFlight; + } + + private async doRefresh(): Promise { + let secrets: Record | null; + try { + secrets = await fetchVaultSecrets(); + } catch (err) { + this.logger.warn( + `Vault credential refresh failed, keeping current credentials: ${(err as Error).message}`, + ); + return false; + } + if (!secrets) return false; + + const changed = DATABASE_CREDENTIAL_KEYS.filter( + (key) => secrets[key] !== undefined && secrets[key] !== process.env[key], + ); + if (changed.length === 0) return false; + + // Log key names only — never values. + this.logger.log(`Rotated credential(s) detected in Vault: ${changed.join(', ')}`); + const previous = Object.fromEntries(changed.map((key) => [key, process.env[key]])); + for (const key of changed) { + process.env[key] = secrets[key]; + } + + try { + return await this.prisma.rotateCredentials(); + } catch { + // PrismaService already logged why. Restore the old values so + // process.env keeps describing the connections actually in use and the + // next tick sees the change again and retries. + for (const key of changed) { + if (previous[key] === undefined) delete process.env[key]; + else process.env[key] = previous[key]; + } + return false; + } + } +} diff --git a/src/prisma/prisma.module.ts b/src/prisma/prisma.module.ts index 11883c5..9725758 100644 --- a/src/prisma/prisma.module.ts +++ b/src/prisma/prisma.module.ts @@ -1,9 +1,10 @@ import { Global, Module } from '@nestjs/common'; import { PrismaService } from './prisma.service'; +import { DatabaseCredentialRotationService } from './database-credential-rotation.service'; @Global() @Module({ - providers: [PrismaService], - exports: [PrismaService], + providers: [PrismaService, DatabaseCredentialRotationService], + exports: [PrismaService, DatabaseCredentialRotationService], }) export class PrismaModule {} diff --git a/src/prisma/prisma.service.spec.ts b/src/prisma/prisma.service.spec.ts index 8154aaa..6465b12 100644 --- a/src/prisma/prisma.service.spec.ts +++ b/src/prisma/prisma.service.spec.ts @@ -1,3 +1,4 @@ +import { Logger } from '@nestjs/common'; import { PrismaService } from './prisma.service'; describe('PrismaService', () => { @@ -150,3 +151,152 @@ describe('PrismaService', () => { }); }); + +describe('PrismaService credential rotation', () => { + const originalEnv = process.env; + const OLD_URL = 'postgresql://old:old-pass@localhost:5432/db'; + const NEW_URL = 'postgresql://new:new-pass@localhost:5432/db'; + const REPLICA_URL = 'postgresql://ro:ro-pass@replica:5432/db'; + + type FakeClient = { + url: string | undefined; + $connect: jest.Mock; + $disconnect: jest.Mock; + $transaction: jest.Mock; + policy: { findMany: jest.Mock }; + }; + + let clients: FakeClient[]; + + function fakeClient(url: string | undefined): FakeClient { + const client: FakeClient = { + url, + $connect: jest.fn().mockResolvedValue(undefined), + $disconnect: jest.fn().mockResolvedValue(undefined), + $transaction: jest.fn(function (this: FakeClient) { return Promise.resolve(this.url); }), + policy: { findMany: jest.fn().mockResolvedValue([url]) }, + }; + clients.push(client); + return client; + } + + function clientFor(url: string): FakeClient { + return clients.find((c) => c.url === url)!; + } + + beforeEach(() => { + jest.useFakeTimers(); + process.env = { ...originalEnv, DATABASE_URL: OLD_URL }; + delete process.env.DATABASE_REPLICA_URL; + delete process.env.DATABASE_ROTATION_DRAIN_MS; + clients = []; + jest + .spyOn(PrismaService.prototype as any, 'createClient') + .mockImplementation((url) => fakeClient(url as string | undefined)); + jest.spyOn(Logger.prototype, 'log').mockImplementation(() => undefined); + jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined); + jest.spyOn(Logger.prototype, 'error').mockImplementation(() => undefined); + }); + + afterEach(() => { + process.env = originalEnv; + jest.useRealTimers(); + jest.restoreAllMocks(); + }); + + it('forwards PrismaClient calls to the current client, bound to it', async () => { + const service = new PrismaService(); + + await expect(service.policy.findMany()).resolves.toEqual([OLD_URL]); + await expect(service.$transaction([])).resolves.toBe(OLD_URL); + }); + + it('is a no-op when the credentials have not changed', async () => { + const service = new PrismaService(); + + await expect(service.rotateCredentials()).resolves.toBe(false); + expect(clients).toHaveLength(1); + }); + + it('reconnects with the new credentials and drains the old client', async () => { + process.env.DATABASE_ROTATION_DRAIN_MS = '5000'; + const service = new PrismaService(); + const oldClient = clientFor(OLD_URL); + + process.env.DATABASE_URL = NEW_URL; + await expect(service.rotateCredentials()).resolves.toBe(true); + + const newClient = clientFor(NEW_URL); + expect(newClient.$connect).toHaveBeenCalledTimes(1); + await expect(service.policy.findMany()).resolves.toEqual([NEW_URL]); + // Old pool stays open for in-flight work until the drain period ends. + expect(oldClient.$disconnect).not.toHaveBeenCalled(); + + await jest.advanceTimersByTimeAsync(4999); + expect(oldClient.$disconnect).not.toHaveBeenCalled(); + await jest.advanceTimersByTimeAsync(1); + expect(oldClient.$disconnect).toHaveBeenCalledTimes(1); + expect(newClient.$disconnect).not.toHaveBeenCalled(); + }); + + it('keeps the existing connection when the new credentials are rejected, and retries later', async () => { + const service = new PrismaService(); + const oldClient = clientFor(OLD_URL); + process.env.DATABASE_URL = NEW_URL; + + const createClient = PrismaService.prototype['createClient'] as jest.Mock; + createClient.mockImplementationOnce((url: string) => { + const client = fakeClient(url); + client.$connect.mockRejectedValue(new Error('P1000: Authentication failed')); + return client; + }); + + await expect(service.rotateCredentials()).rejects.toThrow('Authentication failed'); + expect(clients[1].$disconnect).toHaveBeenCalledTimes(1); + await expect(service.policy.findMany()).resolves.toEqual([OLD_URL]); + expect(oldClient.$disconnect).not.toHaveBeenCalled(); + + // The database now accepts the new credentials. + await expect(service.rotateCredentials()).resolves.toBe(true); + await expect(service.policy.findMany()).resolves.toEqual([NEW_URL]); + }); + + it('rotates the read replica independently of the primary', async () => { + process.env.DATABASE_REPLICA_URL = REPLICA_URL; + const service = new PrismaService(); + const oldReplica = clientFor(REPLICA_URL); + + const NEW_REPLICA_URL = 'postgresql://ro:rotated@replica:5432/db'; + process.env.DATABASE_REPLICA_URL = NEW_REPLICA_URL; + await expect(service.rotateCredentials()).resolves.toBe(true); + + expect(service.reader).toBe(clientFor(NEW_REPLICA_URL)); + await expect(service.policy.findMany()).resolves.toEqual([OLD_URL]); + await jest.runOnlyPendingTimersAsync(); + expect(oldReplica.$disconnect).toHaveBeenCalledTimes(1); + expect(clientFor(OLD_URL).$disconnect).not.toHaveBeenCalled(); + }); + + it('serializes concurrent rotations so the swap happens once', async () => { + const service = new PrismaService(); + process.env.DATABASE_URL = NEW_URL; + + const results = await Promise.all([service.rotateCredentials(), service.rotateCredentials()]); + + expect(results).toEqual([true, false]); + expect(clients.filter((c) => c.url === NEW_URL)).toHaveLength(1); + }); + + it('disconnects draining clients immediately on shutdown', async () => { + const service = new PrismaService(); + const oldClient = clientFor(OLD_URL); + process.env.DATABASE_URL = NEW_URL; + await service.rotateCredentials(); + + await service.onModuleDestroy(); + + expect(oldClient.$disconnect).toHaveBeenCalledTimes(1); + expect(clientFor(NEW_URL).$disconnect).toHaveBeenCalledTimes(1); + expect(jest.getTimerCount()).toBe(0); + }); +}); diff --git a/src/prisma/prisma.service.ts b/src/prisma/prisma.service.ts index ea33dde..335dd54 100644 --- a/src/prisma/prisma.service.ts +++ b/src/prisma/prisma.service.ts @@ -75,19 +75,53 @@ function withConnectionPoolParams(url: string | undefined): string | undefined { const DEFAULT_QUERY_CACHE_TTL_SECONDS = 30; +// After a credential rotation the previous clients are kept alive for this +// long so in-flight queries and interactive transactions can finish before +// their pools are closed. Override with DATABASE_ROTATION_DRAIN_MS. +const DEFAULT_ROTATION_DRAIN_MS = 30_000; + +function getRotationDrainMs(): number { + const parsed = parseInt(process.env.DATABASE_ROTATION_DRAIN_MS ?? '', 10); + return Number.isInteger(parsed) && parsed >= 0 ? parsed : DEFAULT_ROTATION_DRAIN_MS; +} + interface CacheEntry { value: unknown; expiresAt: number; } +// Database credentials can rotate at runtime (see rotateCredentials), and a +// PrismaClient's datasource URL is fixed once constructed. PrismaService +// therefore wraps a swappable client instead of extending PrismaClient: the +// interface merge below keeps the full PrismaClient API on PrismaService, and +// the Proxy returned from the constructor forwards everything PrismaService +// doesn't define itself to whichever client is current. +// eslint-disable-next-line @typescript-eslint/no-unsafe-declaration-merging +export interface PrismaService extends PrismaClient {} + @Injectable() -export class PrismaService extends PrismaClient implements OnModuleInit, OnModuleDestroy { +// eslint-disable-next-line @typescript-eslint/no-unsafe-declaration-merging +export class PrismaService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(PrismaService.name); + // Primary client; replaced on credential rotation. + private client: PrismaClient; + // #439 — Read replica client. Instantiated only when DATABASE_REPLICA_URL // is set; otherwise reader falls back to the primary so callers need no // conditional logic and existing code paths are unchanged. - private readonly replicaClient: PrismaClient | null; + private replicaClient: PrismaClient | null; + + // Raw URLs the current clients were built from, compared against + // process.env to detect rotated credentials. + private primaryUrl: string | undefined; + private replicaUrl: string | undefined; + + // Serializes rotations so two refreshes never race on the swap. + private rotation: Promise = Promise.resolve(false); + + // Clients replaced by a rotation, waiting out the drain period. + private readonly retiring = new Map(); // In-process TTL cache for repeated read queries (e.g. product lookups, // config-style tables) that are hit frequently but change rarely. Callers @@ -95,19 +129,31 @@ export class PrismaService extends PrismaClient implements OnModuleInit, OnModul // writes and transactions are never affected. private readonly queryCache = new Map(); - public readonly poolConfig: PrismaPoolConfig; + public poolConfig: PrismaPoolConfig; constructor() { - super({ datasourceUrl: withConnectionPoolParams(process.env.DATABASE_URL) }); + this.primaryUrl = process.env.DATABASE_URL; + this.client = this.createClient(this.primaryUrl); + this.poolConfig = getEffectivePoolConfig(this.primaryUrl); - this.poolConfig = getEffectivePoolConfig(process.env.DATABASE_URL); + this.replicaUrl = process.env.DATABASE_REPLICA_URL; + this.replicaClient = this.replicaUrl ? this.createClient(this.replicaUrl) : null; - const replicaUrl = withConnectionPoolParams(process.env.DATABASE_REPLICA_URL); - this.replicaClient = replicaUrl - ? new PrismaClient({ datasourceUrl: replicaUrl }) - : null; + return new Proxy(this, { + get(target, prop, receiver) { + if (prop in target) { + return Reflect.get(target, prop, receiver); + } + const value = Reflect.get(target.client, prop); + return typeof value === 'function' ? value.bind(target.client) : value; + }, + }); } + /** Builds a client for `url` with the pool parameters applied. */ + protected createClient(url: string | undefined): PrismaClient { + return new PrismaClient({ datasourceUrl: withConnectionPoolParams(url) }); + } /** * Returns the read-replica PrismaClient when DATABASE_REPLICA_URL is @@ -124,6 +170,94 @@ export class PrismaService extends PrismaClient implements OnModuleInit, OnModul return this.replicaClient ?? this; } + /** + * Picks up rotated database credentials without a restart. + * + * Re-reads DATABASE_URL and DATABASE_REPLICA_URL from process.env; for each + * one that changed, a new client is built and connected *before* it is + * swapped in, so a bad or not-yet-active credential never replaces a + * working connection — the error is thrown and the current clients stay in + * place (the next rotation attempt retries). Once swapped, new queries use + * the new credentials, while the replaced clients are disconnected only + * after DATABASE_ROTATION_DRAIN_MS (default 30s) so in-flight queries and + * transactions can complete. + * + * Rotations are serialized. Resolves true when any client was swapped. + */ + rotateCredentials(): Promise { + this.rotation = this.rotation + .catch(() => false) + .then(() => this.swapClientsIfCredentialsChanged()); + return this.rotation; + } + + private async swapClientsIfCredentialsChanged(): Promise { + const primaryUrl = process.env.DATABASE_URL; + const replicaUrl = process.env.DATABASE_REPLICA_URL; + const primaryChanged = primaryUrl !== this.primaryUrl; + const replicaChanged = replicaUrl !== this.replicaUrl; + if (!primaryChanged && !replicaChanged) { + return false; + } + + const nextPrimary = primaryChanged ? this.createClient(primaryUrl) : null; + const nextReplica = replicaChanged && replicaUrl ? this.createClient(replicaUrl) : null; + try { + await nextPrimary?.$connect(); + await nextReplica?.$connect(); + } catch (err) { + await this.disconnectQuietly([nextPrimary, nextReplica]); + const errorMsg = err instanceof Error ? err.message : String(err); + this.logger.error( + `Database credential rotation failed, keeping existing connections: ${errorMsg}`, + ); + throw err; + } + + const retired: PrismaClient[] = []; + const rotated: string[] = []; + if (nextPrimary) { + retired.push(this.client); + this.client = nextPrimary; + this.primaryUrl = primaryUrl; + this.poolConfig = getEffectivePoolConfig(primaryUrl); + rotated.push('primary'); + } + if (replicaChanged) { + if (this.replicaClient) retired.push(this.replicaClient); + this.replicaClient = nextReplica; + this.replicaUrl = replicaUrl; + rotated.push('replica'); + } + + const drainMs = getRotationDrainMs(); + const timer = setTimeout(() => { + this.retiring.delete(timer); + void this.disconnectQuietly(retired); + }, drainMs); + timer.unref?.(); + this.retiring.set(timer, retired); + + this.logger.log( + `Database credentials rotated (${rotated.join(', ')}); ` + + `previous connections close in ${drainMs / 1000}s`, + ); + return true; + } + + private async disconnectQuietly(clients: Array): Promise { + await Promise.all(clients.map(async (client) => { + if (!client) return; + try { + await client.$disconnect(); + } catch (err) { + this.logger.warn( + `Failed to disconnect retired database client: ${err instanceof Error ? err.message : String(err)}`, + ); + } + })); + } + async onModuleInit() { // #696 — Handle SIGTERM / SIGINT signals gracefully process.once('SIGTERM', async () => { @@ -168,6 +302,11 @@ export class PrismaService extends PrismaClient implements OnModuleInit, OnModul } async onModuleDestroy() { + const retiring = [...this.retiring.entries()]; + this.retiring.clear(); + for (const [timer] of retiring) clearTimeout(timer); + await this.disconnectQuietly(retiring.flatMap(([, clients]) => clients)); + await this.$disconnect(); if (this.replicaClient) { await this.replicaClient.$disconnect(); @@ -233,8 +372,8 @@ export class PrismaService extends PrismaClient implements OnModuleInit, OnModul */ async getMetrics(): Promise { try { - if (typeof (this as any).$metrics?.json === 'function') { - return await (this as any).$metrics.json(); + if (typeof (this.client as any).$metrics?.json === 'function') { + return await (this.client as any).$metrics.json(); } } catch (err) { this.logger.warn(`Failed to collect Prisma metrics: ${err instanceof Error ? err.message : String(err)}`);