From 2ffcb1797ca478f0acb931e70ee0016edf4f5c47 Mon Sep 17 00:00:00 2001 From: Umar faruk Date: Tue, 30 Jun 2026 07:20:33 +0000 Subject: [PATCH] feat(aml): add configurable AML screening service with velocity checks - Add migration for aml_rules and aml_screening_results tables - aml_rules: DB-backed configurable rules (no hardcoded thresholds) - aml_screening_results: per-transaction rule evaluation audit log - Seeds 5 default rules: amount thresholds (1M/500K XAF), velocity checks (3/hr, 10/24hr), phone blacklist - Add AmlScreeningResultModel (src/models/amlScreeningResult.ts) - createBulk(), findByTransactionId(), list() with pagination - Add AmlScreeningService (src/services/amlScreening.ts) - Evaluates amount_threshold, velocity_check (Redis counters), and blacklisted_phone rules loaded from DB (1-min cache) - Velocity uses time-bucketed Redis keys shared across instances - Fails open on Redis unavailability (never blocks transactions) - Persists all evaluation results fire-and-forget - Integrate screening synchronously into processTransactionRequest - Runs before transaction creation; flagged txns get status=review and tags [aml-screened, aml-flagged] - Velocity counters incremented after successful commit - Admin notes populated with matched rule names - Add 32 unit tests (all passing) covering all rule types, edge cases, Redis failure isolation, caching, and DB persistence --- .../20260630_create_aml_screening_tables.sql | 141 +++++ src/controllers/transactionController.ts | 87 ++- src/models/amlScreeningResult.ts | 236 +++++++++ src/services/__tests__/amlScreening.test.ts | 499 ++++++++++++++++++ src/services/amlScreening.ts | 403 ++++++++++++++ 5 files changed, 1364 insertions(+), 2 deletions(-) create mode 100644 migrations/20260630_create_aml_screening_tables.sql create mode 100644 src/models/amlScreeningResult.ts create mode 100644 src/services/__tests__/amlScreening.test.ts create mode 100644 src/services/amlScreening.ts diff --git a/migrations/20260630_create_aml_screening_tables.sql b/migrations/20260630_create_aml_screening_tables.sql new file mode 100644 index 00000000..e1c7ffbe --- /dev/null +++ b/migrations/20260630_create_aml_screening_tables.sql @@ -0,0 +1,141 @@ +-- Migration: 20260630_create_aml_screening_tables +-- Description: Create aml_rules and aml_screening_results tables for +-- the configurable AML screening service with velocity counters. + +-- ─── aml_rules ─────────────────────────────────────────────────────────────── +-- Each row defines one screening rule. +-- rule_type drives which evaluation logic is used: +-- amount_threshold – block/flag if transaction amount >= config.threshold_xaf +-- velocity_check – block/flag if phone number has > config.max_count +-- transactions within config.window_seconds seconds +-- blacklisted_phone – block/flag if phone_number is in config.numbers[] + +CREATE TYPE aml_rule_type AS ENUM ( + 'amount_threshold', + 'velocity_check', + 'blacklisted_phone' +); + +CREATE TABLE IF NOT EXISTS aml_rules ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + rule_type aml_rule_type NOT NULL, + name VARCHAR(255) NOT NULL, + description TEXT, + -- JSONB payload; schema depends on rule_type (see docs per rule below) + config JSONB NOT NULL DEFAULT '{}', + enabled BOOLEAN NOT NULL DEFAULT TRUE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS idx_aml_rules_rule_type ON aml_rules (rule_type); +CREATE INDEX IF NOT EXISTS idx_aml_rules_enabled ON aml_rules (enabled); + +-- ─── aml_screening_results ─────────────────────────────────────────────────── +-- One row per (transaction × rule) evaluation. +-- triggered = TRUE means the rule matched and the transaction was flagged. + +CREATE TABLE IF NOT EXISTS aml_screening_results ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + transaction_id UUID NOT NULL, -- FK to transactions.id (soft – no FK for perf) + rule_id UUID NOT NULL, -- FK to aml_rules.id (soft) + rule_name VARCHAR(255) NOT NULL, -- snapshot of rule name at evaluation time + rule_type aml_rule_type NOT NULL, -- snapshot of rule type + triggered BOOLEAN NOT NULL DEFAULT FALSE, + -- Evaluation details: observed values, thresholds, etc. + details JSONB NOT NULL DEFAULT '{}', + screened_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS idx_aml_screening_transaction + ON aml_screening_results (transaction_id); +CREATE INDEX IF NOT EXISTS idx_aml_screening_rule + ON aml_screening_results (rule_id); +CREATE INDEX IF NOT EXISTS idx_aml_screening_triggered + ON aml_screening_results (triggered, screened_at DESC); +CREATE INDEX IF NOT EXISTS idx_aml_screening_screened_at + ON aml_screening_results (screened_at DESC); + +-- Auto-update updated_at on aml_rules +CREATE OR REPLACE FUNCTION update_aml_rules_updated_at() +RETURNS TRIGGER AS $$ +BEGIN + NEW.updated_at = NOW(); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS trg_aml_rules_updated_at ON aml_rules; +CREATE TRIGGER trg_aml_rules_updated_at + BEFORE UPDATE ON aml_rules + FOR EACH ROW EXECUTE FUNCTION update_aml_rules_updated_at(); + +-- ─── Seed default rules ─────────────────────────────────────────────────────── +-- These rules are active from first deployment. Operators can toggle enabled or +-- adjust config via the aml_rules table without code changes. + +-- Rule 1: Large single-transaction threshold (1 000 000 XAF) +INSERT INTO aml_rules (rule_type, name, description, config, enabled) +VALUES ( + 'amount_threshold', + 'Large Transaction Threshold', + 'Flag any single transaction whose amount meets or exceeds 1 000 000 XAF (approx. $1 500 USD). ' + 'Matches the Central Bank of Central African States (BEAC) reporting threshold.', + '{"threshold_xaf": 1000000}', + TRUE +) +ON CONFLICT DO NOTHING; + +-- Rule 2: Medium single-transaction alert (500 000 XAF) – lower severity flag +INSERT INTO aml_rules (rule_type, name, description, config, enabled) +VALUES ( + 'amount_threshold', + 'Elevated Amount Alert', + 'Flag transactions >= 500 000 XAF for enhanced due diligence. ' + 'Below the hard reporting threshold but warrants review.', + '{"threshold_xaf": 500000}', + TRUE +) +ON CONFLICT DO NOTHING; + +-- Rule 3: Velocity check – more than 3 transactions from same phone number in 1 hour +INSERT INTO aml_rules (rule_type, name, description, config, enabled) +VALUES ( + 'velocity_check', + 'High-Frequency Velocity Check', + 'Flag when the same phone number initiates more than 3 transactions within a 1-hour ' + 'rolling window. Detects rapid structuring or compromised-account abuse.', + '{"max_count": 3, "window_seconds": 3600}', + TRUE +) +ON CONFLICT DO NOTHING; + +-- Rule 4: Velocity check – more than 10 transactions from same phone in 24 hours +INSERT INTO aml_rules (rule_type, name, description, config, enabled) +VALUES ( + 'velocity_check', + 'Daily Velocity Check', + 'Flag when the same phone number initiates more than 10 transactions within 24 hours.', + '{"max_count": 10, "window_seconds": 86400}', + TRUE +) +ON CONFLICT DO NOTHING; + +-- Rule 5: Blacklisted phone numbers (sample list – replace/extend via DB update) +INSERT INTO aml_rules (rule_type, name, description, config, enabled) +VALUES ( + 'blacklisted_phone', + 'Blacklisted Phone Numbers', + 'Reject any transaction from a phone number on the AML blacklist. ' + 'The numbers array is managed operationally and updated without a code deployment.', + '{"numbers": ["+237600000000", "+237611111111", "+237622222222"]}', + TRUE +) +ON CONFLICT DO NOTHING; + +-- ─── Down migration (reference) ────────────────────────────────────────────── +-- To roll back: +-- DROP TABLE IF EXISTS aml_screening_results; +-- DROP TABLE IF EXISTS aml_rules; +-- DROP TYPE IF EXISTS aml_rule_type; +-- DROP FUNCTION IF EXISTS update_aml_rules_updated_at(); diff --git a/src/controllers/transactionController.ts b/src/controllers/transactionController.ts index af494792..8ce9740f 100644 --- a/src/controllers/transactionController.ts +++ b/src/controllers/transactionController.ts @@ -18,6 +18,7 @@ import { } from "../config/providers"; import type { TransactionJobData } from "../queue/transactionQueue"; import { amlService } from "../services/aml"; +import { amlScreeningService } from "../services/amlScreening"; import { generateFlaggedTransactionComplianceReport } from "../services/complianceReportService"; import { twoFactorWithdrawalService } from "../services/twoFactorWithdrawalService"; import { @@ -585,14 +586,36 @@ async function processTransactionRequest( } } + // ── AML Screening (synchronous, before transaction creation) ───── + // Evaluate every enabled AML rule against this request. If any rule + // fires, the transaction is created with status=pending_review so it + // cannot be processed until a compliance officer approves it. + // screenTransaction() logs all evaluations to aml_screening_results + // after we have a real transaction ID (see persistTransactionId below). + const screeningOutcome = await amlScreeningService.screenTransaction({ + transactionId: "pre-create", // placeholder; real ID is used after create + userId, + amount: requestAmount, + phoneNumber, + type, + }); + + const initialStatus = screeningOutcome.shouldFlag + ? TransactionStatus.Review + : TransactionStatus.Pending; + + const amlTags = screeningOutcome.shouldFlag + ? ["aml-screened", "aml-flagged"] + : ["aml-screened"]; + const transaction = await transactionModel.create({ type, amount: String(amount), phoneNumber, provider, stellarAddress, - status: TransactionStatus.Pending, - tags: [], + status: initialStatus, + tags: amlTags, notes, userId, idempotencyKey, @@ -601,6 +624,66 @@ async function processTransactionRequest( : null, locationMetadata: (req as any).geoLocation ?? null, }); + + // Now that we have a real transaction ID, persist the screening results. + // This call is fire-and-forget inside screenTransaction; re-trigger with + // the real ID so the DB rows are correctly linked. + void amlScreeningService.screenTransaction( + { + transactionId: transaction.id, + userId, + amount: requestAmount, + phoneNumber, + type, + }, + transaction.id, + ).catch((err) => { + console.error( + `[AmlScreening] Failed to persist results for transaction ${transaction.id}:`, + err, + ); + }); + + // Increment velocity counters for future screenings. + // Collect all unique window_seconds values from velocity rules so + // we only do one Redis round-trip per window. + void (async () => { + try { + const velocityRules = await amlScreeningService.loadRules(); + const windows = [ + ...new Set( + velocityRules + .filter((r) => r.ruleType === "velocity_check") + .map((r) => Number(r.config["window_seconds"])) + .filter((w) => Number.isFinite(w) && w > 0), + ), + ]; + if (windows.length > 0) { + await amlScreeningService.incrementVelocityCounters(phoneNumber, windows); + } + } catch (err) { + console.error("[AmlScreening] Failed to increment velocity counters:", err); + } + })(); + + if (screeningOutcome.shouldFlag) { + const matchedNames = screeningOutcome.matchedRules + .map((e) => e.rule.name) + .join(" | "); + + void transactionModel + .updateAdminNotes( + transaction.id, + `[AML-SCREEN] Flagged by: ${matchedNames}`.slice(0, 1000), + ) + .catch((err) => { + console.error( + "[AmlScreening] Failed to set admin notes:", + err, + ); + }); + } + void monitorTransactionForAML(transaction); void applyTravelRule(transaction); diff --git a/src/models/amlScreeningResult.ts b/src/models/amlScreeningResult.ts new file mode 100644 index 00000000..0f18ddd5 --- /dev/null +++ b/src/models/amlScreeningResult.ts @@ -0,0 +1,236 @@ +import { pool } from "../config/database"; + +// ─── Types ─────────────────────────────────────────────────────────────────── + +export type AmlRuleType = "amount_threshold" | "velocity_check" | "blacklisted_phone"; + +/** + * A single rule evaluation result that gets persisted to aml_screening_results. + */ +export interface AmlScreeningResult { + id: string; + transactionId: string; + ruleId: string; + ruleName: string; + ruleType: AmlRuleType; + triggered: boolean; + /** Free-form details: observed values, thresholds, velocity counters, etc. */ + details: Record; + screenedAt: Date; +} + +export interface CreateAmlScreeningResultInput { + transactionId: string; + ruleId: string; + ruleName: string; + ruleType: AmlRuleType; + triggered: boolean; + details?: Record; +} + +export interface AmlScreeningResultFilter { + transactionId?: string; + triggered?: boolean; + limit?: number; + offset?: number; +} + +// ─── Model ─────────────────────────────────────────────────────────────────── + +export class AmlScreeningResultModel { + /** + * Insert a single screening result row. + */ + async create(input: CreateAmlScreeningResultInput): Promise { + const query = ` + INSERT INTO aml_screening_results ( + transaction_id, + rule_id, + rule_name, + rule_type, + triggered, + details + ) + VALUES ($1, $2, $3, $4, $5, $6) + RETURNING + id, + transaction_id AS "transactionId", + rule_id AS "ruleId", + rule_name AS "ruleName", + rule_type AS "ruleType", + triggered, + details, + screened_at AS "screenedAt" + `; + + const result = await pool.query(query, [ + input.transactionId, + input.ruleId, + input.ruleName, + input.ruleType, + input.triggered, + JSON.stringify(input.details ?? {}), + ]); + + return this.mapRow(result.rows[0]); + } + + /** + * Bulk-insert multiple screening results in a single statement for efficiency. + */ + async createBulk(inputs: CreateAmlScreeningResultInput[]): Promise { + if (inputs.length === 0) return []; + + const values: unknown[] = []; + const placeholders = inputs.map((input, i) => { + const base = i * 6; + values.push( + input.transactionId, + input.ruleId, + input.ruleName, + input.ruleType, + input.triggered, + JSON.stringify(input.details ?? {}), + ); + return `($${base + 1}, $${base + 2}, $${base + 3}, $${base + 4}, $${base + 5}, $${base + 6})`; + }); + + const query = ` + INSERT INTO aml_screening_results ( + transaction_id, rule_id, rule_name, rule_type, triggered, details + ) + VALUES ${placeholders.join(", ")} + RETURNING + id, + transaction_id AS "transactionId", + rule_id AS "ruleId", + rule_name AS "ruleName", + rule_type AS "ruleType", + triggered, + details, + screened_at AS "screenedAt" + `; + + const result = await pool.query(query, values); + return result.rows.map((row) => this.mapRow(row)); + } + + /** + * Retrieve all screening results for a given transaction. + */ + async findByTransactionId(transactionId: string): Promise { + const query = ` + SELECT + id, + transaction_id AS "transactionId", + rule_id AS "ruleId", + rule_name AS "ruleName", + rule_type AS "ruleType", + triggered, + details, + screened_at AS "screenedAt" + FROM aml_screening_results + WHERE transaction_id = $1 + ORDER BY screened_at ASC + `; + + const result = await pool.query(query, [transactionId]); + return result.rows.map((row) => this.mapRow(row)); + } + + /** + * Retrieve only the triggered results for a transaction (the rule matches). + */ + async findTriggeredByTransactionId(transactionId: string): Promise { + const query = ` + SELECT + id, + transaction_id AS "transactionId", + rule_id AS "ruleId", + rule_name AS "ruleName", + rule_type AS "ruleType", + triggered, + details, + screened_at AS "screenedAt" + FROM aml_screening_results + WHERE transaction_id = $1 + AND triggered = TRUE + ORDER BY screened_at ASC + `; + + const result = await pool.query(query, [transactionId]); + return result.rows.map((row) => this.mapRow(row)); + } + + /** + * List screening results with optional filtering and pagination. + */ + async list(filter: AmlScreeningResultFilter = {}): Promise<{ + results: AmlScreeningResult[]; + total: number; + }> { + const conditions: string[] = []; + const params: unknown[] = []; + let idx = 1; + + if (filter.transactionId !== undefined) { + conditions.push(`transaction_id = $${idx++}`); + params.push(filter.transactionId); + } + + if (filter.triggered !== undefined) { + conditions.push(`triggered = $${idx++}`); + params.push(filter.triggered); + } + + const where = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : ""; + + const countResult = await pool.query( + `SELECT COUNT(*) AS count FROM aml_screening_results ${where}`, + params, + ); + const total = parseInt(countResult.rows[0].count, 10); + + const limit = filter.limit ?? 50; + const offset = filter.offset ?? 0; + + const dataResult = await pool.query( + `SELECT + id, + transaction_id AS "transactionId", + rule_id AS "ruleId", + rule_name AS "ruleName", + rule_type AS "ruleType", + triggered, + details, + screened_at AS "screenedAt" + FROM aml_screening_results + ${where} + ORDER BY screened_at DESC + LIMIT $${idx++} OFFSET $${idx++}`, + [...params, limit, offset], + ); + + return { + results: dataResult.rows.map((row) => this.mapRow(row)), + total, + }; + } + + // ─── Private helpers ─────────────────────────────────────────────────────── + + private mapRow(row: Record): AmlScreeningResult { + return { + id: row.id as string, + transactionId: row.transactionId as string, + ruleId: row.ruleId as string, + ruleName: row.ruleName as string, + ruleType: row.ruleType as AmlRuleType, + triggered: row.triggered as boolean, + details: (row.details ?? {}) as Record, + screenedAt: row.screenedAt instanceof Date + ? row.screenedAt + : new Date(row.screenedAt as string), + }; + } +} diff --git a/src/services/__tests__/amlScreening.test.ts b/src/services/__tests__/amlScreening.test.ts new file mode 100644 index 00000000..ec39369f --- /dev/null +++ b/src/services/__tests__/amlScreening.test.ts @@ -0,0 +1,499 @@ +import { AmlScreeningService, AmlRule, ScreenTransactionInput } from "../amlScreening"; +import { AmlScreeningResultModel } from "../../models/amlScreeningResult"; +import { pool } from "../../config/database"; +import { redisClient } from "../../config/redis"; + +// ─── Mocks ─────────────────────────────────────────────────────────────────── + +jest.mock("../../config/database", () => ({ + pool: { + query: jest.fn(), + }, +})); + +jest.mock("../../config/redis", () => ({ + redisClient: { + get: jest.fn(), + incr: jest.fn(), + expire: jest.fn(), + }, +})); + +jest.mock("../../models/amlScreeningResult"); + +// ─── Helpers ───────────────────────────────────────────────────────────────── + +function makeRule(overrides: Partial & { ruleType: AmlRule["ruleType"]; config: Record }): AmlRule { + return { + id: "rule-" + Math.random().toString(36).slice(2), + name: "Test Rule", + description: null, + enabled: true, + ...overrides, + }; +} + +function makeInput(overrides: Partial = {}): ScreenTransactionInput { + return { + transactionId: "tx-test-001", + userId: "user-test-001", + amount: 100_000, + phoneNumber: "+237600000001", + type: "deposit", + ...overrides, + }; +} + +// ─── Tests ─────────────────────────────────────────────────────────────────── + +describe("AmlScreeningService", () => { + let service: AmlScreeningService; + let mockResultModel: jest.Mocked; + + beforeEach(() => { + jest.clearAllMocks(); + mockResultModel = new AmlScreeningResultModel() as jest.Mocked; + mockResultModel.createBulk = jest.fn().mockResolvedValue([]); + service = new AmlScreeningService(mockResultModel); + }); + + // ── loadRules ────────────────────────────────────────────────────────────── + + describe("loadRules()", () => { + it("returns enabled rules from the database", async () => { + const fakeRows = [ + { id: "r1", rule_type: "amount_threshold", name: "Big TX", description: null, config: { threshold_xaf: 1_000_000 }, enabled: true }, + { id: "r2", rule_type: "blacklisted_phone", name: "Blacklist", description: null, config: { numbers: [] }, enabled: true }, + ]; + (pool.query as jest.fn).mockResolvedValueOnce({ rows: fakeRows }); + + const rules = await service.loadRules(); + + expect(rules).toHaveLength(2); + expect(rules[0].ruleType).toBe("amount_threshold"); + expect(rules[1].ruleType).toBe("blacklisted_phone"); + }); + + it("caches results and avoids a second DB query within TTL", async () => { + (pool.query as jest.fn).mockResolvedValue({ rows: [] }); + + await service.loadRules(); + await service.loadRules(); + + expect(pool.query).toHaveBeenCalledTimes(1); + }); + + it("re-fetches after invalidateRuleCache() is called", async () => { + (pool.query as jest.fn).mockResolvedValue({ rows: [] }); + + await service.loadRules(); + service.invalidateRuleCache(); + await service.loadRules(); + + expect(pool.query).toHaveBeenCalledTimes(2); + }); + }); + + // ── amount_threshold ─────────────────────────────────────────────────────── + + describe("amount_threshold rule", () => { + it("triggers when amount equals the threshold", async () => { + const rule = makeRule({ ruleType: "amount_threshold", config: { threshold_xaf: 500_000 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + + const outcome = await service.screenTransaction(makeInput({ amount: 500_000 })); + + expect(outcome.shouldFlag).toBe(true); + expect(outcome.matchedRules).toHaveLength(1); + expect(outcome.matchedRules[0].rule.ruleType).toBe("amount_threshold"); + expect(outcome.matchedRules[0].details).toMatchObject({ + observed_amount: 500_000, + threshold_xaf: 500_000, + }); + }); + + it("triggers when amount exceeds the threshold", async () => { + const rule = makeRule({ ruleType: "amount_threshold", config: { threshold_xaf: 500_000 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + + const outcome = await service.screenTransaction(makeInput({ amount: 1_000_000 })); + + expect(outcome.shouldFlag).toBe(true); + }); + + it("does NOT trigger when amount is below threshold", async () => { + const rule = makeRule({ ruleType: "amount_threshold", config: { threshold_xaf: 500_000 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + + const outcome = await service.screenTransaction(makeInput({ amount: 499_999 })); + + expect(outcome.shouldFlag).toBe(false); + expect(outcome.matchedRules).toHaveLength(0); + }); + + it("does NOT trigger and returns error detail when config is invalid", async () => { + const rule = makeRule({ ruleType: "amount_threshold", config: { threshold_xaf: -100 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + + const outcome = await service.screenTransaction(makeInput({ amount: 500_000 })); + + expect(outcome.shouldFlag).toBe(false); + expect(outcome.allEvaluations[0].details).toMatchObject({ error: expect.any(String) }); + }); + + it("evaluates multiple amount_threshold rules independently", async () => { + const ruleA = makeRule({ ruleType: "amount_threshold", config: { threshold_xaf: 500_000 } }); + const ruleB = makeRule({ ruleType: "amount_threshold", config: { threshold_xaf: 1_000_000 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(ruleA), toDbRow(ruleB)] }); + + // amount triggers ruleA but not ruleB + const outcome = await service.screenTransaction(makeInput({ amount: 750_000 })); + + expect(outcome.shouldFlag).toBe(true); + expect(outcome.matchedRules).toHaveLength(1); + expect(outcome.allEvaluations).toHaveLength(2); + }); + }); + + // ── velocity_check ───────────────────────────────────────────────────────── + + describe("velocity_check rule", () => { + it("triggers when current Redis count equals max_count", async () => { + const rule = makeRule({ ruleType: "velocity_check", config: { max_count: 3, window_seconds: 3600 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + (redisClient.get as jest.fn).mockResolvedValueOnce("3"); // already 3 → this is the 4th + + const outcome = await service.screenTransaction(makeInput()); + + expect(outcome.shouldFlag).toBe(true); + expect(outcome.matchedRules[0].details).toMatchObject({ + current_count: 3, + max_count: 3, + }); + }); + + it("triggers when current Redis count exceeds max_count", async () => { + const rule = makeRule({ ruleType: "velocity_check", config: { max_count: 3, window_seconds: 3600 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + (redisClient.get as jest.fn).mockResolvedValueOnce("5"); + + const outcome = await service.screenTransaction(makeInput()); + + expect(outcome.shouldFlag).toBe(true); + }); + + it("does NOT trigger when count is below max_count", async () => { + const rule = makeRule({ ruleType: "velocity_check", config: { max_count: 3, window_seconds: 3600 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + (redisClient.get as jest.fn).mockResolvedValueOnce("2"); + + const outcome = await service.screenTransaction(makeInput()); + + expect(outcome.shouldFlag).toBe(false); + }); + + it("does NOT trigger when Redis key is missing (first transaction in window)", async () => { + const rule = makeRule({ ruleType: "velocity_check", config: { max_count: 3, window_seconds: 3600 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + (redisClient.get as jest.fn).mockResolvedValueOnce(null); + + const outcome = await service.screenTransaction(makeInput()); + + expect(outcome.shouldFlag).toBe(false); + }); + + it("fails open (does NOT trigger) when Redis is unavailable", async () => { + const rule = makeRule({ ruleType: "velocity_check", config: { max_count: 3, window_seconds: 3600 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + (redisClient.get as jest.fn).mockRejectedValueOnce(new Error("Redis connection refused")); + + const outcome = await service.screenTransaction(makeInput()); + + expect(outcome.shouldFlag).toBe(false); + expect(outcome.allEvaluations[0].details).toMatchObject({ error: expect.stringContaining("Redis unavailable") }); + }); + + it("does NOT trigger and returns error detail when config is invalid", async () => { + const rule = makeRule({ ruleType: "velocity_check", config: { max_count: -1, window_seconds: 0 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + + const outcome = await service.screenTransaction(makeInput()); + + expect(outcome.shouldFlag).toBe(false); + expect(outcome.allEvaluations[0].details).toMatchObject({ error: expect.any(String) }); + }); + + it("uses phone number in the Redis key", async () => { + const rule = makeRule({ ruleType: "velocity_check", config: { max_count: 3, window_seconds: 3600 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + (redisClient.get as jest.fn).mockResolvedValueOnce("0"); + + await service.screenTransaction(makeInput({ phoneNumber: "+237699887766" })); + + const calledKey = (redisClient.get as jest.fn).mock.calls[0][0] as string; + expect(calledKey).toContain("+237699887766"); + }); + }); + + // ── blacklisted_phone ────────────────────────────────────────────────────── + + describe("blacklisted_phone rule", () => { + it("triggers when the phone number is in the blacklist", async () => { + const rule = makeRule({ + ruleType: "blacklisted_phone", + config: { numbers: ["+237600000000", "+237611111111"] }, + }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + + const outcome = await service.screenTransaction(makeInput({ phoneNumber: "+237600000000" })); + + expect(outcome.shouldFlag).toBe(true); + expect(outcome.matchedRules[0].details).toMatchObject({ matched: true }); + }); + + it("does NOT trigger for a clean phone number", async () => { + const rule = makeRule({ + ruleType: "blacklisted_phone", + config: { numbers: ["+237600000000"] }, + }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + + const outcome = await service.screenTransaction(makeInput({ phoneNumber: "+237699887766" })); + + expect(outcome.shouldFlag).toBe(false); + }); + + it("normalises phone numbers before comparison (removes spaces)", async () => { + const rule = makeRule({ + ruleType: "blacklisted_phone", + config: { numbers: ["+237 600 000 000"] }, // stored with spaces + }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + + // Input without spaces should still match + const outcome = await service.screenTransaction(makeInput({ phoneNumber: "+237600000000" })); + + expect(outcome.shouldFlag).toBe(true); + }); + + it("does NOT trigger and returns error detail when config.numbers is not an array", async () => { + const rule = makeRule({ + ruleType: "blacklisted_phone", + config: { numbers: "not-an-array" }, + }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + + const outcome = await service.screenTransaction(makeInput()); + + expect(outcome.shouldFlag).toBe(false); + expect(outcome.allEvaluations[0].details).toMatchObject({ error: expect.any(String) }); + }); + }); + + // ── combined / multi-rule scenarios ─────────────────────────────────────── + + describe("combined rule evaluation", () => { + it("flags if ANY rule is triggered (OR semantics)", async () => { + const amountRule = makeRule({ ruleType: "amount_threshold", config: { threshold_xaf: 500_000 } }); + const blacklistRule = makeRule({ + ruleType: "blacklisted_phone", + config: { numbers: ["+237600000000"] }, + }); + (pool.query as jest.fn).mockResolvedValueOnce({ + rows: [toDbRow(amountRule), toDbRow(blacklistRule)], + }); + + // Amount is below threshold, but phone is blacklisted + const outcome = await service.screenTransaction( + makeInput({ amount: 100_000, phoneNumber: "+237600000000" }), + ); + + expect(outcome.shouldFlag).toBe(true); + expect(outcome.matchedRules).toHaveLength(1); + expect(outcome.matchedRules[0].rule.ruleType).toBe("blacklisted_phone"); + }); + + it("collects all triggered rules when multiple fire", async () => { + const amountRule = makeRule({ ruleType: "amount_threshold", config: { threshold_xaf: 500_000 } }); + const blacklistRule = makeRule({ + ruleType: "blacklisted_phone", + config: { numbers: ["+237600000000"] }, + }); + (pool.query as jest.fn).mockResolvedValueOnce({ + rows: [toDbRow(amountRule), toDbRow(blacklistRule)], + }); + + const outcome = await service.screenTransaction( + makeInput({ amount: 1_000_000, phoneNumber: "+237600000000" }), + ); + + expect(outcome.shouldFlag).toBe(true); + expect(outcome.matchedRules).toHaveLength(2); + }); + + it("does NOT flag when no rules are defined", async () => { + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [] }); + + const outcome = await service.screenTransaction(makeInput()); + + expect(outcome.shouldFlag).toBe(false); + expect(outcome.allEvaluations).toHaveLength(0); + }); + + it("does NOT flag when no rules trigger", async () => { + const amountRule = makeRule({ ruleType: "amount_threshold", config: { threshold_xaf: 1_000_000 } }); + const blacklistRule = makeRule({ + ruleType: "blacklisted_phone", + config: { numbers: ["+237600000000"] }, + }); + (pool.query as jest.fn).mockResolvedValueOnce({ + rows: [toDbRow(amountRule), toDbRow(blacklistRule)], + }); + (redisClient.get as jest.fn).mockResolvedValue(null); + + const outcome = await service.screenTransaction( + makeInput({ amount: 100_000, phoneNumber: "+237699887766" }), + ); + + expect(outcome.shouldFlag).toBe(false); + expect(outcome.matchedRules).toHaveLength(0); + expect(outcome.allEvaluations).toHaveLength(2); + }); + }); + + // ── result persistence ───────────────────────────────────────────────────── + + describe("result persistence", () => { + it("calls createBulk with one result per evaluated rule", async () => { + const rule = makeRule({ ruleType: "amount_threshold", config: { threshold_xaf: 500_000 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + + await service.screenTransaction(makeInput({ amount: 1_000_000 }), "tx-real-id"); + + // Allow fire-and-forget microtasks to flush + await flushPromises(); + + expect(mockResultModel.createBulk).toHaveBeenCalledWith( + expect.arrayContaining([ + expect.objectContaining({ + transactionId: "tx-real-id", + ruleId: rule.id, + ruleType: "amount_threshold", + triggered: true, + }), + ]), + ); + }); + + it("persists non-triggered results as well (all evaluations logged)", async () => { + const rule = makeRule({ ruleType: "amount_threshold", config: { threshold_xaf: 1_000_000 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + + // Amount below threshold → not triggered + await service.screenTransaction(makeInput({ amount: 100_000 }), "tx-real-id"); + await flushPromises(); + + expect(mockResultModel.createBulk).toHaveBeenCalledWith( + expect.arrayContaining([ + expect.objectContaining({ + transactionId: "tx-real-id", + triggered: false, + }), + ]), + ); + }); + + it("does not throw when createBulk fails (fire-and-forget error isolation)", async () => { + const rule = makeRule({ ruleType: "amount_threshold", config: { threshold_xaf: 500_000 } }); + (pool.query as jest.fn).mockResolvedValueOnce({ rows: [toDbRow(rule)] }); + mockResultModel.createBulk.mockRejectedValueOnce(new Error("DB write failed")); + + // Should not throw + await expect( + service.screenTransaction(makeInput({ amount: 1_000_000 })), + ).resolves.not.toThrow(); + }); + }); + + // ── incrementVelocityCounters ────────────────────────────────────────────── + + describe("incrementVelocityCounters()", () => { + it("increments Redis counter for each provided window", async () => { + (redisClient.incr as jest.fn) + .mockResolvedValueOnce(1) // first window + .mockResolvedValueOnce(2); // second window + (redisClient.expire as jest.fn).mockResolvedValue(1); + + const result = await service.incrementVelocityCounters("+237699887766", [3600, 86400]); + + expect(redisClient.incr).toHaveBeenCalledTimes(2); + expect(result[3600]).toBe(1); + expect(result[86400]).toBe(2); + }); + + it("sets TTL on first increment (count === 1)", async () => { + (redisClient.incr as jest.fn).mockResolvedValue(1); + (redisClient.expire as jest.fn).mockResolvedValue(1); + + await service.incrementVelocityCounters("+237699887766", [3600]); + + expect(redisClient.expire).toHaveBeenCalledWith(expect.any(String), 3600); + }); + + it("does NOT reset TTL on subsequent increments (count > 1)", async () => { + (redisClient.incr as jest.fn).mockResolvedValue(3); // not the first + + await service.incrementVelocityCounters("+237699887766", [3600]); + + expect(redisClient.expire).not.toHaveBeenCalled(); + }); + + it("returns 0 for a window when Redis incr throws", async () => { + (redisClient.incr as jest.fn).mockRejectedValueOnce(new Error("Redis error")); + + const result = await service.incrementVelocityCounters("+237699887766", [3600]); + + expect(result[3600]).toBe(0); + }); + + it("returns empty object for empty window list", async () => { + const result = await service.incrementVelocityCounters("+237699887766", []); + + expect(result).toEqual({}); + expect(redisClient.incr).not.toHaveBeenCalled(); + }); + }); + + // ── invalidateRuleCache ──────────────────────────────────────────────────── + + describe("invalidateRuleCache()", () => { + it("forces a fresh DB query on next loadRules()", async () => { + (pool.query as jest.fn).mockResolvedValue({ rows: [] }); + + await service.loadRules(); // populates cache + service.invalidateRuleCache(); // invalidates + await service.loadRules(); // should re-query + + expect(pool.query).toHaveBeenCalledTimes(2); + }); + }); +}); + +// ─── Utilities ──────────────────────────────────────────────────────────────── + +/** Convert an AmlRule object back to the raw DB row shape expected by loadRules(). */ +function toDbRow(rule: AmlRule) { + return { + id: rule.id, + rule_type: rule.ruleType, + name: rule.name, + description: rule.description, + config: rule.config, + enabled: rule.enabled, + }; +} + +/** Flush all pending microtasks / resolved promises. */ +function flushPromises(): Promise { + return new Promise((resolve) => setImmediate(resolve)); +} diff --git a/src/services/amlScreening.ts b/src/services/amlScreening.ts new file mode 100644 index 00000000..c4059a26 --- /dev/null +++ b/src/services/amlScreening.ts @@ -0,0 +1,403 @@ +/** + * AML Screening Service + * + * Evaluates a transaction synchronously against configurable rules stored in the + * `aml_rules` table before the transaction is created. Rule results are logged + * to `aml_screening_results`. + * + * Supported rule types: + * • amount_threshold – flag if amount >= config.threshold_xaf + * • velocity_check – flag if phone sent > config.max_count txns in + * config.window_seconds seconds (Redis counter) + * • blacklisted_phone – flag if phone number is in config.numbers[] + * + * Any triggered rule causes the transaction to be created with status + * `pending_review` instead of `pending`. + */ + +import { pool } from "../config/database"; +import { redisClient } from "../config/redis"; +import { + AmlScreeningResultModel, + AmlRuleType, + CreateAmlScreeningResultInput, +} from "../models/amlScreeningResult"; + +// ─── Types ─────────────────────────────────────────────────────────────────── + +/** Shape of a row in aml_rules */ +export interface AmlRule { + id: string; + ruleType: AmlRuleType; + name: string; + description: string | null; + config: Record; + enabled: boolean; +} + +/** Per-rule evaluation result (before writing to DB) */ +export interface RuleEvaluation { + rule: AmlRule; + triggered: boolean; + details: Record; +} + +/** Return value of screenTransaction() */ +export interface ScreeningOutcome { + /** True when at least one rule was triggered */ + shouldFlag: boolean; + /** Rules that fired */ + matchedRules: RuleEvaluation[]; + /** All rules that were checked (including non-triggered) */ + allEvaluations: RuleEvaluation[]; +} + +/** Input required to screen a transaction */ +export interface ScreenTransactionInput { + /** Temporary ID used for Redis keys only; the real DB id is set after creation */ + transactionId: string; + userId: string; + amount: number; + phoneNumber: string; + type: "deposit" | "withdraw"; +} + +// ─── Config ────────────────────────────────────────────────────────────────── + +/** TTL for the in-memory rules cache (ms). */ +const RULE_CACHE_TTL_MS = Number(process.env.AML_RULE_CACHE_TTL_MS ?? 60_000); + +/** Redis key prefix for velocity counters. */ +const VELOCITY_KEY_PREFIX = "aml:velocity:"; + +// ─── Service ───────────────────────────────────────────────────────────────── + +export class AmlScreeningService { + private readonly resultModel: AmlScreeningResultModel; + + /** In-process rule cache to avoid hitting PG on every transaction. */ + private cachedRules: AmlRule[] | null = null; + private cacheLoadedAt = 0; + + constructor(resultModel?: AmlScreeningResultModel) { + this.resultModel = resultModel ?? new AmlScreeningResultModel(); + } + + // ─── Public API ──────────────────────────────────────────────────────────── + + /** + * Screen a transaction against all enabled AML rules. + * + * This is the primary synchronous entry point called before the transaction + * record is written to the database. + * + * @param input - Transaction data to evaluate + * @param persistTransactionId - ID to use when writing to aml_screening_results. + * Pass the newly created transaction ID here when you have it, or the same + * `input.transactionId` if you call this before DB creation and update later. + */ + async screenTransaction( + input: ScreenTransactionInput, + persistTransactionId?: string, + ): Promise { + const rules = await this.loadRules(); + const evaluations: RuleEvaluation[] = []; + + // Evaluate each rule, running velocity checks in parallel where possible. + // We serialise for simplicity; the Redis calls are fast enough. + for (const rule of rules) { + const evaluation = await this.evaluateRule(rule, input); + evaluations.push(evaluation); + } + + const matchedRules = evaluations.filter((e) => e.triggered); + const shouldFlag = matchedRules.length > 0; + + // Persist all evaluations (fire-and-forget errors; screening must not block) + const txId = persistTransactionId ?? input.transactionId; + this.persistResults(txId, evaluations).catch((err) => { + console.error("[AmlScreening] Failed to persist screening results:", err); + }); + + return { shouldFlag, matchedRules, allEvaluations: evaluations }; + } + + /** + * Increment the velocity counter for a phone number. + * Call this after a transaction is successfully committed so the counter + * reflects only completed/pending transactions. + * + * Returns the new counter value for each window that was updated. + */ + async incrementVelocityCounters( + phoneNumber: string, + windowsSeconds: number[], + ): Promise> { + const results: Record = {}; + + for (const window of windowsSeconds) { + const key = this.velocityKey(phoneNumber, window); + try { + const newCount = await redisClient.incr(key); + // Set expiry only on first increment to avoid resetting the window + if (newCount === 1) { + await redisClient.expire(key, window); + } + results[window] = newCount; + } catch (err) { + console.error(`[AmlScreening] Redis incr failed for key ${key}:`, err); + results[window] = 0; + } + } + + return results; + } + + /** + * Force-flush the in-memory rule cache on the next load. + * Useful after an operator updates a rule via the DB. + */ + invalidateRuleCache(): void { + this.cachedRules = null; + this.cacheLoadedAt = 0; + } + + // ─── Rule loading ────────────────────────────────────────────────────────── + + /** + * Load enabled rules from Postgres, backed by a short in-process cache. + */ + async loadRules(): Promise { + const now = Date.now(); + if (this.cachedRules !== null && now - this.cacheLoadedAt < RULE_CACHE_TTL_MS) { + return this.cachedRules; + } + + const result = await pool.query<{ + id: string; + rule_type: AmlRuleType; + name: string; + description: string | null; + config: Record; + enabled: boolean; + }>(` + SELECT id, rule_type, name, description, config, enabled + FROM aml_rules + WHERE enabled = TRUE + ORDER BY created_at ASC + `); + + this.cachedRules = result.rows.map((row) => ({ + id: row.id, + ruleType: row.rule_type, + name: row.name, + description: row.description, + config: row.config, + enabled: row.enabled, + })); + this.cacheLoadedAt = now; + + return this.cachedRules; + } + + // ─── Rule evaluation ─────────────────────────────────────────────────────── + + private async evaluateRule( + rule: AmlRule, + input: ScreenTransactionInput, + ): Promise { + switch (rule.ruleType) { + case "amount_threshold": + return this.evaluateAmountThreshold(rule, input); + case "velocity_check": + return this.evaluateVelocityCheck(rule, input); + case "blacklisted_phone": + return this.evaluateBlacklistedPhone(rule, input); + default: + // Unknown rule type – do not trigger, just log + return { + rule, + triggered: false, + details: { error: `Unknown rule type: ${(rule as AmlRule).ruleType}` }, + }; + } + } + + /** + * amount_threshold rule + * + * config shape: + * { "threshold_xaf": number } + * + * Triggers when input.amount >= threshold_xaf. + */ + private evaluateAmountThreshold( + rule: AmlRule, + input: ScreenTransactionInput, + ): RuleEvaluation { + const threshold = Number(rule.config["threshold_xaf"]); + + if (!Number.isFinite(threshold) || threshold <= 0) { + return { + rule, + triggered: false, + details: { error: "Invalid threshold_xaf in rule config", config: rule.config }, + }; + } + + const triggered = input.amount >= threshold; + + return { + rule, + triggered, + details: { + observed_amount: input.amount, + threshold_xaf: threshold, + }, + }; + } + + /** + * velocity_check rule + * + * config shape: + * { "max_count": number, "window_seconds": number } + * + * Uses a Redis INCR counter keyed by phone + window. + * Triggers when the current count (before this transaction) >= max_count, + * meaning this transaction would be the (max_count + 1)th. + * + * Falls back gracefully if Redis is unavailable. + */ + private async evaluateVelocityCheck( + rule: AmlRule, + input: ScreenTransactionInput, + ): Promise { + const maxCount = Number(rule.config["max_count"]); + const windowSeconds = Number(rule.config["window_seconds"]); + + if (!Number.isFinite(maxCount) || !Number.isFinite(windowSeconds) + || maxCount <= 0 || windowSeconds <= 0) { + return { + rule, + triggered: false, + details: { error: "Invalid velocity rule config", config: rule.config }, + }; + } + + const key = this.velocityKey(input.phoneNumber, windowSeconds); + + let currentCount = 0; + try { + const raw = await redisClient.get(key); + currentCount = raw ? parseInt(raw, 10) : 0; + if (!Number.isFinite(currentCount)) currentCount = 0; + } catch (err) { + // Redis unavailable – fail open (don't block transactions) + console.error(`[AmlScreening] Redis GET failed for velocity key ${key}:`, err); + return { + rule, + triggered: false, + details: { + error: "Redis unavailable; velocity check skipped", + phone: input.phoneNumber, + window_seconds: windowSeconds, + max_count: maxCount, + }, + }; + } + + // Trigger if the existing count (before adding this transaction) >= max_count. + const triggered = currentCount >= maxCount; + + return { + rule, + triggered, + details: { + phone_number: input.phoneNumber, + current_count: currentCount, + max_count: maxCount, + window_seconds: windowSeconds, + redis_key: key, + }, + }; + } + + /** + * blacklisted_phone rule + * + * config shape: + * { "numbers": string[] } + * + * Normalises numbers to E.164 before comparison. + */ + private evaluateBlacklistedPhone( + rule: AmlRule, + input: ScreenTransactionInput, + ): RuleEvaluation { + const numbers: unknown = rule.config["numbers"]; + + if (!Array.isArray(numbers)) { + return { + rule, + triggered: false, + details: { error: "Invalid blacklist config – 'numbers' must be an array" }, + }; + } + + const normalised = normalisePhone(input.phoneNumber); + const blacklist: string[] = (numbers as unknown[]) + .filter((n): n is string => typeof n === "string") + .map(normalisePhone); + + const triggered = blacklist.includes(normalised); + + return { + rule, + triggered, + details: { + phone_number: input.phoneNumber, + matched: triggered, + }, + }; + } + + // ─── Persistence ─────────────────────────────────────────────────────────── + + private async persistResults( + transactionId: string, + evaluations: RuleEvaluation[], + ): Promise { + const inputs: CreateAmlScreeningResultInput[] = evaluations.map((e) => ({ + transactionId, + ruleId: e.rule.id, + ruleName: e.rule.name, + ruleType: e.rule.ruleType, + triggered: e.triggered, + details: e.details, + })); + + await this.resultModel.createBulk(inputs); + } + + // ─── Helpers ─────────────────────────────────────────────────────────────── + + private velocityKey(phoneNumber: string, windowSeconds: number): string { + // Normalise the phone so "+237 600 000 000" and "+237600000000" map to the same key + const normalised = normalisePhone(phoneNumber); + // Round window to an absolute time bucket so all service instances share the same key + const bucket = Math.floor(Date.now() / 1000 / windowSeconds); + return `${VELOCITY_KEY_PREFIX}${normalised}:${windowSeconds}:${bucket}`; + } +} + +// ─── Helpers ───────────────────────────────────────────────────────────────── + +/** Strip non-digit characters (except leading +) for consistent key comparison. */ +function normalisePhone(phone: string): string { + return phone.replace(/[^+\d]/g, ""); +} + +// ─── Singleton ─────────────────────────────────────────────────────────────── + +export const amlScreeningService = new AmlScreeningService();