lux/broker
2
0
Fork 0
mirror of https://github.com/luxfi/broker.git synced 2026-09-24 00:25:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: compliance imports from lux/compliance library (thin HTTP layer)

Browse Source 
Replace local compliance types, store, MemoryStore, and jube package with
imports from github.com/luxfi/compliance. Broker's pkg/compliance/ is now a
thin HTTP adapter layer:

- compliance.go: type aliases re-exporting all types from library
- store.go, jube/: removed (moved to library)
- postgres_store.go: kept (pgx dependency stays in broker)
- All HTTP handlers unchanged (types are aliases, fully compatible)
- Router imports jube from library instead of local subpackage

go.mod adds: require github.com/luxfi/compliance v0.1.0
             replace github.com/luxfi/compliance => ../compliance

All 50+ compliance tests pass. No handler code changes needed.
This commit is contained in:
zeekay
2025-12-24 17:32:14 -07:00
parent 18a7205912
commit 9282ad33b0
13 changed files with 120 additions and 3170 deletions
Download Patch File Download Diff File Expand all files Collapse all files
LLM.md
+17 -10
View File
@@ -94,24 +94,31 @@ to avoid import cycles: `provider/alpaca` has compile-time interface assertions
import `provider`, so `provider` cannot import `provider/alpaca`.
### Compliance (pkg/compliance/)
KYC/KYB, onboarding, fund management, eSign, RBAC, and reporting.
Thin HTTP adapter layer over `github.com/luxfi/compliance` library.
All domain types, store interface, MemoryStore, Jube client, RBAC, and onboarding
logic live in the library. This package re-exports types as aliases and provides:
- HTTP handlers (chi router with RBAC guards, rate limiting, CORS)
- PostgresStore (pgx-backed, implements library's ComplianceStore interface)
- Webhook dispatcher for cross-BD trade events (`pkg/compliance/webhooks/`)
- Seed data for development mode
Mounted at `/compliance` on the main server. Supports in-memory store (default) or
PostgreSQL via `DATABASE_URL` env var. Includes Jube AML/fraud client (`pkg/compliance/jube/`)
and webhook dispatcher (`pkg/compliance/webhooks/`).
PostgreSQL via `DATABASE_URL` env var. Jube client imported from library.
Key endpoint groups under `/compliance`:
- `/kyc` -- Identity verification, document upload, status tracking (list by user, update status)
- `/aml` -- AML screening (Jube integration), risk assessment, flagged review, sanctions checks
- `/applications` -- 5-step onboarding flow (basic info, identity, documents, compliance, submit)
- `/kyc` -- Identity verification, document upload, status tracking
- `/aml` -- AML screening (Jube integration), risk assessment, flagged review
- `/applications` -- 5-step onboarding flow (SSN hashed with HMAC-SHA256)
- `/pipelines` + `/sessions` -- Configurable onboarding pipeline engine
- `/funds` -- Fund management with investor tracking
- `/esign` -- Envelope/template eSign workflow
- `/roles` + `/modules` -- RBAC permission matrix
New files (2026-03-27):
- `aml.go` -- AML screening/review/risk-assessment handlers
- `application.go` -- 5-step onboarding application handlers (SSN hashed with SHA-256)
- `schemas/compliance_tables.sql` -- PostgreSQL migration for aml_screenings, applications, document_uploads
Library dependency (go.mod):
```
require github.com/luxfi/compliance v0.1.0
replace github.com/luxfi/compliance => ../compliance
```
### Database (pkg/db/)
PostgreSQL connection pool and auto-migrations. Used by compliance when `DATABASE_URL` is set.
go.mod
+3
View File
@@ -7,11 +7,14 @@ require (
github.com/go-chi/cors v1.2.2
github.com/hanzoai/commerce v1.36.5
github.com/jackc/pgx/v5 v5.8.0
github.com/luxfi/compliance v0.1.0
github.com/nats-io/nats.go v1.49.0
github.com/rs/zerolog v1.34.0
google.golang.org/grpc v1.79.1
)
replace github.com/luxfi/compliance => ../compliance
require (
github.com/jackc/pgpassfile v1.0.0 // indirect
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
pkg/compliance/aml.go
+1 -1
View File
@@ -7,7 +7,7 @@ import (
"github.com/go-chi/chi/v5"
"github.com/luxfi/broker/pkg/admin"
"github.com/luxfi/broker/pkg/compliance/jube"
"github.com/luxfi/compliance/pkg/jube"
"github.com/rs/zerolog/log"
)
pkg/compliance/compliance.go
+98 -389
View File
@@ -1,425 +1,134 @@
// Package compliance provides KYC/KYB verification, investor onboarding
// pipelines, fund management, eSign, and role-based access control for
// broker-dealer compliance workflows.
//
// All domain types and the store interface are re-exported from
// github.com/luxfi/compliance. This package adds only HTTP handlers
// and broker-specific wiring.
package compliance
import "time"
// --- KYC/KYB Status Types ---
// KYCStatus tracks identity verification state.
type KYCStatus string
const (
KYCPending KYCStatus = "pending"
KYCVerified KYCStatus = "verified"
KYCFailed KYCStatus = "failed"
KYCExpired KYCStatus = "expired"
import (
"github.com/luxfi/compliance/pkg/store"
"github.com/luxfi/compliance/pkg/types"
)
// KYBStatus tracks business verification state.
type KYBStatus string
// --- Type aliases: re-export all types from the compliance library ---
type KYCStatus = types.KYCStatus
const (
KYBPending KYBStatus = "pending"
KYBApproved KYBStatus = "approved"
KYBRejected KYBStatus = "rejected"
KYCPending = types.KYCPending
KYCVerified = types.KYCVerified
KYCFailed = types.KYCFailed
KYCExpired = types.KYCExpired
)
// SessionStatus tracks onboarding session state.
type SessionStatus string
type KYBStatus = types.KYBStatus
const (
SessionPending SessionStatus = "pending"
SessionInProgress SessionStatus = "in_progress"
SessionCompleted SessionStatus = "completed"
SessionFailed SessionStatus = "failed"
SessionArchived SessionStatus = "archived"
KYBPending = types.KYBPending
KYBApproved = types.KYBApproved
KYBRejected = types.KYBRejected
)
// --- Core Types ---
// Identity represents a KYC-verified individual identity.
type Identity struct {
ID string `json:"id"`
UserID string `json:"user_id"`
Provider string `json:"provider"` // onfido, berbix, idmerit, etc.
Status KYCStatus `json:"status"`
Data map[string]interface{} `json:"data,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// Document is a verification document (passport, license, etc.).
type Document struct {
ID string `json:"id"`
Type string `json:"type"` // passport, drivers_license, utility_bill, ein_letter
Name string `json:"name,omitempty"`
MimeType string `json:"mime_type,omitempty"`
Content string `json:"content,omitempty"` // base64
Status string `json:"status,omitempty"`
CreatedAt string `json:"created_at,omitempty"`
}
// BusinessKYB represents a business verification record.
type BusinessKYB struct {
ID string `json:"id"`
BusinessID string `json:"business_id"`
Name string `json:"name"`
EIN string `json:"ein"`
Status KYBStatus `json:"status"`
Documents []Document `json:"documents,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// --- Pipeline / Session Types ---
// PipelineStep defines a single step in an onboarding pipeline.
type PipelineStep struct {
ID string `json:"id"`
Name string `json:"name"`
Type string `json:"type"` // kyc, accreditation, esign, payment, review
Required bool `json:"required"`
Order int `json:"order"`
}
// Pipeline is a configurable investor onboarding flow.
type Pipeline struct {
ID string `json:"id"`
Name string `json:"name"`
BusinessID string `json:"business_id"`
Steps []PipelineStep `json:"steps"`
Status string `json:"status"` // active, draft, archived
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// SessionStep tracks completion of a single pipeline step within a session.
type SessionStep struct {
StepID string `json:"step_id"`
Name string `json:"name"`
Type string `json:"type"`
Status string `json:"status"` // pending, completed, failed, skipped
CompletedAt time.Time `json:"completed_at,omitempty"`
}
// Session is an investor's progress through an onboarding pipeline.
type Session struct {
ID string `json:"id"`
PipelineID string `json:"pipeline_id"`
InvestorEmail string `json:"investor_email"`
InvestorName string `json:"investor_name"`
Status SessionStatus `json:"status"`
KYCStatus KYCStatus `json:"kyc_status"`
Steps []SessionStep `json:"steps,omitempty"`
CreatedAt time.Time `json:"created_at"`
CompletedAt time.Time `json:"completed_at,omitempty"`
}
// --- Fund Types ---
// Fund represents an investment fund.
type Fund struct {
ID string `json:"id"`
Name string `json:"name"`
BusinessID string `json:"business_id"`
Type string `json:"type"` // equity, debt, real_estate
MinInvestment float64 `json:"min_investment"`
TotalRaised float64 `json:"total_raised"`
InvestorCount int `json:"investor_count"`
Status string `json:"status"` // open, closed, raising
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// FundInvestor represents an investor's participation in a fund.
type FundInvestor struct {
ID string `json:"id"`
FundID string `json:"fund_id"`
InvestorID string `json:"investor_id"`
Name string `json:"name"`
Email string `json:"email"`
Amount float64 `json:"amount"`
Status string `json:"status"` // committed, funded, withdrawn
CreatedAt time.Time `json:"created_at"`
}
// --- eSign Types ---
// EnvelopeStatus tracks document signing state.
type EnvelopeStatus string
type SessionStatus = types.SessionStatus
const (
EnvelopePending EnvelopeStatus = "pending"
EnvelopeSent EnvelopeStatus = "sent"
EnvelopeViewed EnvelopeStatus = "viewed"
EnvelopeSigned EnvelopeStatus = "signed"
EnvelopeCompleted EnvelopeStatus = "completed"
EnvelopeDeclined EnvelopeStatus = "declined"
EnvelopeVoided EnvelopeStatus = "voided"
SessionPending = types.SessionPending
SessionInProgress = types.SessionInProgress
SessionCompleted = types.SessionCompleted
SessionFailed = types.SessionFailed
SessionArchived = types.SessionArchived
)
// Signer represents a person who must sign an envelope.
type Signer struct {
ID string `json:"id"`
Name string `json:"name"`
Email string `json:"email"`
Role string `json:"role,omitempty"` // investor, issuer, witness
Status string `json:"status"` // pending, signed, declined
SignedAt string `json:"signed_at,omitempty"`
}
// Envelope is a document package sent for signatures.
type Envelope struct {
ID string `json:"id"`
TemplateID string `json:"template_id,omitempty"`
Subject string `json:"subject"`
Message string `json:"message,omitempty"`
Status EnvelopeStatus `json:"status"`
Signers []Signer `json:"signers"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// Template is a reusable eSign document template.
type Template struct {
ID string `json:"id"`
Name string `json:"name"`
Description string `json:"description,omitempty"`
Content string `json:"content,omitempty"` // base64 PDF or URL
Roles []string `json:"roles,omitempty"` // expected signer roles
CreatedAt time.Time `json:"created_at"`
}
// --- RBAC Types ---
// Permission defines a single allowed action on a module.
type Permission struct {
Module string `json:"module"` // kyc, funds, esign, pipelines, sessions, roles
Action string `json:"action"` // read, write, delete, admin
}
// Role is a named set of permissions.
type Role struct {
ID string `json:"id"`
Name string `json:"name"`
Description string `json:"description,omitempty"`
Permissions []Permission `json:"permissions"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// Module describes a compliance module for the permission matrix.
type Module struct {
Name string `json:"name"`
Description string `json:"description"`
Actions []string `json:"actions"` // available actions
}
// --- User Management Types ---
// User represents an admin/compliance platform user.
type User struct {
ID string `json:"id"`
Name string `json:"name"`
Email string `json:"email"`
Role string `json:"role"` // owner, admin, manager, developer, agent
Status string `json:"status"` // active, inactive, suspended
LastLogin time.Time `json:"last_login"`
CreatedAt time.Time `json:"created_at"`
}
// --- Transaction Types ---
// Transaction represents a financial transaction record.
type Transaction struct {
ID string `json:"id"`
Type string `json:"type"` // deposit, withdrawal, trade, dividend
Asset string `json:"asset"` // USD, BTC, ETH, etc.
Amount float64 `json:"amount"`
Fee float64 `json:"fee"`
Status string `json:"status"` // pending, completed, failed, cancelled
Date string `json:"date"`
}
// --- Settings Types ---
// Settings holds business configuration for the compliance tenant.
type Settings struct {
BusinessName string `json:"business_name"`
Timezone string `json:"timezone"`
Currency string `json:"currency"`
NotificationEmail string `json:"notification_email"`
}
// --- Credential Types ---
// Credential represents an API key visible to the compliance admin.
type Credential struct {
ID string `json:"id"`
Name string `json:"name"`
KeyPrefix string `json:"key_prefix"` // first 8 chars, for display only
Permissions []string `json:"permissions"`
CreatedAt time.Time `json:"created_at"`
ExpiresAt string `json:"expires_at,omitempty"`
}
// --- AML Types ---
// AMLStatus tracks anti-money laundering screening state.
type AMLStatus string
type AMLStatus = types.AMLStatus
const (
AMLPending AMLStatus = "pending"
AMLCleared AMLStatus = "cleared"
AMLFlagged AMLStatus = "flagged"
AMLBlocked AMLStatus = "blocked"
AMLExpired AMLStatus = "expired"
AMLPending = types.AMLPending
AMLCleared = types.AMLCleared
AMLFlagged = types.AMLFlagged
AMLBlocked = types.AMLBlocked
AMLExpired = types.AMLExpired
)
// RiskLevel categorizes account risk.
type RiskLevel string
type RiskLevel = types.RiskLevel
const (
RiskLow RiskLevel = "low"
RiskMedium RiskLevel = "medium"
RiskHigh RiskLevel = "high"
RiskCritical RiskLevel = "critical"
RiskLow = types.RiskLow
RiskMedium = types.RiskMedium
RiskHigh = types.RiskHigh
RiskCritical = types.RiskCritical
)
// AMLScreening represents a single AML screening record for an account.
type AMLScreening struct {
ID string `json:"id"`
AccountID string `json:"account_id"`
UserID string `json:"user_id"`
Type string `json:"type"` // sanctions, pep, adverse_media, transaction
Status AMLStatus `json:"status"`
RiskLevel RiskLevel `json:"risk_level"`
RiskScore float64 `json:"risk_score"`
Provider string `json:"provider"` // jube, manual
Details string `json:"details,omitempty"`
SanctionsHit bool `json:"sanctions_hit"`
PEPMatch bool `json:"pep_match"`
ReviewedBy string `json:"reviewed_by,omitempty"`
ReviewedAt time.Time `json:"reviewed_at,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// --- Onboarding Application Types ---
// ApplicationStatus tracks an onboarding application's overall state.
type ApplicationStatus string
type ApplicationStatus = types.ApplicationStatus
const (
AppDraft ApplicationStatus = "draft"
AppInProgress ApplicationStatus = "in_progress"
AppSubmitted ApplicationStatus = "submitted"
AppUnderReview ApplicationStatus = "under_review"
AppApproved ApplicationStatus = "approved"
AppRejected ApplicationStatus = "rejected"
AppDraft = types.AppDraft
AppInProgress = types.AppInProgress
AppSubmitted = types.AppSubmitted
AppUnderReview = types.AppUnderReview
AppApproved = types.AppApproved
AppRejected = types.AppRejected
)
// ApplicationStep represents completion of a single onboarding step.
type ApplicationStep struct {
Step int `json:"step"`
Name string `json:"name"`
Status string `json:"status"` // pending, completed, failed, skipped
Data map[string]interface{} `json:"data,omitempty"`
CompletedAt time.Time `json:"completed_at,omitempty"`
type EnvelopeStatus = types.EnvelopeStatus
const (
EnvelopePending = types.EnvelopePending
EnvelopeSent = types.EnvelopeSent
EnvelopeViewed = types.EnvelopeViewed
EnvelopeSigned = types.EnvelopeSigned
EnvelopeCompleted = types.EnvelopeCompleted
EnvelopeDeclined = types.EnvelopeDeclined
EnvelopeVoided = types.EnvelopeVoided
)
type (
Identity = types.Identity
Document = types.Document
BusinessKYB = types.BusinessKYB
PipelineStep = types.PipelineStep
Pipeline = types.Pipeline
SessionStep = types.SessionStep
Session = types.Session
Fund = types.Fund
FundInvestor = types.FundInvestor
Signer = types.Signer
Envelope = types.Envelope
Template = types.Template
Permission = types.Permission
Role = types.Role
Module = types.Module
User = types.User
Transaction = types.Transaction
Settings = types.Settings
Credential = types.Credential
AMLScreening = types.AMLScreening
ApplicationStep = types.ApplicationStep
Application = types.Application
DocumentUpload = types.DocumentUpload
Invoice = types.Invoice
BillingInfo = types.BillingInfo
DashboardStats = types.DashboardStats
ESignStats = types.ESignStats
)
// --- Store: re-export from the compliance library ---
type ComplianceStore = store.ComplianceStore
type MemoryStore = store.MemoryStore
// NewMemoryStore creates an in-memory compliance store.
func NewMemoryStore() *store.MemoryStore {
return store.NewMemoryStore()
}
// Application is an investor's onboarding application through the 5-step flow:
// Step 1: Basic info + Contact
// Step 2: Identity verification
// Step 3: Document upload
// Step 4: Compliance/AML screening
// Step 5: Review + Submit
type Application struct {
ID string `json:"id"`
UserID string `json:"user_id"`
Email string `json:"email"`
FirstName string `json:"first_name"`
LastName string `json:"last_name"`
Phone string `json:"phone,omitempty"`
DateOfBirth string `json:"date_of_birth,omitempty"`
SSNHash string `json:"-"` // hashed, never exposed via API
SSNLast4 string `json:"ssn_last4,omitempty"` // last 4 digits for display
AddressLine1 string `json:"address_line1,omitempty"`
AddressLine2 string `json:"address_line2,omitempty"`
City string `json:"city,omitempty"`
State string `json:"state,omitempty"`
ZipCode string `json:"zip_code,omitempty"`
Country string `json:"country,omitempty"`
Status ApplicationStatus `json:"status"`
CurrentStep int `json:"current_step"`
KYCStatus KYCStatus `json:"kyc_status"`
AMLStatus AMLStatus `json:"aml_status"`
Steps []ApplicationStep `json:"steps"`
Documents []Document `json:"documents,omitempty"`
RiskLevel RiskLevel `json:"risk_level,omitempty"`
RiskScore float64 `json:"risk_score,omitempty"`
SubmittedAt time.Time `json:"submitted_at,omitempty"`
ReviewedBy string `json:"reviewed_by,omitempty"`
ReviewedAt time.Time `json:"reviewed_at,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
// NewStore is an alias for NewMemoryStore, preserving backward compatibility.
func NewStore() *store.MemoryStore {
return store.NewMemoryStore()
}
// DocumentUpload represents a document uploaded during onboarding.
type DocumentUpload struct {
ID string `json:"id"`
ApplicationID string `json:"application_id"`
UserID string `json:"user_id"`
Type string `json:"type"` // passport, drivers_license, utility_bill, proof_of_address, tax_return
Name string `json:"name"`
MimeType string `json:"mime_type"`
Size int64 `json:"size"`
Status string `json:"status"` // pending, accepted, rejected
ReviewNote string `json:"review_note,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// --- Billing Types ---
// Invoice represents a billing invoice.
type Invoice struct {
ID string `json:"id"`
Date string `json:"date"`
Amount float64 `json:"amount"`
Status string `json:"status"` // paid, pending, overdue
}
// BillingInfo holds billing and subscription details.
type BillingInfo struct {
Plan string `json:"plan"` // starter, professional, enterprise
PaymentMethod string `json:"payment_method"` // visa_4242, etc.
NextBilling string `json:"next_billing"`
MonthlyUsage float64 `json:"monthly_usage"`
Invoices []Invoice `json:"invoices"`
}
// --- Dashboard Types ---
// DashboardStats holds aggregate compliance dashboard statistics.
type DashboardStats struct {
ActiveSessions int `json:"activeSessions"`
PendingKYC int `json:"pendingKYC"`
TotalFunds int `json:"totalFunds"`
MonthlyTransactions int `json:"monthlyTransactions"`
RecentSessions []*Session `json:"recentSessions"`
RecentTransactions []*Transaction `json:"recentTransactions"`
}
// --- eSign Dashboard Types ---
// ESignStats holds aggregate eSign statistics.
type ESignStats struct {
Pending int `json:"pending"`
Completed int `json:"completed"`
Draft int `json:"draft"`
Templates int `json:"templates"`
Recent []*Envelope `json:"recent"`
// generateID delegates to the library's GenerateID.
func generateID() string {
return store.GenerateID()
}
pkg/compliance/jube/client.go
-225
View File
@@ -1,225 +0,0 @@
package jube
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"time"
"github.com/nats-io/nats.go"
"github.com/rs/zerolog/log"
)
// DefaultBaseURL is the default Jube sidecar address.
const DefaultBaseURL = "http://jube:5001"
// DefaultTimeout is the HTTP request timeout for Jube API calls.
const DefaultTimeout = 10 * time.Second
// NATSSubjectAlerts is the NATS subject for async AML alerts from Jube.
const NATSSubjectAlerts = "compliance.jube.alerts"
// Config holds Jube client configuration.
type Config struct {
BaseURL string // Jube sidecar URL (default: http://jube:5001)
Timeout time.Duration // HTTP timeout (default: 10s)
NATSAddr string // NATS server address for async alerts (e.g. nats://localhost:4222)
}
// Client is an HTTP client for the Jube AML/fraud detection sidecar.
type Client struct {
baseURL string
httpClient *http.Client
natsConn *nats.Conn
natsSub *nats.Subscription
}
// New creates a new Jube client. If cfg.NATSAddr is set, it connects to NATS
// for async alert subscriptions. Returns an error if NATS connection fails.
func New(cfg Config) (*Client, error) {
baseURL := cfg.BaseURL
if baseURL == "" {
baseURL = DefaultBaseURL
}
timeout := cfg.Timeout
if timeout == 0 {
timeout = DefaultTimeout
}
c := &Client{
baseURL: baseURL,
httpClient: &http.Client{
Timeout: timeout,
},
}
if cfg.NATSAddr != "" {
nc, err := nats.Connect(cfg.NATSAddr)
if err != nil {
return nil, fmt.Errorf("jube: nats connect %s: %w", cfg.NATSAddr, err)
}
c.natsConn = nc
}
return c, nil
}
// Close releases resources held by the client.
func (c *Client) Close() error {
if c.natsSub != nil {
if err := c.natsSub.Unsubscribe(); err != nil {
return fmt.Errorf("jube: nats unsubscribe: %w", err)
}
}
if c.natsConn != nil {
c.natsConn.Close()
}
return nil
}
// ScreenTransaction submits a transaction to Jube for real-time risk scoring.
// Calls POST /api/EntityAnalysisModel/Invoke.
func (c *Client) ScreenTransaction(ctx context.Context, req TransactionRequest) (*TransactionResponse, error) {
var resp TransactionResponse
if err := c.post(ctx, "/api/EntityAnalysisModel/Invoke", req, &resp); err != nil {
return nil, fmt.Errorf("jube: screen transaction: %w", err)
}
return &resp, nil
}
// CheckSanctions performs sanctions screening against Jube's lists.
// Calls GET /api/Sanction with name and country query params.
func (c *Client) CheckSanctions(ctx context.Context, name, country string) (*SanctionResult, error) {
params := url.Values{}
params.Set("name", name)
if country != "" {
params.Set("country", country)
}
var resp SanctionResult
if err := c.get(ctx, "/api/Sanction?"+params.Encode(), &resp); err != nil {
return nil, fmt.Errorf("jube: check sanctions: %w", err)
}
return &resp, nil
}
// CreateCase creates a new compliance case in Jube.
// Calls POST /api/CaseManagement.
func (c *Client) CreateCase(ctx context.Context, req CaseRequest) (*Case, error) {
var resp Case
if err := c.post(ctx, "/api/CaseManagement", req, &resp); err != nil {
return nil, fmt.Errorf("jube: create case: %w", err)
}
return &resp, nil
}
// GetCases retrieves compliance cases matching the given filters.
// Calls GET /api/CaseManagement with optional query params.
func (c *Client) GetCases(ctx context.Context, filter CaseFilter) ([]Case, error) {
params := url.Values{}
if filter.AccountID != "" {
params.Set("accountId", filter.AccountID)
}
if filter.Type != "" {
params.Set("type", filter.Type)
}
if filter.Status != "" {
params.Set("status", filter.Status)
}
path := "/api/CaseManagement"
if len(params) > 0 {
path += "?" + params.Encode()
}
var resp []Case
if err := c.get(ctx, path, &resp); err != nil {
return nil, fmt.Errorf("jube: get cases: %w", err)
}
return resp, nil
}
// Search performs an exhaustive search across Jube entities.
// Calls POST /api/ExhaustiveSearchInstance.
func (c *Client) Search(ctx context.Context, req SearchRequest) ([]SearchResult, error) {
var resp []SearchResult
if err := c.post(ctx, "/api/ExhaustiveSearchInstance", req, &resp); err != nil {
return nil, fmt.Errorf("jube: search: %w", err)
}
return resp, nil
}
// SubscribeAlerts subscribes to async AML/fraud alerts via NATS.
// The handler is called for each alert received. Returns an error if no
// NATS connection was configured.
func (c *Client) SubscribeAlerts(handler func(Alert)) error {
if c.natsConn == nil {
return fmt.Errorf("jube: nats not configured, cannot subscribe to alerts")
}
sub, err := c.natsConn.Subscribe(NATSSubjectAlerts, func(msg *nats.Msg) {
var alert Alert
if err := json.Unmarshal(msg.Data, &alert); err != nil {
log.Error().Err(err).Msg("jube: failed to unmarshal NATS alert")
return
}
handler(alert)
})
if err != nil {
return fmt.Errorf("jube: nats subscribe: %w", err)
}
c.natsSub = sub
return nil
}
// post sends a POST request with a JSON body and decodes the response.
func (c *Client) post(ctx context.Context, path string, body, dst interface{}) error {
payload, err := json.Marshal(body)
if err != nil {
return fmt.Errorf("marshal request: %w", err)
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+path, bytes.NewReader(payload))
if err != nil {
return fmt.Errorf("create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
return c.doJSON(req, dst)
}
// get sends a GET request and decodes the JSON response.
func (c *Client) get(ctx context.Context, path string, dst interface{}) error {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+path, nil)
if err != nil {
return fmt.Errorf("create request: %w", err)
}
req.Header.Set("Accept", "application/json")
return c.doJSON(req, dst)
}
// doJSON executes an HTTP request and decodes the JSON response into dst.
func (c *Client) doJSON(req *http.Request, dst interface{}) error {
resp, err := c.httpClient.Do(req)
if err != nil {
return fmt.Errorf("http %s %s: %w", req.Method, req.URL.Path, err)
}
defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
body, _ := io.ReadAll(resp.Body)
return fmt.Errorf("http %s %s: status %d: %s", req.Method, req.URL.Path, resp.StatusCode, string(body))
}
if dst != nil {
if err := json.NewDecoder(resp.Body).Decode(dst); err != nil {
return fmt.Errorf("decode response: %w", err)
}
}
return nil
}
pkg/compliance/jube/client_test.go
-551
View File
@@ -1,551 +0,0 @@
package jube
import (
"context"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
)
// --- Client Tests ---
func TestNewClientDefaults(t *testing.T) {
c, err := New(Config{})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer c.Close()
if c.baseURL != DefaultBaseURL {
t.Fatalf("baseURL = %q, want %q", c.baseURL, DefaultBaseURL)
}
if c.httpClient.Timeout != DefaultTimeout {
t.Fatalf("timeout = %v, want %v", c.httpClient.Timeout, DefaultTimeout)
}
if c.natsConn != nil {
t.Fatal("natsConn should be nil when NATSAddr not set")
}
}
func TestNewClientCustomBaseURL(t *testing.T) {
c, err := New(Config{BaseURL: "http://localhost:9999", Timeout: 5 * time.Second})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer c.Close()
if c.baseURL != "http://localhost:9999" {
t.Fatalf("baseURL = %q, want http://localhost:9999", c.baseURL)
}
if c.httpClient.Timeout != 5*time.Second {
t.Fatalf("timeout = %v, want 5s", c.httpClient.Timeout)
}
}
func TestNewClientBadNATS(t *testing.T) {
_, err := New(Config{NATSAddr: "nats://127.0.0.1:19999"})
if err == nil {
t.Fatal("expected error connecting to bad NATS address")
}
if !strings.Contains(err.Error(), "nats connect") {
t.Fatalf("unexpected error: %v", err)
}
}
func TestScreenTransaction(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
t.Errorf("method = %s, want POST", r.Method)
}
if r.URL.Path != "/api/EntityAnalysisModel/Invoke" {
t.Errorf("path = %s, want /api/EntityAnalysisModel/Invoke", r.URL.Path)
}
if r.Header.Get("Content-Type") != "application/json" {
t.Errorf("Content-Type = %s, want application/json", r.Header.Get("Content-Type"))
}
var req TransactionRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
t.Fatalf("decode request: %v", err)
}
if req.EntityAnalysisModelID != 1 {
t.Errorf("modelId = %d, want 1", req.EntityAnalysisModelID)
}
json.NewEncoder(w).Encode(TransactionResponse{
Score: 0.85,
Action: ActionBlock,
Alerts: []Alert{{ID: "a1", RuleName: "high-value", Severity: "high"}},
})
}))
defer srv.Close()
c, err := New(Config{BaseURL: srv.URL})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer c.Close()
resp, err := c.ScreenTransaction(context.Background(), TransactionRequest{
EntityAnalysisModelID: 1,
EntityInstanceEntryPayload: map[string]interface{}{
"AccountId": "acct-123",
"Amount": 50000,
"Currency": "USD",
},
})
if err != nil {
t.Fatalf("ScreenTransaction() error: %v", err)
}
if resp.Score != 0.85 {
t.Fatalf("score = %f, want 0.85", resp.Score)
}
if resp.Action != ActionBlock {
t.Fatalf("action = %q, want %q", resp.Action, ActionBlock)
}
if len(resp.Alerts) != 1 {
t.Fatalf("alerts len = %d, want 1", len(resp.Alerts))
}
if resp.Alerts[0].RuleName != "high-value" {
t.Fatalf("alert ruleName = %q, want high-value", resp.Alerts[0].RuleName)
}
}
func TestCheckSanctions(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
t.Errorf("method = %s, want GET", r.Method)
}
if !strings.HasPrefix(r.URL.Path, "/api/Sanction") {
t.Errorf("path = %s, want /api/Sanction", r.URL.Path)
}
name := r.URL.Query().Get("name")
if name != "John Doe" {
t.Errorf("name param = %q, want 'John Doe'", name)
}
country := r.URL.Query().Get("country")
if country != "US" {
t.Errorf("country param = %q, want 'US'", country)
}
json.NewEncoder(w).Encode(SanctionResult{
Hit: true,
Matches: []SanctionMatch{
{ListName: "OFAC SDN", EntityName: "John Doe", Score: 0.95, Country: "US"},
},
})
}))
defer srv.Close()
c, err := New(Config{BaseURL: srv.URL})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer c.Close()
result, err := c.CheckSanctions(context.Background(), "John Doe", "US")
if err != nil {
t.Fatalf("CheckSanctions() error: %v", err)
}
if !result.Hit {
t.Fatal("expected sanctions hit")
}
if len(result.Matches) != 1 {
t.Fatalf("matches len = %d, want 1", len(result.Matches))
}
if result.Matches[0].ListName != "OFAC SDN" {
t.Fatalf("listName = %q, want 'OFAC SDN'", result.Matches[0].ListName)
}
}
func TestCheckSanctionsNoCountry(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Query().Get("country") != "" {
t.Error("expected no country param when empty string passed")
}
json.NewEncoder(w).Encode(SanctionResult{Hit: false})
}))
defer srv.Close()
c, err := New(Config{BaseURL: srv.URL})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer c.Close()
result, err := c.CheckSanctions(context.Background(), "Clean Person", "")
if err != nil {
t.Fatalf("CheckSanctions() error: %v", err)
}
if result.Hit {
t.Fatal("expected no sanctions hit")
}
}
func TestCreateCase(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
t.Errorf("method = %s, want POST", r.Method)
}
if r.URL.Path != "/api/CaseManagement" {
t.Errorf("path = %s, want /api/CaseManagement", r.URL.Path)
}
var req CaseRequest
json.NewDecoder(r.Body).Decode(&req)
w.WriteHeader(http.StatusCreated)
json.NewEncoder(w).Encode(Case{
ID: "case-001",
AccountID: req.AccountID,
Type: req.Type,
Severity: req.Severity,
Status: "open",
Description: req.Description,
CreatedAt: time.Now(),
})
}))
defer srv.Close()
c, err := New(Config{BaseURL: srv.URL})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer c.Close()
cas, err := c.CreateCase(context.Background(), CaseRequest{
AccountID: "acct-123",
Type: "aml",
Severity: "high",
Description: "Suspicious large wire transfer",
})
if err != nil {
t.Fatalf("CreateCase() error: %v", err)
}
if cas.ID != "case-001" {
t.Fatalf("case ID = %q, want case-001", cas.ID)
}
if cas.Status != "open" {
t.Fatalf("case status = %q, want open", cas.Status)
}
}
func TestGetCases(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
t.Errorf("method = %s, want GET", r.Method)
}
if r.URL.Query().Get("accountId") != "acct-123" {
t.Errorf("accountId = %q, want acct-123", r.URL.Query().Get("accountId"))
}
if r.URL.Query().Get("type") != "aml" {
t.Errorf("type = %q, want aml", r.URL.Query().Get("type"))
}
json.NewEncoder(w).Encode([]Case{
{ID: "case-001", AccountID: "acct-123", Type: "aml", Status: "open"},
{ID: "case-002", AccountID: "acct-123", Type: "aml", Status: "investigating"},
})
}))
defer srv.Close()
c, err := New(Config{BaseURL: srv.URL})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer c.Close()
cases, err := c.GetCases(context.Background(), CaseFilter{
AccountID: "acct-123",
Type: "aml",
})
if err != nil {
t.Fatalf("GetCases() error: %v", err)
}
if len(cases) != 2 {
t.Fatalf("cases len = %d, want 2", len(cases))
}
}
func TestGetCasesNoFilters(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.RawQuery != "" {
t.Errorf("expected no query params, got %q", r.URL.RawQuery)
}
json.NewEncoder(w).Encode([]Case{})
}))
defer srv.Close()
c, err := New(Config{BaseURL: srv.URL})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer c.Close()
cases, err := c.GetCases(context.Background(), CaseFilter{})
if err != nil {
t.Fatalf("GetCases() error: %v", err)
}
if len(cases) != 0 {
t.Fatalf("cases len = %d, want 0", len(cases))
}
}
func TestSearch(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
t.Errorf("method = %s, want POST", r.Method)
}
if r.URL.Path != "/api/ExhaustiveSearchInstance" {
t.Errorf("path = %s, want /api/ExhaustiveSearchInstance", r.URL.Path)
}
json.NewEncoder(w).Encode([]SearchResult{
{EntityID: "e1", EntityType: "account", Score: 0.9},
})
}))
defer srv.Close()
c, err := New(Config{BaseURL: srv.URL})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer c.Close()
results, err := c.Search(context.Background(), SearchRequest{
Query: "suspicious activity",
Limit: 10,
})
if err != nil {
t.Fatalf("Search() error: %v", err)
}
if len(results) != 1 {
t.Fatalf("results len = %d, want 1", len(results))
}
if results[0].EntityID != "e1" {
t.Fatalf("entityId = %q, want e1", results[0].EntityID)
}
}
func TestScreenTransactionHTTPError(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(`{"error":"model not found"}`))
}))
defer srv.Close()
c, err := New(Config{BaseURL: srv.URL})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer c.Close()
_, err = c.ScreenTransaction(context.Background(), TransactionRequest{EntityAnalysisModelID: 999})
if err == nil {
t.Fatal("expected error for 500 response")
}
if !strings.Contains(err.Error(), "status 500") {
t.Fatalf("error should mention status 500, got: %v", err)
}
}
func TestScreenTransactionContextCancelled(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
time.Sleep(5 * time.Second)
}))
defer srv.Close()
c, err := New(Config{BaseURL: srv.URL, Timeout: 100 * time.Millisecond})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer c.Close()
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
_, err = c.ScreenTransaction(ctx, TransactionRequest{EntityAnalysisModelID: 1})
if err == nil {
t.Fatal("expected error for cancelled context")
}
}
func TestSubscribeAlertsNoNATS(t *testing.T) {
c, err := New(Config{})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer c.Close()
err = c.SubscribeAlerts(func(a Alert) {})
if err == nil {
t.Fatal("expected error when NATS not configured")
}
if !strings.Contains(err.Error(), "nats not configured") {
t.Fatalf("unexpected error: %v", err)
}
}
// --- Webhook Tests ---
func TestFireWebhookSuccess(t *testing.T) {
var receivedSig string
var receivedBody []byte
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
receivedSig = r.Header.Get("X-Webhook-Signature")
var err error
receivedBody, err = io.ReadAll(r.Body)
if err != nil {
t.Fatalf("read body: %v", err)
}
w.WriteHeader(http.StatusOK)
}))
defer srv.Close()
event := WebhookEvent{
Event: EventAMLFlagged,
Timestamp: time.Now(),
Data: AMLFlaggedData{
AccountID: "acct-123",
TransactionID: "tx-456",
RiskScore: 0.92,
SanctionsHit: true,
Action: ActionBlock,
},
}
err := FireWebhook(context.Background(), event, srv.URL, "test-secret")
if err != nil {
t.Fatalf("FireWebhook() error: %v", err)
}
if receivedSig == "" {
t.Fatal("expected X-Webhook-Signature header")
}
if !VerifySignature(receivedBody, receivedSig, "test-secret") {
t.Fatal("signature verification failed")
}
}
func TestFireWebhookRetries(t *testing.T) {
var attempts int32
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
n := atomic.AddInt32(&attempts, 1)
if n < 3 {
w.WriteHeader(http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusOK)
}))
defer srv.Close()
event := WebhookEvent{
Event: EventAMLCleared,
Timestamp: time.Now(),
Data: AMLClearedData{AccountID: "acct-123", CaseID: "case-001"},
}
err := FireWebhook(context.Background(), event, srv.URL, "secret")
if err != nil {
t.Fatalf("FireWebhook() error: %v (attempts: %d)", err, atomic.LoadInt32(&attempts))
}
got := atomic.LoadInt32(&attempts)
if got != 3 {
t.Fatalf("attempts = %d, want 3", got)
}
}
func TestFireWebhookAllRetriesFail(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusBadGateway)
}))
defer srv.Close()
event := WebhookEvent{
Event: EventAMLFlagged,
Timestamp: time.Now(),
Data: AMLFlaggedData{AccountID: "acct-123"},
}
err := FireWebhook(context.Background(), event, srv.URL, "secret")
if err == nil {
t.Fatal("expected error when all retries fail")
}
if !strings.Contains(err.Error(), "delivery failed after") {
t.Fatalf("unexpected error: %v", err)
}
}
func TestFireWebhookContextCancelled(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusInternalServerError)
}))
defer srv.Close()
ctx, cancel := context.WithCancel(context.Background())
cancel() // cancel immediately
event := WebhookEvent{
Event: EventAMLFlagged,
Timestamp: time.Now(),
Data: AMLFlaggedData{AccountID: "acct-123"},
}
err := FireWebhook(ctx, event, srv.URL, "secret")
if err == nil {
t.Fatal("expected error for cancelled context")
}
}
func TestVerifySignature(t *testing.T) {
payload := []byte(`{"event":"aml.flagged","data":{"accountId":"123"}}`)
secret := "my-hmac-secret"
sig := signPayload(payload, secret)
if !VerifySignature(payload, sig, secret) {
t.Fatal("valid signature rejected")
}
if VerifySignature(payload, sig, "wrong-secret") {
t.Fatal("invalid secret accepted")
}
if VerifySignature(payload, "deadbeef", secret) {
t.Fatal("invalid signature accepted")
}
if VerifySignature([]byte("tampered"), sig, secret) {
t.Fatal("tampered payload accepted")
}
}
func TestActionConstants(t *testing.T) {
if ActionAllow != "allow" {
t.Fatalf("ActionAllow = %q", ActionAllow)
}
if ActionBlock != "block" {
t.Fatalf("ActionBlock = %q", ActionBlock)
}
if ActionReview != "review" {
t.Fatalf("ActionReview = %q", ActionReview)
}
}
func TestEventConstants(t *testing.T) {
if EventAMLFlagged != "aml.flagged" {
t.Fatalf("EventAMLFlagged = %q", EventAMLFlagged)
}
if EventAMLCleared != "aml.cleared" {
t.Fatalf("EventAMLCleared = %q", EventAMLCleared)
}
if EventKYCApproved != "kyc.approved" {
t.Fatalf("EventKYCApproved = %q", EventKYCApproved)
}
}
pkg/compliance/jube/e2e_test.go
-299
View File
@@ -1,299 +0,0 @@
//go:build integration
package jube
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/httptest"
"os"
"sync"
"testing"
"time"
)
// jubeURL returns the Jube endpoint from env or defaults to http://jube:5001.
func jubeURL() string {
if u := os.Getenv("JUBE_URL"); u != "" {
return u
}
return DefaultBaseURL
}
// --- Smoke Tests ---
func TestSmoke_JubeReady(t *testing.T) {
c := &http.Client{Timeout: 5 * time.Second}
resp, err := c.Get(jubeURL() + "/api/Ready")
if err != nil {
t.Fatalf("GET /api/Ready: %v", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
t.Fatalf("GET /api/Ready: status %d, body: %s", resp.StatusCode, string(body))
}
t.Log("Jube is ready")
}
func TestSmoke_SanctionsHit(t *testing.T) {
c := &http.Client{Timeout: 10 * time.Second}
url := jubeURL() + "/api/Invoke/Sanction?multiPartString=OSAMA+BIN+LADEN&distance=1"
resp, err := c.Post(url, "application/json", nil)
if err != nil {
t.Fatalf("POST /api/Invoke/Sanction: %v", err)
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
if resp.StatusCode != http.StatusOK {
t.Fatalf("POST /api/Invoke/Sanction: status %d, body: %s", resp.StatusCode, string(body))
}
// The response should contain sanctions match data.
// Jube returns different structures depending on config, so we just
// verify we got a non-empty JSON response.
if len(body) < 2 {
t.Fatal("expected non-empty sanctions response")
}
t.Logf("Sanctions hit response: %s", string(body))
}
// --- E2E: Structuring Detection ---
func TestE2E_StructuringDetection(t *testing.T) {
client, err := New(Config{BaseURL: jubeURL(), Timeout: 15 * time.Second})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer client.Close()
// Submit a pattern of rapid-fire $9,500 transactions.
// This is the classic structuring pattern: multiple transactions just
// below the $10,000 CTR reporting threshold.
accountID := fmt.Sprintf("e2e-structuring-%d", time.Now().UnixNano())
var highestScore float64
var gotBlock bool
var allAlerts []Alert
for i := 0; i < 5; i++ {
resp, err := client.ScreenTransaction(context.Background(), TransactionRequest{
EntityAnalysisModelID: 1,
EntityInstanceEntryPayload: map[string]interface{}{
"AccountId": accountID,
"TransactionId": fmt.Sprintf("tx-%s-%d", accountID, i),
"Amount": 9500,
"Currency": "USD",
"Side": "buy",
"Symbol": "BTC-USD",
"IP": "192.168.1.100",
},
})
if err != nil {
t.Fatalf("ScreenTransaction #%d error: %v", i, err)
}
t.Logf("Transaction #%d: score=%.2f action=%s alerts=%d",
i, resp.Score, resp.Action, len(resp.Alerts))
if resp.Score > highestScore {
highestScore = resp.Score
}
if resp.Action == ActionBlock {
gotBlock = true
}
allAlerts = append(allAlerts, resp.Alerts...)
}
// Jube should detect structuring with escalating scores.
// We require at least one of:
// - A block action
// - A high risk score (> 0.5)
// - At least one alert
detected := gotBlock || highestScore > 0.5 || len(allAlerts) > 0
if !detected {
t.Fatalf("structuring not detected: highest_score=%.2f blocked=%v alerts=%d",
highestScore, gotBlock, len(allAlerts))
}
t.Logf("Structuring detection: highest_score=%.2f blocked=%v total_alerts=%d",
highestScore, gotBlock, len(allAlerts))
}
// --- E2E: Pre-Trade Screen Integration ---
func TestE2E_PreTradeScreenBlock(t *testing.T) {
client, err := New(Config{BaseURL: jubeURL(), Timeout: 15 * time.Second})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer client.Close()
screen := NewPreTradeScreen(client, PreTradeConfig{
ModelID: 1,
AllowOnReview: false, // strict mode
AllowOnError: false,
})
accountID := fmt.Sprintf("e2e-pretrade-%d", time.Now().UnixNano())
// Submit rapid structuring pattern through the pre-trade screen.
var lastResult PreTradeResult
for i := 0; i < 5; i++ {
lastResult = screen.Screen(context.Background(), ScreenRequest{
AccountID: accountID,
OrderID: fmt.Sprintf("order-%s-%d", accountID, i),
Provider: "alpaca",
Symbol: "BTC-USD",
Side: "buy",
Qty: "1",
Price: "9500",
Currency: "USD",
IP: "10.0.0.1",
})
t.Logf("PreTrade #%d: action=%s allowed=%v score=%.2f errors=%v warnings=%v",
i, lastResult.Action, lastResult.Allowed, lastResult.Score,
lastResult.Errors, lastResult.Warnings)
}
// After 5 structuring transactions, Jube should have escalated.
if lastResult.Score < 0.3 {
t.Logf("WARNING: Jube did not escalate score after 5 structuring txns (score=%.2f)", lastResult.Score)
t.Log("This may indicate Jube's structuring model needs configuration.")
t.Log("Ensure EntityAnalysisModel ID=1 has structuring rules enabled.")
}
}
// --- E2E: Webhook Fires on AML Flag ---
func TestE2E_WebhookFiresOnFlag(t *testing.T) {
var mu sync.Mutex
var received []WebhookEvent
webhookSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var event WebhookEvent
json.NewDecoder(r.Body).Decode(&event)
mu.Lock()
received = append(received, event)
mu.Unlock()
w.WriteHeader(http.StatusOK)
}))
defer webhookSrv.Close()
client, err := New(Config{BaseURL: jubeURL(), Timeout: 15 * time.Second})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer client.Close()
screen := NewPreTradeScreen(client, PreTradeConfig{
ModelID: 1,
AllowOnReview: true,
AllowOnError: true,
WebhookURL: webhookSrv.URL,
WebhookHMACSecret: "e2e-test-secret",
})
accountID := fmt.Sprintf("e2e-webhook-%d", time.Now().UnixNano())
// Submit structuring pattern to trigger flags.
for i := 0; i < 5; i++ {
screen.Screen(context.Background(), ScreenRequest{
AccountID: accountID,
OrderID: fmt.Sprintf("order-%s-%d", accountID, i),
Provider: "alpaca",
Symbol: "BTC-USD",
Side: "buy",
Qty: "1",
Price: "9500",
Currency: "USD",
})
}
// Wait for async webhook deliveries.
time.Sleep(2 * time.Second)
mu.Lock()
defer mu.Unlock()
if len(received) == 0 {
t.Log("No webhooks received. This is expected if Jube does not flag structuring by default.")
t.Log("Configure Jube with structuring rules (EntityAnalysisModel ID=1) to trigger aml.flagged events.")
return
}
for i, event := range received {
if event.Event != EventAMLFlagged {
t.Errorf("webhook #%d: event=%q, want %q", i, event.Event, EventAMLFlagged)
}
}
t.Logf("Received %d aml.flagged webhooks", len(received))
}
// --- E2E: Sanctions Screen via Client ---
func TestE2E_SanctionsScreenSDN(t *testing.T) {
client, err := New(Config{BaseURL: jubeURL(), Timeout: 15 * time.Second})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer client.Close()
result, err := client.CheckSanctions(context.Background(), "OSAMA BIN LADEN", "")
if err != nil {
t.Fatalf("CheckSanctions() error: %v", err)
}
t.Logf("Sanctions result: hit=%v matches=%d", result.Hit, len(result.Matches))
if !result.Hit {
t.Log("WARNING: No sanctions hit for 'OSAMA BIN LADEN'. Check Jube sanctions list configuration.")
}
}
// --- E2E: Case Management ---
func TestE2E_CreateCaseForFlaggedTransaction(t *testing.T) {
client, err := New(Config{BaseURL: jubeURL(), Timeout: 15 * time.Second})
if err != nil {
t.Fatalf("New() error: %v", err)
}
defer client.Close()
accountID := fmt.Sprintf("e2e-case-%d", time.Now().UnixNano())
cas, err := client.CreateCase(context.Background(), CaseRequest{
AccountID: accountID,
Type: "aml",
Severity: "high",
Description: "E2E test: structuring detected in rapid $9,500 transactions",
})
if err != nil {
t.Fatalf("CreateCase() error: %v", err)
}
if cas.ID == "" {
t.Fatal("expected case to have an ID")
}
if cas.Status != "open" {
t.Logf("case status = %q (expected 'open', but Jube may use different defaults)", cas.Status)
}
t.Logf("Created case: id=%s status=%s", cas.ID, cas.Status)
// Verify we can retrieve the case.
cases, err := client.GetCases(context.Background(), CaseFilter{AccountID: accountID, Type: "aml"})
if err != nil {
t.Fatalf("GetCases() error: %v", err)
}
if len(cases) == 0 {
t.Log("WARNING: GetCases returned 0 results. Jube may not support case listing with these filters.")
} else {
t.Logf("Retrieved %d cases for account %s", len(cases), accountID)
}
}
pkg/compliance/jube/models.go
-101
View File
@@ -1,101 +0,0 @@
// Package jube provides an HTTP client for the Jube AML/fraud detection sidecar.
// Jube runs as a sidecar service and exposes REST endpoints for real-time
// transaction risk scoring, sanctions screening, and case management.
package jube
import "time"
// Action constants returned by Jube risk scoring.
const (
ActionAllow = "allow"
ActionBlock = "block"
ActionReview = "review"
)
// TransactionRequest is sent to Jube for real-time risk scoring.
type TransactionRequest struct {
EntityAnalysisModelID int `json:"entityAnalysisModelId"`
EntityInstanceEntryPayload map[string]interface{} `json:"entityInstanceEntryPayload"`
}
// TransactionResponse is the risk assessment returned by Jube.
type TransactionResponse struct {
Score float64 `json:"score"`
Alerts []Alert `json:"alerts"`
Action string `json:"action"` // allow, block, review
ModelTags map[string]string `json:"modelTags,omitempty"`
}
// Alert represents a single AML/fraud alert from Jube.
type Alert struct {
ID string `json:"id"`
RuleName string `json:"ruleName"`
Severity string `json:"severity"` // low, medium, high, critical
Description string `json:"description"`
Score float64 `json:"score"`
CreatedAt time.Time `json:"createdAt"`
}
// SanctionCheckRequest contains the parameters for a sanctions screening query.
type SanctionCheckRequest struct {
Name string `json:"name"`
Country string `json:"country,omitempty"`
}
// SanctionResult is the response from a sanctions screening query.
type SanctionResult struct {
Hit bool `json:"hit"`
Matches []SanctionMatch `json:"matches,omitempty"`
}
// SanctionMatch is a single match from the sanctions list.
type SanctionMatch struct {
ListName string `json:"listName"`
EntityName string `json:"entityName"`
Score float64 `json:"score"`
Country string `json:"country,omitempty"`
}
// CaseRequest is used to create a new compliance case.
type CaseRequest struct {
AccountID string `json:"accountId"`
TransactionID string `json:"transactionId,omitempty"`
Type string `json:"type"` // aml, fraud, sanctions
Severity string `json:"severity"`
Description string `json:"description"`
}
// Case represents a compliance case in Jube.
type Case struct {
ID string `json:"id"`
AccountID string `json:"accountId"`
TransactionID string `json:"transactionId,omitempty"`
Type string `json:"type"`
Severity string `json:"severity"`
Status string `json:"status"` // open, investigating, escalated, closed
Description string `json:"description"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
}
// CaseFilter contains query parameters for listing cases.
type CaseFilter struct {
AccountID string `json:"accountId,omitempty"`
Type string `json:"type,omitempty"`
Status string `json:"status,omitempty"`
}
// SearchRequest is sent to the exhaustive search endpoint.
type SearchRequest struct {
Query string `json:"query"`
Filters map[string]string `json:"filters,omitempty"`
Limit int `json:"limit,omitempty"`
}
// SearchResult is a single result from the exhaustive search.
type SearchResult struct {
EntityID string `json:"entityId"`
EntityType string `json:"entityType"`
Score float64 `json:"score"`
Data map[string]interface{} `json:"data,omitempty"`
}
pkg/compliance/jube/pretrade.go
-214
View File
@@ -1,214 +0,0 @@
package jube
import (
"context"
"encoding/json"
"fmt"
"strconv"
"time"
"github.com/rs/zerolog/log"
)
// PreTradeAction describes what the broker should do with the order.
type PreTradeAction string
const (
PreTradeAllow PreTradeAction = "allow"
PreTradeBlock PreTradeAction = "block"
PreTradeReview PreTradeAction = "review"
)
// PreTradeConfig controls how the Jube pre-trade screen behaves.
type PreTradeConfig struct {
// ModelID is the Jube EntityAnalysisModel to invoke.
ModelID int
// AllowOnReview controls whether orders flagged "review" are allowed
// to proceed (true) or held (false). Default: true (allow, flag async).
AllowOnReview bool
// AllowOnError controls whether orders proceed if Jube is unreachable.
// Default: true (fail-open). Set false for fail-closed.
AllowOnError bool
// WebhookURL is the target for aml.flagged webhook events. Empty = no webhook.
WebhookURL string
// WebhookHMACSecret is the HMAC-SHA256 secret for signing webhooks.
// Must come from KMS, never hardcoded.
WebhookHMACSecret string
}
// PreTradeResult is the outcome of a Jube AML screen on an order.
type PreTradeResult struct {
Action PreTradeAction `json:"action"`
Allowed bool `json:"allowed"`
Score float64 `json:"score"`
Alerts []Alert `json:"alerts,omitempty"`
Errors []string `json:"errors,omitempty"`
Warnings []string `json:"warnings,omitempty"`
}
// PreTradeScreen wraps a Jube client and provides pre-trade AML screening
// for the broker's order flow. Call Screen() before executing any trade.
type PreTradeScreen struct {
client *Client
cfg PreTradeConfig
}
// NewPreTradeScreen creates a pre-trade screener using the given Jube client.
func NewPreTradeScreen(client *Client, cfg PreTradeConfig) *PreTradeScreen {
if cfg.ModelID == 0 {
cfg.ModelID = 1 // default model
}
return &PreTradeScreen{
client: client,
cfg: cfg,
}
}
// Screen submits a transaction to Jube for AML/fraud scoring and returns
// a decision. This should be called in the broker's order creation path
// after risk checks pass but before the order reaches the provider.
func (s *PreTradeScreen) Screen(ctx context.Context, req ScreenRequest) PreTradeResult {
txReq := TransactionRequest{
EntityAnalysisModelID: s.cfg.ModelID,
EntityInstanceEntryPayload: map[string]interface{}{
"AccountId": req.AccountID,
"TransactionId": req.OrderID,
"Amount": req.Amount(),
"Currency": req.Currency,
"Symbol": req.Symbol,
"Side": req.Side,
"Provider": req.Provider,
"IP": req.IP,
},
}
resp, err := s.client.ScreenTransaction(ctx, txReq)
if err != nil {
log.Error().Err(err).
Str("account", req.AccountID).
Str("symbol", req.Symbol).
Msg("jube: pre-trade screen failed")
if s.cfg.AllowOnError {
return PreTradeResult{
Action: PreTradeAllow,
Allowed: true,
Warnings: []string{fmt.Sprintf("jube unavailable: %v", err)},
}
}
return PreTradeResult{
Action: PreTradeBlock,
Allowed: false,
Errors: []string{fmt.Sprintf("compliance check unavailable: %v", err)},
}
}
result := PreTradeResult{
Score: resp.Score,
Alerts: resp.Alerts,
}
switch resp.Action {
case ActionBlock:
result.Action = PreTradeBlock
result.Allowed = false
result.Errors = append(result.Errors, "transaction blocked by AML screening")
for _, a := range resp.Alerts {
result.Errors = append(result.Errors, fmt.Sprintf("[%s] %s (score=%.2f)", a.Severity, a.RuleName, a.Score))
}
case ActionReview:
result.Action = PreTradeReview
result.Allowed = s.cfg.AllowOnReview
result.Warnings = append(result.Warnings, "transaction flagged for manual review")
for _, a := range resp.Alerts {
result.Warnings = append(result.Warnings, fmt.Sprintf("[%s] %s (score=%.2f)", a.Severity, a.RuleName, a.Score))
}
default: // allow
result.Action = PreTradeAllow
result.Allowed = true
}
// Fire webhook async for flagged/blocked transactions.
if resp.Action == ActionBlock || resp.Action == ActionReview {
if s.cfg.WebhookURL != "" && s.cfg.WebhookHMACSecret != "" {
go s.fireAMLWebhook(req, resp)
}
// Publish to NATS if connected.
if s.client.natsConn != nil && len(resp.Alerts) > 0 {
go s.publishNATSAlert(req, resp)
}
}
return result
}
// ScreenRequest contains the order details for AML screening.
type ScreenRequest struct {
AccountID string
OrderID string
Provider string
Symbol string
Side string
Qty string
Price string
Currency string
IP string
}
// Amount returns the estimated USD value of the order.
func (r ScreenRequest) Amount() float64 {
q, _ := strconv.ParseFloat(r.Qty, 64)
p, _ := strconv.ParseFloat(r.Price, 64)
if p == 0 {
p = 1
}
return q * p
}
// fireAMLWebhook sends an aml.flagged webhook event.
func (s *PreTradeScreen) fireAMLWebhook(req ScreenRequest, resp *TransactionResponse) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
event := WebhookEvent{
Event: EventAMLFlagged,
Timestamp: time.Now().UTC(),
Data: AMLFlaggedData{
AccountID: req.AccountID,
TransactionID: req.OrderID,
RiskScore: resp.Score,
Alerts: resp.Alerts,
Action: resp.Action,
},
}
if err := FireWebhook(ctx, event, s.cfg.WebhookURL, s.cfg.WebhookHMACSecret); err != nil {
log.Error().Err(err).
Str("account", req.AccountID).
Msg("jube: failed to fire aml.flagged webhook")
}
}
// publishNATSAlert publishes an alert to NATS for async processing.
func (s *PreTradeScreen) publishNATSAlert(req ScreenRequest, resp *TransactionResponse) {
for _, alert := range resp.Alerts {
data, err := json.Marshal(alert)
if err != nil {
log.Error().Err(err).Msg("jube: failed to marshal alert for NATS")
continue
}
if err := s.client.natsConn.Publish(NATSSubjectAlerts, data); err != nil {
log.Error().Err(err).
Str("account", req.AccountID).
Str("alert", alert.ID).
Msg("jube: failed to publish NATS alert")
}
}
}
pkg/compliance/jube/pretrade_test.go
-396
View File
@@ -1,396 +0,0 @@
package jube
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"sync"
"testing"
"time"
)
// --- PreTradeScreen Unit Tests ---
func newTestScreen(t *testing.T, handler http.HandlerFunc, cfg PreTradeConfig) (*PreTradeScreen, func()) {
t.Helper()
srv := httptest.NewServer(handler)
c, err := New(Config{BaseURL: srv.URL})
if err != nil {
t.Fatalf("New() error: %v", err)
}
screen := NewPreTradeScreen(c, cfg)
return screen, func() {
c.Close()
srv.Close()
}
}
func TestScreenAllowCleanTransaction(t *testing.T) {
screen, cleanup := newTestScreen(t, func(w http.ResponseWriter, r *http.Request) {
json.NewEncoder(w).Encode(TransactionResponse{
Score: 0.1,
Action: ActionAllow,
})
}, PreTradeConfig{})
defer cleanup()
result := screen.Screen(context.Background(), ScreenRequest{
AccountID: "acct-clean",
Symbol: "AAPL",
Side: "buy",
Qty: "10",
Price: "150.00",
Currency: "USD",
})
if !result.Allowed {
t.Fatalf("expected allowed=true, got false; errors: %v", result.Errors)
}
if result.Action != PreTradeAllow {
t.Fatalf("action = %q, want %q", result.Action, PreTradeAllow)
}
if result.Score != 0.1 {
t.Fatalf("score = %f, want 0.1", result.Score)
}
if len(result.Errors) != 0 {
t.Fatalf("expected no errors, got %v", result.Errors)
}
}
func TestScreenBlockHighRisk(t *testing.T) {
screen, cleanup := newTestScreen(t, func(w http.ResponseWriter, r *http.Request) {
json.NewEncoder(w).Encode(TransactionResponse{
Score: 0.95,
Action: ActionBlock,
Alerts: []Alert{
{ID: "a1", RuleName: "structuring", Severity: "critical", Score: 0.95},
},
})
}, PreTradeConfig{})
defer cleanup()
result := screen.Screen(context.Background(), ScreenRequest{
AccountID: "acct-sus",
Symbol: "BTC-USD",
Side: "buy",
Qty: "1",
Price: "9500",
Currency: "USD",
})
if result.Allowed {
t.Fatal("expected allowed=false for blocked transaction")
}
if result.Action != PreTradeBlock {
t.Fatalf("action = %q, want %q", result.Action, PreTradeBlock)
}
if result.Score != 0.95 {
t.Fatalf("score = %f, want 0.95", result.Score)
}
if len(result.Errors) < 2 {
t.Fatalf("expected at least 2 errors (block msg + alert), got %d: %v", len(result.Errors), result.Errors)
}
if len(result.Alerts) != 1 {
t.Fatalf("expected 1 alert, got %d", len(result.Alerts))
}
}
func TestScreenReviewAllowedByDefault(t *testing.T) {
screen, cleanup := newTestScreen(t, func(w http.ResponseWriter, r *http.Request) {
json.NewEncoder(w).Encode(TransactionResponse{
Score: 0.65,
Action: ActionReview,
Alerts: []Alert{
{ID: "a1", RuleName: "velocity", Severity: "medium", Score: 0.65},
},
})
}, PreTradeConfig{AllowOnReview: true})
defer cleanup()
result := screen.Screen(context.Background(), ScreenRequest{
AccountID: "acct-review",
Symbol: "MSFT",
Side: "buy",
Qty: "100",
Price: "300",
Currency: "USD",
})
if !result.Allowed {
t.Fatal("expected allowed=true when AllowOnReview=true")
}
if result.Action != PreTradeReview {
t.Fatalf("action = %q, want %q", result.Action, PreTradeReview)
}
if len(result.Warnings) == 0 {
t.Fatal("expected warnings for review action")
}
}
func TestScreenReviewBlockedWhenConfigured(t *testing.T) {
screen, cleanup := newTestScreen(t, func(w http.ResponseWriter, r *http.Request) {
json.NewEncoder(w).Encode(TransactionResponse{
Score: 0.70,
Action: ActionReview,
Alerts: []Alert{
{ID: "a1", RuleName: "suspicious-pattern", Severity: "high", Score: 0.70},
},
})
}, PreTradeConfig{AllowOnReview: false})
defer cleanup()
result := screen.Screen(context.Background(), ScreenRequest{
AccountID: "acct-strict",
Symbol: "ETH-USD",
Side: "sell",
Qty: "50",
Price: "2000",
Currency: "USD",
})
if result.Allowed {
t.Fatal("expected allowed=false when AllowOnReview=false")
}
if result.Action != PreTradeReview {
t.Fatalf("action = %q, want %q", result.Action, PreTradeReview)
}
}
func TestScreenFailOpenOnError(t *testing.T) {
screen, cleanup := newTestScreen(t, func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(`{"error":"internal"}`))
}, PreTradeConfig{AllowOnError: true})
defer cleanup()
result := screen.Screen(context.Background(), ScreenRequest{
AccountID: "acct-err",
Symbol: "GOOG",
Side: "buy",
Qty: "5",
Price: "100",
Currency: "USD",
})
if !result.Allowed {
t.Fatal("expected allowed=true (fail-open) when Jube returns error")
}
if result.Action != PreTradeAllow {
t.Fatalf("action = %q, want %q", result.Action, PreTradeAllow)
}
if len(result.Warnings) == 0 {
t.Fatal("expected warning about Jube unavailability")
}
}
func TestScreenFailClosedOnError(t *testing.T) {
screen, cleanup := newTestScreen(t, func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(`{"error":"internal"}`))
}, PreTradeConfig{AllowOnError: false})
defer cleanup()
result := screen.Screen(context.Background(), ScreenRequest{
AccountID: "acct-err",
Symbol: "GOOG",
Side: "buy",
Qty: "5",
Price: "100",
Currency: "USD",
})
if result.Allowed {
t.Fatal("expected allowed=false (fail-closed) when Jube returns error")
}
if result.Action != PreTradeBlock {
t.Fatalf("action = %q, want %q", result.Action, PreTradeBlock)
}
if len(result.Errors) == 0 {
t.Fatal("expected error about compliance unavailability")
}
}
func TestScreenSendsCorrectPayload(t *testing.T) {
var received TransactionRequest
screen, cleanup := newTestScreen(t, func(w http.ResponseWriter, r *http.Request) {
json.NewDecoder(r.Body).Decode(&received)
json.NewEncoder(w).Encode(TransactionResponse{Action: ActionAllow, Score: 0.05})
}, PreTradeConfig{ModelID: 42})
defer cleanup()
screen.Screen(context.Background(), ScreenRequest{
AccountID: "acct-payload",
OrderID: "order-789",
Provider: "alpaca",
Symbol: "AAPL",
Side: "buy",
Qty: "10",
Price: "150",
Currency: "USD",
IP: "10.0.0.1",
})
if received.EntityAnalysisModelID != 42 {
t.Fatalf("modelId = %d, want 42", received.EntityAnalysisModelID)
}
payload := received.EntityInstanceEntryPayload
if payload["AccountId"] != "acct-payload" {
t.Fatalf("AccountId = %v, want acct-payload", payload["AccountId"])
}
if payload["TransactionId"] != "order-789" {
t.Fatalf("TransactionId = %v, want order-789", payload["TransactionId"])
}
if payload["Symbol"] != "AAPL" {
t.Fatalf("Symbol = %v, want AAPL", payload["Symbol"])
}
// Amount should be 10 * 150 = 1500
if amt, ok := payload["Amount"].(float64); !ok || amt != 1500 {
t.Fatalf("Amount = %v, want 1500", payload["Amount"])
}
}
func TestScreenWebhookFiredOnBlock(t *testing.T) {
var webhookReceived bool
var webhookBody WebhookEvent
var mu sync.Mutex
webhookSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
defer mu.Unlock()
webhookReceived = true
json.NewDecoder(r.Body).Decode(&webhookBody)
w.WriteHeader(http.StatusOK)
}))
defer webhookSrv.Close()
screen, cleanup := newTestScreen(t, func(w http.ResponseWriter, r *http.Request) {
json.NewEncoder(w).Encode(TransactionResponse{
Score: 0.90,
Action: ActionBlock,
Alerts: []Alert{{ID: "a1", RuleName: "high-risk", Severity: "critical"}},
})
}, PreTradeConfig{
WebhookURL: webhookSrv.URL,
WebhookHMACSecret: "test-webhook-secret",
})
defer cleanup()
screen.Screen(context.Background(), ScreenRequest{
AccountID: "acct-webhook",
OrderID: "order-wh",
Symbol: "BTC-USD",
Side: "buy",
Qty: "1",
Price: "50000",
Currency: "USD",
})
// Wait for async webhook delivery.
time.Sleep(500 * time.Millisecond)
mu.Lock()
defer mu.Unlock()
if !webhookReceived {
t.Fatal("expected webhook to be fired on block")
}
if webhookBody.Event != EventAMLFlagged {
t.Fatalf("webhook event = %q, want %q", webhookBody.Event, EventAMLFlagged)
}
}
func TestScreenDefaultModelID(t *testing.T) {
screen := NewPreTradeScreen(&Client{}, PreTradeConfig{})
if screen.cfg.ModelID != 1 {
t.Fatalf("default ModelID = %d, want 1", screen.cfg.ModelID)
}
}
func TestScreenCustomModelID(t *testing.T) {
screen := NewPreTradeScreen(&Client{}, PreTradeConfig{ModelID: 99})
if screen.cfg.ModelID != 99 {
t.Fatalf("ModelID = %d, want 99", screen.cfg.ModelID)
}
}
func TestScreenRequestAmount(t *testing.T) {
tests := []struct {
qty, price string
want float64
}{
{"10", "150", 1500},
{"0.5", "100", 50},
{"100", "", 100}, // no price = market order, p defaults to 1
{"", "100", 0}, // no qty = 0
{"abc", "100", 0}, // invalid qty
}
for _, tt := range tests {
r := ScreenRequest{Qty: tt.qty, Price: tt.price}
got := r.Amount()
if got != tt.want {
t.Errorf("Amount(%q, %q) = %f, want %f", tt.qty, tt.price, got, tt.want)
}
}
}
func TestScreenMultipleAlerts(t *testing.T) {
screen, cleanup := newTestScreen(t, func(w http.ResponseWriter, r *http.Request) {
json.NewEncoder(w).Encode(TransactionResponse{
Score: 0.88,
Action: ActionBlock,
Alerts: []Alert{
{ID: "a1", RuleName: "structuring", Severity: "critical", Score: 0.88},
{ID: "a2", RuleName: "velocity", Severity: "high", Score: 0.75},
{ID: "a3", RuleName: "geography", Severity: "medium", Score: 0.60},
},
})
}, PreTradeConfig{})
defer cleanup()
result := screen.Screen(context.Background(), ScreenRequest{
AccountID: "acct-multi",
Symbol: "BTC-USD",
Side: "buy",
Qty: "1",
Price: "9500",
Currency: "USD",
})
if result.Allowed {
t.Fatal("expected blocked")
}
if len(result.Alerts) != 3 {
t.Fatalf("expected 3 alerts, got %d", len(result.Alerts))
}
// 1 block message + 3 alert details = 4 errors
if len(result.Errors) != 4 {
t.Fatalf("expected 4 errors, got %d: %v", len(result.Errors), result.Errors)
}
}
func TestScreenContextCancelled(t *testing.T) {
screen, cleanup := newTestScreen(t, func(w http.ResponseWriter, r *http.Request) {
time.Sleep(5 * time.Second)
}, PreTradeConfig{AllowOnError: false})
defer cleanup()
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
result := screen.Screen(ctx, ScreenRequest{
AccountID: "acct-timeout",
Symbol: "AAPL",
Side: "buy",
Qty: "1",
Price: "100",
Currency: "USD",
})
if result.Allowed {
t.Fatal("expected blocked on timeout with fail-closed")
}
if len(result.Errors) == 0 {
t.Fatal("expected error message")
}
}
pkg/compliance/jube/webhook.go
-132
View File
@@ -1,132 +0,0 @@
package jube
import (
"bytes"
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"math"
"net/http"
"time"
)
// Webhook event type constants.
const (
EventAMLFlagged = "aml.flagged"
EventAMLCleared = "aml.cleared"
EventKYCApproved = "kyc.approved"
EventTradeExecuted = "trade.executed"
)
// maxWebhookRetries is the maximum number of delivery attempts.
const maxWebhookRetries = 5
// webhookInitialBackoff is the base delay for exponential backoff.
const webhookInitialBackoff = 500 * time.Millisecond
// WebhookEvent is the envelope sent to webhook targets.
type WebhookEvent struct {
Event string `json:"event"`
Timestamp time.Time `json:"timestamp"`
Data interface{} `json:"data"`
}
// AMLFlaggedData is the payload for aml.flagged events.
type AMLFlaggedData struct {
AccountID string `json:"accountId"`
TransactionID string `json:"transactionId"`
RiskScore float64 `json:"riskScore"`
Alerts []Alert `json:"alerts"`
SanctionsHit bool `json:"sanctionsHit"`
PEPMatch bool `json:"pepMatch"`
Action string `json:"action"`
}
// AMLClearedData is the payload for aml.cleared events.
type AMLClearedData struct {
AccountID string `json:"accountId"`
TransactionID string `json:"transactionId"`
CaseID string `json:"caseId"`
ClearedBy string `json:"clearedBy"`
}
// KYCApprovedData is the payload for kyc.approved events.
type KYCApprovedData struct {
AccountID string `json:"accountId"`
UserID string `json:"userId"`
Provider string `json:"provider"`
VerifiedAt string `json:"verifiedAt"`
}
// FireWebhook sends a signed webhook event to the target URL.
// The payload is signed with HMAC-SHA256 using hmacSecret, and the signature
// is sent in the X-Webhook-Signature header. Delivery is retried up to
// maxWebhookRetries times with exponential backoff on failure.
func FireWebhook(ctx context.Context, event WebhookEvent, targetURL string, hmacSecret string) error {
payload, err := json.Marshal(event)
if err != nil {
return fmt.Errorf("jube webhook: marshal event: %w", err)
}
sig := signPayload(payload, hmacSecret)
var lastErr error
for attempt := 0; attempt < maxWebhookRetries; attempt++ {
if attempt > 0 {
backoff := time.Duration(float64(webhookInitialBackoff) * math.Pow(2, float64(attempt-1)))
select {
case <-ctx.Done():
return fmt.Errorf("jube webhook: context cancelled after %d attempts: %w", attempt, ctx.Err())
case <-time.After(backoff):
}
}
lastErr = sendWebhook(ctx, targetURL, payload, sig)
if lastErr == nil {
return nil
}
}
return fmt.Errorf("jube webhook: delivery failed after %d attempts: %w", maxWebhookRetries, lastErr)
}
// sendWebhook performs a single webhook delivery attempt.
func sendWebhook(ctx context.Context, targetURL string, payload []byte, signature string) error {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, targetURL, bytes.NewReader(payload))
if err != nil {
return fmt.Errorf("create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Webhook-Signature", signature)
client := &http.Client{Timeout: 15 * time.Second}
resp, err := client.Do(req)
if err != nil {
return fmt.Errorf("http post: %w", err)
}
defer resp.Body.Close()
io.Copy(io.Discard, resp.Body)
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return fmt.Errorf("target returned status %d", resp.StatusCode)
}
return nil
}
// signPayload computes HMAC-SHA256 of payload using the given secret.
func signPayload(payload []byte, secret string) string {
mac := hmac.New(sha256.New, []byte(secret))
mac.Write(payload)
return hex.EncodeToString(mac.Sum(nil))
}
// VerifySignature checks that the given signature matches the expected
// HMAC-SHA256 of payload. Use this on the receiving end of webhooks.
func VerifySignature(payload []byte, signature string, secret string) bool {
expected := signPayload(payload, secret)
return hmac.Equal([]byte(expected), []byte(signature))
}
pkg/compliance/router.go
+1 -1
View File
@@ -12,7 +12,7 @@ import (
"github.com/go-chi/cors"
"github.com/luxfi/broker/pkg/admin"
"github.com/luxfi/broker/pkg/auth"
"github.com/luxfi/broker/pkg/compliance/jube"
"github.com/luxfi/compliance/pkg/jube"
)
// RouterOption configures optional dependencies for the compliance router.
pkg/compliance/store.go
-851
View File
@@ -1,851 +0,0 @@
package compliance
import (
"crypto/rand"
"encoding/hex"
"fmt"
"sync"
"time"
)
// ComplianceStore is the interface for compliance data persistence.
// MemoryStore and PostgresStore both implement this interface.
type ComplianceStore interface {
// Identity
SaveIdentity(id *Identity) error
GetIdentity(id string) (*Identity, error)
// Pipeline
SavePipeline(p *Pipeline) error
GetPipeline(id string) (*Pipeline, error)
ListPipelines() []*Pipeline
DeletePipeline(id string) error
// Session
SaveSession(sess *Session) error
GetSession(id string) (*Session, error)
ListSessions() []*Session
// Fund
SaveFund(f *Fund) error
GetFund(id string) (*Fund, error)
ListFunds() []*Fund
DeleteFund(id string) error
AddFundInvestor(inv *FundInvestor) error
ListFundInvestors(fundID string) []*FundInvestor
// Envelope
SaveEnvelope(env *Envelope) error
GetEnvelope(id string) (*Envelope, error)
ListEnvelopes() []*Envelope
// Template
SaveTemplate(t *Template) error
GetTemplate(id string) (*Template, error)
ListTemplates() []*Template
// Role
SaveRole(role *Role) error
GetRole(id string) (*Role, error)
GetRoleByName(name string) (*Role, error)
ListRoles() []*Role
DeleteRole(id string) error
// User
SaveUser(u *User) error
GetUser(id string) (*User, error)
ListUsers() []*User
// Transaction
SaveTransaction(tx *Transaction) error
ListTransactions() []*Transaction
// Credential
SaveCredential(c *Credential) error
ListCredentials() []*Credential
DeleteCredential(id string) error
// AML Screening
SaveAMLScreening(s *AMLScreening) error
GetAMLScreening(id string) (*AMLScreening, error)
ListAMLScreeningsByAccount(accountID string) []*AMLScreening
ListAMLScreeningsByStatus(status AMLStatus) []*AMLScreening
// Application (onboarding)
SaveApplication(app *Application) error
GetApplication(id string) (*Application, error)
GetApplicationByUser(userID string) (*Application, error)
ListApplications() []*Application
ListApplicationsByStatus(status ApplicationStatus) []*Application
// Document Upload
SaveDocumentUpload(doc *DocumentUpload) error
GetDocumentUpload(id string) (*DocumentUpload, error)
ListDocumentUploads(applicationID string) []*DocumentUpload
// Identity listing
ListIdentitiesByUser(userID string) []*Identity
// Settings
GetSettings() *Settings
SaveSettings(settings *Settings)
// Dashboard
ComputeDashboard() *DashboardStats
ComputeESignStats() *ESignStats
ListEnvelopesByDirection(direction string) []*Envelope
}
// MemoryStore is an in-memory compliance data store.
// Use PostgresStore in production.
type MemoryStore struct {
mu sync.RWMutex
identities map[string]*Identity // id -> Identity
businesses map[string]*BusinessKYB // id -> BusinessKYB
pipelines map[string]*Pipeline // id -> Pipeline
sessions map[string]*Session // id -> Session
funds map[string]*Fund // id -> Fund
investors map[string][]*FundInvestor // fundID -> investors
envelopes map[string]*Envelope // id -> Envelope
templates map[string]*Template // id -> Template
roles map[string]*Role // id -> Role
users map[string]*User // id -> User
transactions map[string]*Transaction // id -> Transaction
credentials map[string]*Credential // id -> Credential
amlScreenings map[string]*AMLScreening // id -> AMLScreening
applications map[string]*Application // id -> Application
documentUploads map[string]*DocumentUpload // id -> DocumentUpload
settings *Settings
}
// NewMemoryStore creates an empty in-memory compliance store.
func NewMemoryStore() *MemoryStore {
return &MemoryStore{
identities: make(map[string]*Identity),
businesses: make(map[string]*BusinessKYB),
pipelines: make(map[string]*Pipeline),
sessions: make(map[string]*Session),
funds: make(map[string]*Fund),
investors: make(map[string][]*FundInvestor),
envelopes: make(map[string]*Envelope),
templates: make(map[string]*Template),
roles: make(map[string]*Role),
users: make(map[string]*User),
transactions: make(map[string]*Transaction),
credentials: make(map[string]*Credential),
amlScreenings: make(map[string]*AMLScreening),
applications: make(map[string]*Application),
documentUploads: make(map[string]*DocumentUpload),
settings: &Settings{
BusinessName: "Your Company",
Timezone: "America/New_York",
Currency: "USD",
NotificationEmail: "[email protected]",
},
}
}
// NewStore is an alias for NewMemoryStore, preserving backward compatibility.
func NewStore() *MemoryStore {
return NewMemoryStore()
}
// --- Identity ---
func (s *MemoryStore) SaveIdentity(id *Identity) error {
s.mu.Lock()
defer s.mu.Unlock()
if id.ID == "" {
id.ID = generateID()
}
if id.CreatedAt.IsZero() {
id.CreatedAt = time.Now().UTC()
}
id.UpdatedAt = time.Now().UTC()
s.identities[id.ID] = id
return nil
}
func (s *MemoryStore) GetIdentity(id string) (*Identity, error) {
s.mu.RLock()
defer s.mu.RUnlock()
ident, ok := s.identities[id]
if !ok {
return nil, fmt.Errorf("identity not found")
}
return ident, nil
}
// --- Pipeline ---
func (s *MemoryStore) SavePipeline(p *Pipeline) error {
s.mu.Lock()
defer s.mu.Unlock()
if p.ID == "" {
p.ID = generateID()
}
if p.CreatedAt.IsZero() {
p.CreatedAt = time.Now().UTC()
}
p.UpdatedAt = time.Now().UTC()
s.pipelines[p.ID] = p
return nil
}
func (s *MemoryStore) GetPipeline(id string) (*Pipeline, error) {
s.mu.RLock()
defer s.mu.RUnlock()
p, ok := s.pipelines[id]
if !ok {
return nil, fmt.Errorf("pipeline not found")
}
return p, nil
}
func (s *MemoryStore) ListPipelines() []*Pipeline {
s.mu.RLock()
defer s.mu.RUnlock()
out := make([]*Pipeline, 0, len(s.pipelines))
for _, p := range s.pipelines {
out = append(out, p)
}
return out
}
func (s *MemoryStore) DeletePipeline(id string) error {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.pipelines[id]; !ok {
return fmt.Errorf("pipeline not found")
}
delete(s.pipelines, id)
return nil
}
// --- Session ---
func (s *MemoryStore) SaveSession(sess *Session) error {
s.mu.Lock()
defer s.mu.Unlock()
if sess.ID == "" {
sess.ID = generateID()
}
if sess.CreatedAt.IsZero() {
sess.CreatedAt = time.Now().UTC()
}
s.sessions[sess.ID] = sess
return nil
}
func (s *MemoryStore) GetSession(id string) (*Session, error) {
s.mu.RLock()
defer s.mu.RUnlock()
sess, ok := s.sessions[id]
if !ok {
return nil, fmt.Errorf("session not found")
}
return sess, nil
}
func (s *MemoryStore) ListSessions() []*Session {
s.mu.RLock()
defer s.mu.RUnlock()
out := make([]*Session, 0, len(s.sessions))
for _, sess := range s.sessions {
out = append(out, sess)
}
return out
}
// --- Fund ---
func (s *MemoryStore) SaveFund(f *Fund) error {
s.mu.Lock()
defer s.mu.Unlock()
if f.ID == "" {
f.ID = generateID()
}
if f.CreatedAt.IsZero() {
f.CreatedAt = time.Now().UTC()
}
f.UpdatedAt = time.Now().UTC()
s.funds[f.ID] = f
return nil
}
func (s *MemoryStore) GetFund(id string) (*Fund, error) {
s.mu.RLock()
defer s.mu.RUnlock()
f, ok := s.funds[id]
if !ok {
return nil, fmt.Errorf("fund not found")
}
return f, nil
}
func (s *MemoryStore) ListFunds() []*Fund {
s.mu.RLock()
defer s.mu.RUnlock()
out := make([]*Fund, 0, len(s.funds))
for _, f := range s.funds {
out = append(out, f)
}
return out
}
func (s *MemoryStore) DeleteFund(id string) error {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.funds[id]; !ok {
return fmt.Errorf("fund not found")
}
delete(s.funds, id)
delete(s.investors, id)
return nil
}
func (s *MemoryStore) AddFundInvestor(inv *FundInvestor) error {
s.mu.Lock()
defer s.mu.Unlock()
if inv.ID == "" {
inv.ID = generateID()
}
if inv.CreatedAt.IsZero() {
inv.CreatedAt = time.Now().UTC()
}
f, ok := s.funds[inv.FundID]
if !ok {
return fmt.Errorf("fund not found")
}
s.investors[inv.FundID] = append(s.investors[inv.FundID], inv)
f.InvestorCount = len(s.investors[inv.FundID])
f.TotalRaised += inv.Amount
return nil
}
func (s *MemoryStore) ListFundInvestors(fundID string) []*FundInvestor {
s.mu.RLock()
defer s.mu.RUnlock()
return s.investors[fundID]
}
// --- Envelope ---
func (s *MemoryStore) SaveEnvelope(env *Envelope) error {
s.mu.Lock()
defer s.mu.Unlock()
if env.ID == "" {
env.ID = generateID()
}
if env.CreatedAt.IsZero() {
env.CreatedAt = time.Now().UTC()
}
env.UpdatedAt = time.Now().UTC()
s.envelopes[env.ID] = env
return nil
}
func (s *MemoryStore) GetEnvelope(id string) (*Envelope, error) {
s.mu.RLock()
defer s.mu.RUnlock()
env, ok := s.envelopes[id]
if !ok {
return nil, fmt.Errorf("envelope not found")
}
return env, nil
}
func (s *MemoryStore) ListEnvelopes() []*Envelope {
s.mu.RLock()
defer s.mu.RUnlock()
out := make([]*Envelope, 0, len(s.envelopes))
for _, env := range s.envelopes {
out = append(out, env)
}
return out
}
// --- Template ---
func (s *MemoryStore) SaveTemplate(t *Template) error {
s.mu.Lock()
defer s.mu.Unlock()
if t.ID == "" {
t.ID = generateID()
}
if t.CreatedAt.IsZero() {
t.CreatedAt = time.Now().UTC()
}
s.templates[t.ID] = t
return nil
}
func (s *MemoryStore) GetTemplate(id string) (*Template, error) {
s.mu.RLock()
defer s.mu.RUnlock()
t, ok := s.templates[id]
if !ok {
return nil, fmt.Errorf("template not found")
}
return t, nil
}
func (s *MemoryStore) ListTemplates() []*Template {
s.mu.RLock()
defer s.mu.RUnlock()
out := make([]*Template, 0, len(s.templates))
for _, t := range s.templates {
out = append(out, t)
}
return out
}
// --- Role ---
func (s *MemoryStore) SaveRole(role *Role) error {
s.mu.Lock()
defer s.mu.Unlock()
if role.ID == "" {
role.ID = generateID()
}
if role.CreatedAt.IsZero() {
role.CreatedAt = time.Now().UTC()
}
role.UpdatedAt = time.Now().UTC()
s.roles[role.ID] = role
return nil
}
func (s *MemoryStore) GetRole(id string) (*Role, error) {
s.mu.RLock()
defer s.mu.RUnlock()
role, ok := s.roles[id]
if !ok {
return nil, fmt.Errorf("role not found")
}
return role, nil
}
func (s *MemoryStore) GetRoleByName(name string) (*Role, error) {
s.mu.RLock()
defer s.mu.RUnlock()
for _, role := range s.roles {
if role.Name == name {
return role, nil
}
}
return nil, fmt.Errorf("role not found: %s", name)
}
func (s *MemoryStore) ListRoles() []*Role {
s.mu.RLock()
defer s.mu.RUnlock()
out := make([]*Role, 0, len(s.roles))
for _, role := range s.roles {
out = append(out, role)
}
return out
}
func (s *MemoryStore) DeleteRole(id string) error {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.roles[id]; !ok {
return fmt.Errorf("role not found")
}
delete(s.roles, id)
return nil
}
// --- User ---
func (s *MemoryStore) SaveUser(u *User) error {
s.mu.Lock()
defer s.mu.Unlock()
if u.ID == "" {
u.ID = generateID()
}
if u.CreatedAt.IsZero() {
u.CreatedAt = time.Now().UTC()
}
s.users[u.ID] = u
return nil
}
func (s *MemoryStore) GetUser(id string) (*User, error) {
s.mu.RLock()
defer s.mu.RUnlock()
u, ok := s.users[id]
if !ok {
return nil, fmt.Errorf("user not found")
}
return u, nil
}
func (s *MemoryStore) ListUsers() []*User {
s.mu.RLock()
defer s.mu.RUnlock()
out := make([]*User, 0, len(s.users))
for _, u := range s.users {
out = append(out, u)
}
return out
}
// --- Transaction ---
func (s *MemoryStore) SaveTransaction(tx *Transaction) error {
s.mu.Lock()
defer s.mu.Unlock()
if tx.ID == "" {
tx.ID = generateID()
}
s.transactions[tx.ID] = tx
return nil
}
func (s *MemoryStore) ListTransactions() []*Transaction {
s.mu.RLock()
defer s.mu.RUnlock()
out := make([]*Transaction, 0, len(s.transactions))
for _, tx := range s.transactions {
out = append(out, tx)
}
return out
}
// --- Credential ---
func (s *MemoryStore) SaveCredential(c *Credential) error {
s.mu.Lock()
defer s.mu.Unlock()
if c.ID == "" {
c.ID = generateID()
}
if c.CreatedAt.IsZero() {
c.CreatedAt = time.Now().UTC()
}
s.credentials[c.ID] = c
return nil
}
func (s *MemoryStore) ListCredentials() []*Credential {
s.mu.RLock()
defer s.mu.RUnlock()
out := make([]*Credential, 0, len(s.credentials))
for _, c := range s.credentials {
out = append(out, c)
}
return out
}
func (s *MemoryStore) DeleteCredential(id string) error {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.credentials[id]; !ok {
return fmt.Errorf("credential not found")
}
delete(s.credentials, id)
return nil
}
// --- Settings ---
func (s *MemoryStore) GetSettings() *Settings {
s.mu.RLock()
defer s.mu.RUnlock()
cp := *s.settings
return &cp
}
func (s *MemoryStore) SaveSettings(settings *Settings) {
s.mu.Lock()
defer s.mu.Unlock()
s.settings = settings
}
// --- Dashboard ---
// ComputeDashboard aggregates store data into dashboard statistics.
func (s *MemoryStore) ComputeDashboard() *DashboardStats {
s.mu.RLock()
defer s.mu.RUnlock()
stats := &DashboardStats{
TotalFunds: len(s.funds),
MonthlyTransactions: len(s.transactions),
}
for _, sess := range s.sessions {
if sess.Status == SessionPending || sess.Status == SessionInProgress {
stats.ActiveSessions++
}
if sess.KYCStatus == KYCPending {
stats.PendingKYC++
}
}
stats.RecentSessions = make([]*Session, 0, 5)
count := 0
for _, sess := range s.sessions {
if count >= 5 {
break
}
stats.RecentSessions = append(stats.RecentSessions, sess)
count++
}
stats.RecentTransactions = make([]*Transaction, 0, 5)
count = 0
for _, tx := range s.transactions {
if count >= 5 {
break
}
stats.RecentTransactions = append(stats.RecentTransactions, tx)
count++
}
return stats
}
// --- eSign Dashboard ---
// ComputeESignStats aggregates eSign envelope/template data.
func (s *MemoryStore) ComputeESignStats() *ESignStats {
s.mu.RLock()
defer s.mu.RUnlock()
stats := &ESignStats{
Templates: len(s.templates),
}
for _, env := range s.envelopes {
switch env.Status {
case EnvelopePending, EnvelopeSent, EnvelopeViewed:
stats.Pending++
case EnvelopeCompleted, EnvelopeSigned:
stats.Completed++
}
}
stats.Draft = stats.Pending
stats.Recent = make([]*Envelope, 0, 5)
count := 0
for _, env := range s.envelopes {
if count >= 5 {
break
}
stats.Recent = append(stats.Recent, env)
count++
}
return stats
}
// ListEnvelopesByDirection returns envelopes filtered by direction.
// "inbox" returns envelopes with status sent/pending (received for signing).
// "sent" returns envelopes that were created/sent out.
func (s *MemoryStore) ListEnvelopesByDirection(direction string) []*Envelope {
s.mu.RLock()
defer s.mu.RUnlock()
out := make([]*Envelope, 0)
for _, env := range s.envelopes {
switch direction {
case "inbox":
if env.Status == EnvelopeSent || env.Status == EnvelopePending || env.Status == EnvelopeViewed {
out = append(out, env)
}
case "sent":
if env.Status == EnvelopeSigned || env.Status == EnvelopeCompleted || env.Status == EnvelopeDeclined || env.Status == EnvelopeVoided {
out = append(out, env)
}
}
}
return out
}
// --- AML Screening ---
func (s *MemoryStore) SaveAMLScreening(sc *AMLScreening) error {
s.mu.Lock()
defer s.mu.Unlock()
if sc.ID == "" {
sc.ID = generateID()
}
if sc.CreatedAt.IsZero() {
sc.CreatedAt = time.Now().UTC()
}
sc.UpdatedAt = time.Now().UTC()
s.amlScreenings[sc.ID] = sc
return nil
}
func (s *MemoryStore) GetAMLScreening(id string) (*AMLScreening, error) {
s.mu.RLock()
defer s.mu.RUnlock()
sc, ok := s.amlScreenings[id]
if !ok {
return nil, fmt.Errorf("aml screening not found")
}
return sc, nil
}
func (s *MemoryStore) ListAMLScreeningsByAccount(accountID string) []*AMLScreening {
s.mu.RLock()
defer s.mu.RUnlock()
var out []*AMLScreening
for _, sc := range s.amlScreenings {
if sc.AccountID == accountID {
out = append(out, sc)
}
}
if out == nil {
out = make([]*AMLScreening, 0)
}
return out
}
func (s *MemoryStore) ListAMLScreeningsByStatus(status AMLStatus) []*AMLScreening {
s.mu.RLock()
defer s.mu.RUnlock()
var out []*AMLScreening
for _, sc := range s.amlScreenings {
if sc.Status == status {
out = append(out, sc)
}
}
if out == nil {
out = make([]*AMLScreening, 0)
}
return out
}
// --- Application ---
func (s *MemoryStore) SaveApplication(app *Application) error {
s.mu.Lock()
defer s.mu.Unlock()
if app.ID == "" {
app.ID = generateID()
}
if app.CreatedAt.IsZero() {
app.CreatedAt = time.Now().UTC()
}
app.UpdatedAt = time.Now().UTC()
s.applications[app.ID] = app
return nil
}
func (s *MemoryStore) GetApplication(id string) (*Application, error) {
s.mu.RLock()
defer s.mu.RUnlock()
app, ok := s.applications[id]
if !ok {
return nil, fmt.Errorf("application not found")
}
return app, nil
}
func (s *MemoryStore) GetApplicationByUser(userID string) (*Application, error) {
s.mu.RLock()
defer s.mu.RUnlock()
for _, app := range s.applications {
if app.UserID == userID {
return app, nil
}
}
return nil, fmt.Errorf("application not found")
}
func (s *MemoryStore) ListApplications() []*Application {
s.mu.RLock()
defer s.mu.RUnlock()
out := make([]*Application, 0, len(s.applications))
for _, app := range s.applications {
out = append(out, app)
}
return out
}
func (s *MemoryStore) ListApplicationsByStatus(status ApplicationStatus) []*Application {
s.mu.RLock()
defer s.mu.RUnlock()
var out []*Application
for _, app := range s.applications {
if app.Status == status {
out = append(out, app)
}
}
if out == nil {
out = make([]*Application, 0)
}
return out
}
// --- Document Upload ---
func (s *MemoryStore) SaveDocumentUpload(doc *DocumentUpload) error {
s.mu.Lock()
defer s.mu.Unlock()
if doc.ID == "" {
doc.ID = generateID()
}
if doc.CreatedAt.IsZero() {
doc.CreatedAt = time.Now().UTC()
}
doc.UpdatedAt = time.Now().UTC()
s.documentUploads[doc.ID] = doc
return nil
}
func (s *MemoryStore) GetDocumentUpload(id string) (*DocumentUpload, error) {
s.mu.RLock()
defer s.mu.RUnlock()
doc, ok := s.documentUploads[id]
if !ok {
return nil, fmt.Errorf("document not found")
}
return doc, nil
}
func (s *MemoryStore) ListDocumentUploads(applicationID string) []*DocumentUpload {
s.mu.RLock()
defer s.mu.RUnlock()
var out []*DocumentUpload
for _, doc := range s.documentUploads {
if doc.ApplicationID == applicationID {
out = append(out, doc)
}
}
if out == nil {
out = make([]*DocumentUpload, 0)
}
return out
}
// --- Identity listing ---
func (s *MemoryStore) ListIdentitiesByUser(userID string) []*Identity {
s.mu.RLock()
defer s.mu.RUnlock()
var out []*Identity
for _, ident := range s.identities {
if ident.UserID == userID {
out = append(out, ident)
}
}
if out == nil {
out = make([]*Identity, 0)
}
return out
}
// generateID returns a random hex ID. Panics if the system CSPRNG fails,
// which indicates a catastrophic OS-level problem.
func generateID() string {
b := make([]byte, 16)
if _, err := rand.Read(b); err != nil {
panic("crypto/rand: " + err.Error())
}
return hex.EncodeToString(b)
}