[DripTide CI check] #234 Implement Event Emitter Domain Event Handlers for Transaction Risk Scoring

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',

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<EventBusService, 'emit'> & { emit: ReturnType<typeof vi.fn> };
-}
+  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<typeof vi.fn>;
-    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<typeof vi.fn>;
-    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<EventBusService, 'emit'> & { emit: ReturnType<typeof vi.fn> };
+}

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<string>();
+
   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<void> {
+    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);
   }