a3s_flow/store/
mod.rs

1use async_trait::async_trait;
2
3use crate::error::Result;
4use crate::model::{project_run, ActiveHookSnapshot, FlowEvent, FlowEventEnvelope, HookStatus};
5
6mod local_file;
7mod memory;
8#[cfg(any(feature = "postgres", feature = "sqlite"))]
9mod migrations;
10#[cfg(feature = "postgres")]
11mod postgres;
12mod retention;
13#[cfg(feature = "sqlite")]
14mod sqlite;
15
16pub use local_file::LocalFileEventStore;
17pub use memory::InMemoryEventStore;
18#[cfg(feature = "postgres")]
19pub(crate) use migrations::postgres_migrations;
20#[cfg(feature = "sqlite")]
21pub(crate) use migrations::sqlite_migrations;
22#[cfg(feature = "postgres")]
23pub use postgres::PostgresEventStore;
24#[cfg(any(feature = "postgres", feature = "sqlite"))]
25pub use retention::{
26    FlowHistoryHold, FlowHistoryRetentionPolicy, FlowHistoryRetentionReport, FlowHistoryTombstone,
27};
28#[cfg(feature = "sqlite")]
29pub use sqlite::SqliteEventStore;
30
31/// Append-only event store for durable workflow runs.
32#[async_trait]
33pub trait FlowEventStore: Send + Sync {
34    async fn append(&self, run_id: &str, event: FlowEvent) -> Result<FlowEventEnvelope>;
35
36    async fn append_if_sequence(
37        &self,
38        run_id: &str,
39        expected_sequence: u64,
40        event: FlowEvent,
41    ) -> Result<FlowEventEnvelope>;
42
43    async fn list(&self, run_id: &str) -> Result<Vec<FlowEventEnvelope>>;
44
45    async fn list_run_ids(&self) -> Result<Vec<String>>;
46
47    /// Find active hooks that own an external callback token.
48    ///
49    /// The default implementation replays every run for compatibility with
50    /// custom stores. SQL stores override it with their indexed projection.
51    async fn find_active_hooks_by_token(&self, token: &str) -> Result<Vec<ActiveHookSnapshot>> {
52        Ok(self
53            .list_active_hooks()
54            .await?
55            .into_iter()
56            .filter(|active| active.hook.token == token)
57            .collect())
58    }
59
60    /// List active external callback hooks in stable run/hook order.
61    ///
62    /// The default implementation preserves the append-only store contract by
63    /// projecting histories. Durable SQL adapters provide a materialized path.
64    async fn list_active_hooks(&self) -> Result<Vec<ActiveHookSnapshot>> {
65        let mut hooks = Vec::new();
66        for run_id in self.list_run_ids().await? {
67            let history = self.list(&run_id).await?;
68            let snapshot = project_run(&run_id, &history)?;
69            if snapshot.status.is_terminal() {
70                continue;
71            }
72            for hook in snapshot.hooks.values() {
73                if hook.status == HookStatus::Active {
74                    hooks.push(ActiveHookSnapshot {
75                        run_id: run_id.clone(),
76                        hook: hook.clone(),
77                    });
78                }
79            }
80        }
81        hooks.sort_by(|left, right| {
82            (left.run_id.as_str(), left.hook.hook_id.as_str())
83                .cmp(&(right.run_id.as_str(), right.hook.hook_id.as_str()))
84        });
85        Ok(hooks)
86    }
87}
a3s_flow/store/mod.rs

a3s_flow/store/
mod.rs
