aboutsummaryrefslogtreecommitdiffstats
path: root/crates/daemon/src/daemon.rs
diff options
context:
space:
mode:
Diffstat (limited to 'crates/daemon/src/daemon.rs')
-rw-r--r--crates/daemon/src/daemon.rs290
1 files changed, 290 insertions, 0 deletions
diff --git a/crates/daemon/src/daemon.rs b/crates/daemon/src/daemon.rs
new file mode 100644
index 00000000..70e65c1e
--- /dev/null
+++ b/crates/daemon/src/daemon.rs
@@ -0,0 +1,290 @@
+//! 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<DaemonEvent>,
+
+ // Configuration (mutable - can be reloaded)
+ settings: RwLock<Settings>,
+
+ // 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<DaemonState>,
+}
+
+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}");
+ }
+ }
+ pub(crate) async fn wait_for(&self, matches: fn(&DaemonEvent) -> bool) -> Result<DaemonEvent> {
+ let mut rx = self.subscribe();
+ loop {
+ match rx.recv().await {
+ Ok(e) if matches(&e) => {
+ return Ok(e);
+ }
+ Err(err) => {
+ return Err(err).context("while waiting for events");
+ }
+ Ok(_) => (),
+ }
+ }
+ }
+
+ /// Subscribe to the event bus.
+ ///
+ /// Returns a receiver that will receive all events emitted after this call.
+ pub(crate) fn subscribe(&self) -> broadcast::Receiver<DaemonEvent> {
+ 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
+ }
+
+ /// 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.
+ pub(crate) async fn wait_for_shutdown(&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, "event received");
+ }
+ 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(())
+ }
+}
+
+// ============================================================================
+// 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<SqliteStore>,
+ history_db: Option<HistoryDatabase>,
+}
+
+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<Daemon> {
+ 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 })
+ }
+}