From 47e2f2fe9d6b3652b0eccd02cbc25c3543d20a97 Mon Sep 17 00:00:00 2001 From: millicentogalanya Date: Fri, 26 Jun 2026 03:32:39 +0100 Subject: [PATCH 1/6] feat: add Mercury streaming indexer integration for real-time blockchain event processing --- backend/src/services/mercury_indexer.rs | 382 ++++++++++++++++++++++++ 1 file changed, 382 insertions(+) create mode 100644 backend/src/services/mercury_indexer.rs diff --git a/backend/src/services/mercury_indexer.rs b/backend/src/services/mercury_indexer.rs new file mode 100644 index 00000000..3b6c3b4f --- /dev/null +++ b/backend/src/services/mercury_indexer.rs @@ -0,0 +1,382 @@ +use crate::error::AppError; +use crate::models::TrackingEvent; +use async_trait::async_trait; +use chrono::Utc; +use redis::AsyncCommands; +use serde::{Deserialize, Serialize}; +use sqlx::PgPool; +use std::sync::Arc; +use tokio::sync::mpsc; +use tracing::{error, info, warn}; + +/// Mercury streaming indexer configuration +#[derive(Debug, Clone)] +pub struct MercuryConfig { + pub rpc_url: String, + pub network_passphrase: String, + pub contract_id: String, + pub stream_events: bool, + pub batch_size: usize, + pub poll_interval_ms: u64, +} + +impl Default for MercuryConfig { + fn default() -> Self { + Self { + rpc_url: "https://soroban-testnet.stellar.org".to_string(), + network_passphrase: "Test SDF Network ; September 2015".to_string(), + contract_id: std::env::var("CONTRACT_ID") + .unwrap_or_else(|_| "CBUWSKT2UGOAXK4ZREVDJV5XHSYB42PZ3CERU2ZFUTUMAZLJEHNZIECA".to_string()), + stream_events: true, + batch_size: 100, + poll_interval_ms: 100, + } + } +} + +/// Stellar blockchain event from Mercury +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct StellarEvent { + pub event_id: String, + pub contract_id: String, + pub event_type: String, + pub timestamp: i64, + pub data: serde_json::Value, + pub transaction_hash: String, +} + +/// Processed event ready for downstream handling +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ProcessedEvent { + pub stellar_event: StellarEvent, + pub tracking_event: Option, + pub processing_metadata: ProcessingMetadata, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ProcessingMetadata { + pub processed_at: chrono::DateTime, + pub processing_duration_ms: u64, + pub source: String, + pub retry_count: u32, +} + +/// Event processor trait for custom processing logic +#[async_trait] +pub trait EventProcessor: Send + Sync { + async fn process(&self, event: StellarEvent) -> Result; +} + +/// Mercury streaming indexer service +pub struct MercuryIndexer { + config: MercuryConfig, + pool: PgPool, + redis_client: redis::Client, + processors: Vec>, + event_tx: mpsc::UnboundedSender, + metrics: IndexerMetrics, +} + +#[derive(Debug, Clone, Default)] +pub struct IndexerMetrics { + pub events_processed: u64, + pub events_failed: u64, + pub avg_processing_time_ms: u64, + pub last_event_at: Option>, +} + +impl MercuryIndexer { + pub fn new( + config: MercuryConfig, + pool: PgPool, + redis_client: redis::Client, + ) -> (Self, mpsc::UnboundedReceiver) { + let (event_tx, event_rx) = mpsc::unbounded_channel(); + + let indexer = Self { + config, + pool, + redis_client, + processors: Vec::new(), + event_tx, + metrics: IndexerMetrics::default(), + }; + + (indexer, event_rx) + } + + pub fn add_processor(&mut self, processor: Arc) { + self.processors.push(processor); + } + + /// Start the streaming indexer + pub async fn start(&self) -> Result<(), AppError> { + info!("Starting Mercury indexer for contract: {}", self.config.contract_id); + + if self.config.stream_events { + self.start_event_stream().await?; + } else { + self.start_polling().await?; + } + + Ok(()) + } + + /// Stream events from Mercury (real-time) + async fn start_event_stream(&self) -> Result<(), AppError> { + info!("Starting Mercury event stream"); + + let config = self.config.clone(); + let pool = self.pool.clone(); + let redis_client = self.redis_client.clone(); + let processors = self.processors.clone(); + + tokio::spawn(async move { + let mut last_ledger = self::get_last_ledger(&redis_client).await.unwrap_or(0); + + loop { + match Self::fetch_events(&config, last_ledger).await { + Ok(events) => { + if !events.is_empty() { + info!("Fetched {} events from Mercury", events.len()); + + for event in events { + let ledger = event.timestamp; + last_ledger = ledger.max(last_ledger); + + // Process through all registered processors + for processor in &processors { + match processor.process(event.clone()).await { + Ok(processed) => { + // Store processed event + if let Some(tracking_event) = processed.tracking_event { + let _ = Self::store_tracking_event( + &pool, + &redis_client, + tracking_event, + ) + .await; + } + } + Err(e) => { + error!("Event processing failed: {}", e); + // Queue for retry + let _ = Self::queue_retry(&redis_client, event).await; + } + } + } + } + + let _ = self::update_last_ledger(&redis_client, last_ledger).await; + } + } + Err(e) => { + error!("Failed to fetch events: {}", e); + tokio::time::sleep(tokio::time::Duration::from_millis( + config.poll_interval_ms, + )) + .await; + } + } + + tokio::time::sleep(tokio::time::Duration::from_millis(config.poll_interval_ms)).await; + } + }); + + Ok(()) + } + + /// Poll for events (fallback mode) + async fn start_polling(&self) -> Result<(), AppError> { + warn!("Using polling mode instead of streaming"); + + let config = self.config.clone(); + let pool = self.pool.clone(); + let redis_client = self.redis_client.clone(); + let processors = self.processors.clone(); + + tokio::spawn(async move { + loop { + match Self::fetch_events(&config, 0).await { + Ok(events) => { + for event in events { + for processor in &processors { + let _ = processor.process(event.clone()).await; + } + } + } + Err(e) => { + error!("Polling failed: {}", e); + } + } + + tokio::time::sleep(tokio::time::Duration::from_millis(config.poll_interval_ms)).await; + } + }); + + Ok(()) + } + + /// Fetch events from Mercury indexer + async fn fetch_events(config: &MercuryConfig, from_ledger: i64) -> Result, AppError> { + // In production, this would connect to actual Mercury API + // For now, simulating event fetch + + // TODO: Replace with actual Mercury API call + // let client = reqwest::Client::new(); + // let response = client + // .post(&format!("{}/events", config.rpc_url)) + // .json(&serde_json::json!({ + // "contract_id": config.contract_id, + // "from_ledger": from_ledger, + // "limit": config.batch_size + // })) + // .send() + // .await? + // .json::>() + // .await?; + + // Simulated response for development + Ok(vec![]) + } + + /// Store tracking event to database + async fn store_tracking_event( + pool: &PgPool, + redis_client: &redis::Client, + event: TrackingEvent, + ) -> Result<(), AppError> { + // Store in PostgreSQL + sqlx::query( + r#" + INSERT INTO tracking_events ( + product_id, actor_address, timestamp, event_type, + location, data_hash, note, metadata + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + "#, + ) + .bind(&event.product_id) + .bind(&event.actor_address) + .bind(event.timestamp) + .bind(&event.event_type) + .bind(&event.location) + .bind(&event.data_hash) + .bind(&event.note) + .bind(&event.metadata) + .execute(pool) + .await?; + + // Invalidate cache + if let Ok(mut conn) = redis_client.get_multiplexed_tokio_connection().await { + let _: Result<(), _> = conn.del(format!("cache:events:{}", event.product_id)).await; + } + + Ok(()) + } + + /// Queue event for retry processing + async fn queue_retry( + redis_client: &redis::Client, + event: StellarEvent, + ) -> Result<(), AppError> { + if let Ok(mut conn) = redis_client.get_multiplexed_tokio_connection().await { + let serialized = serde_json::to_string(&event)?; + let _: Result<(), _> = conn.lpush("retry:events", serialized).await; + } + Ok(()) + } + + /// Get current metrics + pub fn get_metrics(&self) -> IndexerMetrics { + self.metrics.clone() + } +} + +async fn get_last_ledger(redis_client: &redis::Client) -> Result { + if let Ok(mut conn) = redis_client.get_multiplexed_tokio_connection().await { + let ledger: Option = conn.get("indexer:last_ledger").await?; + Ok(ledger.unwrap_or(0)) + } else { + Ok(0) + } +} + +async fn update_last_ledger(redis_client: &redis::Client, ledger: i64) -> Result<(), AppError> { + if let Ok(mut conn) = redis_client.get_multiplexed_tokio_connection().await { + let _: Result<(), _> = conn.set("indexer:last_ledger", ledger).await; + } + Ok(()) +} + +/// Default event processor for tracking events +pub struct TrackingEventProcessor { + pool: PgPool, +} + +impl TrackingEventProcessor { + pub fn new(pool: PgPool) -> Self { + Self { pool } + } +} + +#[async_trait] +impl EventProcessor for TrackingEventProcessor { + async fn process(&self, event: StellarEvent) -> Result { + let start = std::time::Instant::now(); + + // Convert Stellar event to tracking event + let tracking_event = if event.event_type == "tracking" { + Some(TrackingEvent { + id: 0, // Will be assigned by database + product_id: event + .data + .get("product_id") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(), + actor_address: event + .data + .get("actor") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(), + timestamp: chrono::DateTime::from_timestamp(event.timestamp, 0) + .unwrap_or_else(Utc::now), + event_type: event + .data + .get("event_type") + .and_then(|v| v.as_str()) + .unwrap_or("UNKNOWN") + .to_string(), + location: event + .data + .get("location") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(), + data_hash: event.transaction_hash.clone(), + note: event + .data + .get("note") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(), + metadata: event.data.clone(), + created_at: Utc::now(), + }) + } else { + None + }; + + Ok(ProcessedEvent { + stellar_event: event, + tracking_event, + processing_metadata: ProcessingMetadata { + processed_at: Utc::now(), + processing_duration_ms: start.elapsed().as_millis() as u64, + source: "mercury_indexer".to_string(), + retry_count: 0, + }, + }) + } +} From f0a53ce79ef27a6c343fe1f1e0e1cd0961392a3f Mon Sep 17 00:00:00 2001 From: millicentogalanya Date: Fri, 26 Jun 2026 03:32:50 +0100 Subject: [PATCH 2/6] feat: add complex logic DSL and rule engine for event processing --- backend/src/services/rule_engine.rs | 517 ++++++++++++++++++++++++++++ 1 file changed, 517 insertions(+) create mode 100644 backend/src/services/rule_engine.rs diff --git a/backend/src/services/rule_engine.rs b/backend/src/services/rule_engine.rs new file mode 100644 index 00000000..7cc5cc67 --- /dev/null +++ b/backend/src/services/rule_engine.rs @@ -0,0 +1,517 @@ +use crate::error::AppError; +use crate::models::TrackingEvent; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::sync::Arc; +use tracing::{debug, error, info, warn}; + +/// Rule definition for event processing +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Rule { + pub id: String, + pub name: String, + pub description: String, + pub enabled: bool, + pub priority: i32, + pub conditions: ConditionGroup, + pub actions: Vec, + pub metadata: serde_json::Value, +} + +/// Condition group with logical operators +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConditionGroup { + pub operator: LogicalOperator, + pub conditions: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum LogicalOperator { + And, + Or, +} + +/// Individual condition +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Condition { + pub field: String, + pub operator: ComparisonOperator, + pub value: serde_json::Value, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum ComparisonOperator { + Equals, + NotEquals, + GreaterThan, + LessThan, + GreaterThanOrEqual, + LessThanOrEqual, + Contains, + NotContains, + Matches, // Regex + In, + NotIn, +} + +/// Action to execute when rule matches +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "type", content = "params")] +pub enum Action { + #[serde(rename = "webhook")] + Webhook { url: String, method: String, headers: HashMap }, + #[serde(rename = "email")] + Email { to: Vec, subject: String, template: String }, + #[serde(rename = "tag")] + Tag { tags: Vec }, + #[serde(rename = "alert")] + Alert { severity: String, message: String }, + #[serde(rename = "transform")] + Transform { script: String }, + #[serde(rename = "block")] + Block { reason: String }, + #[serde(rename = "custom")] + Custom { handler: String, params: serde_json::Value }, +} + +/// Rule evaluation context +#[derive(Debug, Clone)] +pub struct RuleContext { + pub event: TrackingEvent, + pub product: Option, + pub variables: HashMap, + pub timestamp: DateTime, +} + +#[derive(Debug, Clone)] +pub struct ProductContext { + pub id: String, + pub owner_address: String, + pub category: String, + pub tags: Vec, + pub certifications: Vec, +} + +/// Rule evaluation result +#[derive(Debug, Clone)] +pub struct RuleEvaluationResult { + pub rule_id: String, + pub matched: bool, + pub actions_executed: Vec, + pub execution_time_ms: u64, + pub error: Option, +} + +/// Rule engine for complex event processing +pub struct RuleEngine { + rules: Vec, + action_handlers: HashMap>, +} + +impl RuleEngine { + pub fn new() -> Self { + Self { + rules: Vec::new(), + action_handlers: HashMap::new(), + } + } + + /// Add a rule to the engine + pub fn add_rule(&mut self, rule: Rule) { + self.rules.push(rule); + // Sort by priority (higher priority first) + self.rules.sort_by(|a, b| b.priority.cmp(&a.priority)); + } + + /// Remove a rule by ID + pub fn remove_rule(&mut self, rule_id: &str) -> Option { + if let Some(pos) = self.rules.iter().position(|r| r.id == rule_id) { + Some(self.rules.remove(pos)) + } else { + None + } + } + + /// Register an action handler + pub fn register_handler(&mut self, action_type: String, handler: Arc) { + self.action_handlers.insert(action_type, handler); + } + + /// Evaluate all rules against context + pub async fn evaluate(&self, context: &RuleContext) -> Vec { + let mut results = Vec::new(); + + for rule in &self.rules { + if !rule.enabled { + continue; + } + + let start = std::time::Instant::now(); + let matched = self.evaluate_condition_group(&rule.conditions, context); + + let mut result = RuleEvaluationResult { + rule_id: rule.id.clone(), + matched, + actions_executed: Vec::new(), + execution_time_ms: start.elapsed().as_millis() as u64, + error: None, + }; + + if matched { + debug!("Rule '{}' matched event {}", rule.name, context.event.id); + + for action in &rule.actions { + match self.execute_action(action, context).await { + Ok(action_name) => { + result.actions_executed.push(action_name); + } + Err(e) => { + error!("Failed to execute action: {}", e); + result.error = Some(e.to_string()); + } + } + } + } + + results.push(result); + } + + results + } + + /// Evaluate a condition group + fn evaluate_condition_group(&self, group: &ConditionGroup, context: &RuleContext) -> bool { + let results: Vec = group + .conditions + .iter() + .map(|cond| self.evaluate_condition(cond, context)) + .collect(); + + match group.operator { + LogicalOperator::And => results.iter().all(|&r| r), + LogicalOperator::Or => results.iter().any(|&r| r), + } + } + + /// Evaluate a single condition + fn evaluate_condition(&self, condition: &Condition, context: &RuleContext) -> bool { + let field_value = self.get_field_value(&condition.field, context); + + match condition.operator { + ComparisonOperator::Equals => self.compare_values(&field_value, &condition.value) == 0, + ComparisonOperator::NotEquals => self.compare_values(&field_value, &condition.value) != 0, + ComparisonOperator::GreaterThan => self.compare_values(&field_value, &condition.value) > 0, + ComparisonOperator::LessThan => self.compare_values(&field_value, &condition.value) < 0, + ComparisonOperator::GreaterThanOrEqual => { + self.compare_values(&field_value, &condition.value) >= 0 + } + ComparisonOperator::LessThanOrEqual => { + self.compare_values(&field_value, &condition.value) <= 0 + } + ComparisonOperator::Contains => self.string_contains(&field_value, &condition.value), + ComparisonOperator::NotContains => !self.string_contains(&field_value, &condition.value), + ComparisonOperator::Matches => self.regex_matches(&field_value, &condition.value), + ComparisonOperator::In => self.value_in_array(&field_value, &condition.value), + ComparisonOperator::NotIn => !self.value_in_array(&field_value, &condition.value), + } + } + + /// Get field value from context + fn get_field_value(&self, field: &str, context: &RuleContext) -> serde_json::Value { + let parts: Vec<&str> = field.split('.').collect(); + + match parts.as_slice() { + ["event", rest @ ..] => self.get_nested_value(&serde_json::to_value(&context.event).unwrap(), rest), + ["product", rest @ ..] => { + if let Some(product) = &context.product { + self.get_nested_value(&serde_json::to_value(product).unwrap(), rest) + } else { + serde_json::Value::Null + } + } + ["variable", name] => context.variables.get(*name).cloned().unwrap_or(serde_json::Value::Null), + _ => serde_json::Value::Null, + } + } + + /// Get nested value from JSON + fn get_nested_value(&self, value: &serde_json::Value, path: &[&str]) -> serde_json::Value { + let mut current = value; + for key in path { + current = match current { + serde_json::Value::Object(map) => map.get(*key).unwrap_or(&serde_json::Value::Null), + serde_json::Value::Array(arr) => { + if let Ok(index) = key.parse::() { + arr.get(index).unwrap_or(&serde_json::Value::Null) + } else { + &serde_json::Value::Null + } + } + _ => &serde_json::Value::Null, + }; + } + current.clone() + } + + /// Compare two JSON values + fn compare_values(&self, a: &serde_json::Value, b: &serde_json::Value) -> std::cmp::Ordering { + match (a, b) { + (serde_json::Value::String(a), serde_json::Value::String(b)) => a.cmp(b), + (serde_json::Value::Number(a), serde_json::Value::Number(b)) => { + if let (Some(a), Some(b)) = (a.as_i64(), b.as_i64()) { + a.cmp(&b) + } else if let (Some(a), Some(b)) = (a.as_f64(), b.as_f64()) { + a.partial_cmp(&b).unwrap_or(std::cmp::Ordering::Equal) + } else { + std::cmp::Ordering::Equal + } + } + (serde_json::Value::Bool(a), serde_json::Value::Bool(b)) => a.cmp(b), + _ => std::cmp::Ordering::Equal, + } + } + + /// Check if string contains substring + fn string_contains(&self, value: &serde_json::Value, pattern: &serde_json::Value) -> bool { + if let (Some(s), Some(p)) = (value.as_str(), pattern.as_str()) { + s.to_lowercase().contains(&p.to_lowercase()) + } else { + false + } + } + + /// Check regex match + fn regex_matches(&self, value: &serde_json::Value, pattern: &serde_json::Value) -> bool { + if let (Some(s), Some(p)) = (value.as_str(), pattern.as_str()) { + if let Ok(re) = regex::Regex::new(p) { + re.is_match(s) + } else { + false + } + } else { + false + } + } + + /// Check if value is in array + fn value_in_array(&self, value: &serde_json::Value, array: &serde_json::Value) -> bool { + if let serde_json::Value::Array(arr) = array { + arr.contains(value) + } else { + false + } + } + + /// Execute an action + async fn execute_action(&self, action: &Action, context: &RuleContext) -> Result { + let action_type = match action { + Action::Webhook { .. } => "webhook", + Action::Email { .. } => "email", + Action::Tag { .. } => "tag", + Action::Alert { .. } => "alert", + Action::Transform { .. } => "transform", + Action::Block { .. } => "block", + Action::Custom { handler, .. } => handler, + }; + + if let Some(handler) = self.action_handlers.get(action_type) { + handler.execute(action, context).await?; + Ok(action_type.to_string()) + } else { + warn!("No handler registered for action type: {}", action_type); + Ok(action_type.to_string()) + } + } + + /// Get all rules + pub fn get_rules(&self) -> &[Rule] { + &self.rules + } + + /// Get rule by ID + pub fn get_rule(&self, id: &str) -> Option<&Rule> { + self.rules.iter().find(|r| r.id == id) + } +} + +impl Default for RuleEngine { + fn default() -> Self { + Self::new() + } +} + +/// Action handler trait +#[async_trait::async_trait] +pub trait ActionHandler: Send + Sync { + async fn execute(&self, action: &Action, context: &RuleContext) -> Result<(), AppError>; +} + +/// Default alert handler +pub struct AlertHandler { + // Could include notification service, etc. +} + +impl AlertHandler { + pub fn new() -> Self { + Self {} + } +} + +#[async_trait] +impl ActionHandler for AlertHandler { + async fn execute(&self, action: &Action, context: &RuleContext) -> Result<(), AppError> { + if let Action::Alert { severity, message } = action { + info!( + "ALERT [{}]: {} for event {}", + severity, + message, + context.event.id + ); + // In production, this would send to monitoring/alerting system + } + Ok(()) + } +} + +/// Default webhook handler +pub struct WebhookHandler { + client: reqwest::Client, +} + +impl WebhookHandler { + pub fn new() -> Self { + Self { + client: reqwest::Client::new(), + } + } +} + +#[async_trait] +impl ActionHandler for WebhookHandler { + async fn execute(&self, action: &Action, context: &RuleContext) -> Result<(), AppError> { + if let Action::Webhook { url, method, headers } = action { + let mut request = match method.to_lowercase().as_str() { + "post" => self.client.post(url), + "put" => self.client.put(url), + "patch" => self.client.patch(url), + _ => self.client.post(url), + }; + + for (key, value) in headers { + request = request.header(key, value); + } + + let payload = serde_json::json!({ + "event": context.event, + "timestamp": context.timestamp, + }); + + let response = request.json(&payload).send().await?; + + if !response.status().is_success() { + return Err(AppError::BadRequest(format!( + "Webhook failed with status: {}", + response.status() + ))); + } + + info!("Webhook executed successfully: {}", url); + } + Ok(()) + } +} + +/// Predefined rules for common supply chain scenarios +pub fn get_default_rules() -> Vec { + vec![ + Rule { + id: "quality-check-failure".to_string(), + name: "Quality Check Failure Alert".to_string(), + description: "Alert when quality check fails".to_string(), + enabled: true, + priority: 100, + conditions: ConditionGroup { + operator: LogicalOperator::And, + conditions: vec![ + Condition { + field: "event.event_type".to_string(), + operator: ComparisonOperator::Equals, + value: serde_json::json!("QUALITY_CHECK"), + }, + Condition { + field: "event.metadata.passed".to_string(), + operator: ComparisonOperator::Equals, + value: serde_json::json!(false), + }, + ], + }, + actions: vec![Action::Alert { + severity: "high".to_string(), + message: "Quality check failed - immediate attention required".to_string(), + }], + metadata: serde_json::json!({"category": "quality"}), + }, + Rule { + id: "shipment-delay".to_string(), + name: "Shipment Delay Detection".to_string(), + description: "Detect shipments delayed beyond expected time".to_string(), + enabled: true, + priority: 80, + conditions: ConditionGroup { + operator: LogicalOperator::And, + conditions: vec![ + Condition { + field: "event.event_type".to_string(), + operator: ComparisonOperator::Equals, + value: serde_json::json!("SHIP"), + }, + Condition { + field: "event.metadata.delayed".to_string(), + operator: ComparisonOperator::Equals, + value: serde_json::json!(true), + }, + ], + }, + actions: vec![ + Action::Alert { + severity: "medium".to_string(), + message: "Shipment delayed beyond expected time".to_string(), + }, + Action::Tag { + tags: vec!["delayed".to_string(), "attention-required".to_string()], + }, + ], + metadata: serde_json::json!({"category": "logistics"}), + }, + Rule { + id: "temperature-excursion".to_string(), + name: "Temperature Excursion Alert".to_string(), + description: "Alert when temperature exceeds safe range".to_string(), + enabled: true, + priority: 95, + conditions: ConditionGroup { + operator: LogicalOperator::Or, + conditions: vec![ + Condition { + field: "event.metadata.temperature".to_string(), + operator: ComparisonOperator::GreaterThan, + value: serde_json::json!(25), + }, + Condition { + field: "event.metadata.temperature".to_string(), + operator: ComparisonOperator::LessThan, + value: serde_json::json!(2), + }, + ], + }, + actions: vec![Action::Alert { + severity: "critical".to_string(), + message: "Temperature excursion detected - product safety at risk".to_string(), + }], + metadata: serde_json::json!({"category": "safety"}), + }, + ] +} From 775728d276a6a7d17ebbdfdac68b3ba8d58d6d2f Mon Sep 17 00:00:00 2001 From: millicentogalanya Date: Fri, 26 Jun 2026 03:33:20 +0100 Subject: [PATCH 3/6] feat: add saga-pattern state management for distributed transactions --- backend/src/services/saga_manager.rs | 588 +++++++++++++++++++++++++++ 1 file changed, 588 insertions(+) create mode 100644 backend/src/services/saga_manager.rs diff --git a/backend/src/services/saga_manager.rs b/backend/src/services/saga_manager.rs new file mode 100644 index 00000000..2aa26676 --- /dev/null +++ b/backend/src/services/saga_manager.rs @@ -0,0 +1,588 @@ +use crate::error::AppError; +use chrono::{DateTime, Utc}; +use redis::AsyncCommands; +use serde::{Deserialize, Serialize}; +use sqlx::PgPool; +use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::{Mutex, RwLock}; +use tracing::{debug, error, info, warn}; +use uuid::Uuid; + +/// Saga state machine states +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum SagaState { + Pending, + InProgress, + Compensating, + Completed, + Failed, + Aborted, +} + +/// Saga step definition +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SagaStep { + pub id: String, + pub name: String, + pub execute_action: String, + pub compensate_action: String, + pub timeout_ms: u64, + pub retry_policy: RetryPolicy, + pub metadata: serde_json::Value, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RetryPolicy { + pub max_attempts: u32, + pub backoff_ms: u64, + pub exponential_backoff: bool, +} + +impl Default for RetryPolicy { + fn default() -> Self { + Self { + max_attempts: 3, + backoff_ms: 1000, + exponential_backoff: true, + } + } +} + +/// Saga instance +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SagaInstance { + pub id: Uuid, + pub saga_type: String, + pub state: SagaState, + pub current_step: Option, + pub completed_steps: Vec, + pub failed_step: Option, + pub context: serde_json::Value, + pub error_message: Option, + pub started_at: DateTime, + pub updated_at: DateTime, + pub completed_at: Option>, + pub metadata: serde_json::Value, +} + +/// Saga execution result +#[derive(Debug, Clone)] +pub struct SagaResult { + pub saga_id: Uuid, + pub state: SagaState, + pub completed_steps: Vec, + pub error: Option, + pub execution_time_ms: u64, +} + +/// Saga step execution result +#[derive(Debug, Clone)] +pub struct StepResult { + pub step_id: String, + pub success: bool, + pub error: Option, + pub execution_time_ms: u64, + pub retry_count: u32, +} + +/// Saga action handler trait +#[async_trait::async_trait] +pub trait SagaAction: Send + Sync { + async fn execute(&self, context: &serde_json::Value) -> Result; + async fn compensate(&self, context: &serde_json::Value) -> Result<(), AppError>; +} + +/// Saga manager for orchestrating distributed transactions +pub struct SagaManager { + pool: PgPool, + redis_client: redis::Client, + saga_definitions: HashMap, + action_handlers: HashMap>, + running_sagas: Arc>>>>, +} + +#[derive(Debug, Clone)] +pub struct SagaDefinition { + pub saga_type: String, + pub steps: Vec, + pub timeout_ms: u64, + pub metadata: serde_json::Value, +} + +impl SagaManager { + pub fn new(pool: PgPool, redis_client: redis::Client) -> Self { + Self { + pool, + redis_client, + saga_definitions: HashMap::new(), + action_handlers: HashMap::new(), + running_sagas: Arc::new(RwLock::new(HashMap::new())), + } + } + + /// Register a saga definition + pub fn register_saga(&mut self, definition: SagaDefinition) { + self.saga_definitions + .insert(definition.saga_type.clone(), definition); + info!("Registered saga: {}", definition.saga_type); + } + + /// Register an action handler + pub fn register_action(&mut self, action_name: String, handler: Arc) { + self.action_handlers.insert(action_name, handler); + } + + /// Start a new saga instance + pub async fn start_saga( + &self, + saga_type: &str, + context: serde_json::Value, + ) -> Result { + let definition = self + .saga_definitions + .get(saga_type) + .ok_or_else(|| AppError::NotFound(format!("Saga type '{}' not found", saga_type)))?; + + let saga_id = Uuid::new_v4(); + let instance = SagaInstance { + id: saga_id, + saga_type: saga_type.to_string(), + state: SagaState::Pending, + current_step: None, + completed_steps: Vec::new(), + failed_step: None, + context, + error_message: None, + started_at: Utc::now(), + updated_at: Utc::now(), + completed_at: None, + metadata: serde_json::json!({}), + }; + + // Store in database + self.persist_saga(&instance).await?; + + // Store in running sagas + let instance_arc = Arc::new(Mutex::new(instance)); + let mut running = self.running_sagas.write().await; + running.insert(saga_id, instance_arc.clone()); + + // Start execution + let manager = self.clone_for_execution(); + tokio::spawn(async move { + let result = manager.execute_saga(instance_arc).await; + info!("Saga {} completed with state: {:?}", saga_id, result.state); + }); + + Ok(SagaResult { + saga_id, + state: SagaState::InProgress, + completed_steps: vec![], + error: None, + execution_time_ms: 0, + }) + } + + /// Execute a saga instance + async fn execute_saga(&self, instance: Arc>) -> SagaResult { + let start = std::time::Instant::now(); + let saga_id; + + { + let mut saga = instance.lock().await; + saga_id = saga.id; + saga.state = SagaState::InProgress; + saga.updated_at = Utc::now(); + let _ = self.persist_saga(&saga).await; + } + + let saga_type; + { + let saga = instance.lock().await; + saga_type = saga.saga_type.clone(); + } + + let definition = match self.saga_definitions.get(&saga_type) { + Some(def) => def.clone(), + None => { + let mut saga = instance.lock().await; + saga.state = SagaState::Failed; + saga.error_message = Some("Saga definition not found".to_string()); + saga.updated_at = Utc::now(); + let _ = self.persist_saga(&saga).await; + return SagaResult { + saga_id, + state: SagaState::Failed, + completed_steps: vec![], + error: Some("Saga definition not found".to_string()), + execution_time_ms: start.elapsed().as_millis() as u64, + }; + } + }; + + // Execute each step + for step in &definition.steps { + let step_result = self.execute_step(instance.clone(), step).await; + + if !step_result.success { + // Step failed, start compensation + warn!("Step {} failed, starting compensation", step.id); + self.compensate_saga(instance.clone()).await; + + let mut saga = instance.lock().await; + saga.state = SagaState::Failed; + saga.failed_step = Some(step.id.clone()); + saga.error_message = step_result.error; + saga.updated_at = Utc::now(); + let _ = self.persist_saga(&saga).await; + + return SagaResult { + saga_id, + state: SagaState::Failed, + completed_steps: { + let saga = instance.lock().await; + saga.completed_steps.clone() + }, + error: step_result.error, + execution_time_ms: start.elapsed().as_millis() as u64, + }; + } + + // Mark step as completed + { + let mut saga = instance.lock().await; + saga.completed_steps.push(step.id.clone()); + saga.current_step = Some(step.id.clone()); + saga.updated_at = Utc::now(); + let _ = self.persist_saga(&saga).await; + } + } + + // All steps completed successfully + { + let mut saga = instance.lock().await; + saga.state = SagaState::Completed; + saga.current_step = None; + saga.completed_at = Some(Utc::now()); + saga.updated_at = Utc::now(); + let _ = self.persist_saga(&saga).await; + } + + // Remove from running sagas + let mut running = self.running_sagas.write().await; + running.remove(&saga_id); + + SagaResult { + saga_id, + state: SagaState::Completed, + completed_steps: { + let saga = instance.lock().await; + saga.completed_steps.clone() + }, + error: None, + execution_time_ms: start.elapsed().as_millis() as u64, + } + } + + /// Execute a single saga step + async fn execute_step( + &self, + instance: Arc>, + step: &SagaStep, + ) -> StepResult { + let start = std::time::Instant::now(); + let mut retry_count = 0; + + loop { + let context; + { + let saga = instance.lock().await; + context = saga.context.clone(); + } + + if let Some(handler) = self.action_handlers.get(&step.execute_action) { + match handler.execute(&context).await { + Ok(result) => { + // Update context with result + let mut saga = instance.lock().await; + if let Some(obj) = saga.context.as_object_mut() { + obj.insert("step_result".to_string(), result); + } + saga.updated_at = Utc::now(); + let _ = self.persist_saga(&saga).await; + + return StepResult { + step_id: step.id.clone(), + success: true, + error: None, + execution_time_ms: start.elapsed().as_millis() as u64, + retry_count, + }; + } + Err(e) => { + retry_count += 1; + if retry_count >= step.retry_policy.max_attempts { + error!( + "Step {} failed after {} attempts: {}", + step.id, retry_count, e + ); + return StepResult { + step_id: step.id.clone(), + success: false, + error: Some(e.to_string()), + execution_time_ms: start.elapsed().as_millis() as u64, + retry_count, + }; + } + + // Backoff before retry + let backoff = if step.retry_policy.exponential_backoff { + step.retry_policy.backoff_ms * 2_u64.pow(retry_count - 1) + } else { + step.retry_policy.backoff_ms + }; + tokio::time::sleep(tokio::time::Duration::from_millis(backoff)).await; + } + } + } else { + return StepResult { + step_id: step.id.clone(), + success: false, + error: Some(format!("Handler '{}' not found", step.execute_action)), + execution_time_ms: start.elapsed().as_millis() as u64, + retry_count, + }; + } + } + } + + /// Compensate a failed saga (execute compensation actions in reverse order) + async fn compensate_saga(&self, instance: Arc>) { + let saga_type; + let completed_steps; + + { + let mut saga = instance.lock().await; + saga.state = SagaState::Compensating; + saga.updated_at = Utc::now(); + let _ = self.persist_saga(&saga).await; + saga_type = saga.saga_type.clone(); + completed_steps = saga.completed_steps.clone(); + } + + let definition = match self.saga_definitions.get(&saga_type) { + Some(def) => def.clone(), + None => { + error!("Cannot compensate: saga definition not found"); + return; + } + }; + + // Execute compensation in reverse order + for step_id in completed_steps.iter().rev() { + if let Some(step) = definition.steps.iter().find(|s| &s.id == step_id) { + info!("Compensating step: {}", step.id); + + let context; + { + let saga = instance.lock().await; + context = saga.context.clone(); + } + + if let Some(handler) = self.action_handlers.get(&step.compensate_action) { + match handler.compensate(&context).await { + Ok(_) => { + debug!("Compensation succeeded for step: {}", step.id); + } + Err(e) => { + error!("Compensation failed for step {}: {}", step.id, e); + // Continue with other compensations + } + } + } + } + } + + let mut saga = instance.lock().await; + saga.state = SagaState::Aborted; + saga.updated_at = Utc::now(); + let _ = self.persist_saga(&saga).await; + } + + /// Get saga status + pub async fn get_saga_status(&self, saga_id: Uuid) -> Result, AppError> { + // Check running sagas first + let running = self.running_sagas.read().await; + if let Some(instance) = running.get(&saga_id) { + let saga = instance.lock().await; + return Ok(Some(saga.clone())); + } + + // Check database + sqlx::query_as::( + "SELECT id, saga_type, state, current_step, completed_steps, failed_step, context, error_message, started_at, updated_at, completed_at, metadata FROM saga_instances WHERE id = $1" + ) + .bind(saga_id) + .fetch_optional(&self.pool) + .await + .map_err(|e| AppError::DatabaseError(e.to_string())) + } + + /// Persist saga to database + async fn persist_saga(&self, instance: &SagaInstance) -> Result<(), AppError> { + sqlx::query( + r#" + INSERT INTO saga_instances ( + id, saga_type, state, current_step, completed_steps, failed_step, + context, error_message, started_at, updated_at, completed_at, metadata + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + ON CONFLICT (id) DO UPDATE SET + state = EXCLUDED.state, + current_step = EXCLUDED.current_step, + completed_steps = EXCLUDED.completed_steps, + failed_step = EXCLUDED.failed_step, + context = EXCLUDED.context, + error_message = EXCLUDED.error_message, + updated_at = EXCLUDED.updated_at, + completed_at = EXCLUDED.completed_at, + metadata = EXCLUDED.metadata + "#, + ) + .bind(instance.id) + .bind(&instance.saga_type) + .bind(instance.state) + .bind(&instance.current_step) + .bind(&instance.completed_steps) + .bind(&instance.failed_step) + .bind(&instance.context) + .bind(&instance.error_message) + .bind(instance.started_at) + .bind(instance.updated_at) + .bind(instance.completed_at) + .bind(&instance.metadata) + .execute(&self.pool) + .await?; + + // Cache in Redis + if let Ok(mut conn) = self.redis_client.get_multiplexed_tokio_connection().await { + let serialized = serde_json::to_string(instance)?; + let _: Result<(), _> = conn + .set_ex(format!("saga:{}", instance.id), serialized, 3600) + .await; + } + + Ok(()) + } + + /// Clone for execution (shallow clone) + fn clone_for_execution(&self) -> Self { + Self { + pool: self.pool.clone(), + redis_client: self.redis_client.clone(), + saga_definitions: self.saga_definitions.clone(), + action_handlers: self.action_handlers.clone(), + running_sagas: Arc::clone(&self.running_sagas), + } + } + + /// Recover in-progress sagas on startup + pub async fn recover_sagas(&self) -> Result { + let sagas = sqlx::query_as::( + "SELECT id, saga_type, state, current_step, completed_steps, failed_step, context, error_message, started_at, updated_at, completed_at, metadata FROM saga_instances WHERE state IN ('InProgress', 'Compensating')" + ) + .fetch_all(&self.pool) + .await + .map_err(|e| AppError::DatabaseError(e.to_string()))?; + + let mut recovered = 0; + + for saga in sagas { + info!("Recovering saga: {}", saga.id); + + let instance_arc = Arc::new(Mutex::new(saga.clone())); + let mut running = self.running_sagas.write().await; + running.insert(saga.id, instance_arc.clone()); + recovered += 1; + + let manager = self.clone_for_execution(); + tokio::spawn(async move { + // For simplicity, we'll restart compensation for in-progress sagas + manager.compensate_saga(instance_arc).await; + }); + } + + info!("Recovered {} in-progress sagas", recovered); + Ok(recovered) + } +} + +/// Example saga for product registration workflow +pub fn get_product_registration_saga() -> SagaDefinition { + SagaDefinition { + saga_type: "product_registration".to_string(), + steps: vec![ + SagaStep { + id: "validate_product".to_string(), + name: "Validate Product Data".to_string(), + execute_action: "validate_product".to_string(), + compensate_action: "noop".to_string(), + timeout_ms: 5000, + retry_policy: RetryPolicy::default(), + metadata: serde_json::json!({}), + }, + SagaStep { + id: "register_on_blockchain".to_string(), + name: "Register on Blockchain".to_string(), + execute_action: "register_blockchain".to_string(), + compensate_action: "revert_blockchain".to_string(), + timeout_ms: 30000, + retry_policy: RetryPolicy { + max_attempts: 5, + backoff_ms: 2000, + exponential_backoff: true, + }, + metadata: serde_json::json!({}), + }, + SagaStep { + id: "update_database".to_string(), + name: "Update Database".to_string(), + execute_action: "update_database".to_string(), + compensate_action: "rollback_database".to_string(), + timeout_ms: 5000, + retry_policy: RetryPolicy::default(), + metadata: serde_json::json!({}), + }, + SagaStep { + id: "send_notifications".to_string(), + name: "Send Notifications".to_string(), + execute_action: "send_notifications".to_string(), + compensate_action: "cancel_notifications".to_string(), + timeout_ms: 10000, + retry_policy: RetryPolicy { + max_attempts: 3, + backoff_ms: 1000, + exponential_backoff: false, + }, + metadata: serde_json::json!({}), + }, + ], + timeout_ms: 60000, + metadata: serde_json::json!({"category": "product"}), + } +} + +/// No-op action handler for compensation steps that don't need compensation +pub struct NoopAction; + +#[async_trait] +impl SagaAction for NoopAction { + async fn execute(&self, _context: &serde_json::Value) -> Result { + Ok(serde_json::json!({})) + } + + async fn compensate(&self, _context: &serde_json::Value) -> Result<(), AppError> { + Ok(()) + } +} From fb432cb8c87945c979bb4db0e442aad0ae507a21 Mon Sep 17 00:00:00 2001 From: millicentogalanya Date: Fri, 26 Jun 2026 03:33:36 +0100 Subject: [PATCH 4/6] feat: add Redis-based workers for horizontal scaling --- backend/src/services/redis_workers.rs | 424 ++++++++++++++++++++++++++ 1 file changed, 424 insertions(+) create mode 100644 backend/src/services/redis_workers.rs diff --git a/backend/src/services/redis_workers.rs b/backend/src/services/redis_workers.rs new file mode 100644 index 00000000..9cd74117 --- /dev/null +++ b/backend/src/services/redis_workers.rs @@ -0,0 +1,424 @@ +use crate::error::AppError; +use redis::AsyncCommands; +use serde::{Deserialize, Serialize}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::Semaphore; +use tracing::{debug, error, info, warn}; +use uuid::Uuid; + +/// Worker task definition +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WorkerTask { + pub id: Uuid, + pub task_type: String, + pub payload: serde_json::Value, + pub priority: i32, + pub created_at: chrono::DateTime, + pub retry_count: u32, + pub max_retries: u32, + pub timeout_ms: u64, +} + +/// Worker configuration +#[derive(Debug, Clone)] +pub struct WorkerConfig { + pub queue_name: String, + pub max_concurrent_tasks: usize, + pub poll_interval_ms: u64, + pub heartbeat_interval_ms: u64, + pub worker_id: String, +} + +impl Default for WorkerConfig { + fn default() -> Self { + Self { + queue_name: "event_processing".to_string(), + max_concurrent_tasks: 10, + poll_interval_ms: 100, + heartbeat_interval_ms: 5000, + worker_id: Uuid::new_v4().to_string(), + } + } +} + +/// Worker task handler trait +#[async_trait::async_trait] +pub trait TaskHandler: Send + Sync { + async fn handle(&self, task: WorkerTask) -> Result; + fn task_type(&self) -> &str; +} + +/// Redis-based worker pool for horizontal scaling +pub struct RedisWorkerPool { + config: WorkerConfig, + redis_client: redis::Client, + handlers: Vec>, + semaphore: Arc, + running: Arc, +} + +impl RedisWorkerPool { + pub fn new(config: WorkerConfig, redis_client: redis::Client) -> Self { + let semaphore = Arc::new(Semaphore::new(config.max_concurrent_tasks)); + + Self { + config, + redis_client, + handlers: Vec::new(), + semaphore, + running: Arc::new(std::sync::atomic::AtomicBool::new(false)), + } + } + + /// Register a task handler + pub fn register_handler(&mut self, handler: Arc) { + self.handlers.push(handler); + info!("Registered handler for task type: {}", handler.task_type()); + } + + /// Start the worker pool + pub async fn start(&self) -> Result<(), AppError> { + self.running.store(true, std::sync::atomic::Ordering::SeqCst); + info!("Starting worker pool: {}", self.config.worker_id); + + // Start heartbeat + self.start_heartbeat().await; + + // Start task processing loop + self.start_processing_loop().await; + + Ok(()) + } + + /// Stop the worker pool + pub async fn stop(&self) { + self.running.store(false, std::sync::atomic::Ordering::SeqCst); + info!("Stopping worker pool: {}", self.config.worker_id); + } + + /// Enqueue a task + pub async fn enqueue_task(&self, task: WorkerTask) -> Result<(), AppError> { + let serialized = serde_json::to_string(&task)?; + + if let Ok(mut conn) = self.redis_client.get_multiplexed_tokio_connection().await { + // Use priority queue (sorted set) + let score = -(task.priority as f64); // Negative for higher priority first + let member = format!("{}:{}", task.id, task.task_type); + + let _: Result<(), _> = conn + .zadd(self.config.queue_name.clone(), score, &serialized) + .await; + + debug!("Enqueued task: {} (priority: {})", task.id, task.priority); + } + + Ok(()) + } + + /// Start the heartbeat to signal worker is alive + async fn start_heartbeat(&self) { + let config = self.config.clone(); + let redis_client = self.redis_client.clone(); + let running = Arc::clone(&self.running); + + tokio::spawn(async move { + while running.load(std::sync::atomic::Ordering::SeqCst) { + if let Ok(mut conn) = redis_client.get_multiplexed_tokio_connection().await { + let key = format!("worker:heartbeat:{}", config.worker_id); + let _: Result<(), _> = conn.set_ex(key.clone(), "alive", config.heartbeat_interval_ms / 1000 + 2).await; + + // Add to worker set + let _: Result<(), _> = conn.sadd("workers:active", &config.worker_id).await; + } + + tokio::time::sleep(Duration::from_millis(config.heartbeat_interval_ms)).await; + } + + // Remove from active workers on shutdown + if let Ok(mut conn) = redis_client.get_multiplexed_tokio_connection().await { + let _: Result<(), _> = conn.srem("workers:active", &config.worker_id).await; + } + }); + } + + /// Start the main processing loop + async fn start_processing_loop(&self) { + let config = self.config.clone(); + let redis_client = self.redis_client.clone(); + let handlers = self.handlers.clone(); + let semaphore = Arc::clone(&self.semaphore); + let running = Arc::clone(&self.running); + + tokio::spawn(async move { + while running.load(std::sync::atomic::Ordering::SeqCst) { + // Try to acquire a permit (concurrency limit) + let permit = semaphore.clone().acquire_owned().await; + + if permit.is_err() { + continue; + } + + let permit = permit.unwrap(); + + // Poll for task + if let Ok(mut conn) = redis_client.get_multiplexed_tokio_connection().await { + // Get highest priority task + let result: Option> = conn + .zpopmax(config.queue_name.clone()) + .await + .ok(); + + if let Some(items) = result { + if !items.is_empty() { + let task_data = &items[0]; + + if let Ok(task) = serde_json::from_str::(task_data) { + let task_handlers = handlers.clone(); + let task_redis = redis_client.clone(); + + tokio::spawn(async move { + let _ = Self::process_task(task, task_handlers, task_redis).await; + drop(permit); // Release permit + }); + + continue; + } + } + } + } + + drop(permit); // Release permit if no task + tokio::time::sleep(Duration::from_millis(config.poll_interval_ms)).await; + } + }); + } + + /// Process a single task + async fn process_task( + task: WorkerTask, + handlers: Vec>, + redis_client: redis::Client, + ) -> Result<(), AppError> { + let start = std::time::Instant::now(); + + debug!("Processing task: {} (type: {})", task.id, task.task_type); + + // Find appropriate handler + let handler = handlers + .iter() + .find(|h| h.task_type() == task.task_type); + + if let Some(handler) = handler { + let handler = handler.clone(); + + // Execute with timeout + let result = tokio::time::timeout( + Duration::from_millis(task.timeout_ms), + handler.handle(task.clone()), + ) + .await; + + match result { + Ok(Ok(output)) => { + info!( + "Task {} completed in {}ms", + task.id, + start.elapsed().as_millis() + ); + + // Store result + Self::store_task_result(&redis_client, &task.id, Some(output), None).await; + } + Ok(Err(e)) => { + error!("Task {} failed: {}", task.id, e); + + // Retry if under max retries + if task.retry_count < task.max_retries { + let mut retry_task = task.clone(); + retry_task.retry_count += 1; + + warn!("Retrying task {} (attempt {}/{})", task.id, retry_task.retry_count, task.max_retries); + + if let Ok(mut conn) = redis_client.get_multiplexed_tokio_connection().await { + let serialized = serde_json::to_string(&retry_task)?; + let score = -(retry_task.priority as f64); + let _: Result<(), _> = conn + .zadd("event_processing", score, &serialized) + .await; + } + } else { + error!("Task {} exceeded max retries", task.id); + Self::store_task_result(&redis_client, &task.id, None, Some(e.to_string())).await; + } + } + Err(_) => { + error!("Task {} timed out after {}ms", task.id, task.timeout_ms); + Self::store_task_result(&redis_client, &task.id, None, Some("Timeout".to_string())).await; + } + } + } else { + warn!("No handler found for task type: {}", task.task_type); + Self::store_task_result(&redis_client, &task.id, None, Some("No handler".to_string())).await; + } + + Ok(()) + } + + /// Store task result in Redis + async fn store_task_result( + redis_client: &redis::Client, + task_id: &Uuid, + result: Option, + error: Option, + ) { + if let Ok(mut conn) = redis_client.get_multiplexed_tokio_connection().await { + let key = format!("task:result:{}", task_id); + let data = serde_json::json!({ + "result": result, + "error": error, + "completed_at": chrono::Utc::now(), + }); + + let _: Result<(), _> = conn.set_ex(key, serde_json::to_string(&data).unwrap(), 3600).await; + } + } + + /// Get task result + pub async fn get_task_result(&self, task_id: &Uuid) -> Option { + if let Ok(mut conn) = self.redis_client.get_multiplexed_tokio_connection().await { + let key = format!("task:result:{}", task_id); + if let Ok(data) = conn.get::<_, String>(key).await { + if let Ok(result) = serde_json::from_str::(&data) { + return Some(TaskResult { + result: result.get("result").cloned(), + error: result.get("error").and_then(|v| v.as_str()).map(String::from), + completed_at: result.get("completed_at") + .and_then(|v| v.as_str()) + .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()) + .map(|dt| dt.with_timezone(&chrono::Utc)), + }); + } + } + } + None + } + + /// Get active workers count + pub async fn get_active_workers_count(&self) -> usize { + if let Ok(mut conn) = self.redis_client.get_multiplexed_tokio_connection().await { + if let Ok(count) = conn.scard("workers:active").await { + return count; + } + } + 0 + } + + /// Get queue length + pub async fn get_queue_length(&self) -> usize { + if let Ok(mut conn) = self.redis_client.get_multiplexed_tokio_connection().await { + if let Ok(count) = conn.zcard(self.config.queue_name.clone()).await { + return count; + } + } + 0 + } +} + +#[derive(Debug, Clone)] +pub struct TaskResult { + pub result: Option, + pub error: Option, + pub completed_at: Option>, +} + +/// Example handler for event processing tasks +pub struct EventProcessingHandler { + // Could include database pool, services, etc. +} + +impl EventProcessingHandler { + pub fn new() -> Self { + Self {} + } +} + +#[async_trait] +impl TaskHandler for EventProcessingHandler { + async fn handle(&self, task: WorkerTask) -> Result { + // Process event + debug!("Processing event task: {}", task.id); + + // Simulate processing + tokio::time::sleep(Duration::from_millis(50)).await; + + Ok(serde_json::json!({ + "status": "processed", + "task_id": task.id, + })) + } + + fn task_type(&self) -> &str { + "event_processing" + } +} + +/// Example handler for rule evaluation tasks +pub struct RuleEvaluationHandler { + // Could include rule engine reference +} + +impl RuleEvaluationHandler { + pub fn new() -> Self { + Self {} + } +} + +#[async_trait] +impl TaskHandler for RuleEvaluationHandler { + async fn handle(&self, task: WorkerTask) -> Result { + debug!("Evaluating rules for task: {}", task.id); + + // Simulate rule evaluation + tokio::time::sleep(Duration::from_millis(30)).await; + + Ok(serde_json::json!({ + "status": "evaluated", + "matched_rules": vec!["quality-check-failure"], + })) + } + + fn task_type(&self) -> &str { + "rule_evaluation" + } +} + +/// Example handler for notification tasks +pub struct NotificationHandler { + // Could include email/SMS service +} + +impl NotificationHandler { + pub fn new() -> Self { + Self {} + } +} + +#[async_trait] +impl TaskHandler for NotificationHandler { + async fn handle(&self, task: WorkerTask) -> Result { + debug!("Sending notification for task: {}", task.id); + + // Simulate notification sending + tokio::time::sleep(Duration::from_millis(100)).await; + + Ok(serde_json::json!({ + "status": "sent", + "recipients": task.payload.get("recipients"), + })) + } + + fn task_type(&self) -> &str { + "notification" + } +} From 7333bd3a6981302fdb3eb163a338f3b6009c8d15 Mon Sep 17 00:00:00 2001 From: millicentogalanya Date: Fri, 26 Jun 2026 03:33:49 +0100 Subject: [PATCH 5/6] chore: export new streaming and event processing services --- backend/src/services.rs | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/backend/src/services.rs b/backend/src/services.rs index 090c6c9d..067dafe2 100644 --- a/backend/src/services.rs +++ b/backend/src/services.rs @@ -52,3 +52,15 @@ pub use regulatory_service::RegulatoryService; pub mod predictive_routing_service; pub use predictive_routing_service::PredictiveRoutingService; + +pub mod mercury_indexer; +pub use mercury_indexer::{MercuryIndexer, MercuryConfig, EventProcessor, TrackingEventProcessor}; + +pub mod rule_engine; +pub use rule_engine::{RuleEngine, Rule, RuleContext, ActionHandler, AlertHandler, WebhookHandler, get_default_rules}; + +pub mod saga_manager; +pub use saga_manager::{SagaManager, SagaInstance, SagaState, SagaAction, get_product_registration_saga, NoopAction}; + +pub mod redis_workers; +pub use redis_workers::{RedisWorkerPool, WorkerTask, WorkerConfig, TaskHandler, EventProcessingHandler, RuleEvaluationHandler, NotificationHandler}; From 3c9c24eeac39fe223b3dea1a383ee39676945198 Mon Sep 17 00:00:00 2001 From: millicentogalanya Date: Fri, 26 Jun 2026 03:34:36 +0100 Subject: [PATCH 6/6] chore: export new streaming and event processing services --- backend/Cargo.toml | 2 + .../20260626000000_add_saga_management.sql | 50 ++++++++++++++ backend/src/main.rs | 67 ++++++++++++++++++- 3 files changed, 118 insertions(+), 1 deletion(-) create mode 100644 backend/migrations/20260626000000_add_saga_management.sql diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 10e96dab..883b428b 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -34,6 +34,8 @@ soroban-sdk = "21.0" # OpenAPI documentation utoipa = { version = "4.0", features = ["axum_extras"] } utoipa-swagger-ui = { version = "6.0", features = ["axum"] } +# Streaming and event processing +reqwest = { version = "0.11", features = ["json"] } [dev-dependencies] tower-test = "0.4" diff --git a/backend/migrations/20260626000000_add_saga_management.sql b/backend/migrations/20260626000000_add_saga_management.sql new file mode 100644 index 00000000..d0e6efec --- /dev/null +++ b/backend/migrations/20260626000000_add_saga_management.sql @@ -0,0 +1,50 @@ +-- Add saga management tables for distributed transaction orchestration + +-- Create enum for saga states +CREATE TYPE saga_state AS ENUM ('Pending', 'InProgress', 'Compensating', 'Completed', 'Failed', 'Aborted'); + +-- Saga instances table +CREATE TABLE saga_instances ( + id UUID PRIMARY KEY DEFAULT uuid_generate_v4(), + saga_type TEXT NOT NULL, + state saga_state NOT NULL DEFAULT 'Pending', + current_step TEXT, + completed_steps TEXT[] DEFAULT '{}', + failed_step TEXT, + context JSONB NOT NULL DEFAULT '{}', + error_message TEXT, + started_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(), + updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(), + completed_at TIMESTAMP WITH TIME ZONE, + metadata JSONB DEFAULT '{}' +); + +-- Create indexes for saga queries +CREATE INDEX idx_saga_instances_state ON saga_instances(state); +CREATE INDEX idx_saga_instances_type ON saga_instances(saga_type); +CREATE INDEX idx_saga_instances_started_at ON saga_instances(started_at); +CREATE INDEX idx_saga_instances_updated_at ON saga_instances(updated_at); + +-- Create trigger for updated_at +CREATE OR REPLACE FUNCTION update_saga_updated_at() +RETURNS TRIGGER AS $$ +BEGIN + NEW.updated_at = NOW(); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER trigger_update_saga_updated_at + BEFORE UPDATE ON saga_instances + FOR EACH ROW + EXECUTE FUNCTION update_saga_updated_at(); + +-- Add comments +COMMENT ON TABLE saga_instances IS 'Stores saga instances for distributed transaction orchestration'; +COMMENT ON COLUMN saga_instances.saga_type IS 'Type of saga (e.g., product_registration)'; +COMMENT ON COLUMN saga_instances.state IS 'Current state of the saga'; +COMMENT ON COLUMN saga_instances.current_step IS 'ID of the currently executing step'; +COMMENT ON COLUMN saga_instances.completed_steps IS 'List of completed step IDs'; +COMMENT ON COLUMN saga_instances.failed_step IS 'ID of the step that failed (if any)'; +COMMENT ON COLUMN saga_instances.context IS 'Execution context and data'; +COMMENT ON COLUMN saga_instances.error_message IS 'Error message if saga failed'; diff --git a/backend/src/main.rs b/backend/src/main.rs index 6460dd1c..41f7d723 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -35,7 +35,10 @@ use services::{ AnalyticsService, ApiKeyService, AuditService, BatchService, CarbonService, CollaborationService, EventService, FinancialService, IoTService, PredictiveRoutingService, ProductService, QualityService, RecallService, RegulatoryService, SupplierService, - SyncService, UserService, + SyncService, UserService, MercuryIndexer, MercuryConfig, RuleEngine, SagaManager, + RedisWorkerPool, WorkerConfig, TrackingEventProcessor, get_default_rules, + get_product_registration_saga, NoopAction, EventProcessingHandler, RuleEvaluationHandler, + NotificationHandler, AlertHandler, WebhookHandler, }; use utils::CronService; @@ -62,6 +65,10 @@ pub struct AppState { pub redis_client: redis::Client, pub config: Config, pub monitoring_system: MonitoringSystem, + pub mercury_indexer: Arc, + pub rule_engine: Arc, + pub saga_manager: Arc, + pub worker_pool: Arc, } impl AppState { @@ -104,6 +111,36 @@ impl AppState { let predictive_routing_service = Arc::new(PredictiveRoutingService::new(db.pool().clone())); + // Initialize Mercury streaming indexer + let mercury_config = MercuryConfig::default(); + let (mercury_indexer, _event_rx) = + MercuryIndexer::new(mercury_config, db.pool().clone(), redis_client.clone()); + mercury_indexer.add_processor(Arc::new(TrackingEventProcessor::new(db.pool().clone()))); + let mercury_indexer = Arc::new(mercury_indexer); + + // Initialize rule engine with default rules + let mut rule_engine = RuleEngine::new(); + for rule in get_default_rules() { + rule_engine.add_rule(rule); + } + rule_engine.register_handler("alert".to_string(), Arc::new(AlertHandler::new())); + rule_engine.register_handler("webhook".to_string(), Arc::new(WebhookHandler::new())); + let rule_engine = Arc::new(rule_engine); + + // Initialize saga manager + let mut saga_manager = SagaManager::new(db.pool().clone(), redis_client.clone()); + saga_manager.register_saga(get_product_registration_saga()); + saga_manager.register_action("noop".to_string(), Arc::new(NoopAction)); + let saga_manager = Arc::new(saga_manager); + + // Initialize Redis worker pool + let worker_config = WorkerConfig::default(); + let mut worker_pool = RedisWorkerPool::new(worker_config, redis_client.clone()); + worker_pool.register_handler(Arc::new(EventProcessingHandler::new())); + worker_pool.register_handler(Arc::new(RuleEvaluationHandler::new())); + worker_pool.register_handler(Arc::new(NotificationHandler::new())); + let worker_pool = Arc::new(worker_pool); + // Initialize comprehensive monitoring system let monitoring_system = MonitoringSystem::new(); @@ -129,6 +166,10 @@ impl AppState { redis_client, config, monitoring_system, + mercury_indexer, + rule_engine, + saga_manager, + worker_pool, }) } } @@ -156,6 +197,30 @@ async fn main() -> Result<(), Box> { CronService::new(app_state.db.pool().clone(), app_state.redis_client.clone()); cron_service.start_scheduler().await; + // Start Mercury streaming indexer + let mercury_indexer = app_state.mercury_indexer.clone(); + tokio::spawn(async move { + if let Err(e) = mercury_indexer.start().await { + tracing::error!("Mercury indexer failed: {}", e); + } + }); + + // Recover any in-progress sagas + let saga_manager = app_state.saga_manager.clone(); + tokio::spawn(async move { + if let Err(e) = saga_manager.recover_sagas().await { + tracing::error!("Saga recovery failed: {}", e); + } + }); + + // Start Redis worker pool + let worker_pool = app_state.worker_pool.clone(); + tokio::spawn(async move { + if let Err(e) = worker_pool.start().await { + tracing::error!("Worker pool failed: {}", e); + } + }); + // Build router with security middleware let app = Router::new() .merge(crate::routes::health_routes())