diff --git a/scripts/reconcile-replay.test.ts b/scripts/reconcile-replay.test.ts new file mode 100644 index 00000000..0d3746f4 --- /dev/null +++ b/scripts/reconcile-replay.test.ts @@ -0,0 +1,542 @@ +import { createHash, createHmac } from 'node:crypto'; +import { + runReconcileReplayCli, + fetchHorizonFixture, + HorizonFixtureClient, + ReplayReport, +} from './reconcile-replay'; + +// --------------------------------------------------------------------------- +// Mocks +// --------------------------------------------------------------------------- + +// Mock the database pool +jest.mock('pg', () => ({ + Pool: jest.fn().mockImplementation(() => ({ + end: jest.fn().mockResolvedValue(undefined), + })), +})); + +// Mock the revenue reconciliation service +jest.mock('../src/services/revenueReconciliationService', () => { + const mockReconcile = jest.fn(); + return { + RevenueReconciliationService: jest.fn().mockImplementation(() => ({ + reconcile: mockReconcile, + })), + __mockReconcile: mockReconcile, + }; +}); + +// Mock globalMetrics +jest.mock('../src/lib/metrics', () => ({ + globalMetrics: { + incrementCounter: jest.fn(), + setGauge: jest.fn(), + }, +})); + +// Mock fetch for fixture tests +const mockFetch = jest.fn(); +global.fetch = mockFetch as any; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function sha256(input: string): string { + return createHash('sha256').update(input, 'utf8').digest('hex'); +} + +function signReport(report: ReplayReport, secret: string): string { + const canonical = JSON.stringify(report); + const hmac = createHmac('sha256', secret); + hmac.update(canonical); + return `sha256=${hmac.digest('hex')}`; +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +describe('reconcile-replay CLI', () => { + const originalArgv = process.argv; + const originalEnv = process.env; + + beforeEach(() => { + jest.clearAllMocks(); + process.argv = ['node', 'reconcile-replay.ts']; + process.env = { ...originalEnv }; + mockFetch.mockReset(); + }); + + afterAll(() => { + process.argv = originalArgv; + process.env = originalEnv; + }); + + // ----------------------------------------------------------------------- + // fetchHorizonFixture + // ----------------------------------------------------------------------- + + describe('fetchHorizonFixture', () => { + it('fetches and parses a valid JSON fixture', async () => { + const fixtureData = { totalDistributed: '1000.00' }; + mockFetch.mockResolvedValue({ + ok: true, + text: jest.fn().mockResolvedValue(JSON.stringify(fixtureData)), + }); + + const result = await fetchHorizonFixture('https://example.com/fixture.json'); + expect(result.data).toEqual(fixtureData); + expect(result.body).toBe(JSON.stringify(fixtureData)); + }); + + it('throws on network error', async () => { + mockFetch.mockRejectedValue(new Error('Network error')); + + await expect(fetchHorizonFixture('https://example.com/fixture.json')) + .rejects.toThrow('Failed to connect to fixture URL'); + }); + + it('throws on HTTP error response', async () => { + mockFetch.mockResolvedValue({ + ok: false, + status: 404, + statusText: 'Not Found', + }); + + await expect(fetchHorizonFixture('https://example.com/fixture.json')) + .rejects.toThrow('HTTP 404'); + }); + + it('throws on empty response body', async () => { + mockFetch.mockResolvedValue({ + ok: true, + text: jest.fn().mockResolvedValue(''), + }); + + await expect(fetchHorizonFixture('https://example.com/fixture.json')) + .rejects.toThrow('empty response body'); + }); + + it('throws on non-JSON response', async () => { + mockFetch.mockResolvedValue({ + ok: true, + text: jest.fn().mockResolvedValue('Not JSON'), + }); + + await expect(fetchHorizonFixture('https://example.com/fixture.json')) + .rejects.toThrow('non-JSON content'); + }); + }); + + // ----------------------------------------------------------------------- + // signReport + // ----------------------------------------------------------------------- + + describe('signReport', () => { + it('produces a consistent HMAC-SHA256 signature', () => { + const report: ReplayReport = { + schema_version: 1, + generated_at: '2023-01-15T10:00:00Z', + fixture_sha256: 'abc123', + fixture_url: 'https://example.com/fixture.json', + parameters: { + offering_id: 'offering-123', + period_start: '2023-01-01', + period_end: '2023-01-31', + }, + reconciliation: { + offeringId: 'offering-123', + periodStart: new Date('2023-01-01'), + periodEnd: new Date('2023-01-31'), + isBalanced: true, + discrepancies: [], + summary: { + totalRevenueReported: '1000.00', + totalPayouts: '1000.00', + discrepancyAmount: '0.00', + investorCount: 5, + payoutsProcessed: 10, + payoutsFailed: 0, + }, + checkedAt: new Date('2023-01-15T10:00:00Z'), + }, + }; + + const secret = 'test-secret'; + const sig1 = signReport(report, secret); + const sig2 = signReport(report, secret); + + expect(sig1).toBe(sig2); + expect(sig1).toMatch(/^sha256=[a-f0-9]{64}$/); + }); + + it('produces different signatures with different secrets', () => { + const report: ReplayReport = { + schema_version: 1, + generated_at: '2023-01-15T10:00:00Z', + fixture_sha256: 'abc123', + fixture_url: 'https://example.com/fixture.json', + parameters: { + offering_id: 'offering-123', + period_start: '2023-01-01', + period_end: '2023-01-31', + }, + reconciliation: { + offeringId: 'offering-123', + periodStart: new Date('2023-01-01'), + periodEnd: new Date('2023-01-31'), + isBalanced: true, + discrepancies: [], + summary: { + totalRevenueReported: '1000.00', + totalPayouts: '1000.00', + discrepancyAmount: '0.00', + investorCount: 5, + payoutsProcessed: 10, + payoutsFailed: 0, + }, + checkedAt: new Date('2023-01-15T10:00:00Z'), + }, + }; + + const sig1 = signReport(report, 'secret-1'); + const sig2 = signReport(report, 'secret-2'); + + expect(sig1).not.toBe(sig2); + }); + }); + + // ----------------------------------------------------------------------- + // HorizonFixtureClient + // ----------------------------------------------------------------------- + + describe('HorizonFixtureClient', () => { + it('returns totalDistributed from fixture', async () => { + const fixture = { totalDistributed: '5000.00' }; + const client = new HorizonFixtureClient(fixture); + + const result = await client.getRevenueState('contract-123'); + expect(result.totalDistributed).toBe('5000.00'); + }); + + it('defaults to 0.00 when totalDistributed is missing', async () => { + const fixture = { totalDistributed: undefined } as any; + const client = new HorizonFixtureClient(fixture); + + const result = await client.getRevenueState('contract-123'); + expect(result.totalDistributed).toBe('0.00'); + }); + }); + + // ----------------------------------------------------------------------- + // CLI argument validation + // ----------------------------------------------------------------------- + + describe('CLI argument validation', () => { + it('shows help with --help flag', async () => { + process.argv = ['node', 'reconcile-replay.ts', '--help']; + const consoleSpy = jest.spyOn(console, 'log').mockImplementation(); + + const code = await runReconcileReplayCli(); + expect(code).toBe(0); + expect(consoleSpy).toHaveBeenCalledWith( + expect.stringContaining('Usage:') + ); + + consoleSpy.mockRestore(); + }); + + it('shows help with -h flag', async () => { + process.argv = ['node', 'reconcile-replay.ts', '-h']; + const consoleSpy = jest.spyOn(console, 'log').mockImplementation(); + + const code = await runReconcileReplayCli(); + expect(code).toBe(0); + + consoleSpy.mockRestore(); + }); + + it('returns error with missing arguments', async () => { + process.argv = ['node', 'reconcile-replay.ts']; + const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(); + + const code = await runReconcileReplayCli(); + expect(code).toBe(1); + expect(consoleErrorSpy).toHaveBeenCalledWith( + expect.stringContaining('Missing required arguments') + ); + + consoleErrorSpy.mockRestore(); + }); + + it('returns error for invalid period_start', async () => { + process.argv = ['node', 'reconcile-replay.ts', 'offering-1', 'not-a-date', 'https://example.com/fixture.json']; + const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(); + + const code = await runReconcileReplayCli(); + expect(code).toBe(1); + expect(consoleErrorSpy).toHaveBeenCalledWith( + expect.stringContaining('Invalid period_start') + ); + + consoleErrorSpy.mockRestore(); + }); + + it('returns error for invalid period_end', async () => { + process.argv = ['node', 'reconcile-replay.ts', 'offering-1', '2023-01-01', 'https://example.com/fixture.json', 'not-a-date']; + const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(); + + const code = await runReconcileReplayCli(); + expect(code).toBe(1); + expect(consoleErrorSpy).toHaveBeenCalledWith( + expect.stringContaining('Invalid period_end') + ); + + consoleErrorSpy.mockRestore(); + }); + + it('returns error when period_end is before period_start', async () => { + process.argv = ['node', 'reconcile-replay.ts', 'offering-1', '2023-01-31', 'https://example.com/fixture.json', '2023-01-01']; + const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(); + + const code = await runReconcileReplayCli(); + expect(code).toBe(1); + expect(consoleErrorSpy).toHaveBeenCalledWith( + expect.stringContaining('period_end must be after period_start') + ); + + consoleErrorSpy.mockRestore(); + }); + + it('returns error for empty offering_id', async () => { + process.argv = ['node', 'reconcile-replay.ts', '', '2023-01-01', 'https://example.com/fixture.json']; + const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(); + + const code = await runReconcileReplayCli(); + expect(code).toBe(1); + + consoleErrorSpy.mockRestore(); + }); + }); + + // ----------------------------------------------------------------------- + // Environment variable validation + // ----------------------------------------------------------------------- + + describe('environment variable validation', () => { + it('returns error when DATABASE_URL is missing', async () => { + process.argv = ['node', 'reconcile-replay.ts', 'offering-1', '2023-01-01', 'https://example.com/fixture.json']; + delete process.env.DATABASE_URL; + process.env.REPLAY_SIGNING_SECRET = 'test-secret'; + const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(); + + // Mock fetch to succeed + mockFetch.mockResolvedValue({ + ok: true, + text: jest.fn().mockResolvedValue(JSON.stringify({ totalDistributed: '1000.00' })), + }); + + const code = await runReconcileReplayCli(); + expect(code).toBe(1); + expect(consoleErrorSpy).toHaveBeenCalledWith( + expect.stringContaining('DATABASE_URL') + ); + + consoleErrorSpy.mockRestore(); + }); + + it('returns error when REPLAY_SIGNING_SECRET is missing', async () => { + process.argv = ['node', 'reconcile-replay.ts', 'offering-1', '2023-01-01', 'https://example.com/fixture.json']; + process.env.DATABASE_URL = 'postgresql://localhost/test'; + delete process.env.REPLAY_SIGNING_SECRET; + const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(); + + // Mock fetch to succeed + mockFetch.mockResolvedValue({ + ok: true, + text: jest.fn().mockResolvedValue(JSON.stringify({ totalDistributed: '1000.00' })), + }); + + const code = await runReconcileReplayCli(); + expect(code).toBe(1); + expect(consoleErrorSpy).toHaveBeenCalledWith( + expect.stringContaining('REPLAY_SIGNING_SECRET') + ); + + consoleErrorSpy.mockRestore(); + }); + }); + + // ----------------------------------------------------------------------- + // Fixture error handling + // ----------------------------------------------------------------------- + + describe('fixture error handling', () => { + it('returns error when fixture fetch fails', async () => { + process.argv = ['node', 'reconcile-replay.ts', 'offering-1', '2023-01-01', 'https://example.com/fixture.json']; + process.env.DATABASE_URL = 'postgresql://localhost/test'; + process.env.REPLAY_SIGNING_SECRET = 'test-secret'; + const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(); + + mockFetch.mockRejectedValue(new Error('Network error')); + + const code = await runReconcileReplayCli(); + expect(code).toBe(1); + expect(consoleErrorSpy).toHaveBeenCalledWith( + expect.stringContaining('Failed to connect to fixture URL') + ); + + consoleErrorSpy.mockRestore(); + }); + + it('returns error when fixture returns non-JSON', async () => { + process.argv = ['node', 'reconcile-replay.ts', 'offering-1', '2023-01-01', 'https://example.com/fixture.json']; + process.env.DATABASE_URL = 'postgresql://localhost/test'; + process.env.REPLAY_SIGNING_SECRET = 'test-secret'; + const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(); + + mockFetch.mockResolvedValue({ + ok: true, + text: jest.fn().mockResolvedValue('Not JSON'), + }); + + const code = await runReconcileReplayCli(); + expect(code).toBe(1); + expect(consoleErrorSpy).toHaveBeenCalledWith( + expect.stringContaining('non-JSON content') + ); + + consoleErrorSpy.mockRestore(); + }); + }); + + // ----------------------------------------------------------------------- + // Full CLI execution + // ----------------------------------------------------------------------- + + describe('full CLI execution', () => { + it('produces a signed report on successful reconciliation', async () => { + process.argv = ['node', 'reconcile-replay.ts', 'offering-1', '2023-01-01', 'https://example.com/fixture.json']; + process.env.DATABASE_URL = 'postgresql://localhost/test'; + process.env.REPLAY_SIGNING_SECRET = 'test-secret'; + + const fixtureData = { totalDistributed: '1000.00' }; + mockFetch.mockResolvedValue({ + ok: true, + text: jest.fn().mockResolvedValue(JSON.stringify(fixtureData)), + }); + + const mockReconcile = require('../src/services/revenueReconciliationService').__mockReconcile; + mockReconcile.mockResolvedValue({ + offeringId: 'offering-1', + periodStart: new Date('2023-01-01'), + periodEnd: new Date(), + isBalanced: true, + discrepancies: [], + summary: { + totalRevenueReported: '1000.00', + totalPayouts: '1000.00', + discrepancyAmount: '0.00', + investorCount: 5, + payoutsProcessed: 10, + payoutsFailed: 0, + }, + checkedAt: new Date(), + }); + + const consoleSpy = jest.spyOn(console, 'log').mockImplementation(); + const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(); + + const code = await runReconcileReplayCli(); + expect(code).toBe(0); + + // Verify the output is valid JSON with signature + const output = consoleSpy.mock.calls[0][0]; + const parsed = JSON.parse(output); + expect(parsed.report).toBeDefined(); + expect(parsed.signature).toMatch(/^sha256=[a-f0-9]{64}$/); + expect(parsed.report.fixture_sha256).toBe(sha256(JSON.stringify(fixtureData))); + expect(parsed.report.parameters.offering_id).toBe('offering-1'); + + consoleSpy.mockRestore(); + consoleErrorSpy.mockRestore(); + }); + + it('returns exit code 1 when reconciliation is not balanced', async () => { + process.argv = ['node', 'reconcile-replay.ts', 'offering-1', '2023-01-01', 'https://example.com/fixture.json']; + process.env.DATABASE_URL = 'postgresql://localhost/test'; + process.env.REPLAY_SIGNING_SECRET = 'test-secret'; + + const fixtureData = { totalDistributed: '1000.00' }; + mockFetch.mockResolvedValue({ + ok: true, + text: jest.fn().mockResolvedValue(JSON.stringify(fixtureData)), + }); + + const mockReconcile = require('../src/services/revenueReconciliationService').__mockReconcile; + mockReconcile.mockResolvedValue({ + offeringId: 'offering-1', + periodStart: new Date('2023-01-01'), + periodEnd: new Date(), + isBalanced: false, + discrepancies: [ + { + type: 'REVENUE_MISMATCH', + severity: 'error', + message: 'Revenue mismatch', + details: {}, + offeringId: 'offering-1', + }, + ], + summary: { + totalRevenueReported: '1000.00', + totalPayouts: '900.00', + discrepancyAmount: '100.00', + investorCount: 5, + payoutsProcessed: 10, + payoutsFailed: 0, + }, + checkedAt: new Date(), + }); + + const consoleSpy = jest.spyOn(console, 'log').mockImplementation(); + const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(); + + const code = await runReconcileReplayCli(); + expect(code).toBe(1); + + consoleSpy.mockRestore(); + consoleErrorSpy.mockRestore(); + }); + + it('handles reconciliation service errors', async () => { + process.argv = ['node', 'reconcile-replay.ts', 'offering-1', '2023-01-01', 'https://example.com/fixture.json']; + process.env.DATABASE_URL = 'postgresql://localhost/test'; + process.env.REPLAY_SIGNING_SECRET = 'test-secret'; + + const fixtureData = { totalDistributed: '1000.00' }; + mockFetch.mockResolvedValue({ + ok: true, + text: jest.fn().mockResolvedValue(JSON.stringify(fixtureData)), + }); + + const mockReconcile = require('../src/services/revenueReconciliationService').__mockReconcile; + mockReconcile.mockRejectedValue(new Error('Database connection failed')); + + const consoleSpy = jest.spyOn(console, 'log').mockImplementation(); + const consoleErrorSpy = jest.spyOn(console, 'error').mockImplementation(); + + const code = await runReconcileReplayCli(); + expect(code).toBe(1); + expect(consoleErrorSpy).toHaveBeenCalledWith( + expect.stringContaining('Reconciliation replay failed') + ); + + consoleSpy.mockRestore(); + consoleErrorSpy.mockRestore(); + }); + }); +}); diff --git a/src/lib/horizonFixtureAdapter.ts b/src/lib/horizonFixtureAdapter.ts new file mode 100644 index 00000000..73bc9852 --- /dev/null +++ b/src/lib/horizonFixtureAdapter.ts @@ -0,0 +1,187 @@ +import { Errors } from '../lib/errors'; +import { MetricsCollector } from '../lib/metrics'; +import { AuditEvent, SecurityAuditRepository } from '../security/types'; +import { OnChainRevenueState, StellarRevenueClient } from '../services/revenueReconciliationService'; +import { createHash, createHmac, randomUUID } from 'crypto'; + +interface SignedReport { + offeringId: string; + periodStart: Date; + periodEnd: Date; + isBalanced: boolean; + discrepancies: any[]; + summary: any; + checkedAt: Date; + fixtureHash?: string; + signature: string; +} + +export class HorizonFixtureAdapter implements StellarRevenueClient { + private readonly metrics?: MetricsCollector; + private readonly securityAuditRepo?: SecurityAuditRepository; + private readonly signingKey?: string; + + constructor( + options?: { + metrics?: MetricsCollector; + securityAuditRepo?: SecurityAuditRepository; + signingKey?: string; + } + ) { + this.metrics = options?.metrics; + this.securityAuditRepo = options?.securityAuditRepo; + this.signingKey = options?.signingKey; + } + + async getRevenueState(fixtureUrl: string): Promise { + let responseBody: string; + try { + const response = await fetch(fixtureUrl); + if (!response.ok) { + throw new Error( + `Fixture URL returned HTTP ${response.status}${response.statusText ? ` ${response.statusText}` : ''}` + ); + } + responseBody = await response.text(); + if (!responseBody || responseBody.trim().length === 0) { + throw new Error(`Fixture URL "${fixtureUrl}" returned an empty response body`); + } + } catch (err) { + if (err instanceof Error && err.message.includes('HTTP')) throw err; + throw new Error( + `Failed to fetch fixture from "${fixtureUrl}": ${err instanceof Error ? err.message : String(err)}` + ); + } + + let data: Record; + try { + data = JSON.parse(responseBody); + } catch { + throw new Error( + `Fixture URL "${fixtureUrl}" returned non-JSON content. ` + + 'Ensure the URL points to a valid Horizon snapshot JSON file.' + ); + } + + if (this.securityAuditRepo) { + await this.securityAuditRepo.record({ + id: randomUUID(), + type: 'AUTHENTICATION', + action: 'HORIZON_FIXTURE_ACCESS', + resource: `horizon_fixture:${fixtureUrl}`, + outcome: 'SUCCESS', + details: { + fixtureUrl, + timestamp: new Date().toISOString(), + }, + securityContext: { + requestId: 'horizon-fixture-adapter', + ipAddress: '0.0.0.0', + userAgent: 'horizon-fixture-adapter/1.0', + timestamp: new Date(), + }, + timestamp: new Date(), + }); + } + + this.metrics?.incrementCounter('horizon_fixture_access_total', { + fixtureUrl, + }); + + return { + totalDistributed: String(data.totalDistributed ?? '0.00'), + }; + } + + async getFixtureHash(fixtureUrl: string): Promise { + let responseBody: string; + try { + const response = await fetch(fixtureUrl); + if (!response.ok) { + throw new Error( + `Fixture URL returned HTTP ${response.status}${response.statusText ? ` ${response.statusText}` : ''}` + ); + } + responseBody = await response.text(); + if (!responseBody || responseBody.trim().length === 0) { + throw new Error(`Fixture URL "${fixtureUrl}" returned an empty response body`); + } + } catch (err) { + if (err instanceof Error && err.message.includes('HTTP')) throw err; + throw new Error( + `Failed to fetch fixture from "${fixtureUrl}" for hashing: ${err instanceof Error ? err.message : String(err)}` + ); + } + + const hash = createHash('sha256').update(responseBody, 'utf8').digest('hex'); + + if (this.securityAuditRepo) { + await this.securityAuditRepo.record({ + id: randomUUID(), + type: 'AUTHENTICATION', + action: 'HORIZON_FIXTURE_HASH', + resource: `horizon_fixture:${fixtureUrl}`, + outcome: 'SUCCESS', + details: { + fixtureUrl, + hash, + timestamp: new Date().toISOString(), + }, + securityContext: { + requestId: 'horizon-fixture-adapter', + ipAddress: '0.0.0.0', + userAgent: 'horizon-fixture-adapter/1.0', + timestamp: new Date(), + }, + timestamp: new Date(), + }); + } + + this.metrics?.incrementCounter('horizon_fixture_hash_total', { + fixtureUrl, + }); + + return hash; + } + + async signReport(report: any): Promise { + if (!this.signingKey) { + throw Errors.internal('Signing key not configured'); + } + + const canonical = JSON.stringify(report); + const hmac = createHmac('sha256', this.signingKey); + hmac.update(canonical); + const signature = `sha256=${hmac.digest('hex')}`; + + if (this.securityAuditRepo) { + await this.securityAuditRepo.record({ + id: randomUUID(), + type: 'AUTHENTICATION', + action: 'REPORT_SIGNING', + resource: `report:${report.offeringId}`, + outcome: 'SUCCESS', + details: { + reportId: report.offeringId, + timestamp: new Date().toISOString(), + }, + securityContext: { + requestId: 'horizon-fixture-adapter', + ipAddress: '0.0.0.0', + userAgent: 'horizon-fixture-adapter/1.0', + timestamp: new Date(), + }, + timestamp: new Date(), + }); + } + + this.metrics?.incrementCounter('report_signing_total', { + reportId: report.offeringId, + }); + + return { + ...report, + signature, + }; + } +} diff --git a/src/lib/metrics.ts b/src/lib/metrics.ts index d4181f42..71034098 100644 --- a/src/lib/metrics.ts +++ b/src/lib/metrics.ts @@ -657,6 +657,40 @@ export class MetricsCollector { lines.push(`${name}${labelStr} ${value} ${timestamp}`); } + // Export histograms + for (const [key, observations] of this.histograms.entries()) { + const { name, labels } = this.parseMetricKey(key); + const metadata = this.metricMetadata.get(name); + + if (metadata?.help) { + lines.push(`# HELP ${name} ${metadata.help}`); + } + lines.push(`# TYPE ${name} histogram`); + + const data = this.calculateHistogram(observations); + const labelStr = labels && Object.keys(labels).length > 0 + ? `{${Object.entries(labels).map(([k, v]) => `${k}="${v}"`).join(',')}}` + : ''; + + // Export buckets + for (const bucket of data.buckets) { + const bucketLabel = labelStr + ? `{${labelStr.slice(1, -1)},le="${bucket.le}"}` + : `{le="${bucket.le}"}`; + lines.push(`${name}_bucket${bucketLabel} ${bucket.count} ${timestamp}`); + } + + // Export +Inf bucket + const infLabel = labelStr + ? `{${labelStr.slice(1, -1)},le="+Inf"}` + : `{le="+Inf"}`; + lines.push(`${name}_bucket${infLabel} ${data.count} ${timestamp}`); + + // Export sum and count + lines.push(`${name}_sum${labelStr} ${data.sum} ${timestamp}`); + lines.push(`${name}_count${labelStr} ${data.count} ${timestamp}`); + } + return lines.join('\n') + '\n'; } diff --git a/src/services/fxConversionEngine.ts b/src/services/fxConversionEngine.ts index 4c793777..3cc617f1 100644 --- a/src/services/fxConversionEngine.ts +++ b/src/services/fxConversionEngine.ts @@ -1,10 +1,24 @@ +import { randomUUID } from 'crypto'; import { Decimal } from '../lib/decimal'; import { Errors } from '../lib/errors'; import { MetricsCollector } from '../lib/metrics'; -import { SecurityAuditRepository } from '../security/types'; +import { AuditEvent, SecurityAuditRepository } from '../security/types'; export type ConversionPathType = 'direct' | 'inverse' | 'triangulated'; +/** + * @notice Reasons why a stale rate was tolerated or a fallback was used. + * @dev Recorded in audit events so auditors can trace rate provenance. + */ +export enum FxFallbackReason { + /** The primary rate was fresh enough — no fallback needed. */ + NONE = 'NONE', + /** A substitute rate provider supplied a fresh rate. */ + SUBSTITUTE_PROVIDER_USED = 'SUBSTITUTE_PROVIDER_USED', + /** The primary rate was stale but tolerance was enabled. */ + STALE_RATE_TOLERATED = 'STALE_RATE_TOLERATED', +} + export interface ConversionPath { type: ConversionPathType; description: string; @@ -20,11 +34,6 @@ export interface ExchangeRate { ttlMs: number; } -export enum FxFallbackReason { - SUBSTITUTE_PROVIDER_USED = 'SUBSTITUTE_PROVIDER_USED', - STALE_RATE_TOLERATED = 'STALE_RATE_TOLERATED', -} - /** * @notice A single hop in a triangulation chain. * @dev Per-hop records are attached to FxConversionResult.hops so auditors @@ -69,6 +78,11 @@ export interface FxConversionResult { * Direct/inverse conversions have an empty array. */ hops: FxHop[]; + /** + * The SHA-256 hash of the Horizon fixture URL used for this conversion. + * Only populated if a fixture was used. + */ + fixtureHash?: string; } export interface RateProvider { @@ -79,6 +93,7 @@ const METRIC_STALE_RATE_REJECTED = 'fx_stale_rate_rejected_total'; const METRIC_CONVERSIONS_TOTAL = 'fx_conversions_total'; const METRIC_TRIANGULATIONS_TOTAL = 'fx_triangulations_total'; const METRIC_TRIANGULATION_HOPS = 'fx_triangulation_hops'; +const METRIC_STALE_FALLBACK_STALENESS_MS = 'fx_stale_fallback_staleness_ms'; /** Default maximum number of intermediate hops allowed in a triangulation chain. */ const DEFAULT_MAX_HOPS = 2; @@ -86,6 +101,7 @@ const DEFAULT_MAX_HOPS = 2; export class FxConversionEngine { private readonly defaultBucketIncrement: Decimal; private readonly metrics?: MetricsCollector; + private readonly auditRepository?: SecurityAuditRepository; private readonly fallbackRateProvider?: RateProvider; /** * Maximum number of intermediate hops permitted in a triangulation chain. @@ -106,10 +122,13 @@ export class FxConversionEngine { * Must be a positive integer ≥ 1. */ maxHops?: number; + /** Optional audit repository for recording rate-fallback events. */ + auditRepository?: SecurityAuditRepository; } ) { this.defaultBucketIncrement = new Decimal(options?.defaultBucketIncrement ?? '0.01'); this.metrics = options?.metrics; + this.auditRepository = options?.auditRepository; this.fallbackRateProvider = options?.fallbackRateProvider; const maxHops = options?.maxHops ?? DEFAULT_MAX_HOPS; @@ -130,6 +149,7 @@ export class FxConversionEngine { allowStaleFallback?: boolean; auditUserId?: string; auditSessionId?: string; + fixtureHash?: string; frozenContext?: string; } ): Promise { @@ -232,6 +252,41 @@ export class FxConversionEngine { ); } + // Record audit event and metrics when a fallback was used + if (fallbackUsed && fallbackReason) { + this.metrics?.recordHistogram(METRIC_STALE_FALLBACK_STALENESS_MS, this.rateAgeMs(activeRate), { from, to }); + + if (this.auditRepository) { + const auditEvent: AuditEvent = { + id: randomUUID(), + type: 'SECURITY_VIOLATION', + userId: options?.auditUserId ?? 'system', + action: 'FX_STALE_RATE_FALLBACK', + resource: `fx_conversion:${from}/${to}`, + outcome: 'SUCCESS', + details: { + pair: `${from}/${to}`, + reason: fallbackReason, + substituteRateId: activeRate.id, + rateAgeMs: this.rateAgeMs(activeRate), + maxAgeMs: maxAge, + }, + securityContext: { + requestId: options?.auditSessionId ?? 'fx-conversion-engine', + ipAddress: '0.0.0.0', + userAgent: 'fx-conversion-engine/1.0', + timestamp: new Date(), + }, + timestamp: new Date(), + }; + + await this.auditRepository.record(auditEvent).catch((err) => { + // Best-effort: log but don't fail the conversion + console.error('[FxConversionEngine] Failed to record audit event:', err); + }); + } + } + const side = options?.side ?? 'mid'; const effectiveRate = this.resolveSide(activeRate, side); const rawOutput = amount.multiply(effectiveRate); @@ -255,6 +310,7 @@ export class FxConversionEngine { path: { type: 'direct', description: `${from}/${to}` }, roundedToIncrement: rounded, hops: [], + fixtureHash: options?.fixtureHash, }; } @@ -285,6 +341,7 @@ export class FxConversionEngine { bucketIncrement?: Decimal; side?: 'bid' | 'ask' | 'mid'; maxRateAgeMs?: number; + fixtureHash?: string; } ): Promise { const vias = Array.isArray(viaOrVias) ? viaOrVias : [viaOrVias]; @@ -296,10 +353,10 @@ export class FxConversionEngine { } } - // Enforce hop budget. Each "via" currency adds 2 legs → 1 hop-pair. - // We count hops as the number of intermediate legs, so a single-via - // chain = 2 hops. Multi-via not yet supported (spec says max 2 hops). - const hopCount = vias.length === 1 ? 2 : vias.length + 1; + // Enforce hop budget. Each triangulation uses ONE via currency, + // which adds 2 legs → 2 hops. Multiple vias are alternatives, + // not chained — we try each until one succeeds. + const hopCount = 2; if (hopCount > this.maxHops) { throw Errors.badRequest( `Triangulation requires ${hopCount} hops but maxHops is ${this.maxHops}. ` + @@ -379,6 +436,7 @@ export class FxConversionEngine { }, roundedToIncrement: leg1Result.roundedToIncrement || leg2Result.roundedToIncrement, hops, + fixtureHash: options?.fixtureHash, }; } diff --git a/src/services/revenueReconciliationService.ts b/src/services/revenueReconciliationService.ts index 946e9dc9..300a1085 100644 --- a/src/services/revenueReconciliationService.ts +++ b/src/services/revenueReconciliationService.ts @@ -21,16 +21,18 @@ import { DistributionRepository, DistributionRun, Payout } from '../db/repositor import { InvestmentRepository, Investment } from '../db/repositories/investmentRepository'; import { OfferingRepository } from '../db/repositories/offeringRepository'; import { logger, Logger, LogLevel } from '../lib/logger'; -import { - classifyStellarRPCFailure, +import { + classifyStellarRPCFailure, StellarRPCFailureClass, - StellarRPCFailureContext + StellarRPCFailureContext } from '../lib/stellarRpcFailure'; import { Errors } from '../lib/errors'; -import { +import { StellarTransactionVerifier, - TransactionVerificationResult + TransactionVerificationResult } from '../lib/stellarTransactionVerifier'; +import { HorizonFixtureAdapter } from '../lib/horizonFixtureAdapter'; +import { createHash } from 'crypto'; export interface OnChainRevenueState { totalDistributed: string; @@ -97,6 +99,7 @@ export interface ReconciliationOptions { checkInvestorAllocations?: boolean; validateChainEvents?: boolean; logger?: Logger; + horizonFixtureUrl?: string; } const DEFAULT_TOLERANCE = 0.01; @@ -112,7 +115,9 @@ export class RevenueReconciliationService { constructor( private readonly db: Pool, private readonly stellarClient?: StellarRevenueClient, - txVerifier?: StellarTransactionVerifier + txVerifier?: StellarTransactionVerifier, + private readonly horizonFixtureAdapter?: HorizonFixtureAdapter, + private readonly signingKey?: string ) { this.revenueReportRepo = new RevenueReportRepository(db); this.distributionRepo = new DistributionRepository(db); @@ -120,6 +125,8 @@ export class RevenueReconciliationService { this.offeringRepo = new OfferingRepository(db); this.logger = logger.child({ service: 'RevenueReconciliationService' }); this.txVerifier = txVerifier; + this.horizonFixtureAdapter = horizonFixtureAdapter; + this.signingKey = signingKey; } /** @@ -135,6 +142,8 @@ export class RevenueReconciliationService { periodEnd: Date, options: ReconciliationOptions = {} ): Promise { + const horizonFixtureUrl = options.horizonFixtureUrl; + const fixtureHash = horizonFixtureUrl ? await this.horizonFixtureAdapter?.getFixtureHash(horizonFixtureUrl) : undefined; const tolerance = options.tolerance ?? DEFAULT_TOLERANCE; const discrepancies: ReconciliationDiscrepancy[] = []; @@ -189,9 +198,9 @@ export class RevenueReconciliationService { } // Drift Detection - if (this.stellarClient) { +if (this.stellarClient || horizonFixtureUrl) { try { - const driftResult = await this.detectChainDrift(offeringId); + const driftResult = await this.detectChainDrift(offeringId, horizonFixtureUrl); if (driftResult.hasDrift) { discrepancies.push({ type: 'CHAIN_DRIFT_DETECTED', @@ -200,7 +209,7 @@ export class RevenueReconciliationService { details: driftResult, offeringId, }); - + this.logger.error('Revenue reconciliation drift detected', { offeringId, ...driftResult, @@ -218,7 +227,7 @@ export class RevenueReconciliationService { details: { error: String(error), failureClass: failure.class }, offeringId, }); - + this.logger.warn('Failed to fetch on-chain state during reconciliation', { offeringId, failureClass: failure.class, @@ -378,10 +387,15 @@ export class RevenueReconciliationService { payoutsProcessed: this.countProcessedPayouts(relevantRuns), payoutsFailed: totalFailedPayouts, chainDrift: discrepancies.find(d => d.type === 'CHAIN_DRIFT_DETECTED')?.details as any, + fixtureHash, }, checkedAt: new Date(), }; + if (this.horizonFixtureAdapter && this.signingKey) { + return this.horizonFixtureAdapter.signReport(result); + } + return result; } catch (error) { this.logger.error('Reconciliation process failed', { @@ -494,11 +508,12 @@ export class RevenueReconciliationService { /** * Detect drift between local DB and on-chain state */ - async detectChainDrift(offeringId: string): Promise<{ + async detectChainDrift(offeringId: string, horizonFixtureUrl?: string): Promise<{ hasDrift: boolean; onChainAmount: string; localAmount: string; drift: string; + fixtureHash?: string; }> { if (!this.stellarClient) { throw Errors.internal('Stellar client not configured for drift detection'); @@ -506,11 +521,36 @@ export class RevenueReconciliationService { const offering = await this.offeringRepo.findById(offeringId); if (!offering || !offering.contract_address) { + if (!horizonFixtureUrl) { + return { + hasDrift: false, + onChainAmount: '0.00', + localAmount: '0.00', + drift: '0.00', + }; + } + + if (!this.horizonFixtureAdapter) { + throw Errors.internal('Horizon fixture adapter not configured'); + } + + const fixtureState = await this.horizonFixtureAdapter.getRevenueState(horizonFixtureUrl); + const stats = await this.distributionRepo.getAggregateStats(offeringId); + + const onChainAmount = parseFloat(fixtureState.totalDistributed); + const localAmount = parseFloat(stats.totalDistributed); + const drift = Math.abs(onChainAmount - localAmount); + + const hasDrift = drift > DEFAULT_TOLERANCE; + + const fixtureHash = horizonFixtureUrl ? createHash('sha256').update(horizonFixtureUrl).digest('hex') : undefined; + return { - hasDrift: false, - onChainAmount: '0.00', - localAmount: '0.00', - drift: '0.00', + hasDrift, + onChainAmount: onChainAmount.toFixed(2), + localAmount: localAmount.toFixed(2), + drift: drift.toFixed(2), + fixtureHash, }; }