/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 | | } |