swarm / companies/starter/work/project/src/sync/ingest.rs

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<Pool>,
    max_retries: u32,
}

impl Ingestor {
    pub fn new(pool: Arc<Pool>) -> Self {
        Self {
            pool,
            max_retries: 3,
        }
    }

    pub async fn process_batch(&self, messages: Vec<Message>) -> Result<(), Box<dyn std::error::Error>> {
        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<Outcome, Box<dyn std::error::Error>> {
        // 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<dyn