//! Core daemon infrastructure. //! //! This module provides the foundational types for building the atuin daemon: //! //! - [`DaemonState`]: Shared state owned by the daemon //! - [`DaemonHandle`]: A lightweight, cloneable handle for accessing daemon state //! - [`Daemon`]: The main daemon orchestrator //! - [`DaemonBuilder`]: Builder for constructing and configuring the daemon use std::sync::Arc; use crate::aclient::{ database::ClientSqlite as HistoryDatabase, encryption, record::sqlite_store::SqliteStore, settings::Settings, }; use eyre::{Context, Result}; use tokio::sync::{RwLock, broadcast}; use crate::events::DaemonEvent; // ============================================================================ // DaemonState // ============================================================================ /// Shared state owned by the daemon. /// /// This contains all the resources that services need access to. /// The state is wrapped in an `Arc` and accessed via [`DaemonHandle`]. pub(crate) struct DaemonState { // Event bus event_tx: broadcast::Sender, // Configuration (mutable - can be reloaded) settings: RwLock, // Encryption key (immutable - derived at startup) encryption_key: [u8; 32], // Database handles history_db: HistoryDatabase, store: SqliteStore, } // ============================================================================ // DaemonHandle // ============================================================================ /// A lightweight handle to the daemon's shared state. /// /// This is the primary way for gRPC services, and spawned tasks to /// interact with the daemon. It provides access to: /// /// - Event emission and subscription /// - Configuration (settings, encryption key) /// - Database handles /// /// The handle is cheaply cloneable (wraps an `Arc`) and can be freely passed /// around to any code that needs daemon access. /// /// # Example /// /// ```ignore /// // Emit an event /// handle.emit(DaemonEvent::HistoryPruned); /// /// // Access settings /// let settings = handle.settings().await; /// let sync_freq = settings.daemon.sync_frequency; /// /// // Access database /// let history = handle.history_db().load(id).await?; /// ``` #[derive(Clone)] pub(crate) struct DaemonHandle { state: Arc, } impl DaemonHandle { // ---- Events ---- /// Emit an event to the daemon's event bus. /// /// This is fire-and-forget - if no receivers are listening (which shouldn't /// happen in normal operation), the event is dropped silently. pub(crate) fn emit(&self, event: DaemonEvent) { if let Err(e) = self.state.event_tx.send(event) { tracing::warn!("failed to emit event (no receivers?): {e}"); } } /// Subscribe to the event bus. /// /// Returns a receiver that will receive all events emitted after this call. pub(crate) fn subscribe(&self) -> broadcast::Receiver { self.state.event_tx.subscribe() } /// Request graceful shutdown of the daemon. pub(crate) fn shutdown(&self) { self.emit(DaemonEvent::ShutdownRequested); } // ---- Configuration ---- /// Get the current settings. /// /// This acquires a read lock on the settings. For most use cases, clone /// the settings if you need to hold onto them. pub(crate) async fn settings(&self) -> tokio::sync::RwLockReadGuard<'_, Settings> { self.state.settings.read().await } /// Apply already-loaded settings and emit a [`SettingsReloaded`] event. /// /// Use this when settings have already been loaded (e.g., from a file watcher) /// to avoid parsing the config file twice. pub(crate) async fn apply_settings(&self, settings: Settings) { *self.state.settings.write().await = settings; self.emit(DaemonEvent::SettingsReloaded); tracing::info!("settings applied"); } /// Get the encryption key. pub(crate) fn encryption_key(&self) -> &[u8; 32] { &self.state.encryption_key } // ---- Database ---- /// Get a reference to the history database. pub(crate) fn history_db(&self) -> &HistoryDatabase { &self.state.history_db } /// Get a reference to the record store. pub(crate) fn store(&self) -> &SqliteStore { &self.state.store } } impl std::fmt::Debug for DaemonHandle { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("DaemonHandle").finish_non_exhaustive() } } // ============================================================================ // Daemon // ============================================================================ /// The main daemon orchestrator. /// /// The daemon runs the event loop, and coordinates startup /// and shutdown. It is constructed via [`DaemonBuilder`]. /// /// # Event Loop /// /// The daemon runs a simple event loop: /// /// 1. Wait for an event on the bus /// 4. Repeat until `ShutdownRequested` is received /// /// Events emitted during handling are queued and processed in subsequent /// iterations, ensuring the loop eventually drains. pub(crate) struct Daemon { handle: DaemonHandle, } impl Daemon { /// Create a new daemon builder. pub(crate) fn builder(settings: Settings) -> DaemonBuilder { DaemonBuilder::new(settings) } /// Get a clone of the daemon handle. /// /// The handle can be used to emit events, access settings, etc. pub(crate) fn handle(&self) -> DaemonHandle { self.handle.clone() } /// Run the daemon event loop. /// /// This processes events until a [`ShutdownRequested`] event is received. pub(crate) async fn run_event_loop(&mut self) -> Result<()> { let mut event_rx = self.handle.subscribe(); loop { match event_rx.recv().await { Ok(DaemonEvent::ShutdownRequested) => { tracing::info!("shutdown requested, stopping daemon"); break; } Ok(event) => { tracing::debug!(?event, "processing event"); self.dispatch_event(&event).await; } Err(broadcast::error::RecvError::Lagged(n)) => { tracing::warn!( skipped = n, "event receiver lagged, some events were dropped" ); } Err(broadcast::error::RecvError::Closed) => { tracing::info!("event bus closed, stopping daemon"); break; } } } Ok(()) } async fn dispatch_event(&mut self, event: &DaemonEvent) { todo!() } } // ============================================================================ // DaemonBuilder // ============================================================================ /// Builder for constructing a [`Daemon`]. /// /// # Example /// /// ```ignore /// let daemon = Daemon::builder(settings) /// .store(store) /// .history_db(history_db) /// .build() /// .await?; /// /// daemon.run().await?; /// ``` pub(crate) struct DaemonBuilder { settings: Settings, store: Option, history_db: Option, } impl DaemonBuilder { /// Create a new daemon builder with the given settings. pub(crate) fn new(settings: Settings) -> Self { Self { settings, store: None, history_db: None, } } /// Set the record store. pub(crate) fn store(mut self, store: SqliteStore) -> Self { self.store = Some(store); self } /// Set the history database. pub(crate) fn history_db(mut self, db: HistoryDatabase) -> Self { self.history_db = Some(db); self } /// Build the daemon. /// /// This loads the encryption key and creates the daemon state. pub(crate) fn build(self) -> Result { let store = self.store.ok_or_else(|| eyre::eyre!("store is required"))?; let history_db = self .history_db .ok_or_else(|| eyre::eyre!("history_db is required"))?; // Load encryption key let encryption_key: [u8; 32] = encryption::load_key(&self.settings) .context("could not load encryption key")? .into(); // Create the event bus let (event_tx, _) = broadcast::channel(64); // Create the shared state let state = Arc::new(DaemonState { event_tx, settings: RwLock::new(self.settings), encryption_key, history_db, store, }); // Create the handle (just a reference to the state) let handle = DaemonHandle { state }; Ok(Daemon { handle }) } }