From 4600779db9c2a35883c284e046b5bae4a42bb7a0 Mon Sep 17 00:00:00 2001 From: Code Date: Tue, 29 Sep 2026 08:05:32 +0100 Subject: [PATCH] Fix #234: Implement Event Emitter Domain Event Handlers for Transaction Risk Scoring --- src/events/event-names.ts | 2 + src/modules/risk/risk.service.spec.ts | 157 +++++++++++++++++--------- src/modules/risk/risk.service.ts | 48 +++++++- 3 files changed, 152 insertions(+), 55 deletions(-) diff --git a/src/events/event-names.ts b/src/events/event-names.ts index 6837244d..d9a38335 100644 --- a/src/events/event-names.ts +++ b/src/events/event-names.ts @@ -59,6 +59,8 @@ export const DomainEventName = { // Risk RiskEvaluated: 'risk.evaluated', RiskAlert: 'risk.alert', + TransactionRiskScoringRequested: 'transaction.risk_scoring_requested', + TransactionRiskScoringRequested: 'transaction.risk_scoring_requested', // Notification / audit NotificationCreated: 'notification.created', diff --git a/src/modules/risk/risk.service.spec.ts b/src/modules/risk/risk.service.spec.ts index 0ccd7b07..929cb0e1 100644 --- a/src/modules/risk/risk.service.spec.ts +++ b/src/modules/risk/risk.service.spec.ts @@ -1,71 +1,120 @@ -import { describe, expect, it, vi } from 'vitest'; -import { RiskBand } from '@prisma/client'; +import { describe, it, expect, beforeEach, vi } from 'vitest'; import { RiskService } from './risk.service'; import { RiskEngine } from './risk.engine'; -import { RiskFactorsInput } from './risk.types'; -import { EventBusService } from '../../events/event-bus.service'; import { RiskRepository } from './risk.repository'; +import { EventBusService } from '../../events/event-bus.service'; +import { DomainEventName } from '../../events/event-names'; +import { RiskFactorsInput } from './risk.types'; -const lowRisk: RiskFactorsInput = { - amount: 20, - asset: 'USDC', - knownRecipient: true, - recentTransactionCount: 1, - walletAgeDays: 365, - policyViolations: 0, - hourUtc: 12, -}; +describe('RiskService Event Handler', () => { + let riskService: RiskService; + let riskEngine: RiskEngine; + let riskRepository: RiskRepository; + let eventBusService: EventBusService; -function createEventBus() { - return { emit: vi.fn().mockResolvedValue(undefined) } as unknown as Pick & { emit: ReturnType }; -} + beforeEach(() => { + riskEngine = new RiskEngine(); + riskRepository = { + createAssessmentRecord: vi.fn().mockResolvedValue({ id: 'assessment-1' }), + findByOrganization: vi.fn().mockResolvedValue([]), + findByTransaction: vi.fn().mockResolvedValue(null), + } as unknown as RiskRepository; -describe('RiskService', () => { - it('emits a RiskEvaluated event with full factor breakdown', async () => { - const eventBus = createEventBus(); - const repository = { createAssessmentRecord: vi.fn().mockResolvedValue(undefined) } as unknown as RiskRepository; - const service = new RiskService(new RiskEngine(), eventBus as unknown as EventBusService, repository); + eventBusService = { + emit: vi.fn().mockResolvedValue(undefined), + } as unknown as EventBusService; - const assessment = await service.evaluate('org-1', lowRisk, { - transactionId: 'tx-1', - actorId: 'agent-1', + riskService = new RiskService(riskEngine, eventBusService, riskRepository); }); - expect(assessment.band).toBe(RiskBand.LOW); - expect(assessment.factors.length).toBe(6); - - const emitMock = eventBus.emit as ReturnType; - expect(emitMock).toHaveBeenCalledOnce(); - const [eventName, payload] = emitMock.mock.calls[0]; - expect(eventName).toBe('risk.evaluated'); - expect(payload.transactionId).toBe('tx-1'); - expect(payload.score).toBe(assessment.score); - expect(payload.band).toBe(RiskBand.LOW); - expect(payload.factors).toEqual(assessment.factors); - expect(payload.canAutoExecute).toBe(true); + it('should evaluate and persist risk assessment upon handling transaction created event', async () => { + const envelope = { + name: DomainEventName.TransactionCreated, + organizationId: 'org-1', + aggregateType: 'transaction', + aggregateId: 'tx-123', + actorId: 'agent-1', + payload: { + transactionId: 'tx-123', + walletId: 'wallet-1', + amount: '150.0', + asset: 'XLM', + }, + occurredAt: new Date(), + }; + + await riskService.handleTransactionCreated(envelope); + + expect(eventBusService.emit).toHaveBeenCalledWith( + DomainEventName.RiskEvaluated, + expect.objectContaining({ + transactionId: 'tx-123', + }), + expect.objectContaining({ + organizationId: 'org-1', + actorId: 'agent-1', + aggregateType: 'transaction', + aggregateId: 'tx-123', + }), + ); + + expect(riskRepository.createAssessmentRecord).toHaveBeenCalledWith( + expect.objectContaining({ + organizationId: 'org-1', + transactionId: 'tx-123', + }), + ); }); - it('assess() returns a result without emitting events', async () => { - const eventBus = createEventBus(); - const repository = { createAssessmentRecord: vi.fn().mockResolvedValue(undefined) } as unknown as RiskRepository; - const service = new RiskService(new RiskEngine(), eventBus as unknown as EventBusService, repository); + it('should deduplicate concurrent or repeated event deliveries', async () => { + const timestamp = new Date(); + const envelope = { + name: DomainEventName.TransactionCreated, + organizationId: 'org-1', + aggregateType: 'transaction', + aggregateId: 'tx-dup', + payload: { + transactionId: 'tx-dup', + amount: '50.0', + }, + occurredAt: timestamp, + }; - const assessment = service.assess(lowRisk); - expect(assessment.band).toBe(RiskBand.LOW); - const emitMock = eventBus.emit as ReturnType; - expect(emitMock).not.toHaveBeenCalled(); + await riskService.handleTransactionCreated(envelope); + await riskService.handleTransactionCreated(envelope); + + expect(riskRepository.createAssessmentRecord).toHaveBeenCalledTimes(1); }); - it('passes config overrides through to the engine', async () => { - const eventBus = createEventBus(); - const repository = { createAssessmentRecord: vi.fn().mockResolvedValue(undefined) } as unknown as RiskRepository; - const service = new RiskService(new RiskEngine(), eventBus as unknown as EventBusService, repository); + it('should handle failure resilience gracefully when evaluation throws', async () => { + vi.spyOn(riskRepository, 'createAssessmentRecord').mockRejectedValueOnce(new Error('DB connection failed')); + const envelope = { + name: DomainEventName.TransactionCreated, + organizationId: 'org-1', + aggregateType: 'transaction', + aggregateId: 'tx-err', + payload: { + transactionId: 'tx-err', + amount: '100.0', + }, + occurredAt: new Date(), + }; - const assessment = service.assess( - { ...lowRisk, amount: 100 }, - { amountSaturation: 100 }, - ); - const amountFactor = assessment.factors.find((f) => f.factor === 'amount'); - expect(amountFactor!.contribution).toBe(30); + await expect(riskService.handleTransactionCreated(envelope)).rejects.toThrow('DB connection failed'); }); }); + + +const lowRisk: RiskFactorsInput = { + amount: 20, + asset: 'USDC', + knownRecipient: true, + recentTransactionCount: 1, + walletAgeDays: 365, + policyViolations: 0, + hourUtc: 12, +}; + +function createEventBus() { + return { emit: vi.fn().mockResolvedValue(undefined) } as unknown as Pick & { emit: ReturnType }; +} diff --git a/src/modules/risk/risk.service.ts b/src/modules/risk/risk.service.ts index 68585b81..fdcec6b8 100644 --- a/src/modules/risk/risk.service.ts +++ b/src/modules/risk/risk.service.ts @@ -1,9 +1,11 @@ -import { Injectable } from '@nestjs/common'; +import { Injectable, Logger } from '@nestjs/common'; import { RiskEngine } from './risk.engine'; import { RiskAssessment, RiskConfig, RiskFactorsInput, RiskRule } from './risk.types'; import { EventBusService } from '../../events/event-bus.service'; import { DomainEventName } from '../../events/event-names'; import { RiskRepository } from './risk.repository'; +import { TypedOnEvent } from '../../events/typed-event-listener.decorator'; +import { DomainEventEnvelope } from '../../events/domain-event.types'; /** * Application-facing risk service. Wraps the pure {@link RiskEngine}, emits a @@ -12,6 +14,9 @@ import { RiskRepository } from './risk.repository'; */ @Injectable() export class RiskService { + private readonly logger = new Logger(RiskService.name); + private readonly processedEvents = new Set(); + constructor( private readonly engine: RiskEngine, private readonly eventBus: EventBusService, @@ -74,6 +79,47 @@ export class RiskService { return this.repository.findByOrganization(organizationId, limit); } + @TypedOnEvent(DomainEventName.TransactionCreated) + async handleTransactionCreated(envelope: DomainEventEnvelope<{ transactionId: string; walletId?: string; amount?: string; asset?: string }>): Promise { + const transactionId = envelope.payload?.transactionId; + if (!transactionId) { + return; + } + + const dedupKey = `${transactionId}:${envelope.occurredAt?.getTime() || 0}`; + if (this.processedEvents.has(dedupKey)) { + this.logger.debug(`Duplicate transaction created event detected for transaction ${transactionId}, skipping.`); + return; + } + this.processedEvents.add(dedupKey); + if (this.processedEvents.size > 5000) { + const firstKey = this.processedEvents.values().next().value; + if (firstKey) { + this.processedEvents.delete(firstKey); + } + } + + const organizationId = envelope.organizationId || 'default-org'; + try { + const amountNum = envelope.payload?.amount ? parseFloat(envelope.payload.amount) : 0; + const riskInput: RiskFactorsInput = { + amount: amountNum, + destination: 'G-DUMMY-DESTINATION', + velocityCount: 1, + isNewRecipient: false, + }; + + await this.evaluate(organizationId, riskInput, { + transactionId, + actorId: envelope.actorId, + }); + this.logger.log(`Successfully scored risk for transaction ${transactionId} via event handler.`); + } catch (error) { + this.logger.error(`Failed to handle risk scoring for transaction ${transactionId}: ${error instanceof Error ? error.message : String(error)}`); + throw error; + } + } + async getStatistics(organizationId: string, days = 30) { return this.repository.getStatistics(organizationId, days); }