a3s_flow/store/
memory.rs

1use async_trait::async_trait;
2use chrono::Utc;
3use std::collections::HashMap;
4use std::sync::Arc;
5use tokio::sync::Mutex;
6use uuid::Uuid;
7
8use crate::error::{FlowError, Result};
9use crate::model::{FlowEvent, FlowEventEnvelope};
10
11use super::FlowEventStore;
12
13/// In-memory event store for tests, local development, and embedded hosts.
14#[derive(Debug, Default)]
15pub struct InMemoryEventStore {
16    runs: Arc<Mutex<HashMap<String, Vec<FlowEventEnvelope>>>>,
17}
18
19impl InMemoryEventStore {
20    pub fn new() -> Self {
21        Self::default()
22    }
23}
24
25#[async_trait]
26impl FlowEventStore for InMemoryEventStore {
27    async fn append(&self, run_id: &str, event: FlowEvent) -> Result<FlowEventEnvelope> {
28        let mut runs = self.runs.lock().await;
29        append_in_memory(&mut runs, run_id, event)
30    }
31
32    async fn append_if_sequence(
33        &self,
34        run_id: &str,
35        expected_sequence: u64,
36        event: FlowEvent,
37    ) -> Result<FlowEventEnvelope> {
38        let mut runs = self.runs.lock().await;
39        let actual_sequence = runs
40            .get(run_id)
41            .and_then(|events| events.last())
42            .map_or(0, |event| event.sequence);
43        if actual_sequence != expected_sequence {
44            return Err(FlowError::EventConflict {
45                run_id: run_id.to_string(),
46                expected_sequence,
47                actual_sequence,
48            });
49        }
50        append_in_memory(&mut runs, run_id, event)
51    }
52
53    async fn list(&self, run_id: &str) -> Result<Vec<FlowEventEnvelope>> {
54        let runs = self.runs.lock().await;
55        match runs.get(run_id) {
56            Some(events) => Ok(events.clone()),
57            None => Err(FlowError::RunNotFound(run_id.to_string())),
58        }
59    }
60
61    async fn list_run_ids(&self) -> Result<Vec<String>> {
62        let runs = self.runs.lock().await;
63        let mut ids: Vec<String> = runs.keys().cloned().collect();
64        ids.sort();
65        Ok(ids)
66    }
67}
68
69fn append_in_memory(
70    runs: &mut HashMap<String, Vec<FlowEventEnvelope>>,
71    run_id: &str,
72    event: FlowEvent,
73) -> Result<FlowEventEnvelope> {
74    let events = runs.entry(run_id.to_string()).or_default();
75    let envelope = FlowEventEnvelope {
76        run_id: run_id.to_string(),
77        sequence: events.last().map_or(1, |event| event.sequence + 1),
78        event_id: Uuid::new_v4(),
79        timestamp: Utc::now(),
80        event,
81    };
82    events.push(envelope.clone());
83    Ok(envelope)
84}
a3s_flow/store/memory.rs

a3s_flow/store/
memory.rs
