shared_framework/app/
mod.rs

1//! Application startup and service wiring.
2//!
3//! Provides [`StartupOptions`] for startup flags and tuning knobs, and
4//! [`GenericStartup`] which initializes tracing, environment, routing,
5//! documentation, cache, and gateway registration, then mounts controllers,
6//! jobs, and queue consumers. Use [`GenericStartup::bootstrap`] for full
7//! startup, or [`GenericStartup::new`] plus [`GenericStartup::init`] for manual wiring.
8//!
9//! ```ignore
10//! let startup = GenericStartup::bootstrap(
11//!     StartupOptions::default().with_http(true),
12//!     vec!["/api".to_string()],
13//!     None,
14//!     Some(controller_registrar),
15//!     None,
16//! ).await?;
17//! startup.serve().await?;
18//! ```
19
20use std::net::SocketAddr;
21use std::sync::Arc;
22
23use crate::config::ConfigurationRegistrant;
24use crate::controller::RouteController;
25use crate::data::cache::RedisStorage;
26use crate::doc::DocumentationRegistrant;
27use crate::env::AppEnvironment;
28use crate::gateway::GatewayConnect;
29use crate::job::JobRegistry;
30
31/// Startup flags and tuning knobs for [`GenericStartup`].
32///
33/// Build with [`StartupOptions::default_options`] (via [`Default`]) and the
34/// `with_*` builders. `worker_pool_size` and `event_loop_pool_size` are
35/// recorded and logged; the Tokio runtime itself is owned by the host binary.
36#[derive(Debug, Clone)]
37pub struct StartupOptions {
38    /// Path to the dotenv file loaded at startup. `None` loads `.env` by default lookup.
39    pub env_file: Option<String>,
40    /// Whether HTTP controllers are mounted during [`GenericStartup::bootstrap`]. Defaults to `true`.
41    pub enable_http: bool,
42    /// Whether [`GenericStartup::run_jobs`] starts registered jobs. Defaults to `true`.
43    pub enable_jobs: bool,
44    /// Whether queue consumers from [`ConsumerRegistrar`] are recorded during bootstrap. Defaults to `false`.
45    pub enable_consumers: bool,
46    /// Service scheme sent during gateway registration. Defaults to `"http"`.
47    pub service_protocol: String,
48    /// Service weight sent during gateway registration. Defaults to `1`.
49    pub service_weight: i32,
50    /// Auth type sent during gateway registration. Defaults to `"token"`.
51    pub auth_type: String,
52    /// Maximum expected worker task duration, in minutes. Defaults to `2`.
53    pub worker_max_execute_time_minutes: u64,
54    /// Maximum expected event-loop task duration, in minutes. Defaults to `1`.
55    pub event_loop_max_execute_time_minutes: u64,
56    /// Interval for blocked-thread checks, in milliseconds. Defaults to `750`.
57    pub blocked_thread_check_interval_millis: u64,
58    /// Worker pool size hint, logged at startup and reported with worker counts. Defaults to `20`.
59    pub worker_pool_size: usize,
60    /// Event-loop pool size hint, logged at startup. Defaults to `16`.
61    pub event_loop_pool_size: usize,
62}
63
64impl Default for StartupOptions {
65    fn default() -> Self {
66        Self::default_options()
67    }
68}
69
70impl StartupOptions {
71    /// Returns the default startup flags (HTTP and jobs on, consumers off).
72    pub fn default_options() -> Self {
73        Self {
74            env_file: None,
75            enable_http: true,
76            enable_jobs: true,
77            enable_consumers: false,
78            service_protocol: "http".to_string(),
79            service_weight: 1,
80            auth_type: "token".to_string(),
81            worker_max_execute_time_minutes: 2,
82            event_loop_max_execute_time_minutes: 1,
83            blocked_thread_check_interval_millis: 750,
84            // The default worker pool size is 20; the event-loop pool defaults to 16.
85            // and DEFAULT_EVENT_LOOP_POOL_SIZE (2 * cores, here 16 as a sane default).
86            worker_pool_size: 20,
87            event_loop_pool_size: 16,
88        }
89    }
90
91    /// Sets the dotenv file path used by [`GenericStartup::init`].
92    pub fn with_env_file(mut self, path: impl Into<String>) -> Self {
93        self.env_file = Some(path.into());
94        self
95    }
96    /// Sets whether HTTP controllers are mounted during bootstrap.
97    pub fn with_http(mut self, enabled: bool) -> Self {
98        self.enable_http = enabled;
99        self
100    }
101    /// Sets whether [`GenericStartup::run_jobs`] starts registered jobs.
102    pub fn with_jobs(mut self, enabled: bool) -> Self {
103        self.enable_jobs = enabled;
104        self
105    }
106    /// Sets whether queue consumers are recorded during bootstrap.
107    pub fn with_consumers(mut self, enabled: bool) -> Self {
108        self.enable_consumers = enabled;
109        self
110    }
111}
112
113/// Assembled service state produced by [`GenericStartup::bootstrap`] or manual init.
114///
115/// Holds the startup options, mounted route registrant, background job registry,
116/// gateway connection, cache handle, bound addresses, recorded consumer names,
117/// and whether OpenAPI specs were built. Pass `&GenericStartup` to registrars so
118/// they can build controllers and consumers from this state.
119pub struct GenericStartup {
120    /// The options this instance was created with.
121    pub options: StartupOptions,
122    /// URL path prefixes the service handles; also used for docs base and gateway registration.
123    pub mount_paths: Vec<String>,
124    /// HTTP router holder, set by [`GenericStartup::init`]. Required before mounting or serving.
125    pub registrant: Option<Arc<ConfigurationRegistrant>>,
126    /// Background job registry. Add jobs with [`GenericStartup::add_job`], run with [`GenericStartup::run_jobs`].
127    pub job_registry: JobRegistry,
128    /// Gateway connection, set during [`GenericStartup::bootstrap`] when registration succeeds.
129    pub gateway: Option<Arc<GatewayConnect>>,
130    /// Redis cache handle, set during [`GenericStartup::init`] when Redis is reachable.
131    pub redis: Option<RedisStorage>,
132    /// HTTP listen address derived from the configured server port.
133    pub server_addr: Option<SocketAddr>,
134    /// Socket listen address derived from the configured socket port.
135    pub socket_addr: Option<SocketAddr>,
136    /// Queue names recorded from [`ConsumerRegistrar`] during bootstrap.
137    pub consumer_names: Vec<String>,
138    docs_built: bool,
139}
140
141impl GenericStartup {
142    /// Creates an empty startup with the given options and no mounted state.
143    pub fn new(options: StartupOptions) -> Self {
144        Self {
145            options,
146            mount_paths: Vec::new(),
147            registrant: None,
148            job_registry: JobRegistry::new(),
149            gateway: None,
150            redis: None,
151            server_addr: None,
152            socket_addr: None,
153            consumer_names: Vec::new(),
154            docs_built: false,
155        }
156    }
157
158    /// Sets the URL path prefixes this service handles.
159    pub fn with_mount_paths(mut self, paths: Vec<String>) -> Self {
160        self.mount_paths = paths;
161        self
162    }
163
164    // ── Legacy init (kept for compat) ────────────────────────────────────
165    /// Performs global init: loads the environment, sets up tracing, resolves the
166    /// server and socket addresses, creates the route registrant, binds the docs
167    /// base to `mount_paths`, and connects to Redis on a best-effort basis.
168    ///
169    /// Returns an error if the environment is invalid or an address fails to parse.
170    /// Redis failure only logs a warning and leaves [`GenericStartup::redis`] as `None`.
171    pub async fn init(&mut self) -> anyhow::Result<()> {
172        AppEnvironment::with_env_file(self.options.env_file.as_deref())?;
173        Self::init_tracing();
174        Self::log_pool_options(&self.options);
175        let env = AppEnvironment::get();
176        let addr: SocketAddr = format!("0.0.0.0:{}", env.server_port).parse()?;
177        let socket_addr: SocketAddr = format!("0.0.0.0:{}", env.socket_port).parse()?;
178        self.registrant = Some(Arc::new(ConfigurationRegistrant::new(addr)));
179        self.server_addr = Some(addr);
180        self.socket_addr = Some(socket_addr);
181
182        // Bind the docs base path to the first mount path.
183        if let Ok(mut reg) = DocumentationRegistrant::global().write() {
184            reg.bind_base_list(&self.mount_paths);
185        }
186
187        // Best-effort Redis — warns and continues without cache when unavailable.
188        match RedisStorage::from_env() {
189            Ok(storage) => {
190                tracing::info!(
191                    component = "cache",
192                    backend = "redis",
193                    "Redis storage initialized"
194                );
195                self.redis = Some(storage);
196            }
197            Err(e) => {
198                tracing::warn!(component = "cache", backend = "redis", error = %e, "Redis unavailable; continuing without cache");
199            }
200        }
201        Ok(())
202    }
203
204    fn init_tracing() {
205        use tracing_subscriber::EnvFilter;
206        use tracing_subscriber::layer::SubscriberExt;
207        use tracing_subscriber::util::SubscriberInitExt;
208        // Only init once — ignore error if already set (e.g., tests)
209        let (env_filter, filter_source) = match EnvFilter::try_from_default_env() {
210            Ok(filter) => (filter, "RUST_LOG"),
211            Err(_) => (EnvFilter::new("INFO"), "DEFAULT"),
212        };
213        let _ = tracing_subscriber::registry()
214            .with(env_filter)
215            .with(tracing_subscriber::fmt::layer().with_ansi(true))
216            .try_init();
217        tracing::info!(filter_source, "Tracing initialized");
218    }
219
220    fn log_pool_options(options: &StartupOptions) {
221        // Tokio runtime is owned by the host binary; we retain/validate the knobs
222        // so pool sizing stays explicit; the Tokio runtime itself is owned by the host binary.
223        if options.worker_pool_size == 0 || options.event_loop_pool_size == 0 {
224            tracing::warn!(
225                worker_pool_size = options.worker_pool_size,
226                event_loop_pool_size = options.event_loop_pool_size,
227                "Invalid runtime pool-size hints; Tokio runtime sizing is owned by the host binary"
228            );
229        } else {
230            tracing::info!(
231                worker_max_execute_time_minutes = options.worker_max_execute_time_minutes,
232                event_loop_max_execute_time_minutes = options.event_loop_max_execute_time_minutes,
233                blocked_thread_check_interval_ms = options.blocked_thread_check_interval_millis,
234                worker_pool_size = options.worker_pool_size,
235                event_loop_pool_size = options.event_loop_pool_size,
236                "Configured runtime pool-size hints"
237            );
238        }
239    }
240
241    /// Runs full startup: global init, static registration, gateway registration,
242    /// controller mounting with OpenAPI spec building, and consumer recording.
243    ///
244    /// `options` selects which subsystems run; `mount_paths` sets the handled URL
245    /// prefixes. Each registrar is optional: pass `None` for anything the caller
246    /// wires manually via the exposed `GenericStartup` fields. An empty
247    /// `mount_paths` skips gateway registration; gateway failure is logged and
248    /// does not fail bootstrap.
249    ///
250    /// ```ignore
251    /// let startup = GenericStartup::bootstrap(opts, mount_paths, None, Some(controllers), None).await?;
252    /// ```
253    pub async fn bootstrap(
254        options: StartupOptions,
255        mount_paths: Vec<String>,
256        static_registrar: Option<Arc<dyn StaticRegistrar>>,
257        controller_registrar: Option<Arc<dyn ControllerRegistrar>>,
258        consumer_registrar: Option<Arc<dyn ConsumerRegistrar>>,
259    ) -> anyhow::Result<Self> {
260        let mut startup = Self::new(options);
261        startup.mount_paths = mount_paths.clone();
262        // 1. Global init (tracing, env, router, docs base, redis)
263        startup.init().await?;
264
265        let env = AppEnvironment::get();
266        let cpu_count = std::thread::available_parallelism()
267            .map(|n| n.get())
268            .unwrap_or(1);
269
270        // 2. Static resources via the static registrar.
271        if let Some(sr) = static_registrar.as_ref() {
272            if let Some(reg) = startup.registrant.as_ref() {
273                let handle = reg.router_handle();
274                let mut router = handle.write().await;
275                sr.register_static(&startup, &mut router);
276            }
277        }
278        // Re-bind the docs base after static registration.
279        if let Ok(mut reg) = DocumentationRegistrant::global().write() {
280            reg.bind_base_list(&mount_paths);
281        }
282
283        // 3. Basilisk gateway / event bus — optional; failures are logged, never fatal.
284        match GatewayConnect::set_up(
285            mount_paths.clone(),
286            startup.options.service_protocol.clone(),
287            startup.options.service_weight,
288            startup.options.auth_type.clone(),
289        )
290        .await
291        {
292            Ok(Some(gw)) => {
293                tracing::info!(
294                    component = "gateway",
295                    mount_path_count = mount_paths.len(),
296                    "Registered service configuration"
297                );
298                startup.gateway = Some(gw);
299            }
300            Ok(None) => {
301                tracing::info!(
302                    component = "gateway",
303                    "No mount paths configured; skipped service registration"
304                );
305            }
306            Err(e) => {
307                tracing::error!(component = "gateway", error = %e, "Failed to register service configuration");
308                if !env.is_production() {
309                    eprintln!("{:?}", e);
310                }
311            }
312        }
313
314        // 4. Warn when deployable exceeds CPU cores.
315        let deployable_count = env.server_count + env.socket_count + env.worker_count;
316        if deployable_count > cpu_count {
317            tracing::warn!(
318                deployable_count,
319                cpu_count,
320                "Configured deployables exceed available CPU cores"
321            );
322        }
323
324        // 5. HTTP controllers — auto-wire via ControllerRegistrar.
325        if env.server_count > 0 && startup.options.enable_http {
326            if let Some(cr) = controller_registrar.as_ref() {
327                let controllers = cr.controllers(&startup);
328                for ctrl in controllers {
329                    startup.mount_controller_boxed(ctrl).await?;
330                }
331            }
332            // Always auto-mount the DocumentationController.
333            startup
334                .mount_controller_boxed(Box::new(crate::doc::controller::DocumentationController))
335                .await?;
336            // Build OpenAPI specs at once, after all controllers are mounted.
337            let timer = std::time::Instant::now();
338            crate::doc::controller::DocumentationController::build_specs();
339            startup.docs_built = true;
340            tracing::info!(
341                component = "openapi",
342                duration_ms = timer.elapsed().as_millis() as u64,
343                "Built OpenAPI documentation"
344            );
345        }
346
347        // 6. Workers / jobs — registered via JobRegistry; started explicitly with run_jobs().
348        if env.worker_count > 0 {
349            tracing::info!(
350                worker_count = env.worker_count,
351                worker_pool_size = startup.options.worker_pool_size,
352                "Jobs are available via run_jobs()"
353            );
354        }
355
356        // 7. Consumers — one deployment per consumer: deploying the same queue
357        // consumer twice is rejected with a duplicate-consumer-tag error by RabbitMQ.
358        // The caller spawns the actual lapin loops; descriptors are collected here,
359        // so nothing is silently dropped.
360        if let Some(cons_reg) = consumer_registrar.as_ref() {
361            if startup.options.enable_consumers {
362                let consumers = cons_reg.consumers(&startup);
363                for c in &consumers {
364                    tracing::info!(queue = %c.queue_name(), "Registered consumer");
365                    startup.consumer_names.push(c.queue_name().to_string());
366                }
367                if consumers.is_empty() {
368                    tracing::info!("Consumer registrar provided no consumers");
369                }
370            } else {
371                tracing::info!(enabled = false, "Consumer registration skipped");
372            }
373        }
374
375        // 8. Sockets — expose socket_addr for the caller to bind.
376        if env.socket_count > 0 {
377            tracing::info!(
378                socket_count = env.socket_count,
379                socket_addr = ?startup.socket_addr,
380                "Socket server is available for binding"
381            );
382        }
383
384        Ok(startup)
385    }
386
387    /// Mounts a controller's routes onto the router. Errors if [`GenericStartup::init`] or [`GenericStartup::bootstrap`] has not run.
388    pub async fn mount_controller<C: RouteController + 'static>(&self, c: C) -> anyhow::Result<()> {
389        if let Some(r) = &self.registrant {
390            r.mount_controller(c).await;
391            Ok(())
392        } else {
393            anyhow::bail!("not initialized — call init() or bootstrap() first")
394        }
395    }
396
397    /// Mounts a boxed trait-object controller. Used for controllers built by [`ControllerRegistrar`]. Errors if not initialised.
398    pub async fn mount_controller_boxed(&self, c: Box<dyn RouteController>) -> anyhow::Result<()> {
399        if let Some(r) = &self.registrant {
400            let handle = r.router_handle();
401            let mut router = handle.write().await;
402            tracing::info!(
403                target: "routing",
404                handler = c.type_name(),
405                path = c.base_path(),
406                "Mounted controller '{}' at '{}'",
407                c.type_name(),
408                c.base_path()
409            );
410            c.register_routes(&mut router).await;
411            Ok(())
412        } else {
413            anyhow::bail!("not initialized")
414        }
415    }
416
417    /// Starts the HTTP server in the background and returns the bound address. Errors if not initialised.
418    pub async fn serve(&self) -> anyhow::Result<SocketAddr> {
419        if let Some(r) = self.registrant.clone() {
420            Ok(r.serve().await?)
421        } else {
422            anyhow::bail!("not initialized")
423        }
424    }
425
426    /// Stops background jobs and deregisters from the gateway. Failures are logged as warnings; shutdown itself succeeds.
427    pub async fn shutdown(&mut self) -> anyhow::Result<()> {
428        if let Err(e) = self.job_registry.stop().await {
429            tracing::warn!(component = "jobs", error = %e, "Failed to stop jobs during shutdown");
430        }
431        if let Some(gw) = self.gateway.take() {
432            if let Err(e) = gw.deregister().await {
433                tracing::warn!(component = "gateway", error = %e, "Failed to deregister service during shutdown");
434            }
435        }
436        tracing::info!("Shutdown complete");
437        Ok(())
438    }
439
440    /// Returns the mutable job registry for manual wiring.
441    pub fn job_registry(&mut self) -> &mut JobRegistry {
442        &mut self.job_registry
443    }
444    /// Adds a job to the registry. Jobs start only when [`GenericStartup::run_jobs`] is called.
445    pub fn add_job<J: crate::job::ServiceJob + 'static>(&mut self, job: J) {
446        self.job_registry.add_job(job);
447    }
448    /// Starts all registered jobs. No-ops with a log when `enable_jobs` is false or no jobs are registered.
449    pub async fn run_jobs(&mut self) -> anyhow::Result<()> {
450        if !self.options.enable_jobs {
451            tracing::info!(enabled = false, "Job startup skipped");
452            return Ok(());
453        }
454        if self.job_registry.job_count() == 0 {
455            tracing::info!("No jobs registered; nothing to start");
456            return Ok(());
457        }
458        tracing::info!(job_count = self.job_registry.job_count(), "Starting jobs");
459        self.job_registry.start().await?;
460        Ok(())
461    }
462    /// Alias for [`GenericStartup::run_jobs`].
463    pub async fn start_jobs(&mut self) -> anyhow::Result<()> {
464        self.run_jobs().await
465    }
466    /// Stops all running jobs.
467    pub async fn stop_jobs(&mut self) -> anyhow::Result<()> {
468        self.job_registry.stop().await?;
469        Ok(())
470    }
471
472    // Convenience getters exposing internal state for caller init
473    /// Returns the route registrant, if initialised.
474    pub fn registrant(&self) -> Option<Arc<ConfigurationRegistrant>> {
475        self.registrant.clone()
476    }
477    /// Returns the gateway connection, if registration succeeded.
478    pub fn gateway_client(&self) -> Option<Arc<GatewayConnect>> {
479        self.gateway.clone()
480    }
481    /// Returns the Redis cache handle, if Redis was reachable at init.
482    pub fn redis_client(&self) -> Option<RedisStorage> {
483        self.redis.clone()
484    }
485    /// Returns the resolved HTTP listen address, if initialised.
486    pub fn server_addr(&self) -> Option<SocketAddr> {
487        self.server_addr
488    }
489    /// Returns the resolved socket listen address, if initialised.
490    pub fn socket_addr(&self) -> Option<SocketAddr> {
491        self.socket_addr
492    }
493    /// Returns the shared router handle, if initialised.
494    pub fn router_handle(&self) -> Option<Arc<tokio::sync::RwLock<crate::controller::Router>>> {
495        self.registrant.as_ref().map(|r| r.router_handle())
496    }
497    /// Reports whether the loaded environment is production. Returns `false` when the environment is not provisioned.
498    pub fn is_production(&self) -> bool {
499        AppEnvironment::try_get()
500            .map(|e| e.is_production())
501            .unwrap_or(false)
502    }
503    /// Reports whether OpenAPI specs were built during bootstrap.
504    pub fn docs_built(&self) -> bool {
505        self.docs_built
506    }
507    /// Returns the localhost server URLs for the configured server port (defaults to `8080` when unprovisioned).
508    pub fn server_urls(&self) -> Vec<String> {
509        let port = AppEnvironment::try_get()
510            .map(|e| e.server_port)
511            .unwrap_or(8080);
512        vec![
513            format!("http://localhost:{port}"),
514            format!("http://127.0.0.1:{port}"),
515        ]
516    }
517}
518
519/// Builds the HTTP controllers to mount during bootstrap.
520///
521/// Receives `&GenericStartup` so controllers can be constructed from startup
522/// state such as the registrant, gateway, Redis handle, or server address.
523pub trait ControllerRegistrar: Send + Sync {
524    /// Returns the controllers to mount, in mount order.
525    fn controllers(&self, startup: &GenericStartup) -> Vec<Box<dyn RouteController>>;
526}
527
528/// Builds the queue consumers to record during bootstrap.
529///
530/// Receives `&GenericStartup` so consumers can be constructed from startup
531/// state such as the environment, gateway, or Redis handle.
532pub trait ConsumerRegistrar: Send + Sync {
533    /// Returns the consumer descriptors to record, one entry per queue.
534    fn consumers(
535        &self,
536        startup: &GenericStartup,
537    ) -> Vec<Box<dyn queue_descriptor::QueueDescriptor>>;
538}
539
540/// Mounts static assets onto the router during bootstrap.
541///
542/// Receives `&GenericStartup` and the mutable router so assets are mounted
543/// with full access to startup state.
544pub trait StaticRegistrar: Send + Sync {
545    /// Registers static routes into `router`.
546    fn register_static(&self, startup: &GenericStartup, router: &mut crate::controller::Router);
547}
548
549/// Queue consumer descriptors recorded by [`ConsumerRegistrar`].
550pub mod queue_descriptor {
551    /// Minimal descriptor for a queue consumer.
552    pub trait QueueDescriptor: Send + Sync {
553        /// Returns the queue name this consumer handles.
554        fn queue_name(&self) -> &str;
555    }
556}
shared_framework/app/mod.rs

shared_framework/app/
mod.rs
