diff --git a/.changeset/structured-logging.md b/.changeset/structured-logging.md new file mode 100644 index 00000000..7787fe7e --- /dev/null +++ b/.changeset/structured-logging.md @@ -0,0 +1,21 @@ +--- +'lens': patch +--- + +Route all logging through pino (`src/logger.ts`) instead of `console.*`. Every +module now logs through a component-scoped child logger — `ingester:sdex`, +`job:aggregate-refresh`, `ws:fanout`, `redis`, `config`, `webhook`, +`route:price`, … — emitted as a structured `component` field rather than a +message prefix, so operators can filter by component without regexing text. +Fastify's request logger is built from the same options, so `req.log` carries +the same redaction and formatting. + +The level is configurable via `LOG_LEVEL` (default `info`, documented in +`.env.example`): logs are newline-delimited JSON when `NODE_ENV=production` and +pretty-printed for humans otherwise. Values of secret-looking environment +variables (`DATABASE_URL`, `DIRECT_DATABASE_URL`, `REDIS_URL`, and any name +containing `KEY`/`SECRET`/`TOKEN`/`PASSWORD`) are scrubbed from every +serialized line, including the password inside a connection string and driver +error messages that echo it, while request auth headers are redacted via +pino's `redact`. A test asserts no `console.*` survives in non-test source and +that a connection string or API key value never reaches the output. diff --git a/.env.example b/.env.example index 41e4465f..8e5a1cd8 100644 --- a/.env.example +++ b/.env.example @@ -1,6 +1,9 @@ # --- General --- # Environment: development, test, production NODE_ENV=development +# Structured log level: trace, debug, info, warn, error, fatal, or silent (default: info) +# Logs are newline-delimited JSON when NODE_ENV=production, human-readable otherwise. +LOG_LEVEL=info # --- Server --- # Port to listen on (default: 3002) diff --git a/package-lock.json b/package-lock.json index 59c3e589..bf123327 100644 --- a/package-lock.json +++ b/package-lock.json @@ -26,6 +26,7 @@ "ioredis": "^5.0.0", "mercurius": "^16.9.0", "pg": "^8.0.0", + "pino": "^9.14.0", "prom-client": "^15.1.3", "uuid": "^9.0.0", "ws": "^8.20.0", diff --git a/package.json b/package.json index c25cfd9e..5ce9baf5 100644 --- a/package.json +++ b/package.json @@ -40,6 +40,7 @@ "ioredis": "^5.0.0", "mercurius": "^16.9.0", "pg": "^8.0.0", + "pino": "^9.14.0", "prom-client": "^15.1.3", "uuid": "^9.0.0", "ws": "^8.20.0", diff --git a/src/__tests__/logger.test.ts b/src/__tests__/logger.test.ts new file mode 100644 index 00000000..be8ba3e3 --- /dev/null +++ b/src/__tests__/logger.test.ts @@ -0,0 +1,161 @@ +import { describe, it, expect } from 'vitest' +import { readFileSync, readdirSync } from 'node:fs' +import { resolve, join, relative } from 'node:path' +import type { DestinationStream } from 'pino' +import { buildLogger, redactSecrets, secretValues } from '../logger' + +const rootDir = resolve(__dirname, '../..') +const srcDir = resolve(rootDir, 'src') + +/** Collects serialized log lines in memory instead of writing to stdout. */ +function capture(): { lines: string[]; destination: DestinationStream } { + const lines: string[] = [] + return { lines, destination: { write: (line: string) => { lines.push(line) } } } +} + +/** Every non-test .ts file under src/. */ +function sourceFiles(dir = srcDir): string[] { + const files: string[] = [] + for (const entry of readdirSync(dir, { withFileTypes: true })) { + const full = join(dir, entry.name) + if (entry.isDirectory()) { + if (entry.name !== '__tests__' && entry.name !== 'node_modules') { + files.push(...sourceFiles(full)) + } + } else if ( + entry.isFile() && + entry.name.endsWith('.ts') && + !entry.name.endsWith('.test.ts') && + !entry.name.endsWith('.spec.ts') && + !entry.name.endsWith('.d.ts') + ) { + files.push(full) + } + } + return files +} + +describe('secret redaction', () => { + it('scrubs a connection string and an API key value', () => { + const dbUrl = 'postgresql://lens:CANARY-PASSWORD-123456@localhost:5432/lens' + const apiKey = 'canary-api-key-value-0123456789' + const env = { DATABASE_URL: dbUrl, ADMIN_API_KEY: apiKey } + + const line = `oops ${dbUrl} and ${apiKey} leaked` + const scrubbed = redactSecrets(line, env) + + expect(scrubbed).not.toContain('CANARY-PASSWORD-123456') + expect(scrubbed).not.toContain(apiKey) + expect(scrubbed).toContain('[redacted]') + }) + + it('also scrubs the password embedded in a connection string on its own', () => { + const dbUrl = 'postgresql://lens:S0me-CANARY-Passw0rd-long@db:5432/lens' + const env = { DATABASE_URL: dbUrl } + + expect(secretValues(env)).toContain('S0me-CANARY-Passw0rd-long') + expect(redactSecrets('driver said password=S0me-CANARY-Passw0rd-long', env)).not.toContain( + 'S0me-CANARY-Passw0rd-long' + ) + }) + + it('ignores boolean-looking flag values so ordinary log content survives', () => { + const env = { REQUIRE_API_KEY: 'true' } + expect(secretValues(env)).toEqual([]) + expect(redactSecrets('auth enabled: true', env)).toBe('auth enabled: true') + }) +}) + +describe('structured logger output', () => { + it('emits JSON with a component field and a level label', () => { + const { lines, destination } = capture() + const log = buildLogger({ level: 'info', pretty: false, destination }).child({ + component: 'ingester:amm', + }) + + log.info({ pool: 'CABCDEF', trades: 3 }, 'ingested pool trades') + + expect(lines).toHaveLength(1) + const parsed = JSON.parse(lines[0]) + expect(parsed.component).toBe('ingester:amm') + expect(parsed.level).toBe('info') + expect(parsed.msg).toBe('ingested pool trades') + expect(parsed.pool).toBe('CABCDEF') + expect(parsed.trades).toBe(3) + expect(parsed.service).toBe('lens') + }) + + it('never writes a secret value carried in an error or a field', () => { + const dbUrl = 'postgresql://lens:CANARY-PASSWORD-123456@localhost:5432/lens' + const apiKey = 'canary-api-key-value-0123456789' + process.env.DATABASE_URL = dbUrl + process.env.ADMIN_API_KEY = apiKey + + const { lines, destination } = capture() + const log = buildLogger({ level: 'info', pretty: false, destination }) + + log.error({ err: new Error(`connect failed for ${dbUrl}`) }, 'twap failed') + log.info({ apiKey }, 'debug key') + + const output = lines.join('\n') + expect(output).not.toContain('CANARY-PASSWORD-123456') + expect(output).not.toContain(apiKey) + expect(output).toContain('[redacted]') + }) + + it('renders human-readable lines in development mode', () => { + const { lines, destination } = capture() + const log = buildLogger({ level: 'info', pretty: true, destination }).child({ + component: 'job:snapshot-retention', + }) + + log.info({ network: 'testnet', pruned: 4 }, 'pruned snapshots') + + expect(lines).toHaveLength(1) + expect(lines[0]).toContain('[job:snapshot-retention]') + expect(lines[0]).toContain('pruned snapshots') + expect(lines[0]).toContain('INFO') + expect(() => JSON.parse(lines[0])).toThrow() + }) + + it('respects the level and drops messages below it', () => { + const { lines, destination } = capture() + const log = buildLogger({ level: 'warn', pretty: false, destination }) + + log.info('not emitted') + log.warn('emitted') + + expect(lines).toHaveLength(1) + expect(JSON.parse(lines[0]).msg).toBe('emitted') + }) +}) + +describe('logging hygiene', () => { + it('leaves no console.* calls in non-test source', () => { + const offenders: string[] = [] + for (const file of sourceFiles()) { + const code = readFileSync(file, 'utf8') + code.split('\n').forEach((line, i) => { + if (/\bconsole\.\w+\s*\(/.test(line)) { + offenders.push(`${relative(rootDir, file)}:${i + 1}`) + } + }) + } + expect(offenders).toEqual([]) + }) + + it('never passes a raw connection-string env value into a log call', () => { + const logCall = /\b(?:log|logger|req\.log)\.(?:trace|debug|info|warn|error|fatal)\b/ + const offenders: string[] = [] + for (const file of sourceFiles()) { + const code = readFileSync(file, 'utf8') + code.split('\n').forEach((line, i) => { + if (!logCall.test(line)) return + if (line.includes('process.env.DATABASE_URL') || line.includes('process.env.DIRECT_DATABASE_URL')) { + offenders.push(`${relative(rootDir, file)}:${i + 1}`) + } + }) + } + expect(offenders).toEqual([]) + }) +}) diff --git a/src/__tests__/networkVenueConfig.test.ts b/src/__tests__/networkVenueConfig.test.ts index 94fb0276..81b45b42 100644 --- a/src/__tests__/networkVenueConfig.test.ts +++ b/src/__tests__/networkVenueConfig.test.ts @@ -8,6 +8,27 @@ import { StrKey } from '@stellar/stellar-sdk' +// config.ts logs invalid contract ids through the structured logger; capture +// those calls so the tests can assert on the warning without a live sink. +const { configLogger } = vi.hoisted(() => ({ + configLogger: { + trace: vi.fn(), + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + fatal: vi.fn(), + child: vi.fn(), + }, +})) + +vi.mock('../logger', () => ({ + createLogger: () => configLogger, + logger: configLogger, + loggerOptions: () => ({}), + createDestination: () => ({ write: () => {} }), +})) + const ENV_KEYS = [ 'STELLAR_NETWORK', 'SOROSWAP_ENABLED_TESTNET', @@ -163,27 +184,31 @@ describe('per-network venue config', () => { }) it('rejects the former 55-character mainnet Reflector default, disabling the oracle and naming the env var', async () => { - const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + configLogger.warn.mockClear() process.env.REFLECTOR_CONTRACT_ID_MAINNET = OLD_REFLECTOR_LITERAL const { getNetworkConfig } = await loadConfig() expect(OLD_REFLECTOR_LITERAL).toHaveLength(55) expect(getNetworkConfig('mainnet').oracle.enabled).toBe(false) - expect(warn).toHaveBeenCalledWith(expect.stringContaining('REFLECTOR_CONTRACT_ID_MAINNET')) - warn.mockRestore() + expect(configLogger.warn).toHaveBeenCalledWith( + expect.objectContaining({ envVar: 'REFLECTOR_CONTRACT_ID_MAINNET', value: OLD_REFLECTOR_LITERAL }), + expect.stringContaining('invalid Soroban contract id') + ) }) it('rejects the former 55-character testnet Soroswap default, disabling Soroswap and naming the env var', async () => { - const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + configLogger.warn.mockClear() process.env.SOROSWAP_FACTORY_ADDRESS_TESTNET = OLD_SOROSWAP_TESTNET_LITERAL const { getNetworkConfig } = await loadConfig() expect(OLD_SOROSWAP_TESTNET_LITERAL).toHaveLength(55) expect(getNetworkConfig('testnet').soroswap.enabled).toBe(false) - expect(warn).toHaveBeenCalledWith(expect.stringContaining('SOROSWAP_FACTORY_ADDRESS_TESTNET')) - warn.mockRestore() + expect(configLogger.warn).toHaveBeenCalledWith( + expect.objectContaining({ envVar: 'SOROSWAP_FACTORY_ADDRESS_TESTNET' }), + expect.stringContaining('invalid Soroban contract id') + ) }) it('accepts a valid 56-character Soroswap factory id', async () => { diff --git a/src/__tests__/snapshotRetention.test.ts b/src/__tests__/snapshotRetention.test.ts index b333e65c..6753ea1f 100644 --- a/src/__tests__/snapshotRetention.test.ts +++ b/src/__tests__/snapshotRetention.test.ts @@ -14,6 +14,27 @@ vi.mock('bullmq', () => ({ Worker: class {}, })) +// The retention job logs batch progress through the structured logger; capture +// those calls without touching stdout. +const { retentionLogger } = vi.hoisted(() => ({ + retentionLogger: { + trace: vi.fn(), + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + fatal: vi.fn(), + child: vi.fn(), + }, +})) + +vi.mock('../logger', () => ({ + createLogger: () => retentionLogger, + logger: retentionLogger, + loggerOptions: () => ({}), + createDestination: () => ({ write: () => {} }), +})) + import { pruneOldSnapshots, pruneAllNetworks, @@ -81,21 +102,20 @@ describe('pruneOldSnapshots network argument', () => { .mockResolvedValueOnce({ rowCount: 2 }) .mockResolvedValueOnce({ rowCount: 1 }) .mockResolvedValueOnce({ rowCount: 0 }) - const log = vi.spyOn(console, 'log').mockImplementation(() => {}) + retentionLogger.info.mockClear() const onProgress = vi.fn() - try { - expect(await pruneOldSnapshots(14, 'mainnet', 2, onProgress)).toBe(3) - expect(mockQuery.mock.calls.map(call => call[1])).toEqual([ - ['mainnet', 14, 2], - ['mainnet', 14, 2], - ['mainnet', 14, 2], - ]) - expect(log).toHaveBeenCalledWith(expect.stringContaining('batch 2 (1 rows; 3 total)')) - expect(onProgress).toHaveBeenLastCalledWith({ network: 'mainnet', batch: 2, batchRows: 1, totalPruned: 3 }) - } finally { - log.mockRestore() - } + expect(await pruneOldSnapshots(14, 'mainnet', 2, onProgress)).toBe(3) + expect(mockQuery.mock.calls.map(call => call[1])).toEqual([ + ['mainnet', 14, 2], + ['mainnet', 14, 2], + ['mainnet', 14, 2], + ]) + expect(retentionLogger.info).toHaveBeenCalledWith( + expect.objectContaining({ network: 'mainnet', batch: 2, rows: 1, total: 3 }), + 'pruned retention batch' + ) + expect(onProgress).toHaveBeenLastCalledWith({ network: 'mainnet', batch: 2, batchRows: 1, totalPruned: 3 }) }) it('rejects an invalid batch size before querying', async () => { diff --git a/src/alerts/delivery.ts b/src/alerts/delivery.ts index 20a63914..1c0b28af 100644 --- a/src/alerts/delivery.ts +++ b/src/alerts/delivery.ts @@ -1,4 +1,7 @@ import { createHmac } from 'crypto' +import { createLogger } from '../logger' + +const log = createLogger('webhook') import { fetchWithTimeout } from '../fetch' function retryDelayMs(attempt: number): number { @@ -35,14 +38,14 @@ export async function deliverJsonWithRetries( return deliverJsonWithRetries(url, body, secret, attempt + 1) } - console.warn(`[webhook] delivery failed after ${attempt} attempts to ${url}:`, (err as Error).message) + log.warn({ url, attempt, err }, 'webhook delivery failed after retries') return } if (res.ok) return if (res.status >= 400 && res.status < 500) { - console.warn(`[webhook] client error ${res.status} from ${url} — not retrying`) + log.warn({ url, status: res.status }, 'webhook client error; not retrying') return } @@ -51,5 +54,5 @@ export async function deliverJsonWithRetries( return deliverJsonWithRetries(url, body, secret, attempt + 1) } - console.warn(`[webhook] delivery failed after ${attempt} attempts to ${url}: HTTP ${res.status}`) + log.warn({ url, attempt, status: res.status }, 'webhook delivery failed after retries') } \ No newline at end of file diff --git a/src/app.ts b/src/app.ts index c0d2a117..d1f64451 100644 --- a/src/app.ts +++ b/src/app.ts @@ -27,6 +27,7 @@ import { registerBasketRoutes } from './routes/basket' import { registerDiscoveryRoutes } from './routes/discovery' import { registerSettleRoute } from './routes/facilitator' import { getMetrics } from './metrics' +import { loggerOptions, createDestination } from './logger' export interface BuildAppOptions { /** @@ -51,7 +52,11 @@ export interface BuildAppOptions { * The returned app is NOT listening; the caller calls `app.listen(...)`. */ export async function buildApp(options: BuildAppOptions = {}): Promise { - const app = Fastify({ logger: { level: 'warn' } }) + // Share the structured logger configuration (level, redaction, dev pretty + // printing) with Fastify so `req.log` is scrubbed the same way. + const app = Fastify({ + logger: { ...loggerOptions(), stream: createDestination() }, + }) // Capture route registrations first, so it observes every route below // (including the ones registered by encapsulated plugins such as Mercurius). diff --git a/src/config.ts b/src/config.ts index 81b84827..1af08b8b 100644 --- a/src/config.ts +++ b/src/config.ts @@ -1,7 +1,10 @@ import 'dotenv/config' import { Networks, StrKey } from '@stellar/stellar-sdk' +import { createLogger } from './logger' import type { WatchedPair, AssetId } from './types' +const log = createLogger('config') + // ── Types ───────────────────────────────────────────────────────────────────── export type NetworkName = 'testnet' | 'mainnet' @@ -103,8 +106,9 @@ function parseWatchedPairs(raw: string): WatchedPair[] { function isUsableContractId(id: string, envVar: string): boolean { if (id === '') return false if (StrKey.isValidContract(id)) return true - console.warn( - `[config] ${envVar}="${id}" is not a valid Soroban contract id (expected a 56-character C... address); the feature that needs it is disabled` + log.warn( + { envVar, value: id }, + 'invalid Soroban contract id; the feature that needs it is disabled' ) return false } diff --git a/src/index.ts b/src/index.ts index 18f372c4..d5320762 100644 --- a/src/index.ts +++ b/src/index.ts @@ -6,6 +6,7 @@ if (!process.env.DIRECT_DATABASE_URL && process.env.DATABASE_URL) { } import { config, type NetworkName } from './config' import { getEnabledNetworks } from './network/enabledNetworks' +import { createLogger } from './logger' import { redis } from './redis' import { pgPool } from './db' import { buildApp } from './app' @@ -17,11 +18,13 @@ import { createAggregateQueue, startAggregateWorker, scheduleAggregateRefresh } import { createSnapshotRetentionQueue, startSnapshotRetentionWorker, scheduleSnapshotRetention, pruneAllNetworks, SNAPSHOT_RETENTION_DAYS } from './jobs/snapshotRetention' import { loadPersistedPairs, getActivePairs } from './pairsRegistry' +const log = createLogger('startup') + async function main() { // ── Ensure DB schema is up-to-date ──────────────────────────────────────── - console.log('[lens] Running database migrations…') + log.info('running database migrations') execSync('node node_modules/prisma/build/index.js db push --accept-data-loss', { stdio: 'inherit' }) - console.log('[lens] Database ready.') + log.info('database ready') // ── Connect dependencies ────────────────────────────────────────────────── // Redis is a cache and a job broker, not a source of truth: reads fall back @@ -34,16 +37,13 @@ async function main() { // database, uncached, until Redis comes back. try { await redis.connect() - console.log('[lens] Redis connected') + log.info('redis connected') } catch (err) { - console.warn( - '[lens] Redis unavailable, continuing without cache:', - (err as Error).message, - ) + log.warn({ err }, 'redis unavailable, continuing without cache') } await pgPool.connect() - console.log('[lens] PostgreSQL connected') + log.info('postgresql connected') // ── Load persisted runtime pairs ────────────────────────────────────────── await loadPersistedPairs() @@ -55,15 +55,15 @@ async function main() { const app = await buildApp() await app.listen({ port: config.api.port, host: config.api.host }) - console.log(`[lens] API listening on http://${config.api.host}:${config.api.port}`) - console.log(`[lens] GraphiQL at http://localhost:${config.api.port}/graphiql`) + log.info({ url: `http://${config.api.host}:${config.api.port}` }, 'API listening') + log.info({ url: `http://localhost:${config.api.port}/graphiql` }, 'GraphiQL available') // ── WebSocket fan-out (non-blocking — requires Redis for multi-instance) ─ try { await fanOutManager.initialize() - console.log('[lens] WebSocket fan-out manager initialized') + log.info('websocket fan-out manager initialized') } catch (err) { - console.warn('[lens] WebSocket fan-out init skipped:', (err as Error).message) + log.warn({ err }, 'websocket fan-out init skipped') } // ── Aggregate refresh worker (non-blocking — requires Redis) ───────────── @@ -86,12 +86,12 @@ async function main() { await scheduleAggregateRefresh(queue, network) started.push(network) } catch (err) { - console.warn(`[lens] Aggregate refresh worker skipped for ${network} (Redis unavailable):`, (err as Error).message) + log.warn({ network, err }, 'aggregate refresh worker skipped (redis unavailable)') await Promise.allSettled([queue?.close(), worker?.close()]) } } if (started.length > 0) { - console.log(`[lens] Aggregate refresh worker started for network(s): ${started.join(', ')}`) + log.info({ networks: started }, 'aggregate refresh worker started') } } @@ -109,11 +109,11 @@ async function main() { const counts = await pruneAllNetworks() for (const [network, pruned] of Object.entries(counts)) { if (pruned > 0) { - console.log(`[lens] Pruned ${pruned} ${network} snapshot(s) older than ${SNAPSHOT_RETENTION_DAYS}d`) + log.info({ network, pruned, retentionDays: SNAPSHOT_RETENTION_DAYS }, 'pruned snapshots older than retention window') } } } catch (err) { - console.error('[lens] Snapshot prune failed:', (err as Error).message) + log.error({ err }, 'snapshot prune failed') } } @@ -126,10 +126,10 @@ async function main() { retentionQueue = createSnapshotRetentionQueue() retentionWorker = startSnapshotRetentionWorker() await scheduleSnapshotRetention(retentionQueue) - console.log('[lens] Snapshot retention worker started') + log.info('snapshot retention worker started') } catch (err) { - console.warn('[lens] Snapshot retention worker skipped (Redis unavailable):', (err as Error).message) - console.warn('[lens] Falling back to an in-process hourly prune') + log.warn({ err }, 'snapshot retention worker skipped (redis unavailable)') + log.warn('falling back to an in-process hourly prune') await Promise.allSettled([retentionQueue?.close(), retentionWorker?.close()]) setInterval(() => { void safePrune() }, 60 * 60 * 1000).unref() } @@ -142,13 +142,13 @@ async function main() { // ingesters running on testnet. const restartIngester = (name: string, network: NetworkName, fn: () => Promise) => { fn().catch(err => { - console.error(`[lens] ${name}/${network} ingester crashed, restarting in 10s:`, err.message) + log.error({ ingester: name, network, err }, 'ingester crashed; restarting in 10s') setTimeout(() => restartIngester(name, network, fn), 10_000) }) } const enabledNetworks = getEnabledNetworks() - console.log(`[lens] Starting ingesters for network(s): ${enabledNetworks.join(', ')}`) + log.info({ networks: enabledNetworks }, 'starting ingesters') for (const network of enabledNetworks) { for (const adapter of venueAdapters) { restartIngester(adapter.name, network, () => startVenueAdapter(adapter, network)) @@ -156,10 +156,13 @@ async function main() { restartIngester('Snapshot', network, () => startSnapshotIngester(network)) } - console.log(`[lens] Watching ${getActivePairs().length} pairs: ${getActivePairs().map(p => p.pairKey).join(', ')}`) + log.info( + { count: getActivePairs().length, pairs: getActivePairs().map(p => p.pairKey) }, + 'watching pairs' + ) } main().catch(err => { - console.error('[lens] Fatal startup error:', err) + log.error({ err }, 'fatal startup error') process.exit(1) }) \ No newline at end of file diff --git a/src/ingest/venues/amm.ts b/src/ingest/venues/amm.ts index b715adb0..714eed6b 100644 --- a/src/ingest/venues/amm.ts +++ b/src/ingest/venues/amm.ts @@ -4,6 +4,7 @@ import { upsertPricePoints, getIndexerState, setIndexerCursor, prisma } from '.. import { resolvePageLedgers } from '../../ingesters/toid' import { dispatchPriceUpdate } from '../../webhookDispatcher' import { publishPriceUpdate } from '../../events' +import { createLogger } from '../../logger' import type { WatchedPair } from '../../types' import type { VenueAdapter, VenuePricePoint } from './types' import { @@ -14,6 +15,8 @@ import { } from './runner' import { fetchWithTimeout } from '../../fetch' +const log = createLogger('ingester:amm') + export function _resetLastPrice(): void { resetLastPrice('AMM') } @@ -43,7 +46,7 @@ export async function fetchPools(pair: WatchedPair, network: NetworkName = activ if (!data._embedded?.records) return [] return data._embedded.records } catch (err) { - console.error(`[amm] Failed to fetch pools for ${pair.pairKey}:`, (err as Error).message) + log.error({ pair: pair.pairKey, err }, 'failed to fetch pools') return [] } } @@ -120,10 +123,10 @@ export async function snapshotPool( previousPrice, currentPrice: spotPrice, network, - }).catch((err) => console.error('[amm] webhook dispatch error:', err.message)) + }).catch((err) => log.error({ pool: pool.id, err }, 'webhook dispatch error')) } } catch (err) { - console.error(`[amm] Snapshot error for pool ${pool.id}:`, (err as Error).message) + log.error({ pool: pool.id, err }, 'snapshot error') } } @@ -151,8 +154,9 @@ export async function ingestPoolTrades( state.ledger ) if (ledgers === null) { - console.error( - `[amm] pool ${pool.id.slice(0, 8)}: no ledger derivable from ${records.length} trades; skipping batch` + log.error( + { pool: pool.id, trades: records.length }, + 'no ledger derivable from trades; skipping batch' ) return } @@ -197,7 +201,7 @@ export async function ingestPoolTrades( const lastRecord = records[records.length - 1] await setIndexerCursor(stateId, lastRecord.paging_token, network, ledgers[ledgers.length - 1]) - console.log(`[amm] Pool ${pool.id.slice(0, 8)}: ingested ${points.length} trades`) + log.info({ pair: pair.pairKey, pool: pool.id, trades: points.length }, 'ingested pool trades') publishPriceUpdate({ pair: pair.pairKey, @@ -212,15 +216,15 @@ export async function ingestPoolTrades( previousPrice, currentPrice, network, - }).catch((err) => console.error('[amm] webhook dispatch error:', err.message)) + }).catch((err) => log.error({ pool: pool.id, err }, 'webhook dispatch error')) } catch (err) { - console.error(`[amm] Trade ingest error for pool ${pool.id}:`, (err as Error).message) + log.error({ pool: pool.id, err }, 'trade ingest error') } } export async function ingestPair(pair: WatchedPair, network: NetworkName = activeNetwork): Promise { const pools = await fetchPools(pair, network) - console.log(`[amm] ${pair.pairKey}: found ${pools.length} AMM pools`) + log.info({ pair: pair.pairKey, pools: pools.length }, 'found AMM pools') await Promise.all( pools.map(async (pool) => { diff --git a/src/ingest/venues/aquarius.ts b/src/ingest/venues/aquarius.ts index 3c2255cd..a38f64dc 100644 --- a/src/ingest/venues/aquarius.ts +++ b/src/ingest/venues/aquarius.ts @@ -9,6 +9,7 @@ */ import { config, activeNetwork, getNetworkConfig, type NetworkName } from '../../config' +import { createLogger } from '../../logger' import type { WatchedPair } from '../../types' import type { VenueAdapter, VenuePricePoint } from './types' import { @@ -19,6 +20,8 @@ import { } from './runner' import { fetchWithTimeout } from '../../fetch' +const log = createLogger('ingester:aquarius') + export function _resetLastPrice(): void { resetLastPrice('Aquarius') } @@ -58,7 +61,7 @@ export async function fetchAquariusPools( const data = (await res.json()) as AquariusListResponse return data.results ?? [] } catch (err) { - console.error(`[aquarius] Failed to fetch pools for ${pair.pairKey}:`, (err as Error).message) + log.error({ pair: pair.pairKey, err }, 'failed to fetch pools') return [] } } diff --git a/src/ingest/venues/runner.ts b/src/ingest/venues/runner.ts index ca2f1cf2..20cd3b83 100644 --- a/src/ingest/venues/runner.ts +++ b/src/ingest/venues/runner.ts @@ -3,9 +3,12 @@ import { getActivePairs } from '../../pairsRegistry' import { upsertPricePoints } from '../../db' import { dispatchPriceUpdate } from '../../webhookDispatcher' import { publishPriceUpdate } from '../../events' +import { createLogger } from '../../logger' import type { WatchedPair } from '../../types' import type { VenueAdapter, VenuePricePoint } from './types' +const log = createLogger('ingester:runner') + // ── Centralized Last Price Cache ────────────────────────────────────────────── // Keyed by `${venue}:${network}:${pairKey}` const lastPrice = new Map() @@ -66,7 +69,9 @@ export async function writeVenuePricePoints( previousPrice, currentPrice, network, - }).catch((err) => console.error(`[${venue}] webhook dispatch error:`, (err as Error).message)) + }).catch((err) => + createLogger(`ingester:${venue.toLowerCase()}`).error({ err }, 'webhook dispatch error') + ) } // ── Sleep Helper ────────────────────────────────────────────────────────────── @@ -91,11 +96,11 @@ export async function startVenueAdapter( network: NetworkName = activeNetwork ): Promise { if (!adapter.isEnabled(network)) { - console.log(`[${adapter.name}] ${adapter.name} is disabled on ${network} — ingester not started`) + log.info({ venue: adapter.name, network }, 'venue adapter is disabled — ingester not started') return } - console.log(`[${adapter.name}] Starting ${adapter.name} venue adapter on ${network}`) + log.info({ venue: adapter.name, network }, 'starting venue adapter') let consecutiveErrors = 0 while (true) { @@ -108,9 +113,9 @@ export async function startVenueAdapter( } catch (err) { consecutiveErrors++ const backoffMs = Math.min(10_000 * Math.pow(2, consecutiveErrors - 1), 60_000) - console.error( - `[${adapter.name}] Cycle failed on ${network} (attempt ${consecutiveErrors}), retrying in ${backoffMs}ms:`, - (err as Error).message + log.error( + { venue: adapter.name, network, attempt: consecutiveErrors, backoffMs, err }, + 'venue adapter cycle failed; retrying' ) await sleep(backoffMs) continue @@ -129,10 +134,10 @@ export async function startAllVenueAdapters( adapters: VenueAdapter[], network: NetworkName = activeNetwork ): Promise { - console.log(`[runner] Launching ${adapters.length} venue adapter(s) on ${network}`) + log.info({ count: adapters.length, network }, 'launching venue adapters') for (const adapter of adapters) { void startVenueAdapter(adapter, network).catch((err) => { - console.error(`[runner] Unhandled error starting ${adapter.name} on ${network}:`, err) + log.error({ venue: adapter.name, network, err }, 'unhandled error starting venue adapter') }) } } diff --git a/src/ingest/venues/sdex.ts b/src/ingest/venues/sdex.ts index e8adc156..29875a97 100644 --- a/src/ingest/venues/sdex.ts +++ b/src/ingest/venues/sdex.ts @@ -6,6 +6,7 @@ import { upsertPricePoints, getIndexerState, setIndexerCursor } from '../../db' import { resolvePageLedgers } from '../../ingesters/toid' import { dispatchPriceUpdate } from '../../webhookDispatcher' import { publishPriceUpdate } from '../../events' +import { createLogger } from '../../logger' import type { WatchedPair } from '../../types' import type { VenueAdapter, VenuePricePoint } from './types' import { @@ -15,6 +16,8 @@ import { startVenueAdapter, } from './runner' +const log = createLogger('ingester:sdex') + export function _resetLastPrice(): void { resetLastPrice('SDEX') } @@ -65,7 +68,7 @@ export async function fetchSdexTrades( lastLedger: state.ledger, } } catch (err) { - console.error(`[sdex] Error fetching trades for ${pair.pairKey}:`, (err as Error).message) + log.error({ pair: pair.pairKey, err }, 'error fetching trades') return null } } @@ -82,7 +85,10 @@ export function deriveSdexPricePoints( lastLedger, ) if (ledgers === null) { - console.error(`[sdex] ${pair.pairKey}: no ledger derivable from ${records.length} trades; skipping batch`) + log.error( + { pair: pair.pairKey, trades: records.length }, + 'no ledger derivable from trades; skipping batch' + ) return null } @@ -137,7 +143,7 @@ export async function writeSdexPricePoints( ) await setIndexerCursor(stateId, lastPagingToken, network, finalLedger) - console.log(`[sdex] ${pair.pairKey}: ingested ${points.length} trades`) + log.info({ pair: pair.pairKey, trades: points.length }, 'ingested trades') publishPriceUpdate({ pair: pair.pairKey, @@ -152,7 +158,7 @@ export async function writeSdexPricePoints( previousPrice, currentPrice, network, - }).catch((err) => console.error('[sdex] webhook dispatch error:', err.message)) + }).catch((err) => log.error({ pair: pair.pairKey, err }, 'webhook dispatch error')) } export async function ingestPair(pair: WatchedPair, network: NetworkName = activeNetwork): Promise { diff --git a/src/ingest/venues/soroswap.ts b/src/ingest/venues/soroswap.ts index 1b2b2b42..e5d677e5 100644 --- a/src/ingest/venues/soroswap.ts +++ b/src/ingest/venues/soroswap.ts @@ -24,6 +24,7 @@ import { getRpcServer } from '../../network/clients' import { upsertPricePoints } from '../../db' import { dispatchPriceUpdate } from '../../webhookDispatcher' import { publishPriceUpdate } from '../../events' +import { createLogger } from '../../logger' import { fetchWithTimeout } from '../../fetch' import type { WatchedPair } from '../../types' import type { VenueAdapter, VenuePricePoint } from './types' @@ -34,6 +35,8 @@ import { startVenueAdapter, } from './runner' +const log = createLogger('ingester:soroswap') + // ── Constants ───────────────────────────────────────────────────────────────── // Ephemeral fee-payer account (no real funds needed for simulation) @@ -81,7 +84,7 @@ export async function fetchSoroswapTokenList( const data = (await res.json()) as SoroswapTokenList return Array.isArray(data.tokens) ? data.tokens : [] } catch (err) { - console.error('[soroswap] Failed to fetch token list:', (err as Error).message) + log.error({ err }, 'failed to fetch token list') return [] } } @@ -122,10 +125,7 @@ export async function fetchPoolsFromFactory( const sim = await rpc.simulateTransaction(tx) if (SorobanRpc.Api.isSimulationError(sim)) { - console.debug( - `[soroswap] get_pools simulation error for ${tokenA}/${tokenB}:`, - sim.error - ) + log.debug({ tokenA, tokenB, error: sim.error }, 'get_pools simulation error') return [] } @@ -134,10 +134,7 @@ export async function fetchPoolsFromFactory( const pools = scValToNative(sim.result.retval) as string[] return Array.isArray(pools) ? pools : [] } catch (err) { - console.debug( - '[soroswap] fetchPoolsFromFactory error:', - (err as Error).message - ) + log.debug({ err }, 'fetchPoolsFromFactory error') return [] } } @@ -171,10 +168,7 @@ export async function fetchPoolReserves( const sim = await rpc.simulateTransaction(tx) if (SorobanRpc.Api.isSimulationError(sim)) { - console.debug( - `[soroswap] get_reserves simulation error for pool ${poolAddress}:`, - sim.error - ) + log.debug({ pool: poolAddress, error: sim.error }, 'get_reserves simulation error') return null } @@ -185,10 +179,7 @@ export async function fetchPoolReserves( return [BigInt(reserves[0]), BigInt(reserves[1])] } catch (err) { - console.debug( - `[soroswap] fetchPoolReserves error for ${poolAddress}:`, - (err as Error).message - ) + log.debug({ pool: poolAddress, err }, 'fetchPoolReserves error') return null } } @@ -228,9 +219,7 @@ export async function ingestPool( const spotPrice = calcSpotPrice(reserveA, reserveB) if (spotPrice === 0) { - console.debug( - `[soroswap] Skipping empty pool ${poolEntry.poolAddress} for ${pair.pairKey}` - ) + log.debug({ pool: poolEntry.poolAddress, pair: pair.pairKey }, 'skipping empty pool') return } @@ -269,18 +258,14 @@ export async function ingestPool( previousPrice, currentPrice: spotPrice, network, - }).catch((err) => - console.error('[soroswap] webhook dispatch error:', err.message) - ) + }).catch((err) => log.error({ pool: poolEntry.poolAddress, err }, 'webhook dispatch error')) - console.log( - `[soroswap] Pool ${poolEntry.poolAddress.slice(0, 8)}: price=${spotPrice.toFixed(8)} for ${pair.pairKey}` + log.info( + { pair: pair.pairKey, pool: poolEntry.poolAddress, price: spotPrice }, + 'ingested pool price' ) } catch (err) { - console.error( - `[soroswap] ingestPool error for ${poolEntry.poolAddress}:`, - (err as Error).message - ) + log.error({ pool: poolEntry.poolAddress, err }, 'ingestPool error') } } @@ -310,9 +295,7 @@ export async function ingestPair( ) if (!tokenA || !tokenB) { - console.debug( - `[soroswap] No Soroswap token found for pair ${pair.pairKey} — skipping` - ) + log.debug({ pair: pair.pairKey }, 'no Soroswap token found; skipping') return } @@ -324,9 +307,7 @@ export async function ingestPair( ) if (!poolAddresses.length) { - console.debug( - `[soroswap] No pools found for ${pair.pairKey} — skipping` - ) + log.debug({ pair: pair.pairKey }, 'no pools found; skipping') return } diff --git a/src/ingesters/snapshot.ts b/src/ingesters/snapshot.ts index 84224373..e48df001 100644 --- a/src/ingesters/snapshot.ts +++ b/src/ingesters/snapshot.ts @@ -2,6 +2,9 @@ import { pgPool } from '../db' import { activeNetwork, type NetworkName } from '../config' import { getActivePairs } from '../pairsRegistry' import { price_snapshots_total } from '../metrics' +import { createLogger } from '../logger' + +const log = createLogger('ingester:snapshot') /** * Floors a Date down to the start of its minute (seconds/millis zeroed). @@ -78,7 +81,7 @@ function sleep(ms: number) { * volume window cleanly covers the minute that just elapsed. */ export async function startSnapshotIngester(network: NetworkName = activeNetwork): Promise { - console.log(`[snapshot] Starting 1-minute snapshot ingester for ${getActivePairs().length} pairs on ${network}`) + log.info({ network, pairs: getActivePairs().length }, 'starting 1-minute snapshot ingester') while (true) { // Sleep until the next minute boundary. @@ -87,9 +90,9 @@ export async function startSnapshotIngester(network: NetworkName = activeNetwork try { const n = await appendSnapshots(new Date(), network) - if (n > 0) console.log(`[snapshot] appended ${n} snapshot(s) on ${network}`) + if (n > 0) log.info({ network, snapshots: n }, 'appended snapshots') } catch (err) { - console.error(`[snapshot] Error appending snapshots on ${network}:`, (err as Error).message) + log.error({ network, err }, 'error appending snapshots') } } } diff --git a/src/jobs/aggregateRefresh.ts b/src/jobs/aggregateRefresh.ts index 8048b258..4b49b9b7 100644 --- a/src/jobs/aggregateRefresh.ts +++ b/src/jobs/aggregateRefresh.ts @@ -4,6 +4,9 @@ import { prisma } from '../db' import { setCachedPrice } from '../redis' import { calculateVWAP, calculateOHLCV, getAggregatedPrice } from '../aggregator/vwap' import { getBestRoute } from '../aggregator/bestRoute' +import { createLogger } from '../logger' + +const log = createLogger('job:aggregate-refresh') /** * One queue, worker and job stream per network. Everything a job touches — @@ -91,16 +94,16 @@ export function startAggregateWorker(network: NetworkName) { }) } - console.log(`[aggregator] Refreshed ${network}/${pairKey}: price=${agg.price.toFixed(6)}, route=${route.route}`) + log.info({ network, pair: pairKey, price: agg.price, route: route.route }, 'refreshed aggregate') } catch (err) { - console.error(`[aggregator] Failed for ${network}/${pairKey}:`, (err as Error).message) + log.error({ network, pair: pairKey, err }, 'aggregate refresh failed') } }, { connection: redisConnection(), concurrency: 5 } ) worker.on('failed', (job, err) => { - console.error(`[aggregator] Job failed:`, err.message) + log.error({ err }, 'aggregate refresh job failed') }) return worker diff --git a/src/jobs/snapshotRetention.ts b/src/jobs/snapshotRetention.ts index 2d44641f..ab5c94e3 100644 --- a/src/jobs/snapshotRetention.ts +++ b/src/jobs/snapshotRetention.ts @@ -2,6 +2,9 @@ import { Queue, Worker } from 'bullmq' import { pgPool } from '../db' import { activeNetwork, type NetworkName } from '../config' import { getEnabledNetworks } from '../network/enabledNetworks' +import { createLogger } from '../logger' + +const log = createLogger('job:snapshot-retention') const QUEUE_NAME = `${activeNetwork}:snapshot-retention` @@ -70,7 +73,10 @@ export async function pruneOldSnapshots( totalPruned += pruned batchNumber += 1 - console.log(`[snapshot-retention] ${network}: pruned batch ${batchNumber} (${pruned} rows; ${totalPruned} total)`) + log.info( + { network, batch: batchNumber, rows: pruned, total: totalPruned }, + 'pruned retention batch' + ) onProgress?.({ network, batch: batchNumber, batchRows: pruned, totalPruned }) // Give other jobs and requests a chance to run between database batches. @@ -110,17 +116,22 @@ export function startSnapshotRetentionWorker() { try { const counts = await pruneAllNetworks() for (const [network, pruned] of Object.entries(counts)) { - if (pruned > 0) console.log(`[snapshot-retention] pruned ${pruned} ${network} snapshot(s) older than ${SNAPSHOT_RETENTION_DAYS}d`) + if (pruned > 0) { + log.info( + { network, pruned, retentionDays: SNAPSHOT_RETENTION_DAYS }, + 'pruned snapshots older than retention window' + ) + } } } catch (err) { - console.error('[snapshot-retention] prune failed:', (err as Error).message) + log.error({ err }, 'snapshot prune failed') } }, { connection: redisConnection(), concurrency: 1 } ) worker.on('failed', (_job, err) => { - console.error('[snapshot-retention] Job failed:', err.message) + log.error({ err }, 'snapshot retention job failed') }) return worker diff --git a/src/logger.ts b/src/logger.ts new file mode 100644 index 00000000..f0674ff3 --- /dev/null +++ b/src/logger.ts @@ -0,0 +1,220 @@ +/** + * Structured logging. + * + * Every module logs through a pino logger with a stable `component` field + * (`ingester:amm`, `job:snapshot-retention`, `route:price`, …) so operators can + * filter logs without guessing which `[bracket]` prefix a message came from. + * + * - Level is controlled by `LOG_LEVEL` (default `info`). Standard pino levels: + * trace, debug, info, warn, error, fatal, silent. + * - Production emits newline-delimited JSON; development pretty-prints for + * humans. Switch is `NODE_ENV === 'production'`. + * - Values of secret-looking environment variables (DATABASE_URL, *_KEY, + * *_SECRET, *_TOKEN, *_PASSWORD) are scrubbed from every line before it is + * written, and request auth headers are redacted via pino's `redact`. + * + * `logger.ts` must not import `config.ts`: the config module imports this one, + * so it reads `process.env` directly to avoid a cycle. + */ + +import pino, { type Logger, type DestinationStream } from 'pino' + +/** Default level when `LOG_LEVEL` is unset. */ +export const DEFAULT_LOG_LEVEL = 'info' + +/** True when logs should be rendered for humans rather than emitted as JSON. */ +export const prettyEnabled = process.env.NODE_ENV !== 'production' + +/** + * Secret environment variables scrubbed out of serialized log lines. A name + * counts as secret if it is a known connection-string variable or ends with / + * contains KEY, SECRET, TOKEN or PASSWORD. + */ +const SECRET_URL_NAMES = new Set(['DATABASE_URL', 'DIRECT_DATABASE_URL', 'REDIS_URL']) +const SECRET_NAME_RE = /(API_?KEY|SECRET|TOKEN|PASSWORD|PASSWD)/i + +/** + * Minimum length a value must have before it is treated as a secret worth + * scrubbing. Guards against clobbering ordinary log content with the literal + * "true"/"false" from boolean-looking flags such as `REQUIRE_API_KEY`. + */ +const MIN_SECRET_LENGTH = 16 + +/** + * All secret values currently present in the environment, including the + * password embedded in a connection-string URL so that driver error messages + * (which often echo only the credentials) are covered too. + */ +export function secretValues(env: NodeJS.ProcessEnv = process.env): string[] { + const secrets = new Set() + for (const [name, value] of Object.entries(env)) { + if (!value || value.length < MIN_SECRET_LENGTH) continue + if (SECRET_URL_NAMES.has(name) || SECRET_NAME_RE.test(name)) { + secrets.add(value) + if (SECRET_URL_NAMES.has(name)) { + try { + const password = new URL(value).password + if (password && password.length >= 4) secrets.add(password) + } catch { + // Not a parseable URL — the full value is still scrubbed above. + } + } + } + } + return [...secrets] +} + +/** + * Replaces every secret value in `text` with `[redacted]`. Applied to the + * fully-serialized line so secrets cannot slip through nested fields. + */ +export function redactSecrets(text: string, env: NodeJS.ProcessEnv = process.env): string { + let out = text + for (const secret of secretValues(env)) { + if (out.includes(secret)) out = out.split(secret).join('[redacted]') + } + return out +} + +/** + * Paths pino redacts before serialization. Request auth headers and common + * credential field names, matched at any depth. + */ +const REDACT_PATHS = [ + 'req.headers.authorization', + 'req.headers.cookie', + 'req.headers["x-api-key"]', + 'req.headers["x-admin-token"]', + 'req.headers["x-payment"]', + 'res.headers["set-cookie"]', + 'headers.authorization', + 'headers.cookie', + 'headers["x-api-key"]', + 'headers["x-admin-token"]', + 'headers["x-payment"]', + 'password', + 'secret', + 'token', + 'apiKey', + '*.password', + '*.secret', + '*.token', + '*.apiKey', +] + +interface PrettyLine { + time?: string + level?: string | number + component?: string + msg?: string + err?: { type?: string; message?: string; stack?: string } + [key: string]: unknown +} + +const LEVEL_COLOR: Record = { + trace: '\x1b[90m', + debug: '\x1b[36m', + info: '\x1b[32m', + warn: '\x1b[33m', + error: '\x1b[31m', + fatal: '\x1b[35m', +} +const RESET = '\x1b[0m' + +/** + * Renders one NDJSON pino line as a single human-readable line (development). + * Falls back to the raw line if it is not JSON. + */ +export function formatPretty(line: string): string { + let obj: PrettyLine + try { + obj = JSON.parse(line) as PrettyLine + } catch { + return line + } + + const time = typeof obj.time === 'string' ? obj.time.slice(11, 23) : '' + const level = String(obj.level ?? '').toLowerCase() + const color = LEVEL_COLOR[level] ?? '' + const label = color ? `${color}${level.toUpperCase().padEnd(5)}${RESET}` : level.toUpperCase().padEnd(5) + const component = obj.component ? `[${obj.component}] ` : '' + + const { time: _t, level: _l, component: _c, msg, err, ...extra } = obj + const extraKeys = Object.keys(extra).filter((key) => extra[key] !== undefined) + + let out = `${time} ${label} ${component}${msg ?? ''}` + if (extraKeys.length > 0) { + out += ` ${JSON.stringify(extra)}` + } + if (err) out += `\n${err.stack ?? err.message ?? ''}` + return `${out}\n` +} + +export interface BuildLoggerOptions { + /** pino level; defaults to `LOG_LEVEL` or `info`. */ + level?: string + /** Pretty (human) output. Defaults to `NODE_ENV !== 'production'`. */ + pretty?: boolean + /** Sink for serialized lines. Defaults to `process.stdout`. */ + destination?: DestinationStream +} + +/** + * Builds a pino logger. Factory form is used by tests to capture output; + * application code should use the shared `logger` / `createLogger` below. + */ +export function buildLogger(opts: BuildLoggerOptions = {}): Logger { + const level = opts.level ?? process.env.LOG_LEVEL ?? DEFAULT_LOG_LEVEL + const pretty = opts.pretty ?? prettyEnabled + const destination = createDestination(pretty, opts.destination) + return pino(loggerOptions(level), destination) +} + +/** + * pino options shared by the module-level logger and Fastify's request logger. + * `level` is read from `LOG_LEVEL` when not given. + */ +export function loggerOptions( + level: string = process.env.LOG_LEVEL ?? DEFAULT_LOG_LEVEL +): pino.LoggerOptions { + return { + level, + base: { service: 'lens' }, + formatters: { level: (label) => ({ level: label }) }, + timestamp: pino.stdTimeFunctions.isoTime, + redact: { paths: REDACT_PATHS, censor: '[redacted]' }, + } +} + +/** + * Wraps a sink so every serialized line is scrubbed of secret env values and, + * in development, pretty-printed. `err` keys are serialized by pino itself. + */ +export function createDestination( + pretty: boolean = prettyEnabled, + sink: DestinationStream = process.stdout +): DestinationStream { + return { + write(line: string) { + const safe = redactSecrets(line) + sink.write(pretty ? formatPretty(safe) : safe) + }, + } +} + +/** Shared root logger. */ +export const logger: Logger = buildLogger() + +/** + * Returns a logger tagged with a stable component name. Prefer one child per + * module, created once at import time: + * + * const log = createLogger('ingester:amm') + * log.info({ pool: pool.id }, 'ingested trades') + * + * HTTP handlers should derive from the request instead: + * req.log.child({ component: 'route:price' }) + */ +export function createLogger(component: string): Logger { + return logger.child({ component }) +} diff --git a/src/redis.ts b/src/redis.ts index 41fa9f34..bd5e6afb 100644 --- a/src/redis.ts +++ b/src/redis.ts @@ -1,5 +1,8 @@ import Redis from 'ioredis' import { config, activeNetwork } from './config' +import { createLogger } from './logger' + +const log = createLogger('redis') export const redis = new Redis(config.redis.url, { maxRetriesPerRequest: 3, @@ -40,8 +43,7 @@ redis.on('error', (err) => { suppressedRedisErrors++ return } - const suffix = suppressedRedisErrors > 0 ? ` (${suppressedRedisErrors} more since last report)` : '' - console.error('[redis] Connection error:', err.message + suffix) + log.error({ err, suppressed: suppressedRedisErrors }, 'redis connection error') lastRedisErrorLog = now suppressedRedisErrors = 0 }) diff --git a/src/routes/pairs.ts b/src/routes/pairs.ts index f2208e04..199ada5a 100644 --- a/src/routes/pairs.ts +++ b/src/routes/pairs.ts @@ -60,6 +60,7 @@ export async function registerPairsRoutes(app: FastifyInstance) { app.post<{ Body: { assetA?: string; assetB?: string } }>('/pairs', async (req, reply) => { + const log = req.log.child({ component: 'route:pairs' }) // Auth check — read at request time so env can be set after module load const ADMIN_API_KEY = process.env.ADMIN_API_KEY const key = req.headers['x-admin-key'] ?? req.headers['authorization']?.replace(/^Bearer /, '') @@ -97,7 +98,7 @@ export async function registerPairsRoutes(app: FastifyInstance) { registerPair(pair) await persistPair(pair) - console.log(`[pairs] Added runtime pair: ${pairKey}`) + log.info({ pair: pairKey }, 'added runtime pair') return reply.status(201).send({ pairKey, assetA, assetB }) }) } diff --git a/src/routes/price.ts b/src/routes/price.ts index 241a0609..116567b6 100644 --- a/src/routes/price.ts +++ b/src/routes/price.ts @@ -36,6 +36,7 @@ export async function registerPriceRoutes(app: FastifyInstance) { }>( '/price/twap/:assetA/:assetB', async (req, reply) => { + const log = req.log.child({ component: 'route:price' }) const { assetA, assetB } = req.params const parsed = TwapQuerySchema.safeParse(req.query) if (!parsed.success) { @@ -70,7 +71,7 @@ export async function registerPriceRoutes(app: FastifyInstance) { } catch (err) { // A driver error message can carry the connection string, and so the // Postgres credentials. Log it; never put it in the response body. - console.error('[price] TWAP computation failed:', (err as Error).message) + log.error({ err }, 'TWAP computation failed') return reply.status(500).send({ error: 'TWAP computation failed' }) } } @@ -87,6 +88,7 @@ export async function registerPriceRoutes(app: FastifyInstance) { }>( '/price/vwap/:assetA/:assetB', async (req, reply) => { + const log = req.log.child({ component: 'route:price' }) const { assetA, assetB } = req.params const parsed = VwapQuerySchema.safeParse(req.query) if (!parsed.success) { @@ -118,7 +120,7 @@ export async function registerPriceRoutes(app: FastifyInstance) { } } catch (err) { // Same as TWAP above: the driver message can contain credentials. - console.error('[price] VWAP computation failed:', (err as Error).message) + log.error({ err }, 'VWAP computation failed') return reply.status(500).send({ error: 'VWAP computation failed' }) } } diff --git a/src/routes/screener.ts b/src/routes/screener.ts index 783059d4..067cc26e 100644 --- a/src/routes/screener.ts +++ b/src/routes/screener.ts @@ -25,6 +25,7 @@ export async function registerScreenerRoutes(app: FastifyInstance) { limit?: string } }>('/screener', async (req, reply) => { + const log = req.log.child({ component: 'route:screener' }) try { const q = req.query @@ -189,7 +190,7 @@ export async function registerScreenerRoutes(app: FastifyInstance) { total: null, } } catch (error) { - console.error('[screener]', error) + log.error({ err: error }, 'screener query failed') return reply.status(500).send({ error: 'Internal server error' }) } }) diff --git a/src/ws/fanout.ts b/src/ws/fanout.ts index e708610f..85c414ed 100644 --- a/src/ws/fanout.ts +++ b/src/ws/fanout.ts @@ -3,6 +3,9 @@ import { redis } from '../redis' import { activeNetwork } from '../config' import { priceEmitter, PRICE_UPDATE, PriceUpdateEvent } from '../events' import { register, Gauge, Counter } from 'prom-client' +import { createLogger } from '../logger' + +const log = createLogger('ws:fanout') // ─── Metrics ────────────────────────────────────────────────────────────────── export const ws_clients_total = new Gauge({ @@ -90,9 +93,9 @@ class FanOutManager { // ignore malformed messages } }) - console.log('[fanout] Redis pub/sub subscriber ready') + log.info('redis pub/sub subscriber ready') } catch (err) { - console.warn('[fanout] Redis pub/sub unavailable, running in single-instance mode:', (err as Error).message) + log.warn({ err }, 'redis pub/sub unavailable, running in single-instance mode') this.redisConnected = false } } diff --git a/vitest.setup.ts b/vitest.setup.ts index b939bf6e..e48df3e1 100644 --- a/vitest.setup.ts +++ b/vitest.setup.ts @@ -17,6 +17,10 @@ */ import { beforeEach, afterEach } from 'vitest' +// Keep the shared pino logger quiet during tests unless a run overrides it. +// This is read when src/logger.ts first imports, before any test module runs. +process.env.LOG_LEVEL = process.env.LOG_LEVEL ?? 'silent' + let envSnapshot: Record beforeEach(() => {