Coverage Report

Created: 2025-10-06 12:43

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/home/jgrusewski/Work/foxhunt/config/src/compliance_config.rs
Line
Count
Source
1
//! Compliance rule configuration and hot-reload support
2
//!
3
//! Provides database-backed compliance rule loading with PostgreSQL NOTIFY/LISTEN
4
//! for hot-reload capabilities. Integrates with the ComplianceValidator in the
5
//! risk crate to enable dynamic rule configuration without service restarts.
6
7
use serde::{Deserialize, Serialize};
8
9
#[cfg(feature = "postgres")]
10
use crate::error::ConfigResult;
11
#[cfg(feature = "postgres")]
12
use std::collections::HashMap;
13
#[cfg(feature = "postgres")]
14
use std::sync::Arc;
15
#[cfg(feature = "postgres")]
16
use std::time::Duration;
17
#[cfg(feature = "postgres")]
18
use tokio::sync::RwLock;
19
#[cfg(feature = "postgres")]
20
use tracing::{error, info};
21
22
#[cfg(feature = "postgres")]
23
use sqlx::postgres::{PgListener, PgPool};
24
25
/// Compliance rule loader with PostgreSQL integration and hot-reload support
26
///
27
/// Loads compliance rules from the PostgreSQL database and automatically
28
/// reloads them when changes are detected via PostgreSQL NOTIFY/LISTEN.
29
#[cfg(feature = "postgres")]
30
pub struct PostgresComplianceRuleLoader {
31
    /// Database connection pool
32
    pool: PgPool,
33
    /// PostgreSQL listener for rule change notifications
34
    listener: Arc<RwLock<Option<PgListener>>>,
35
    /// Cached compliance rules by rule_id
36
    rules_cache: Arc<RwLock<HashMap<String, ComplianceRuleConfig>>>,
37
    /// Cache timeout duration
38
    cache_timeout: Duration,
39
}
40
41
/// Compliance rule configuration structure
42
///
43
/// Represents a compliance rule loaded from the database.
44
/// This structure is designed to be compatible with both the database
45
/// schema and the ComplianceRule type in the risk crate.
46
#[derive(Debug, Clone, Serialize, Deserialize)]
47
#[cfg_attr(feature = "postgres", derive(sqlx::FromRow))]
48
pub struct ComplianceRuleConfig {
49
    /// Unique rule identifier
50
    pub rule_id: String,
51
    /// Human-readable rule name
52
    pub name: String,
53
    /// Detailed description
54
    pub description: String,
55
    /// Rule type (POSITION_LIMIT, MARKET_ABUSE, etc.)
56
    pub rule_type: String,
57
    /// Whether the rule is active
58
    pub active: bool,
59
    /// Rule version for audit trail
60
    pub version: i32,
61
    /// Severity level (Info, Low, Medium, High, Critical)
62
    pub severity: String,
63
    /// Priority for evaluation (0-100)
64
    pub priority: i32,
65
    /// Flexible rule parameters as JSON
66
    #[cfg_attr(feature = "postgres", sqlx(json))]
67
    pub parameters: serde_json::Value,
68
    /// Regulatory framework
69
    pub regulatory_framework: Option<String>,
70
    /// Regulatory reference
71
    pub regulatory_reference: Option<String>,
72
}
73
74
#[cfg(feature = "postgres")]
75
impl PostgresComplianceRuleLoader {
76
    /// Creates a new compliance rule loader with PostgreSQL integration
77
    ///
78
    /// # Arguments
79
    ///
80
    /// * `database_url` - PostgreSQL connection URL
81
    ///
82
    /// # Returns
83
    ///
84
    /// Result containing the initialized loader or an error
85
    pub async fn new(database_url: &str) -> ConfigResult<Self> {
86
        let pool = PgPool::connect(database_url).await?;
87
88
        Ok(Self {
89
            pool,
90
            listener: Arc::new(RwLock::new(None)),
91
            rules_cache: Arc::new(RwLock::new(HashMap::new())),
92
            cache_timeout: Duration::from_secs(300), // 5 minutes
93
        })
94
    }
95
96
    /// Creates a loader with an existing connection pool
97
    ///
98
    /// # Arguments
99
    ///
100
    /// * `pool` - Existing PostgreSQL connection pool
101
    ///
102
    /// # Returns
103
    ///
104
    /// Configured compliance rule loader
105
    pub fn with_pool(pool: PgPool) -> Self {
106
        Self {
107
            pool,
108
            listener: Arc::new(RwLock::new(None)),
109
            rules_cache: Arc::new(RwLock::new(HashMap::new())),
110
            cache_timeout: Duration::from_secs(300),
111
        }
112
    }
113
114
    /// Starts listening for rule change notifications
115
    ///
116
    /// Initiates PostgreSQL NOTIFY/LISTEN for hot-reload capabilities.
117
    /// When a rule is changed in the database, the cache will be automatically
118
    /// invalidated and reloaded.
119
    ///
120
    /// # Returns
121
    ///
122
    /// Result indicating success or error
123
    pub async fn start_listener(&self) -> ConfigResult<()> {
124
        let mut listener = PgListener::connect_with(&self.pool)
125
            .await?;
126
127
        listener
128
            .listen("compliance_rules_changed")
129
            .await?;
130
131
        *self.listener.write().await = Some(listener);
132
133
        info!("PostgreSQL NOTIFY/LISTEN started for compliance rule hot-reload");
134
135
        // Spawn background task to handle notifications
136
        let listener_clone = Arc::clone(&self.listener);
137
        let cache_clone = Arc::clone(&self.rules_cache);
138
        let pool_clone = self.pool.clone();
139
140
        tokio::spawn(async move {
141
            loop {
142
                let mut listener_guard = listener_clone.write().await;
143
                if let Some(listener) = listener_guard.as_mut() {
144
                    match listener.try_recv().await {
145
                        Ok(Some(notification)) => {
146
                            info!(
147
                                "Compliance rule change notification received: {}",
148
                                notification.payload()
149
                            );
150
151
                            // Parse notification payload to get rule_id
152
                            if let Ok(payload) = serde_json::from_str::<serde_json::Value>(
153
                                notification.payload(),
154
                            ) {
155
                                if let Some(rule_id) = payload.get("rule_id").and_then(|v| v.as_str()) {
156
                                    // Invalidate cache for this rule
157
                                    cache_clone.write().await.remove(rule_id);
158
                                    info!("Invalidated cache for compliance rule: {}", rule_id);
159
160
                                    // Optionally reload the rule immediately
161
                                    if let Err(e) = Self::reload_rule_static(&pool_clone, &cache_clone, rule_id).await {
162
                                        error!("Failed to reload compliance rule {}: {}", rule_id, e);
163
                                    }
164
                                }
165
                            }
166
                        }
167
                        Ok(None) => {
168
                            // No notification available, continue
169
                            tokio::time::sleep(Duration::from_millis(100)).await;
170
                        }
171
                        Err(e) => {
172
                            error!("Error receiving compliance rule notification: {}", e);
173
                            tokio::time::sleep(Duration::from_secs(1)).await;
174
                        }
175
                    }
176
                }
177
                drop(listener_guard);
178
                tokio::time::sleep(Duration::from_millis(100)).await;
179
            }
180
        });
181
182
        Ok(())
183
    }
184
185
    /// Static helper for reloading a single rule (used in background task)
186
    async fn reload_rule_static(
187
        pool: &PgPool,
188
        cache: &Arc<RwLock<HashMap<String, ComplianceRuleConfig>>>,
189
        rule_id: &str,
190
    ) -> ConfigResult<()> {
191
        let query = "SELECT rule_id, name, description, rule_type::text as rule_type, active, version,
192
                           severity, priority, parameters, regulatory_framework, regulatory_reference
193
                    FROM compliance_rules
194
                    WHERE rule_id = $1 AND active = true";
195
196
        let row = sqlx::query_as::<_, ComplianceRuleConfig>(query)
197
            .bind(rule_id)
198
            .fetch_optional(pool)
199
            .await?;
200
201
        if let Some(rule) = row {
202
            cache.write().await.insert(rule_id.to_string(), rule);
203
            info!("Reloaded compliance rule: {}", rule_id);
204
        }
205
206
        Ok(())
207
    }
208
209
    /// Loads all active compliance rules from the database
210
    ///
211
    /// # Returns
212
    ///
213
    /// Vector of active compliance rules
214
    pub async fn load_all_active_rules(&self) -> ConfigResult<Vec<ComplianceRuleConfig>> {
215
        let query = "SELECT rule_id, name, description, rule_type::text as rule_type, active, version,
216
                           severity, priority, parameters, regulatory_framework, regulatory_reference
217
                    FROM compliance_rules
218
                    WHERE active = true
219
                      AND effective_date <= NOW()
220
                      AND (expiry_date IS NULL OR expiry_date > NOW())
221
                    ORDER BY priority DESC, created_at ASC";
222
223
        let rules = sqlx::query_as::<_, ComplianceRuleConfig>(query)
224
            .fetch_all(&self.pool)
225
            .await?;
226
227
        // Update cache
228
        let mut cache = self.rules_cache.write().await;
229
        for rule in &rules {
230
            cache.insert(rule.rule_id.clone(), rule.clone());
231
        }
232
233
        info!("Loaded {} active compliance rules", rules.len());
234
235
        Ok(rules)
236
    }
237
238
    /// Loads rules filtered by type
239
    ///
240
    /// # Arguments
241
    ///
242
    /// * `rule_type` - Rule type to filter by (e.g., "POSITION_LIMIT")
243
    ///
244
    /// # Returns
245
    ///
246
    /// Vector of rules matching the specified type
247
    pub async fn load_rules_by_type(&self, rule_type: &str) -> ConfigResult<Vec<ComplianceRuleConfig>> {
248
        let query = "SELECT rule_id, name, description, rule_type::text as rule_type, active, version,
249
                           severity, priority, parameters, regulatory_framework, regulatory_reference
250
                    FROM compliance_rules
251
                    WHERE active = true
252
                      AND rule_type::text = $1
253
                      AND effective_date <= NOW()
254
                      AND (expiry_date IS NULL OR expiry_date > NOW())
255
                    ORDER BY priority DESC";
256
257
        let rules = sqlx::query_as::<_, ComplianceRuleConfig>(query)
258
            .bind(rule_type)
259
            .fetch_all(&self.pool)
260
            .await?;
261
262
        Ok(rules)
263
    }
264
265
    /// Gets a specific rule by ID (with caching)
266
    ///
267
    /// # Arguments
268
    ///
269
    /// * `rule_id` - Unique rule identifier
270
    ///
271
    /// # Returns
272
    ///
273
    /// Optional compliance rule configuration
274
    pub async fn get_rule(&self, rule_id: &str) -> ConfigResult<Option<ComplianceRuleConfig>> {
275
        // Check cache first
276
        {
277
            let cache = self.rules_cache.read().await;
278
            if let Some(rule) = cache.get(rule_id) {
279
                return Ok(Some(rule.clone()));
280
            }
281
        }
282
283
        // Load from database if not in cache
284
        let query = "SELECT rule_id, name, description, rule_type::text as rule_type, active, version,
285
                           severity, priority, parameters, regulatory_framework, regulatory_reference
286
                    FROM compliance_rules
287
                    WHERE rule_id = $1 AND active = true";
288
289
        let rule = sqlx::query_as::<_, ComplianceRuleConfig>(query)
290
            .bind(rule_id)
291
            .fetch_optional(&self.pool)
292
            .await?;
293
294
        // Update cache if found
295
        if let Some(ref rule_data) = rule {
296
            self.rules_cache.write().await.insert(rule_id.to_string(), rule_data.clone());
297
        }
298
299
        Ok(rule)
300
    }
301
302
    /// Records a compliance rule execution for audit trail
303
    ///
304
    /// # Arguments
305
    ///
306
    /// * `rule_id` - Rule that was executed
307
    /// * `result` - Execution result (PASS, WARN, FAIL, ERROR)
308
    /// * `violation_detected` - Whether a violation was detected
309
    /// * `order_id` - Optional order ID
310
    /// * `instrument_id` - Optional instrument ID
311
    ///
312
    /// # Returns
313
    ///
314
    /// Result indicating success or error
315
    pub async fn record_execution(
316
        &self,
317
        rule_id: &str,
318
        result: &str,
319
        violation_detected: bool,
320
        order_id: Option<&str>,
321
        instrument_id: Option<&str>,
322
    ) -> ConfigResult<()> {
323
        let query = "SELECT record_compliance_rule_execution($1, $2, $3, $4, $5, NULL, NULL, NULL)";
324
325
        sqlx::query(query)
326
            .bind(rule_id)
327
            .bind(result)
328
            .bind(violation_detected)
329
            .bind(order_id)
330
            .bind(instrument_id)
331
            .execute(&self.pool)
332
            .await?;
333
334
        Ok(())
335
    }
336
337
    /// Clears the rule cache (forces reload on next access)
338
    pub async fn clear_cache(&self) {
339
        self.rules_cache.write().await.clear();
340
        info!("Compliance rule cache cleared");
341
    }
342
343
    /// Gets the current cache size
344
    pub async fn cache_size(&self) -> usize {
345
        self.rules_cache.read().await.len()
346
    }
347
}
348
349
#[cfg(test)]
350
mod tests {
351
    use super::*;
352
353
    #[test]
354
1
    fn test_compliance_rule_config_structure() {
355
1
        let rule = ComplianceRuleConfig {
356
1
            rule_id: "test_rule".to_string(),
357
1
            name: "Test Rule".to_string(),
358
1
            description: "Test description".to_string(),
359
1
            rule_type: "POSITION_LIMIT".to_string(),
360
1
            active: true,
361
1
            version: 1,
362
1
            severity: "High".to_string(),
363
1
            priority: 80,
364
1
            parameters: serde_json::json!({"max_position": 1000000}),
365
1
            regulatory_framework: Some("Basel III".to_string()),
366
1
            regulatory_reference: Some("Article 123".to_string()),
367
1
        };
368
369
1
        assert_eq!(rule.rule_id, "test_rule");
370
1
        assert!(rule.active);
371
1
        assert_eq!(rule.priority, 80);
372
1
    }
373
374
    #[test]
375
1
    fn test_compliance_rule_config_serialization() {
376
1
        let rule = ComplianceRuleConfig {
377
1
            rule_id: "test_rule".to_string(),
378
1
            name: "Test Rule".to_string(),
379
1
            description: "Test description".to_string(),
380
1
            rule_type: "MARKET_ABUSE".to_string(),
381
1
            active: true,
382
1
            version: 1,
383
1
            severity: "Critical".to_string(),
384
1
            priority: 95,
385
1
            parameters: serde_json::json!({"threshold": 1000000}),
386
1
            regulatory_framework: None,
387
1
            regulatory_reference: None,
388
1
        };
389
390
1
        let json = serde_json::to_string(&rule).expect("Failed to serialize");
391
1
        assert!(json.contains("test_rule"));
392
1
        assert!(json.contains("MARKET_ABUSE"));
393
394
1
        let deserialized: ComplianceRuleConfig =
395
1
            serde_json::from_str(&json).expect("Failed to deserialize");
396
1
        assert_eq!(deserialized.rule_id, rule.rule_id);
397
1
        assert_eq!(deserialized.severity, rule.severity);
398
1
    }
399
}