arc_web/helpers/
es_stack.rs

1//! Shared assembly of the event-sourced stack — `EventStore`, in-process
2//! `EventBus`, `ReadModelStore`, `ProjectionEngine`, and `CommandBus`.
3//!
4//! Generic over the domain aggregate `A`: the framework never names a concrete
5//! aggregate. Applications supply `A` and their projectors; the runtime server
6//! (`commands::serve`) and CLI utilities drive the same wiring.
7
8use crate::helpers::config;
9use crate::helpers::config::DatabaseDriver;
10use crate::ProjectorReg;
11use arc_core::aggregate::Aggregate;
12use arc_core::command_bus::{CommandBus, SnapshotPolicy};
13use arc_core::event_bus::{EventBus, InProcessEventBus};
14use arc_core::event_store::EventStore;
15use arc_core::projection::{ProjectionEngine, ProjectionEngineHandler};
16use arc_core::read_model_store::ReadModelStore;
17use arc_es_sqlite::{SqliteEventStore, SqliteReadModelStore};
18use std::sync::Arc;
19
20/// Bundle of constructed components — the parts external code keeps a
21/// handle on after wiring.
22pub struct EsStack<A: Aggregate> {
23    pub command_bus: CommandBus<A>,
24    pub read_model_store: Arc<dyn ReadModelStore>,
25    /// Held so callers that want to drive `rebuild_all()` can do so. CLI
26    /// utilities ignore it; the runtime server keeps a clone.
27    #[allow(dead_code)]
28    pub projection_engine: Arc<ProjectionEngine>,
29}
30
31/// Boxed storage seam shared by every entry point. The command bus and the
32/// projection engine each own an independent handle to the same backend so the
33/// write path and replay path do not contend on one connection wrapper.
34pub struct EsStores {
35    pub command_event_store: Box<dyn EventStore>,
36    pub projection_event_store: Box<dyn EventStore>,
37    pub read_model_store: Arc<dyn ReadModelStore>,
38}
39
40/// Construct the event store and read-model store for the selected driver.
41/// Domain code never sees the concrete type — only `Box<dyn EventStore>` and
42/// `Arc<dyn ReadModelStore>`.
43pub async fn build_stores(
44    driver: DatabaseDriver,
45    database_url: &str,
46) -> Result<EsStores, Box<dyn std::error::Error>> {
47    match driver {
48        DatabaseDriver::Sqlite => build_sqlite_stores(database_url).await,
49        DatabaseDriver::Postgres => build_postgres_stores(database_url).await,
50    }
51}
52
53async fn build_sqlite_stores(database_url: &str) -> Result<EsStores, Box<dyn std::error::Error>> {
54    let event_store = build_sqlite_event_store(database_url).await?;
55    let read_model_store: Arc<dyn ReadModelStore> =
56        Arc::new(SqliteReadModelStore::new(database_url).await?);
57
58    Ok(EsStores {
59        command_event_store: Box::new(event_store.clone()),
60        projection_event_store: Box::new(event_store),
61        read_model_store,
62    })
63}
64
65async fn build_sqlite_event_store(
66    database_url: &str,
67) -> Result<SqliteEventStore, Box<dyn std::error::Error>> {
68    match config::event_integrity_key() {
69        Some(key) => Ok(SqliteEventStore::new_with_integrity_key(
70            database_url,
71            key,
72            config::event_integrity_key_id(),
73        )
74        .await?),
75        None => Ok(SqliteEventStore::new(database_url).await?),
76    }
77}
78
79#[cfg(feature = "postgres")]
80async fn build_postgres_stores(database_url: &str) -> Result<EsStores, Box<dyn std::error::Error>> {
81    use arc_es_postgres::{PostgresEventStore, PostgresReadModelStore};
82
83    let event_store = PostgresEventStore::new(database_url).await?;
84    event_store.initialize_schema().await?;
85
86    let read_model_store_impl = PostgresReadModelStore::new(database_url).await?;
87    read_model_store_impl.initialize_schema().await?;
88    let read_model_store: Arc<dyn ReadModelStore> = Arc::new(read_model_store_impl);
89
90    Ok(EsStores {
91        command_event_store: Box::new(event_store.clone()),
92        projection_event_store: Box::new(event_store),
93        read_model_store,
94    })
95}
96
97#[cfg(not(feature = "postgres"))]
98async fn build_postgres_stores(
99    _database_url: &str,
100) -> Result<EsStores, Box<dyn std::error::Error>> {
101    Err("DATABASE_DRIVER=postgres requires building with the `postgres` cargo feature".into())
102}
103
104/// Build the in-process stack for the configured driver, registering the
105/// supplied projectors against the in-process bus so writes drive their views
106/// synchronously. Used by CLI utilities (`migrate --seed`, `seed`) and tests.
107pub async fn build<A: Aggregate>(
108    database_url: &str,
109    projectors: Vec<ProjectorReg>,
110    snapshot_policy: Option<SnapshotPolicy>,
111) -> Result<EsStack<A>, Box<dyn std::error::Error>> {
112    let stores = build_stores(DatabaseDriver::from_env(), database_url).await?;
113    let read_model_store = stores.read_model_store.clone();
114
115    let mut engine = ProjectionEngine::new(stores.projection_event_store);
116    for reg in projectors {
117        engine.register_projector(reg.projector, read_model_store.clone(), reg.view);
118    }
119    let engine = Arc::new(engine);
120
121    let mut bus = InProcessEventBus::new();
122    bus.subscribe(Box::new(ProjectionEngineHandler::new(engine.clone())))
123        .await?;
124
125    let command_bus = apply_snapshot_policy(
126        CommandBus::<A>::new(stores.command_event_store, Box::new(bus)),
127        snapshot_policy,
128    );
129
130    Ok(EsStack {
131        command_bus,
132        read_model_store,
133        projection_engine: engine,
134    })
135}
136
137/// Apply a snapshot policy to a command bus if one is configured.
138pub fn apply_snapshot_policy<A: Aggregate>(
139    command_bus: CommandBus<A>,
140    policy: Option<SnapshotPolicy>,
141) -> CommandBus<A> {
142    match policy {
143        Some(p) => command_bus.with_snapshot_policy(p),
144        None => command_bus,
145    }
146}
arc_web/helpers/es_stack.rs

arc_web/helpers/
es_stack.rs
