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

src/events/domain-event.types.ts

@@ -59,4 +59,14 @@ export interface RiskEvaluatedPayload {
   transactionId?: string;
   score: number;
   band: string;
+  factors?: Record<string, unknown>;
+  canAutoExecute?: boolean;
+}
+
+export interface TransactionCreatedPayload {
+  transactionId: string;
+  walletId: string;
+  amount: string;
+  asset: string;
+  recipientAddress: string;
 }

src/events/event-names.ts

@@ -79,6 +79,7 @@ export const WEBHOOK_EVENTS: DomainEventNameType[] = [
   DomainEventName.ProposalApproved,
   DomainEventName.ProposalRejected,
   DomainEventName.TransactionCompleted,
+  DomainEventName.TransactionCreated,
   DomainEventName.TransactionFailed,
   DomainEventName.PolicyViolated,
   DomainEventName.BudgetExceeded,

src/modules/risk/risk.module.ts

@@ -2,8 +2,11 @@ import { Module } from '@nestjs/common';
 import { RiskController } from './risk.controller';
 import { RiskService } from './risk.service';
 import { RiskEngine } from './risk.engine';
+import { EventsModule } from '../../events/events.module';
+import { DatabaseModule } from '../../database/database.module';
 
 @Module({
+  imports: [EventsModule, DatabaseModule],
   controllers: [RiskController],
   providers: [RiskService, RiskEngine],
   exports: [RiskService, RiskEngine],

src/modules/risk/risk.service.spec.ts

@@ -19,10 +19,19 @@
   return { emit: vi.fn().mockResolvedValue(undefined) } as unknown as Pick<EventBusService, 'emit'> & { emit: ReturnType<typeof vi.fn> };
 }
 
+function createPrisma() {
+  return {
+    transaction: {
+      update: vi.fn().mockResolvedValue({ id: 'tx-1', riskScore: 10, riskBand: 'LOW' }),
+    },
+  };
+}
+
 describe('RiskService', () => {
   it('emits a RiskEvaluated event with full factor breakdown', async () => {
     const eventBus = createEventBus();
-    const service = new RiskService(new RiskEngine(), eventBus as unknown as EventBusService);
+    const prisma = createPrisma();
+    const service = new RiskService(new RiskEngine(), eventBus as unknown as EventBusService, prisma as any);
 
     const assessment = await service.evaluate('org-1', lowRisk, {
       transactionId: 'tx-1',
@@ -45,7 +54,8 @@
 
   it('assess() returns a result without emitting events', async () => {
     const eventBus = createEventBus();
-    const service = new RiskService(new RiskEngine(), eventBus as unknown as EventBusService);
+    const prisma = createPrisma();
+    const service = new RiskService(new RiskEngine(), eventBus as unknown as EventBusService, prisma as any);
 
     const assessment = service.assess(lowRisk);
     expect(assessment.band).toBe(RiskBand.LOW);
@@ -55,7 +65,8 @@
 
   it('passes config overrides through to the engine', async () => {
     const eventBus = createEventBus();
-    const service = new RiskService(new RiskEngine(), eventBus as unknown as EventBusService);
+    const prisma = createPrisma();
+    const service = new RiskService(new RiskEngine(), eventBus as unknown as EventBusService, prisma as any);
 
     const assessment = service.assess(
       { ...lowRisk, amount: 100 },
@@ -64,4 +75,42 @@
     const amountFactor = assessment.factors.find((f) => f.factor === 'amount');
     expect(amountFactor!.contribution).toBe(30);
   });
+
+  it('handles transaction created event, updates transaction and emits risk evaluated', async () => {
+    const eventBus = createEventBus();
+    const prisma = createPrisma();
+    const service = new RiskService(new RiskEngine(), eventBus as unknown as EventBusService, prisma as any);
+
+    await service.handleTransactionCreated({
+      name: 'transaction.created' as any,
+      organizationId: 'org-1',
+      aggregateType: 'transaction',
+      aggregateId: 'tx-1',
+      actorId: 'agent-1',
+      occurredAt: new Date(),
+      payload: {
+        transactionId: 'tx-1',
+        walletId: 'wallet-1',
+        amount: '50',
+        asset: 'USDC',
+        recipientAddress: 'GA123',
+      },
+    });
+
+    expect(prisma.transaction.update).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: { id: 'tx-1' },
+        data: expect.objectContaining({
+          riskScore: expect.any(Number),
+          riskBand: expect.any(String),
+        }),
+      }),
+    );
+
+    const emitMock = eventBus.emit as ReturnType<typeof vi.fn>;
+    expect(emitMock).toHaveBeenCalled();
+    const [eventName, payload] = emitMock.mock.calls[0];
+    expect(eventName).toBe('risk.evaluated');
+    expect(payload.transactionId).toBe('tx-1');
+  });
 });

src/modules/risk/risk.service.ts

@@ -3,6 +3,10 @@ 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 { OnEvent } from '@nestjs/event-emitter';
+import { DomainEventEnvelope, TransactionCreatedPayload } from '../../events/domain-event.types';
+import { PrismaService } from '../../database/prisma.service';
+import { Logger } from '@nestjs/common';
 
 /**
  * Application-facing risk service. Wraps the pure {@link RiskEngine}, emits a
@@ -11,9 +15,12 @@ import { DomainEventName } from '../../events/event-names';
  */
 @Injectable()
 export class RiskService {
+  private readonly logger = new Logger(RiskService.name);
+
   constructor(
     private readonly engine: RiskEngine,
     private readonly eventBus: EventBusService,
+    private readonly prisma: PrismaService,
   ) {}
 
   /**
@@ -53,4 +60,53 @@ export class RiskService {
   ): RiskAssessment {
     return this.engine.assess(input, config, rules);
   }
+
+  @OnEvent(DomainEventName.TransactionCreated)
+  async handleTransactionCreated(envelope: DomainEventEnvelope<TransactionCreatedPayload>): Promise<void> {
+    try {
+      const { transactionId, amount } = envelope.payload;
+      const organizationId = envelope.organizationId;
+      if (!transactionId || !organizationId) {
+        return;
+      }
+
+      const amountNum = parseFloat(amount) || 0;
+      const assessment = this.engine.assess({
+        amount: amountNum,
+        asset: envelope.payload.asset ?? 'USDC',
+        knownRecipient: true,
+        recentTransactionCount: 1,
+        walletAgeDays: 30,
+        policyViolations: 0,
+        hourUtc: new Date().getUTCHours(),
+      });
+
+      await this.prisma.transaction.update({
+        where: { id: transactionId },
+        data: {
+          riskScore: assessment.score,
+          riskBand: assessment.band,
+        },
+      });
+
+      await this.eventBus.emit(
+        DomainEventName.RiskEvaluated,
+        {
+          transactionId,
+          score: assessment.score,
+          band: assessment.band,
+          factors: assessment.factors,
+          canAutoExecute: assessment.canAutoExecute,
+        },
+        {
+          organizationId,
+          actorId: envelope.actorId,
+          aggregateType: 'transaction',
+          aggregateId: transactionId,
+        },
+      );
+    } catch (error) {
+      this.logger.error(`Failed to process risk scoring for transaction: ${error instanceof Error ? error.message : String(error)}`);
+    }
+  }
 }