use std::sync::Arc; use tokio::time::{sleep, Duration}; use crate::db::Pool; use crate::hl7::Message; #[derive(Debug, Clone)] pub enum Outcome { Success, RetryableError(String), FatalError(String), } pub struct Ingestor { pool: Arc, max_retries: u32, } impl Ingestor { pub fn new(pool: Arc) -> Self { Self { pool, max_retries: 3, } } pub async fn process_batch(&self, messages: Vec) -> Result<(), Box> { for msg in messages { let mut attempts = 0u32; loop { match self.process_single(&msg).await { Ok(outcome) => { self.record_outcome(&msg, &outcome).await?; break; } Err(e) => { attempts += 1; if attempts > self.max_retries { let outcome = Outcome::FatalError(format!("Exceeded max retries: {}", e)); self.record_outcome(&msg, &outcome).await?; break; } // Exponential backoff let delay = Duration::from_millis(100 * (2u64.pow(attempts))); sleep(delay).await; } } } } Ok(()) } async fn process_single(&self, msg: &Message) -> Result> { // Simulate processing; in reality, this parses and persists HL7 if msg.is_valid() { // Persist to DB self.pool.execute("INSERT INTO hl7_messages (id, payload) VALUES ($1, $2)", &[&msg.id, &msg.payload]).await?; Ok(Outcome::Success) } else { Err(Box::new(std::io::Error::new(std::io::ErrorKind::InvalidData, "Malformed HL7 message"))) } } async fn record_outcome(&self, msg: &Message, outcome: &Outcome) -> Result<(), Box