arc_web/
lib.rs

1//! Arc framework runtime.
2//!
3//! Holds the reusable web/runtime machinery — Actix middleware, Tera helpers,
4//! the event-sourced stack wiring, the websocket transport, and the server
5//! bootstrap — behind a builder seam. Applications depend on this crate by
6//! version and plug their own aggregate, projectors, and routes in through
7//! [`ArcApp::builder`]. No concrete domain type lives here: the runtime is
8//! generic over [`arc_core::aggregate::Aggregate`].
9
10use std::any::TypeId;
11use std::future::Future;
12use std::path::PathBuf;
13use std::pin::Pin;
14use std::sync::{Arc, Mutex};
15use std::{env, fs};
16
17use actix_web::web::ServiceConfig;
18use arc_core::aggregate::Aggregate;
19use arc_core::command_bus::SnapshotPolicy;
20use arc_core::projection::Projector;
21use tracing::{debug, error, info};
22
23pub mod helpers {
24    pub mod access_log;
25    pub mod audit_context;
26    pub mod config;
27    pub mod csrf;
28    pub mod database;
29    pub mod es_stack;
30    pub mod general;
31    pub mod jwt;
32    pub mod rate_limit;
33    pub mod session;
34    pub mod template;
35}
36
37pub mod http {
38    pub mod middlewares {
39        pub mod auth_middleware;
40        pub mod idle_timeout_middleware;
41        pub mod jwt_middleware;
42        pub mod rate_limit_middleware;
43    }
44    pub mod errors;
45}
46
47pub mod commands;
48pub mod websocket;
49
50/// Shared application state accessible by request handlers via `web::Data`.
51#[derive(Debug)]
52pub struct AppState {
53    pub app_name: Mutex<String>,
54}
55
56/// A read-model projector plus the view (table) it maintains. Applications
57/// register these against their aggregate so the framework can drive
58/// synchronous, read-after-write projections in single-process mode.
59pub struct ProjectorReg {
60    pub projector: Box<dyn Projector>,
61    pub view: String,
62}
63
64impl ProjectorReg {
65    pub fn new(projector: impl Projector + 'static, view: impl Into<String>) -> Self {
66        Self {
67            projector: Box::new(projector),
68            view: view.into(),
69        }
70    }
71}
72
73/// Entry point to the framework. Applications register one or more aggregate
74/// types, their projectors, and routes, then drive the result with
75/// [`ArcAppBuilder::serve`].
76pub struct ArcApp;
77
78impl ArcApp {
79    pub fn builder() -> ArcAppBuilder {
80        ArcAppBuilder {
81            aggregates: Vec::new(),
82            routes: None,
83        }
84    }
85}
86
87type RoutesFn = dyn Fn(&mut ServiceConfig) + Send + Sync + 'static;
88type AggregateBuildFuture = Pin<
89    Box<
90        dyn Future<Output = std::io::Result<commands::serve::BuiltAggregateRuntime>>
91            + Send
92            + 'static,
93    >,
94>;
95type AggregateBuildFn =
96    dyn FnOnce(Vec<ProjectorReg>, Option<SnapshotPolicy>) -> AggregateBuildFuture + Send + 'static;
97
98struct AggregateRegistration {
99    type_id: TypeId,
100    aggregate_type: &'static str,
101    projectors: Vec<ProjectorReg>,
102    snapshot_policy: Option<SnapshotPolicy>,
103    build: Box<AggregateBuildFn>,
104}
105
106impl AggregateRegistration {
107    fn new<A: Aggregate + 'static>() -> Self {
108        Self {
109            type_id: TypeId::of::<A>(),
110            aggregate_type: A::aggregate_type(),
111            projectors: Vec::new(),
112            snapshot_policy: None,
113            build: Box::new(|projectors, snapshot_policy| {
114                Box::pin(commands::serve::build_aggregate_runtime::<A>(
115                    projectors,
116                    snapshot_policy,
117                ))
118            }),
119        }
120    }
121}
122
123/// Application builder holding type-erased registrations for every aggregate
124/// served by the process. Each registration produces its own typed
125/// `CommandBus<A>` while sharing the configured storage backend.
126pub struct ArcAppBuilder {
127    aggregates: Vec<AggregateRegistration>,
128    routes: Option<Arc<RoutesFn>>,
129}
130
131impl ArcAppBuilder {
132    /// Register a writable aggregate type. Arc creates and injects a distinct
133    /// `CommandBus<A>` for every call.
134    pub fn register_aggregate<A: Aggregate + 'static>(mut self) -> Self {
135        assert!(
136            !self
137                .aggregates
138                .iter()
139                .any(|registration| registration.type_id == TypeId::of::<A>()),
140            "aggregate type {} is already registered",
141            A::aggregate_type()
142        );
143        self.aggregates.push(AggregateRegistration::new::<A>());
144        self
145    }
146
147    /// Register all projectors for the most recently registered aggregate.
148    pub fn register_projectors(mut self, projectors: Vec<ProjectorReg>) -> Self {
149        self.current_aggregate_mut().projectors.extend(projectors);
150        self
151    }
152
153    /// Register a single projector for the most recently registered aggregate.
154    pub fn register_projector(
155        mut self,
156        projector: impl Projector + 'static,
157        view: impl Into<String>,
158    ) -> Self {
159        self.current_aggregate_mut()
160            .projectors
161            .push(ProjectorReg::new(projector, view));
162        self
163    }
164
165    /// Mount the application's own routes. Framework handlers (static assets,
166    /// websocket, health) are exported from this crate for the closure to
167    /// reference; the application owns which routes it mounts.
168    pub fn register_routes<F>(mut self, f: F) -> Self
169    where
170        F: Fn(&mut ServiceConfig) + Send + Sync + 'static,
171    {
172        self.routes = Some(Arc::new(f));
173        self
174    }
175
176    /// Snapshot policy for the most recently registered aggregate.
177    pub fn snapshot_policy(mut self, policy: Option<SnapshotPolicy>) -> Self {
178        self.current_aggregate_mut().snapshot_policy = policy;
179        self
180    }
181
182    fn current_aggregate_mut(&mut self) -> &mut AggregateRegistration {
183        self.aggregates.last_mut().unwrap_or_else(|| {
184            panic!(
185                "register_aggregate::<A>() must be called before registering projectors or a snapshot policy"
186            )
187        })
188    }
189
190    /// Boot the HTTP server with every registered aggregate and the
191    /// application's routes.
192    pub async fn serve(self, app_url: String, app_port: u16) -> std::io::Result<()> {
193        if self.aggregates.is_empty() {
194            return Err(std::io::Error::new(
195                std::io::ErrorKind::InvalidInput,
196                "ArcAppBuilder::serve requires at least one register_aggregate::<A>() call",
197            ));
198        }
199        let routes = self
200            .routes
201            .expect("ArcAppBuilder::serve requires register_routes(..)");
202        commands::serve::run(app_url, app_port, self.aggregates, routes).await
203    }
204}
205
206/// Verifies database availability for the configured driver. For file-backed
207/// drivers (SQLite) this checks the `DATABASE_URL` file exists and exits with
208/// code 1 if missing. Connection-string drivers (Postgres) skip the check.
209pub fn check_database_health() {
210    info!("Checking database health");
211    let driver = helpers::config::DatabaseDriver::from_env();
212
213    if !driver.is_file_backed() {
214        debug!(
215            driver = driver.as_str(),
216            "Database driver uses a connection string; skipping filesystem check"
217        );
218        return;
219    }
220
221    let database: String = helpers::config::database_url();
222    if !fs::exists(PathBuf::from(&database)).unwrap() {
223        error!("Database file not found at: {}", database);
224        error!("Please run `cargo run migrate` to create the database");
225        std::process::exit(1);
226    }
227    debug!("Database file found at: {}", database);
228}
229
230/// Copies `.env.example` to `.env` when no `.env` file is present.
231pub fn check_app_health() {
232    info!("Checking app health");
233    if !fs::exists(PathBuf::from(".env")).unwrap() {
234        info!("Creating .env file from .env.example");
235        fs::copy(PathBuf::from(".env.example"), PathBuf::from(".env"))
236            .expect("Failed to copy .env.example to .env");
237    }
238}
239
240/// Fails fast if required environment variables are missing.
241pub fn validate_environment() {
242    let required_vars = ["APP_URL", "SECRET_KEY", "DATABASE_URL"];
243    let mut missing = Vec::new();
244    for var in required_vars {
245        if env::var(var).is_err() {
246            missing.push(var);
247        }
248    }
249    if !missing.is_empty() {
250        error!(
251            "Missing required environment variables: {}. Check your .env file.",
252            missing.join(", ")
253        );
254        std::process::exit(1);
255    }
256    debug!("All required environment variables present");
257}
arc_web/lib.rs

arc_web/
lib.rs
