diff options
Diffstat (limited to '')
| -rw-r--r-- | crates/daemon/src/aclient/api_client.rs (renamed from crates/turtle/src/atuin_client/api_client.rs) | 34 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/database/mod.rs | 214 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/encryption.rs (renamed from crates/turtle/src/atuin_client/encryption.rs) | 10 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/history/mod.rs (renamed from crates/turtle/src/atuin_client/history.rs) | 400 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/history/store.rs (renamed from crates/turtle/src/atuin_client/history/store.rs) | 239 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/meta.rs (renamed from crates/turtle/src/atuin_client/meta.rs) | 12 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/mod.rs (renamed from crates/turtle/src/atuin_client/mod.rs) | 7 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/record/encryption.rs (renamed from crates/turtle/src/atuin_client/record/encryption.rs) | 14 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/record/mod.rs (renamed from crates/turtle/src/atuin_client/record/mod.rs) | 0 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/record/sqlite_store.rs (renamed from crates/turtle/src/atuin_client/record/sqlite_store.rs) | 157 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/record/sync.rs (renamed from crates/turtle/src/atuin_client/record/sync.rs) | 42 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/settings/meta.rs (renamed from crates/turtle/src/atuin_client/settings/meta.rs) | 4 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/settings/mod.rs | 590 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/utils.rs (renamed from crates/turtle/src/atuin_client/utils.rs) | 17 | ||||
| -rw-r--r-- | crates/daemon/src/api/control.rs (renamed from crates/turtle/src/atuin_daemon/components/sync.rs) | 186 | ||||
| -rw-r--r-- | crates/daemon/src/api/history.rs | 243 | ||||
| -rw-r--r-- | crates/daemon/src/api/mod.rs | 2 | ||||
| -rw-r--r-- | crates/daemon/src/daemon.rs (renamed from crates/turtle/src/atuin_daemon/daemon.rs) | 190 | ||||
| -rw-r--r-- | crates/daemon/src/events.rs (renamed from crates/turtle/src/atuin_daemon/events.rs) | 37 | ||||
| -rw-r--r-- | crates/daemon/src/main.rs | 180 | ||||
| -rw-r--r-- | crates/daemon/src/server.rs | 109 |
21 files changed, 1621 insertions, 1066 deletions
diff --git a/crates/turtle/src/atuin_client/api_client.rs b/crates/daemon/src/aclient/api_client.rs index bd5bf59e..1eba51bd 100644 --- a/crates/turtle/src/atuin_client/api_client.rs +++ b/crates/daemon/src/aclient/api_client.rs @@ -6,11 +6,10 @@ use reqwest::{Response, StatusCode, Url, header::HeaderMap}; use tracing::debug; use uuid::Uuid; -use crate::atuin_common::{api::ErrorResponse, record::RecordStatus}; -use crate::atuin_common::{ +use turtle_common::{api::ErrorResponse, record::RecordStatus}; +use turtle_common::{ api::{ATUIN_CARGO_VERSION, ATUIN_HEADER_VERSION, ATUIN_VERSION}, record::{EncryptedData, HostId, Record, RecordIdx}, - tls::ensure_crypto_provider, }; use semver::Version; @@ -40,7 +39,7 @@ fn make_url(address: &str, path: &str, user_id: Uuid) -> Result<String> { Ok(url.to_string()) } -pub(crate) fn ensure_version(response: &Response) -> Result<bool> { +fn ensure_version(response: &Response) -> Result<bool> { let version = response.headers().get(ATUIN_HEADER_VERSION); let version = if let Some(version) = version { @@ -101,6 +100,22 @@ async fn handle_resp_error(resp: Response) -> Result<Response> { Ok(resp) } +use std::sync::Once; + +static INIT: Once = Once::new(); + +/// Ensure the rustls crypto provider (ring) is installed. +/// +/// Must be called before creating any reqwest clients. Safe to call +/// multiple times — only the first call installs the provider. +fn ensure_crypto_provider() { + INIT.call_once(|| { + rustls::crypto::ring::default_provider() + .install_default() + .expect("Failed to install rustls crypto provider"); + }); +} + impl<'a> Client<'a> { pub(crate) fn new( sync_addr: &'a str, @@ -126,17 +141,6 @@ impl<'a> Client<'a> { }) } - pub(crate) async fn delete_store(&self) -> Result<()> { - let url = make_url(self.sync_addr, "/store", self.user_id)?; - let url = Url::parse(url.as_str())?; - - let resp = self.inner.delete(url).send().await?; - - handle_resp_error(resp).await?; - - Ok(()) - } - pub(crate) async fn post_records(&self, records: &[Record<EncryptedData>]) -> Result<()> { let url = make_url(self.sync_addr, "/record", self.user_id)?; let url = Url::parse(url.as_str())?; diff --git a/crates/daemon/src/aclient/database/mod.rs b/crates/daemon/src/aclient/database/mod.rs new file mode 100644 index 00000000..7cee866e --- /dev/null +++ b/crates/daemon/src/aclient/database/mod.rs @@ -0,0 +1,214 @@ +use std::{path::Path, str::FromStr, time::Duration}; + +use fs_err::{self as fs}; +use sql_builder::{SqlBuilder, SqlName}; +use sqlx::{ + AssertSqlSafe, Result, Row, + sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePool, SqliteRow, SqliteSynchronous}, +}; +use time::OffsetDateTime; +use tracing::debug; +use turtle_api::history::{History, HistoryId}; +use turtle_common::utils; + +use crate::aclient::utils::setup_db; + +// Intended for use on a developer machine and not a sync server. +// TODO: implement IntoIterator +#[derive(Debug, Clone)] +pub(crate) struct ClientSqlite { + pool: SqlitePool, +} + +impl ClientSqlite { + pub(crate) async fn new(path: impl AsRef<Path>, timeout: f64) -> Result<Self> { + fn mk_opts(path: &str) -> Result<SqliteConnectOptions> { + let opts = SqliteConnectOptions::from_str(path)? + .journal_mode(SqliteJournalMode::Wal) + .optimize_on_close(true, None) + .synchronous(SqliteSynchronous::Normal) + .with_regexp() + .create_if_missing(true); + + Ok(opts) + } + + let path = path.as_ref(); + debug!("opening sqlite database at {path:?}"); + + if utils::broken_symlink(path) { + eprintln!( + "Atuin: Sqlite db path ({}) is a broken symlink. Unable to read or create replacement.", + path.display() + ); + std::process::exit(1); + } + + if !path.exists() + && let Some(dir) = path.parent() + { + fs::create_dir_all(dir)?; + } + + let pool = setup_db!(path, timeout, mk_opts, "./db/client-migrations").await?; + Ok(Self { pool }) + } + + async fn save_raw(tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, h: &History) -> Result<()> { + sqlx::query( + " + INSERT OR IGNORE + INTO history (id, timestamp, duration, exit, command, cwd, session, hostname, author, intent, deleted_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) + ", + ) + .bind(h.id.to_string().as_str()) + .bind(h.timestamp.unix_timestamp_nanos() as i64) + .bind(h.duration.as_nanos() as i64) + .bind(h.exit) + .bind(h.command.as_str()) + .bind(h.cwd.as_str()) + .bind(h.session.as_str()) + .bind(h.hostname.as_str()) + .bind(h.author.as_str()) + .bind(h.intent.as_deref()) + .bind(h.deleted_at.map(|t|t.unix_timestamp_nanos() as i64)) + .execute(&mut **tx) + .await?; + + Ok(()) + } + + async fn delete_row_raw( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + id: HistoryId, + ) -> Result<()> { + sqlx::query("delete from history where id = ?1") + .bind(id.to_string().as_str()) + .execute(&mut **tx) + .await?; + + Ok(()) + } + + #[expect(clippy::needless_pass_by_value)] + fn query_history_inner(row: SqliteRow) -> History { + let deleted_at: Option<i64> = row.get("deleted_at"); + let hostname: String = row.get("hostname"); + let author: Option<String> = row.try_get("author").ok().flatten(); + let author = author + .filter(|author| !author.trim().is_empty()) + .unwrap_or_else(|| History::author_from_hostname(hostname.as_str())); + let intent: Option<String> = row.try_get("intent").ok().flatten(); + let intent = intent.filter(|intent| !intent.trim().is_empty()); + + History::from_db() + .id(row.get("id")) + .timestamp( + OffsetDateTime::from_unix_timestamp_nanos(i128::from( + row.get::<i64, _>("timestamp"), + )) + .unwrap(), + ) + .duration(Duration::from_nanos( + u64::try_from(row.get::<i64, _>("duration")).expect("to be small enough"), + )) + .exit(row.get("exit")) + .command(row.get("command")) + .cwd(row.get("cwd")) + .session(row.get("session")) + .hostname(hostname) + .author(author) + .intent(intent) + .deleted_at( + deleted_at + .and_then(|t| OffsetDateTime::from_unix_timestamp_nanos(i128::from(t)).ok()), + ) + .build() + .into() + } +} + +impl ClientSqlite { + pub(crate) async fn save(&self, h: &History) -> Result<()> { + debug!("saving history to sqlite"); + let mut tx = self.pool.begin().await?; + Self::save_raw(&mut tx, h).await?; + tx.commit().await?; + + Ok(()) + } + + /// make a unique list, that only shows the *newest* version of things + pub(crate) async fn list( + &self, + max: Option<usize>, + unique: bool, + include_deleted: bool, + ) -> Result<Vec<History>> { + debug!("listing history"); + + let mut query = SqlBuilder::select_from(SqlName::new("history").alias("h").baquoted()); + query.field("*").order_desc("timestamp"); + if !include_deleted { + query.and_where_is_null("deleted_at"); + } + + if unique { + query.group_by("command").having("max(timestamp)"); + } + + if let Some(max) = max { + let max: usize = max; + query.limit(max); + } + + let query = query.sql().expect("bug in list query. please report"); + + // SAFETY: + // - The query is constructed via sql_bulider, and as such should be safe. + // - The only value, that is directly added to the query is a `usize`. + let res = sqlx::query(AssertSqlSafe(query)) + .map(Self::query_history_inner) + .fetch_all(&self.pool) + .await?; + + Ok(res) + } + + pub(crate) async fn range( + &self, + from: OffsetDateTime, + to: OffsetDateTime, + ) -> Result<Vec<History>> { + debug!("listing history from {:?} to {:?}", from, to); + + let res = sqlx::query( + " + SELECT * + FROM history + WHERE timestamp >= ?1 AND timestamp <= ?2 + ORDER BY timestamp ASC + ", + ) + .bind(from.unix_timestamp_nanos() as i64) + .bind(to.unix_timestamp_nanos() as i64) + .map(Self::query_history_inner) + .fetch_all(&self.pool) + .await?; + + Ok(res) + } + + pub(crate) async fn delete_rows(&self, ids: &[HistoryId]) -> Result<()> { + let mut tx = self.pool.begin().await?; + + for id in ids { + Self::delete_row_raw(&mut tx, id.clone()).await?; + } + + tx.commit().await?; + + Ok(()) + } +} diff --git a/crates/turtle/src/atuin_client/encryption.rs b/crates/daemon/src/aclient/encryption.rs index f1c921cb..45e82ab3 100644 --- a/crates/turtle/src/atuin_client/encryption.rs +++ b/crates/daemon/src/aclient/encryption.rs @@ -11,22 +11,22 @@ use std::io::prelude::Write; use base64::prelude::{BASE64_STANDARD, Engine}; -pub(crate) use crypto_secretbox::Key; +use crypto_secretbox::Key; use crypto_secretbox::{KeyInit, XSalsa20Poly1305, aead::OsRng}; use eyre::{Context, Result, bail, ensure, eyre}; use fs_err as fs; use rmp::Marker; -use crate::atuin_client::settings::Settings; +use crate::aclient::settings::Settings; -pub(crate) fn generate_encoded_key() -> Result<(Key, String)> { +fn generate_encoded_key() -> Result<(Key, String)> { let key = XSalsa20Poly1305::generate_key(&mut OsRng); let encoded = encode_key(&key)?; Ok((key, encoded)) } -pub(crate) fn new_key(settings: &Settings) -> Result<Key> { +fn new_key(settings: &Settings) -> Result<Key> { if settings.sync.encryption_key()?.is_some() { bail!("key already exists! cannot overwrite"); } else if let Some(path) = settings.sync.encryption_key_path.as_ref() { @@ -50,7 +50,7 @@ pub(crate) fn load_key(settings: &Settings) -> Result<Key> { } } -pub(crate) fn encode_key(key: &Key) -> Result<String> { +fn encode_key(key: &Key) -> Result<String> { let mut buf = vec![]; rmp::encode::write_array_len(&mut buf, key.len() as u32) .wrap_err("could not encode key to message pack")?; diff --git a/crates/turtle/src/atuin_client/history.rs b/crates/daemon/src/aclient/history/mod.rs index c38d8ccc..0b5f4982 100644 --- a/crates/turtle/src/atuin_client/history.rs +++ b/crates/daemon/src/aclient/history/mod.rs @@ -1,175 +1,35 @@ -use core::fmt::Formatter; +use std::time::Duration; + use rmp::decode::DecodeStringError; use rmp::decode::ValueReadError; use rmp::{Marker, decode::Bytes}; -use std::env; -use std::fmt::Display; +use turtle_api::history::History; -use crate::atuin_common::record::DecryptedData; -use crate::atuin_common::utils::uuid_v7; +use turtle_common::record::DecryptedData; use eyre::{Result, bail, eyre}; -use crate::atuin_client::secrets::SECRET_PATTERNS_RE; -use crate::atuin_client::settings::Settings; -use crate::atuin_client::utils::get_host_user; use time::OffsetDateTime; -mod builder; pub(crate) mod store; -pub(crate) const HISTORY_VERSION_V0: &str = "v0"; -pub(crate) const HISTORY_VERSION_V1: &str = "v1"; +const HISTORY_VERSION_V0: &str = "v0"; +const HISTORY_VERSION_V1: &str = "v1"; const HISTORY_RECORD_VERSION_V0: u16 = 0; const HISTORY_RECORD_VERSION_V1: u16 = 1; -pub(crate) const HISTORY_VERSION: &str = HISTORY_VERSION_V1; -pub(crate) const HISTORY_TAG: &str = "history"; -const HISTORY_AUTHOR_ENV: &str = "ATUIN_HISTORY_AUTHOR"; -const HISTORY_INTENT_ENV: &str = "ATUIN_HISTORY_INTENT"; - -#[derive(Clone, Debug, Eq, PartialEq, Hash)] -pub(crate) struct HistoryId(pub(crate) String); - -impl Display for HistoryId { - fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", self.0) - } -} - -impl From<String> for HistoryId { - fn from(s: String) -> Self { - Self(s) - } -} - -/// Client-side history entry. -/// -/// Client stores data unencrypted, and only encrypts it before sending to the server. -/// -/// To create a new history entry, use one of the builders: -/// - [`History::import()`] to import an entry from the shell history file -/// - [`History::capture()`] to capture an entry via hook -/// - [`History::from_db()`] to create an instance from the database entry -// -// ## Implementation Notes -// -// New fields must be added to `History::{serialize,deserialize}` in a backwards -// compatible way (sensible defaults and careful `nfields` handling). -#[derive(Debug, Clone, PartialEq, Eq, sqlx::FromRow)] -pub(crate) struct History { - /// A client-generated ID, used to identify the entry when syncing. - /// - /// Stored as `client_id` in the database. - pub(crate) id: HistoryId, - - /// When the command was run. - pub(crate) timestamp: OffsetDateTime, - - /// How long the command took to run. - pub(crate) duration: i64, - - /// The exit code of the command. - pub(crate) exit: i64, - - /// The command that was run. - pub(crate) command: String, - - /// The current working directory when the command was run. - pub(crate) cwd: String, - - /// The session ID, associated with a terminal session. - pub(crate) session: String, - - /// The hostname of the machine the command was run on. - pub(crate) hostname: String, - - /// Who wrote this command (human user or automation/agent identity). - pub(crate) author: String, +const HISTORY_VERSION: &str = HISTORY_VERSION_V1; +const HISTORY_TAG: &str = "history"; - /// Optional rationale for why the command was executed. - pub(crate) intent: Option<String>, - - /// Timestamp, which is set when the entry is deleted, allowing a soft delete. - pub(crate) deleted_at: Option<OffsetDateTime>, -} - -#[derive(Debug, Clone, PartialEq, Eq, sqlx::FromRow)] -pub(crate) struct HistoryStats { - /// The command that was ran after this one in the session - pub(crate) next: Option<History>, - - /// The command that was ran before this one in the session - pub(crate) previous: Option<History>, - - /// How many times has this command been ran? - pub(crate) total: u64, - - pub(crate) average_duration: u64, - - pub(crate) exits: Vec<(i64, i64)>, - - pub(crate) day_of_week: Vec<(String, i64)>, - - pub(crate) duration_over_time: Vec<(String, i64)>, +trait HistoryExt: Sized { + fn serialize(&self) -> Result<DecryptedData>; + fn read_optional_string(bytes: &[u8]) -> Result<(Option<String>, &[u8])>; + fn deserialize_v0(bytes: &[u8]) -> Result<Self>; + fn deserialize_v1(bytes: &[u8]) -> Result<Self>; + fn deserialize(bytes: &[u8], version: &str) -> Result<Self>; } -impl History { - pub(crate) fn author_from_hostname(hostname: &str) -> String { - hostname - .split_once(':') - .map_or_else(|| hostname.to_owned(), |(_, user)| user.to_owned()) - } - - fn normalize_optional_field(field: Option<String>) -> Option<String> { - field.and_then(|value| { - let trimmed = value.trim(); - if trimmed.is_empty() { - None - } else { - Some(trimmed.to_owned()) - } - }) - } - - #[expect(clippy::too_many_arguments)] - fn new( - timestamp: OffsetDateTime, - command: String, - cwd: String, - exit: i64, - duration: i64, - session: Option<String>, - hostname: Option<String>, - author: Option<String>, - intent: Option<String>, - deleted_at: Option<OffsetDateTime>, - ) -> Self { - let session = session - .or_else(|| env::var("ATUIN_SESSION").ok()) - .unwrap_or_else(|| uuid_v7().as_simple().to_string()); - let hostname = hostname.unwrap_or_else(get_host_user); - let author = Self::normalize_optional_field(author) - .or_else(|| Self::normalize_optional_field(env::var(HISTORY_AUTHOR_ENV).ok())) - .unwrap_or_else(|| Self::author_from_hostname(hostname.as_str())); - let intent = Self::normalize_optional_field(intent) - .or_else(|| Self::normalize_optional_field(env::var(HISTORY_INTENT_ENV).ok())); - - Self { - id: uuid_v7().as_simple().to_string().into(), - timestamp, - command, - cwd, - exit, - duration, - session, - hostname, - author, - intent, - deleted_at, - } - } - - pub(crate) fn serialize(&self) -> Result<DecryptedData> { +impl HistoryExt for History { + fn serialize(&self) -> Result<DecryptedData> { // This is pretty much the same as what we used for the old history, with one difference - // it uses integers for timestamps rather than a string format. @@ -182,9 +42,12 @@ impl History { let include_intent = self.intent.is_some(); encode::write_array_len(&mut output, 10 + u32::from(include_intent))?; - encode::write_str(&mut output, &self.id.0)?; + encode::write_str(&mut output, &self.id.to_string())?; encode::write_u64(&mut output, self.timestamp.unix_timestamp_nanos() as u64)?; - encode::write_sint(&mut output, self.duration)?; + encode::write_sint( + &mut output, + i64::try_from(self.duration.as_nanos()).expect("should be small enough"), + )?; encode::write_sint(&mut output, self.exit)?; encode::write_str(&mut output, &self.command)?; encode::write_str(&mut output, &self.cwd)?; @@ -249,7 +112,11 @@ impl History { let mut bytes = Bytes::new(bytes); let timestamp = decode::read_u64(&mut bytes).map_err(error_report)?; - let duration = decode::read_int(&mut bytes).map_err(error_report)?; + let duration = decode::read_int(&mut bytes) + .map(|int: i64| { + Duration::from_nanos(u64::try_from(int).expect("should be small enough")) + }) + .map_err(error_report)?; let exit = decode::read_int(&mut bytes).map_err(error_report)?; let bytes = bytes.remaining_slice(); @@ -313,7 +180,16 @@ impl History { let mut bytes = Bytes::new(bytes); let timestamp = decode::read_u64(&mut bytes).map_err(error_report)?; - let duration = decode::read_int(&mut bytes).map_err(error_report)?; + let duration = decode::read_int(&mut bytes) + .map(|int: i64| { + if int.is_negative() { + // TODO: We should probably handle this case differently <2026-08-23> + Duration::from_nanos(0) + } else { + Duration::from_nanos(u64::try_from(int).expect("to be small enough")) + } + }) + .map_err(error_report)?; let exit = decode::read_int(&mut bytes).map_err(error_report)?; let bytes = bytes.remaining_slice(); @@ -358,7 +234,7 @@ impl History { }) } - pub(crate) fn deserialize(bytes: &[u8], version: &str) -> Result<Self> { + fn deserialize(bytes: &[u8], version: &str) -> Result<Self> { match version { HISTORY_VERSION_V0 => Self::deserialize_v0(bytes), HISTORY_VERSION_V1 => Self::deserialize_v1(bytes), @@ -366,212 +242,24 @@ impl History { _ => bail!("unknown version {version:?}"), } } - - /// Builder for a history entry that is captured via hook. - /// - /// This builder is used only at the `start` step of the hook, - /// so it doesn't have any fields which are known only after - /// the command is finished, such as `exit` or `duration`. - /// - /// ## Examples - /// ```rust - /// use crate::atuin_client::history::History; - /// - /// let history: History = History::capture() - /// .timestamp(time::OffsetDateTime::now_utc()) - /// .command("ls -la") - /// .cwd("/home/user") - /// .build() - /// .into(); - /// ``` - /// - /// Command without any required info cannot be captured, which is forced at compile time: - /// - /// ```compile_fail - /// use crate::atuin_client::history::History; - /// - /// // this will not compile because `cwd` is missing - /// let history: History = History::capture() - /// .timestamp(time::OffsetDateTime::now_utc()) - /// .command("ls -la") - /// .build() - /// .into(); - /// ``` - pub(crate) fn capture() -> builder::HistoryCapturedBuilder { - builder::HistoryCaptured::builder() - } - - /// Builder for a history entry that is captured via hook, and sent to the daemon. - /// - /// This builder is used only at the `start` step of the hook, - /// so it doesn't have any fields which are known only after - /// the command is finished, such as `exit` or `duration`. - /// - /// It does, however, include information that can usually be inferred. - /// - /// This is because the daemon we are sending a request to lacks the context of the command - /// - /// ## Examples - /// ```rust - /// use crate::atuin_client::history::History; - /// - /// let history: History = History::daemon() - /// .timestamp(time::OffsetDateTime::now_utc()) - /// .command("ls -la") - /// .cwd("/home/user") - /// .session("018deb6e8287781f9973ef40e0fde76b") - /// .hostname("computer:ellie") - /// .build() - /// .into(); - /// ``` - /// - /// Command without any required info cannot be captured, which is forced at compile time: - /// - /// ```compile_fail - /// use crate::atuin_client::history::History; - /// - /// // this will not compile because `hostname` is missing - /// let history: History = History::daemon() - /// .timestamp(time::OffsetDateTime::now_utc()) - /// .command("ls -la") - /// .cwd("/home/user") - /// .session("018deb6e8287781f9973ef40e0fde76b") - /// .build() - /// .into(); - /// ``` - pub(crate) fn daemon() -> builder::HistoryDaemonCaptureBuilder { - builder::HistoryDaemonCapture::builder() - } - - /// Builder for a history entry that is imported from the database. - /// - /// All fields are required, as they are all present in the database. - /// - /// ```compile_fail - /// use crate::atuin_client::history::History; - /// - /// // this will not compile because `id` field is missing - /// let history: History = History::from_db() - /// .timestamp(time::OffsetDateTime::now_utc()) - /// .command("ls -la".to_string()) - /// .cwd("/home/user".to_string()) - /// .exit(0) - /// .duration(100) - /// .session("somesession".to_string()) - /// .hostname("localhost".to_string()) - /// .author("user".to_string()) - /// .intent(None) - /// .deleted_at(None) - /// .build() - /// .into(); - /// ``` - pub(crate) fn from_db() -> builder::HistoryFromDbBuilder { - builder::HistoryFromDb::builder() - } - - pub(crate) fn success(&self) -> bool { - self.exit == 0 || self.duration == -1 - } - - pub(crate) fn should_save(&self, settings: &Settings) -> bool { - !(self.command.is_empty() - || settings.history_filter.is_match(&self.command) - || settings.cwd_filter.is_match(&self.cwd) - || (settings.secrets_filter && SECRET_PATTERNS_RE.is_match(&self.command))) - } } #[cfg(test)] mod tests { - use regex::RegexSet; + use std::time::Duration; + use time::macros::datetime; - use crate::atuin_client::{history::HISTORY_VERSION, settings::Settings}; + use crate::aclient::history::{HISTORY_VERSION, HistoryExt}; use super::History; - // Test that we don't save history where necessary - #[test] - fn privacy_test() { - let settings = Settings { - cwd_filter: RegexSet::new(["^/supasecret"]).unwrap(), - history_filter: RegexSet::new(["^psql"]).unwrap(), - ..Settings::default() - }; - - let normal_command: History = History::capture() - .timestamp(time::OffsetDateTime::now_utc()) - .command("echo foo") - .cwd("/") - .build() - .into(); - - let with_space: History = History::capture() - .timestamp(time::OffsetDateTime::now_utc()) - .command(" echo bar") - .cwd("/") - .build() - .into(); - - let empty: History = History::capture() - .timestamp(time::OffsetDateTime::now_utc()) - .command("") - .cwd("/") - .build() - .into(); - - let stripe_key: History = History::capture() - .timestamp(time::OffsetDateTime::now_utc()) - .command("curl foo.com/bar?key=sk_test_1234567890abcdefghijklmnop") - .cwd("/") - .build() - .into(); - - let secret_dir: History = History::capture() - .timestamp(time::OffsetDateTime::now_utc()) - .command("echo ohno") - .cwd("/supasecret") - .build() - .into(); - - let with_psql: History = History::capture() - .timestamp(time::OffsetDateTime::now_utc()) - .command("psql") - .cwd("/supasecret") - .build() - .into(); - - assert!(normal_command.should_save(&settings)); - assert!(!with_space.should_save(&settings)); - assert!(!empty.should_save(&settings)); - assert!(!stripe_key.should_save(&settings)); - assert!(!secret_dir.should_save(&settings)); - assert!(!with_psql.should_save(&settings)); - } - - #[test] - fn disable_secrets() { - let settings = Settings { - secrets_filter: false, - ..Settings::new().unwrap() - }; - - let stripe_key: History = History::capture() - .timestamp(time::OffsetDateTime::now_utc()) - .command("curl foo.com/bar?key=sk_test_1234567890abcdefghijklmnop") - .cwd("/") - .build() - .into(); - - assert!(stripe_key.should_save(&settings)); - } - #[test] fn test_serialize_deserialize() { let history = History { id: "66d16cbee7cd47538e5c5b8b44e9006e".to_owned().into(), timestamp: datetime!(2023-05-28 18:35:40.633872 +00:00), - duration: 49_206_000, + duration: Duration::from_nanos(49_206_000), exit: 0, command: "git status".to_owned(), cwd: "/Users/conrad.ludgate/Documents/code/atuin".to_owned(), @@ -599,7 +287,7 @@ mod tests { let history = History { id: "66d16cbee7cd47538e5c5b8b44e9006e".to_owned().into(), timestamp: datetime!(2023-05-28 18:35:40.633872 +00:00), - duration: 49_206_000, + duration: Duration::from_nanos(49_206_000), exit: 0, command: "git status".to_owned(), cwd: "/Users/conrad.ludgate/Documents/code/atuin".to_owned(), @@ -623,7 +311,7 @@ mod tests { let history = History { id: "66d16cbee7cd47538e5c5b8b44e9006e".to_owned().into(), timestamp: datetime!(2023-05-28 18:35:40.633872 +00:00), - duration: 49_206_000, + duration: Duration::from_nanos(49_206_000), exit: 0, command: "git status".to_owned(), cwd: "/Users/conrad.ludgate/Documents/code/atuin".to_owned(), @@ -665,7 +353,7 @@ mod tests { let current = History { id: "66d16cbee7cd47538e5c5b8b44e9006e".to_owned().into(), timestamp: datetime!(2023-05-28 18:35:40.633872 +00:00), - duration: 49_206_000, + duration: Duration::from_nanos(49_206_000), exit: 0, command: "git status".to_owned(), cwd: "/Users/conrad.ludgate/Documents/code/atuin".to_owned(), diff --git a/crates/turtle/src/atuin_client/history/store.rs b/crates/daemon/src/aclient/history/store.rs index 9c7771cc..a6d6a627 100644 --- a/crates/turtle/src/atuin_client/history/store.rs +++ b/crates/daemon/src/aclient/history/store.rs @@ -1,27 +1,25 @@ -use std::{collections::HashSet, fmt::Write, time::Duration}; - use eyre::{Result, bail, eyre}; -use indicatif::{ProgressBar, ProgressState, ProgressStyle}; use rmp::decode::Bytes; -use tracing::debug; +use turtle_api::history::{History, HistoryId}; -use crate::atuin_client::{ - database::{ClientSqlite, current_context}, +use crate::aclient::{ + database::ClientSqlite, + history::HistoryExt, record::{encryption::PASETO_V4, sqlite_store::SqliteStore}, }; -use crate::atuin_common::record::{DecryptedData, Host, HostId, Record, RecordId, RecordIdx}; +use turtle_common::record::{DecryptedData, Host, HostId, Record, RecordId, RecordIdx}; -use super::{HISTORY_TAG, HISTORY_VERSION, HISTORY_VERSION_V0, History, HistoryId}; +use super::{HISTORY_TAG, HISTORY_VERSION, HISTORY_VERSION_V0}; #[derive(Debug, Clone)] pub(crate) struct HistoryStore { - pub(crate) store: SqliteStore, - pub(crate) host_id: HostId, - pub(crate) encryption_key: [u8; 32], + store: SqliteStore, + host_id: HostId, + encryption_key: [u8; 32], } #[derive(Debug, Eq, PartialEq, Clone)] -pub(crate) enum HistoryRecord { +enum HistoryRecord { Create(History), // Create a history record Delete(HistoryId), // Delete a history record, identified by ID } @@ -39,7 +37,7 @@ impl HistoryRecord { /// twice. /// /// Deletion simply refers to the history by ID - pub(crate) fn serialize(&self) -> Result<DecryptedData> { + fn serialize(&self) -> Result<DecryptedData> { // probably don't actually need to use rmp here, but if we ever need to extend it, it's a // nice wrapper around raw byte stuff use rmp::encode; @@ -58,14 +56,14 @@ impl HistoryRecord { Self::Delete(id) => { // 1 -> a history delete encode::write_u8(&mut output, 1)?; - encode::write_str(&mut output, id.0.as_str())?; + encode::write_str(&mut output, id.to_string().as_str())?; } } Ok(DecryptedData(output)) } - pub(crate) fn deserialize(bytes: &DecryptedData, version: &str) -> Result<Self> { + fn deserialize(bytes: &DecryptedData, version: &str) -> Result<Self> { use rmp::decode; fn error_report<E: std::fmt::Debug>(err: E) -> eyre::Report { @@ -143,58 +141,6 @@ impl HistoryStore { Ok((id, idx)) } - async fn push_batch(&self, records: impl Iterator<Item = HistoryRecord>) -> Result<()> { - let mut ret = Vec::new(); - - let idx = self - .store - .last(self.host_id, HISTORY_TAG) - .await? - .map_or(0, |p| p.idx + 1); - - // Could probably _also_ do this as an iterator, but let's see how this is for now. - // optimizing for minimal sqlite transactions, this code can be optimised later - for (n, record) in records.enumerate() { - let bytes = record.serialize()?; - - let record = Record::builder() - .host(Host::new(self.host_id)) - .version(HISTORY_VERSION.to_string()) - .tag(HISTORY_TAG.to_string()) - .idx(idx + n as u64) - .data(bytes) - .build(); - - let record = record.encrypt::<PASETO_V4>(&self.encryption_key); - - ret.push(record); - } - - self.store.push_batch(ret.iter()).await?; - - Ok(()) - } - - pub(crate) async fn delete(&self, id: HistoryId) -> Result<(RecordId, RecordIdx)> { - let record = HistoryRecord::Delete(id); - - self.push_record(record).await - } - - /// Delete a batch of history entries via the record store. - /// Returns the record IDs so the caller can run `incremental_build` when ready. - pub(crate) async fn delete_entries( - &self, - entries: impl IntoIterator<Item = History>, - ) -> Result<Vec<RecordId>> { - let mut record_ids = Vec::new(); - for entry in entries { - let (id, _) = self.delete(entry.id).await?; - record_ids.push(id); - } - Ok(record_ids) - } - pub(crate) async fn push(&self, history: History) -> Result<(RecordId, RecordIdx)> { // TODO(ellie): move the history store to its own file // it's tiny rn so fine as is @@ -203,60 +149,37 @@ impl HistoryStore { self.push_record(record).await } - pub(crate) async fn history(&self) -> Result<Vec<HistoryRecord>> { - // Atm this loads all history into memory - // Not ideal as that is potentially quite a lot, although history will be small. - let records = self.store.all_tagged(HISTORY_TAG).await?; - let mut ret = Vec::with_capacity(records.len()); - - for record in records { - let hist = match record.version.as_str() { - HISTORY_VERSION_V0 | HISTORY_VERSION => { - let version = record.version.clone(); - let decrypted = record.decrypt::<PASETO_V4>(&self.encryption_key)?; - - HistoryRecord::deserialize(&decrypted.data, version.as_str()) - } - version => bail!("unknown history version {version:?}"), - }?; - - ret.push(hist); - } - - Ok(ret) - } - - pub(crate) async fn build(&self, database: &ClientSqlite) -> Result<()> { - // I'd like to change how we rebuild and not couple this with the database, but need to - // consider the structure more deeply. This will be easy to change. - - // TODO(ellie): page or iterate this - let history = self.history().await?; - - // In theory we could flatten this here - // The current issue is that the database may have history in it already, from the old sync - // This didn't actually delete old history - // If we're sure we have a DB only maintained by the new store, we can flatten - // create/delete before we even get to sqlite - let mut creates = Vec::new(); - let mut deletes = Vec::new(); - - for i in history { - match i { - HistoryRecord::Create(h) => { - creates.push(h); - } - HistoryRecord::Delete(id) => { - deletes.push(id); - } - } - } - - database.save_bulk(&creates).await?; - database.delete_rows(&deletes).await?; - - Ok(()) - } + // async fn build(&self, database: &ClientSqlite) -> Result<()> { + // // I'd like to change how we rebuild and not couple this with the database, but need to + // // consider the structure more deeply. This will be easy to change. + // + // // TODO(ellie): page or iterate this + // let history = self.history().await?; + // + // // In theory we could flatten this here + // // The current issue is that the database may have history in it already, from the old sync + // // This didn't actually delete old history + // // If we're sure we have a DB only maintained by the new store, we can flatten + // // create/delete before we even get to sqlite + // let mut creates = Vec::new(); + // let mut deletes = Vec::new(); + // + // for i in history { + // match i { + // HistoryRecord::Create(h) => { + // creates.push(h); + // } + // HistoryRecord::Delete(id) => { + // deletes.push(id); + // } + // } + // } + // + // database.save_bulk(&creates).await?; + // database.delete_rows(&deletes).await?; + // + // Ok(()) + // } pub(crate) async fn incremental_build( &self, @@ -294,80 +217,16 @@ impl HistoryStore { Ok(()) } - - /// Get a list of history IDs that exist in the store - /// Note: This currently involves loading all history into memory. This is not going to be a - /// large amount in absolute terms, but do not all it in a hot loop. - pub(crate) async fn history_ids(&self) -> Result<HashSet<HistoryId>> { - let history = self.history().await?; - - let ret = history - .iter() - .map(|h| match h { - HistoryRecord::Create(h) => h.id.clone(), - HistoryRecord::Delete(id) => id.clone(), - }) - .collect::<HashSet<_>>(); - - Ok(ret) - } - - pub(crate) async fn init_store(&self, db: &ClientSqlite) -> Result<()> { - let pb = ProgressBar::new_spinner(); - pb.set_style( - ProgressStyle::with_template("{spinner:.blue} {msg}") - .unwrap() - .with_key("eta", |state: &ProgressState, w: &mut dyn Write| { - write!(w, "{:.1}s", state.eta().as_secs_f64()).unwrap(); - }) - .progress_chars("#>-"), - ); - pb.enable_steady_tick(Duration::from_millis(500)); - - pb.set_message("Fetching history from old database"); - - let context = current_context().await?; - let history = db.list(&[], &context, None, false, true).await?; - - pb.set_message("Fetching history already in store"); - let store_ids = self.history_ids().await?; - - pb.set_message("Converting old history to new store"); - let mut records = Vec::new(); - - for i in history { - debug!("loaded {}", i.id); - - if store_ids.contains(&i.id) { - debug!("skipping {} - already exists", i.id); - continue; - } - - if i.deleted_at.is_some() { - records.push(HistoryRecord::Delete(i.id)); - } else { - records.push(HistoryRecord::Create(i)); - } - } - - pb.set_message("Writing to db"); - - if !records.is_empty() { - self.push_batch(records.into_iter()).await?; - } - - pb.finish_with_message("Import complete"); - - Ok(()) - } } #[cfg(test)] mod tests { - use crate::atuin_common::record::DecryptedData; + use std::time::Duration; + use time::macros::datetime; + use turtle_common::record::DecryptedData; - use crate::atuin_client::history::{HISTORY_VERSION, store::HistoryRecord}; + use crate::aclient::history::{HISTORY_VERSION, store::HistoryRecord}; use super::History; @@ -387,7 +246,7 @@ mod tests { let history = History { id: "018cd4fe81757cd2aee65cd7861f9c81".to_owned().into(), timestamp: datetime!(2024-01-04 00:00:00.000000 +00:00), - duration: 100, + duration: Duration::from_nanos(100), exit: 0, command: "ls".to_owned(), cwd: "/Users/ellie/src/github.com/atuinsh/atuin".to_owned(), diff --git a/crates/turtle/src/atuin_client/meta.rs b/crates/daemon/src/aclient/meta.rs index 079c9926..ea660745 100644 --- a/crates/turtle/src/atuin_client/meta.rs +++ b/crates/daemon/src/aclient/meta.rs @@ -2,12 +2,12 @@ use std::path::Path; use std::str::FromStr; use std::time::Duration; -use crate::atuin_common::record::HostId; use eyre::{Result, eyre}; use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePool, SqlitePoolOptions}; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use tokio::sync::OnceCell; use tracing::debug; +use turtle_common::record::HostId; use uuid::Uuid; const KEY_HOST_ID: &str = "host_id"; @@ -71,7 +71,7 @@ impl MetaStore { // Generic key-value operations - pub(crate) async fn get(&self, key: &str) -> Result<Option<String>> { + async fn get(&self, key: &str) -> Result<Option<String>> { let row: Option<(String,)> = sqlx::query_as("SELECT value FROM meta WHERE key = ?1") .bind(key) .fetch_optional(&self.pool) @@ -80,7 +80,7 @@ impl MetaStore { Ok(row.map(|r| r.0)) } - pub(crate) async fn set(&self, key: &str, value: &str) -> Result<()> { + async fn set(&self, key: &str, value: &str) -> Result<()> { sqlx::query( " INSERT INTO meta (key, value, updated_at) @@ -108,7 +108,7 @@ impl MetaStore { return Ok(HostId(parsed)); } - let uuid = crate::atuin_common::utils::uuid_v7(); + let uuid = turtle_common::utils::uuid_v7(); self.set(KEY_HOST_ID, uuid.as_simple().to_string().as_ref()) .await?; @@ -154,8 +154,8 @@ mod tests { store.set("foo", "baz").await.unwrap(); assert_eq!(store.get("foo").await.unwrap(), Some("baz".to_string())); - store.delete("foo").await.unwrap(); - assert_eq!(store.get("foo").await.unwrap(), None); + // store.delete("foo").await.unwrap(); + // assert_eq!(store.get("foo").await.unwrap(), None); } #[tokio::test] diff --git a/crates/turtle/src/atuin_client/mod.rs b/crates/daemon/src/aclient/mod.rs index 851dfbdb..2c445945 100644 --- a/crates/turtle/src/atuin_client/mod.rs +++ b/crates/daemon/src/aclient/mod.rs @@ -1,12 +1,9 @@ -pub(crate) mod api_client; pub(crate) mod database; pub(crate) mod encryption; pub(crate) mod history; -pub(crate) mod meta; -pub(crate) mod ordering; pub(crate) mod record; -pub(crate) mod secrets; pub(crate) mod settings; -pub(crate) mod theme; +mod api_client; +mod meta; mod utils; diff --git a/crates/turtle/src/atuin_client/record/encryption.rs b/crates/daemon/src/aclient/record/encryption.rs index 96ab463e..11de96d5 100644 --- a/crates/turtle/src/atuin_client/record/encryption.rs +++ b/crates/daemon/src/aclient/record/encryption.rs @@ -1,13 +1,13 @@ -use crate::atuin_common::record::{ - AdditionalData, DecryptedData, EncryptedData, Encryption, HostId, RecordId, RecordIdx, -}; use base64::{Engine, engine::general_purpose}; use eyre::{Context, Result, ensure}; use rusty_paserk::{Key, KeyId, Local, PieWrappedKey}; use rusty_paseto::core::{ - ImplicitAssertion, Key as DataKey, Local as LocalPurpose, Paseto, PasetoNonce, Payload, V4, + ImplicitAssertion, Key as DataKey, Local as LocalPurpose, Paseto, PasetoNonce, Payload, V4 }; use serde::{Deserialize, Serialize}; +use turtle_common::record::{ + AdditionalData, DecryptedData, EncryptedData, Encryption, HostId, RecordId, RecordIdx, +}; /// Use PASETO V4 Local encryption using the additional data as an implicit assertion. #[expect(non_camel_case_types)] @@ -68,7 +68,6 @@ impl Encryption for PASETO_V4 { // aka content-encryption-key (CEK) let random_key = Key::<V4, Local>::new_os_random(); - // encode the implicit assertions let assertions = Assertions::from(ad).encode(); // build the payload and encrypt the token @@ -145,11 +144,12 @@ impl PASETO_V4 { fn encrypt_cek(cek: Key<V4, Local>, key: &[u8; 32]) -> String { // aka key-encryption-key (KEK) let wrapping_key = Key::<V4, Local>::from_bytes(*key); + let kid = wrapping_key.to_id(); // wrap the random key so we can decrypt it later let wrapped_cek = AtuinFooter { wpk: cek.wrap_pie(&wrapping_key), - kid: wrapping_key.to_id(), + kid, }; serde_json::to_string(&wrapped_cek).expect("could not serialize wrapped cek") } @@ -201,7 +201,7 @@ impl Assertions<'_> { #[cfg(test)] mod tests { - use crate::atuin_common::{ + use turtle_common::{ record::{Host, Record}, utils::uuid_v7, }; diff --git a/crates/turtle/src/atuin_client/record/mod.rs b/crates/daemon/src/aclient/record/mod.rs index 4e5774ea..4e5774ea 100644 --- a/crates/turtle/src/atuin_client/record/mod.rs +++ b/crates/daemon/src/aclient/record/mod.rs diff --git a/crates/turtle/src/atuin_client/record/sqlite_store.rs b/crates/daemon/src/aclient/record/sqlite_store.rs index 18f5c869..0026690b 100644 --- a/crates/turtle/src/atuin_client/record/sqlite_store.rs +++ b/crates/daemon/src/aclient/record/sqlite_store.rs @@ -14,15 +14,13 @@ use sqlx::{ }; use tracing::debug; -use crate::atuin_client::utils::setup_db; -use crate::atuin_common::record::{ +use crate::aclient::utils::setup_db; +use turtle_common::record::{ EncryptedData, Host, HostId, Record, RecordId, RecordIdx, RecordStatus, }; -use crate::atuin_common::utils; +use turtle_common::utils; use uuid::Uuid; -use super::encryption::PASETO_V4; - #[derive(Debug, Clone)] pub(crate) struct SqliteStore { pool: SqlitePool, @@ -110,15 +108,6 @@ impl SqliteStore { }, } } - - async fn load_all(&self) -> Result<Vec<Record<EncryptedData>>> { - let res = sqlx::query("select * from store ") - .map(Self::query_row) - .fetch_all(&self.pool) - .await?; - - Ok(res) - } } /// A record store stores records @@ -157,21 +146,6 @@ impl SqliteStore { Ok(res) } - pub(crate) async fn delete(&self, id: RecordId) -> Result<()> { - sqlx::query("delete from store where id = ?1") - .bind(id.0.as_hyphenated().to_string()) - .execute(&self.pool) - .await?; - - Ok(()) - } - - pub(crate) async fn delete_all(&self) -> Result<()> { - sqlx::query("delete from store").execute(&self.pool).await?; - - Ok(()) - } - pub(crate) async fn last( &self, host: HostId, @@ -192,26 +166,6 @@ impl SqliteStore { } } - pub(crate) async fn first( - &self, - host: HostId, - tag: &str, - ) -> Result<Option<Record<EncryptedData>>> { - self.idx(host, tag, 0).await - } - - pub(crate) async fn len_tag(&self, tag: &str) -> Result<u64> { - let res: Result<(i64,), sqlx::Error> = - sqlx::query_as("select count(*) from store where tag=?1") - .bind(tag) - .fetch_one(&self.pool) - .await; - match res { - Err(e) => Err(eyre!("failed to fetch local store len: {}", e)), - Ok(v) => Ok(v.0 as u64), - } - } - /// Get the next `limit` records, after and including the given index pub(crate) async fn next( &self, @@ -234,28 +188,6 @@ impl SqliteStore { Ok(res) } - /// Get the first record for a given host and tag - pub(crate) async fn idx( - &self, - host: HostId, - tag: &str, - idx: RecordIdx, - ) -> Result<Option<Record<EncryptedData>>> { - let res = sqlx::query("select * from store where idx = ?1 and host = ?2 and tag = ?3") - .bind(idx as i64) - .bind(host.0.as_hyphenated().to_string()) - .bind(tag) - .map(Self::query_row) - .fetch_one(&self.pool) - .await; - - match res { - Err(sqlx::Error::RowNotFound) => Ok(None), - Err(e) => Err(eyre!("an error occurred: {}", e)), - Ok(v) => Ok(Some(v)), - } - } - pub(crate) async fn status(&self) -> Result<RecordStatus> { let mut status = RecordStatus::new(); @@ -279,89 +211,6 @@ impl SqliteStore { Ok(status) } - - /// Get all records for a given tag - pub(crate) async fn all_tagged(&self, tag: &str) -> Result<Vec<Record<EncryptedData>>> { - let res = sqlx::query("select * from store where tag = ?1 order by timestamp asc") - .bind(tag) - .map(Self::query_row) - .fetch_all(&self.pool) - .await?; - - Ok(res) - } - - /// Reencrypt every single item in this store with a new key - /// Be careful - this may mess with sync. - pub(crate) async fn re_encrypt(&self, old_key: &[u8; 32], new_key: &[u8; 32]) -> Result<()> { - // Load all the records - // In memory like some of the other code here - // This will never be called in a hot loop, and only under the following circumstances - // 1. The user has logged into a new account, with a new key. They are unlikely to have a - // lot of data - // 2. The user has encountered some sort of issue, and runs a maintenance command that - // invokes this - let all = self.load_all().await?; - - let re_encrypted = all - .into_iter() - .map(|record| record.re_encrypt::<PASETO_V4>(old_key, new_key)) - .collect::<Result<Vec<_>>>()?; - - // next up, we delete all the old data and reinsert the new stuff - // do it in one transaction, so if anything fails we rollback OK - - let mut tx = self.pool.begin().await?; - - let res = sqlx::query("delete from store").execute(&mut *tx).await?; - - let rows = res.rows_affected(); - debug!("deleted {rows} rows"); - - // don't call push_batch, as it will start its own transaction - // call the underlying save_raw - - for record in re_encrypted { - Self::save_raw(&mut tx, &record).await?; - } - - tx.commit().await?; - - Ok(()) - } - - /// Verify that every record in this store can be decrypted with the current key - /// Someday maybe also check each tag/record can be deserialized, but not for now. - pub(crate) async fn verify(&self, key: &[u8; 32]) -> Result<()> { - let all = self.load_all().await?; - - all.into_iter() - .map(|record| record.decrypt::<PASETO_V4>(key)) - .collect::<Result<Vec<_>>>()?; - - Ok(()) - } - - /// Verify that every record in this store can be decrypted with the current key - /// Someday maybe also check each tag/record can be deserialized, but not for now. - pub(crate) async fn purge(&self, key: &[u8; 32]) -> Result<()> { - let all = self.load_all().await?; - - for record in &all { - if record.clone().decrypt::<PASETO_V4>(key).is_ok() { - continue; - } - - println!( - "Failed to decrypt {}, deleting", - record.id.0.as_hyphenated() - ); - - self.delete(record.id).await?; - } - - Ok(()) - } } #[cfg(test)] diff --git a/crates/turtle/src/atuin_client/record/sync.rs b/crates/daemon/src/aclient/record/sync.rs index 3057bb10..79239b99 100644 --- a/crates/turtle/src/atuin_client/record/sync.rs +++ b/crates/daemon/src/aclient/record/sync.rs @@ -6,11 +6,11 @@ use thiserror::Error; use tracing::error; use super::encryption::PASETO_V4; -use crate::atuin_client::record::sqlite_store::SqliteStore; -use crate::atuin_client::{api_client::Client, settings::Settings}; +use crate::aclient::record::sqlite_store::SqliteStore; +use crate::aclient::{api_client::Client, settings::Settings}; -use crate::atuin_common::record::{Diff, HostId, RecordId, RecordIdx, RecordStatus}; use indicatif::{ProgressBar, ProgressState, ProgressStyle}; +use turtle_common::record::{Diff, HostId, RecordId, RecordIdx, RecordStatus}; #[derive(Error, Debug)] pub(crate) enum SyncError { @@ -23,7 +23,7 @@ pub(crate) enum SyncError { #[error("operational error: {msg:?}")] OperationalError { msg: String }, - #[error("a request to the sync server failed: {msg:?}")] + #[error("a request to the sync server failed: {msg}")] RemoteRequestError { msg: String }, #[error( @@ -36,7 +36,7 @@ pub(crate) enum SyncError { } #[derive(Debug, Eq, PartialEq)] -pub(crate) enum Operation { +enum Operation { // Either upload or download until the states matches the below Upload { local: RecordIdx, @@ -56,7 +56,7 @@ pub(crate) enum Operation { }, } -pub(crate) fn build_client(settings: &Settings) -> Result<Client<'_>, SyncError> { +fn build_client(settings: &Settings) -> Result<Client<'_>, SyncError> { Client::new( &settings.sync.address, settings.network_connect_timeout, @@ -71,7 +71,7 @@ pub(crate) fn build_client(settings: &Settings) -> Result<Client<'_>, SyncError> .map_err(|e| SyncError::OperationalError { msg: e.to_string() }) } -pub(crate) async fn diff( +async fn diff( client: &Client<'_>, store: &SqliteStore, ) -> Result<(Vec<Diff>, RecordStatus), SyncError> { @@ -94,10 +94,7 @@ pub(crate) async fn diff( // With the store as context, we can determine if a tail exists locally or not and therefore if it needs uploading or download. // In theory this could be done as a part of the diffing stage, but it's easier to reason // about and test this way -pub(crate) fn operations( - diffs: Vec<Diff>, - _store: &SqliteStore, -) -> Result<Vec<Operation>, SyncError> { +fn operations(diffs: Vec<Diff>, _store: &SqliteStore) -> Result<Vec<Operation>, SyncError> { let mut operations = Vec::with_capacity(diffs.len()); for diff in diffs { @@ -283,7 +280,7 @@ async fn sync_download( Ok(ret) } -pub(crate) async fn sync_remote( +async fn sync_remote( client: &Client<'_>, operations: Vec<Operation>, local_store: &SqliteStore, @@ -323,7 +320,7 @@ pub(crate) async fn sync_remote( Ok((uploaded, downloaded)) } -pub(crate) async fn check_encryption_key( +async fn check_encryption_key( client: &Client<'_>, remote_index: &RecordStatus, encryption_key: &[u8; 32], @@ -347,9 +344,10 @@ pub(crate) async fn check_encryption_key( return Ok(()); }; - record - .decrypt::<PASETO_V4>(encryption_key) - .map_err(|_| SyncError::WrongKey)?; + record.decrypt::<PASETO_V4>(encryption_key).map_err(|err| { + error!("Wrong key error: {err}"); + SyncError::WrongKey + })?; Ok(()) } @@ -373,10 +371,10 @@ pub(crate) async fn sync( #[cfg(test)] mod tests { - use crate::atuin_client::record::sync::Operation; - use crate::atuin_common::record::{Diff, EncryptedData, HostId, Record}; + use crate::aclient::record::sync::Operation; + use turtle_common::record::{Diff, EncryptedData, HostId, Record}; - use crate::atuin_client::{ + use crate::aclient::{ record::{ sqlite_store::SqliteStore, sync::{self}, @@ -386,11 +384,11 @@ mod tests { fn test_record() -> Record<EncryptedData> { Record::builder() - .host(crate::atuin_common::record::Host::new(HostId( - crate::atuin_common::utils::uuid_v7(), + .host(turtle_common::record::Host::new(HostId( + turtle_common::utils::uuid_v7(), ))) .version("v1".into()) - .tag(crate::atuin_common::utils::uuid_v7().simple().to_string()) + .tag(turtle_common::utils::uuid_v7().simple().to_string()) .data(EncryptedData { data: String::new(), content_encryption_key: String::new(), diff --git a/crates/turtle/src/atuin_client/settings/meta.rs b/crates/daemon/src/aclient/settings/meta.rs index cc5afcf7..1c9b9cd1 100644 --- a/crates/turtle/src/atuin_client/settings/meta.rs +++ b/crates/daemon/src/aclient/settings/meta.rs @@ -2,12 +2,12 @@ use serde::{Deserialize, Serialize}; #[derive(Debug, Serialize, Deserialize, Clone)] pub(crate) struct Settings { - pub(crate) db_path: String, + pub(super) db_path: String, } impl Default for Settings { fn default() -> Self { - let dir = crate::atuin_common::utils::data_dir(); + let dir = turtle_common::utils::data_dir(); let path = dir.join("meta.db"); Self { diff --git a/crates/daemon/src/aclient/settings/mod.rs b/crates/daemon/src/aclient/settings/mod.rs new file mode 100644 index 00000000..379ee563 --- /dev/null +++ b/crates/daemon/src/aclient/settings/mod.rs @@ -0,0 +1,590 @@ +use crypto_secretbox::Key; +use std::{collections::HashMap, fs::read_to_string, path::PathBuf, sync::OnceLock}; +use tokio::sync::OnceCell; +use tracing::info; +use uuid::Uuid; + +use crate::aclient::encryption::decode_key; +use clap::ValueEnum; +use config::{ + Config, ConfigBuilder, Environment, File as ConfigFile, FileFormat, builder::DefaultState, +}; +use eyre::{Context, Result, eyre}; +use fs_err::create_dir_all; +use serde::{Deserialize, Serialize}; +use time::OffsetDateTime; +use turtle_common::record::HostId; +use turtle_common::utils; + +static DATA_DIR: OnceLock<PathBuf> = OnceLock::new(); +static META_CONFIG: OnceLock<(String, f64)> = OnceLock::new(); +static META_STORE: OnceCell<crate::aclient::meta::MetaStore> = OnceCell::const_new(); + +mod meta; + +// FIXME: Can use upstream Dialect enum if https://github.com/stevedonovan/chrono-english/pull/16 is merged +// FIXME: Above PR was merged, but dependency was changed to interim (fork of chrono-english) in the ... interim +#[derive(Clone, Debug, Deserialize, Copy, Serialize)] +enum Dialect { + #[serde(rename = "us")] + Us, + + #[serde(rename = "uk")] + Uk, +} + +impl From<Dialect> for interim::Dialect { + fn from(d: Dialect) -> Self { + match d { + Dialect::Uk => Self::Uk, + Dialect::Us => Self::Us, + } + } +} + +#[derive(Clone, Debug, Deserialize, Copy, PartialEq, Eq, ValueEnum, Serialize)] +enum KeymapMode { + #[serde(rename = "emacs")] + Emacs, + + #[serde(rename = "vim-normal")] + VimNormal, + + #[serde(rename = "vim-insert")] + VimInsert, + + #[serde(rename = "auto")] + Auto, +} + +// We want to translate the config to crossterm::cursor::SetCursorStyle, but +// the original type does not implement trait serde::Deserialize unfortunately. +// It seems impossible to implement Deserialize for external types when it is +// used in HashMap (https://stackoverflow.com/questions/67142663). We instead +// define an adapter type. +#[derive(Clone, Debug, Deserialize, Copy, PartialEq, Eq, ValueEnum, Serialize)] +enum CursorStyle { + #[serde(rename = "default")] + DefaultUserShape, + + #[serde(rename = "blink-block")] + BlinkingBlock, + + #[serde(rename = "steady-block")] + SteadyBlock, + + #[serde(rename = "blink-underline")] + BlinkingUnderScore, + + #[serde(rename = "steady-underline")] + SteadyUnderScore, + + #[serde(rename = "blink-bar")] + BlinkingBar, + + #[serde(rename = "steady-bar")] + SteadyBar, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub(crate) struct Daemon { + /// The daemon will handle sync on an interval. How often to sync, in seconds. + pub(crate) sync_frequency: u64, + + /// The path to the unix socket used by the daemon + pub(crate) socket_path: String, + + /// Path to the daemon pidfile used for process coordination. + pub(crate) pidfile_path: String, + + /// Use a socket passed via systemd's socket activation protocol, instead of the path + pub(crate) systemd_socket: bool, + + /// The port that should be used for TCP on non unix systems + tcp_port: u64, +} + +impl Default for Daemon { + fn default() -> Self { + Self { + sync_frequency: 300, + socket_path: String::new(), + pidfile_path: String::new(), + systemd_socket: false, + tcp_port: 8889, + } + } +} + +// The preview height strategy also takes max_preview_height into account. +#[derive(Clone, Debug, Deserialize, Copy, PartialEq, Eq, ValueEnum, Serialize)] +enum PreviewStrategy { + // Preview height is calculated for the length of the selected command. + #[serde(rename = "auto")] + Auto, + + // Preview height is calculated for the length of the longest command stored in the history. + #[serde(rename = "static")] + Static, + + // max_preview_height is used as fixed height. + #[serde(rename = "fixed")] + Fixed, +} + +/// Sync-specific settings. +#[derive(Clone, Debug, Default, Deserialize, Serialize)] +pub(crate) struct Sync { + /// The sync address for atuin. + pub(crate) address: String, + + #[serde(default)] + frequency: String, + + #[serde(default)] + pub(crate) auto: bool, + + #[serde(default)] + user_id_path: Option<PathBuf>, + + #[serde(default)] + pub(crate) encryption_key_path: Option<PathBuf>, +} + +impl Sync { + fn try_read_file(file: Option<&PathBuf>) -> Result<Option<String>> { + if let Some(path) = file { + if path.try_exists()? { + let user = read_to_string(path)?; + + if user.is_empty() { + Ok(None) + } else { + Ok(Some(user)) + } + } else { + // It's okay that the file doesn't exist. + // The important part is to error out if we can't access it (e.g. Because of missing + // permissions). + Ok(None) + } + } else { + Ok(None) + } + } + + pub(crate) fn have_sync_user(&self) -> Result<bool> { + let sa = self.user_id()?; + Ok(sa.is_some()) + } + + pub(crate) fn user_id(&self) -> Result<Option<Uuid>> { + Self::try_read_file(self.user_id_path.as_ref())? + .map(|file| { + Uuid::parse_str(file.trim()).context( + "Failed to decode user id as UUID, while trying to decode sync user_id", + ) + }) + .transpose() + } + pub(crate) fn encryption_key(&self) -> Result<Option<Key>> { + Self::try_read_file(self.encryption_key_path.as_ref())? + .as_deref() + .map(str::trim) + .map(decode_key) + .transpose() + } +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub(crate) struct Settings { + pub(crate) db_path: String, + pub(crate) record_store_path: String, + + pub(crate) network_connect_timeout: u64, + pub(crate) network_timeout: u64, + pub(crate) local_timeout: f64, + + #[serde(default)] + pub(crate) sync: Sync, + + #[serde(default)] + pub(crate) daemon: Daemon, + + #[serde(default)] + meta: meta::Settings, +} + +impl Settings { + // -- Meta store: lazily initialized on first access -- + + async fn meta_store() -> Result<&'static crate::aclient::meta::MetaStore> { + META_STORE + .get_or_try_init(|| async { + let (db_path, timeout) = META_CONFIG.get().ok_or_else(|| { + eyre!("meta store config not set — Settings::new() has not been called") + })?; + crate::aclient::meta::MetaStore::new(db_path, *timeout).await + }) + .await + } + + pub(crate) async fn host_id() -> Result<HostId> { + Self::meta_store().await?.host_id().await + } + + async fn last_sync() -> Result<OffsetDateTime> { + Self::meta_store().await?.last_sync().await + } + + pub(crate) async fn save_sync_time() -> Result<()> { + Self::meta_store().await?.save_sync_time().await + } + + fn builder() -> Result<ConfigBuilder<DefaultState>> { + Self::builder_with_data_dir(&utils::data_dir()) + } + + #[expect(clippy::too_many_lines)] + fn builder_with_data_dir(data_dir: &std::path::Path) -> Result<ConfigBuilder<DefaultState>> { + let db_path = data_dir.join("history.db"); + let record_store_path = data_dir.join("records.db"); + let kv_path = data_dir.join("kv.db"); + let scripts_path = data_dir.join("scripts.db"); + let ai_sessions_path = data_dir.join("ai_sessions.db"); + let socket_path = utils::daemon_socket_path(); + let pidfile_path = data_dir.join("atuin-daemon.pid"); + + let key_path = data_dir.join("key"); + let meta_path = data_dir.join("meta.db"); + + Ok(Config::builder() + .set_default("history_format", "{time}\t{command}\t{duration}")? + .set_default("db_path", db_path.to_str())? + .set_default("record_store_path", record_store_path.to_str())? + .set_default("key_path", key_path.to_str())? + .set_default("dialect", "us")? + .set_default("timezone", "local")? + .set_default("auto_sync", true)? + .set_default("sync.address", "https://api.atuin.sh")? + .set_default("sync_frequency", "5m")? + .set_default("search_mode", "fuzzy")? + .set_default("filter_mode", None::<String>)? + .set_default("style", "compact")? + .set_default("inline_height", 40)? + .set_default("show_preview", true)? + .set_default("preview.strategy", "auto")? + .set_default("max_preview_height", 4)? + .set_default("show_help", true)? + .set_default("show_tabs", true)? + .set_default("show_numeric_shortcuts", true)? + .set_default("auto_hide_height", 8)? + .set_default("invert", false)? + .set_default("exit_mode", "return-original")? + .set_default("word_jump_mode", "emacs")? + .set_default( + "word_chars", + "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789", + )? + .set_default("scroll_context_lines", 1)? + .set_default("shell_up_key_binding", false)? + .set_default("workspaces", false)? + .set_default("ctrl_n_shortcuts", false)? + .set_default("secrets_filter", true)? + .set_default("strip_trailing_whitespace", true)? + .set_default("network_connect_timeout", 5)? + .set_default("network_timeout", 30)? + .set_default("local_timeout", 2.0)? + // enter_accept defaults to false here, but true in the default config file. The dissonance is + // intentional! + // Existing users will get the default "False", so we don't mess with any potential + // muscle memory. + // New users will get the new default, that is more similar to what they are used to. + .set_default("enter_accept", false)? + .set_default("keys.scroll_exits", true)? + .set_default("keys.accept_past_line_end", true)? + .set_default("keys.exit_past_line_start", true)? + .set_default("keys.accept_past_line_start", false)? + .set_default("keys.accept_with_backspace", false)? + .set_default("keys.prefix", "a")? + .set_default("keymap_mode", "emacs")? + .set_default("keymap_mode_shell", "auto")? + .set_default("keymap_cursor", HashMap::<String, String>::new())? + .set_default("smart_sort", false)? + .set_default("command_chaining", false)? + .set_default("store_failed", true)? + .set_default("daemon.sync_frequency", 300)? + .set_default("daemon.socket_path", socket_path.to_str())? + .set_default("daemon.pidfile_path", pidfile_path.to_str())? + .set_default("daemon.systemd_socket", false)? + .set_default("daemon.tcp_port", 8889)? + .set_default("logs.enabled", true)? + .set_default("logs.level", "info")? + .set_default("logs.search.file", "search.log")? + .set_default("logs.daemon.file", "daemon.log")? + .set_default("logs.ai.file", "ai.log")? + .set_default("kv.db_path", kv_path.to_str())? + .set_default("scripts.db_path", scripts_path.to_str())? + .set_default("search.recency_score_multiplier", 1.0)? + .set_default("search.frequency_score_multiplier", 1.0)? + .set_default("search.frecency_score_multiplier", 1.0)? + .set_default("meta.db_path", meta_path.to_str())? + .set_default("ai.db_path", ai_sessions_path.to_str())? + .set_default("ai.session_continue_minutes", 60)? + .set_default("ai.send_cwd", false)? + .set_default("ai.opening.send_cwd", false)? + .set_default("ai.opening.send_last_command", false)? + .set_default( + "search.filters", + vec![ + "global", + "host", + "session", + "workspace", + "directory", + "session-preload", + ], + )? + .set_default("theme.name", "default")? + .set_default("theme.debug", None::<bool>)? + .set_default("tmux.enabled", false)? + .set_default("tmux.width", "80%")? + .set_default("tmux.height", "60%")? + .set_default( + "prefers_reduced_motion", + std::env::var("NO_MOTION").ok().map_or_else( + || config::Value::new(None, config::ValueKind::Boolean(false)), + |_| config::Value::new(None, config::ValueKind::Boolean(true)), + ), + )? + .set_default("no_mouse", false)? + .add_source( + Environment::with_prefix("atuin") + .prefix_separator("_") + .separator("__"), + )) + } + + pub(crate) fn get_config_path() -> Result<PathBuf> { + let config_dir = utils::config_dir(); + + create_dir_all(&config_dir) + .wrap_err_with(|| format!("could not create dir {}", config_dir.display()))?; + + let mut config_file = std::env::var("ATUIN_CONFIG_DIR").map_or_else( + |_| { + let mut config_file = PathBuf::new(); + config_file.push(config_dir); + config_file + }, + PathBuf::from, + ); + + config_file.push("config.toml"); + + Ok(config_file) + } + + /// Build a merged `Config` from defaults, config file, and environment. + /// + /// This resolves `data_dir`, initializes the data directory on disk, + /// and layers defaults → config file → env overrides. Both `new()` and + /// `get_config_value()` use this so the resolution logic lives in one place. + fn build_config() -> Result<Config> { + let config_file = Self::get_config_path()?; + + // extract data_dir first so we can use it as the base for other path defaults + let effective_data_dir = if config_file.exists() { + #[derive(Deserialize, Default)] + struct DataDirOnly { + data_dir: Option<String>, + } + + let config_file_str = config_file + .to_str() + .ok_or_else(|| eyre!("config file path is not valid UTF-8"))?; + + let partial_config = Config::builder() + .add_source(ConfigFile::new(config_file_str, FileFormat::Toml)) + .add_source( + Environment::with_prefix("atuin") + .prefix_separator("_") + .separator("__"), + ) + .build() + .ok(); + + let custom_data_dir = partial_config + .and_then(|c| c.try_deserialize::<DataDirOnly>().ok()) + .and_then(|d| d.data_dir); + + match custom_data_dir { + Some(dir) => { + let expanded = shellexpand::full(&dir) + .map_err(|e| eyre!("failed to expand data_dir path: {}", e))?; + PathBuf::from(expanded.as_ref()) + } + None => utils::data_dir(), + } + } else { + utils::data_dir() + }; + + DATA_DIR.set(effective_data_dir.clone()).ok(); + + create_dir_all(&effective_data_dir) + .wrap_err_with(|| format!("could not create dir {}", effective_data_dir.display()))?; + + let mut config_builder = Self::builder_with_data_dir(&effective_data_dir)?; + + config_builder = if config_file.exists() { + let config_file_str = config_file + .to_str() + .ok_or_else(|| eyre!("config file path is not valid UTF-8"))?; + config_builder.add_source(ConfigFile::new(config_file_str, FileFormat::Toml)) + } else { + // TODO(@bpeetz): Rework the config handling, so that we can actually auto-write a + // file with defaults. <2026-06-13> + create_dir_all(config_file.parent().unwrap())?; + + info!( + "No config file at: `{}`. Not adding one.", + config_file.display() + ); + + config_builder + }; + + // all paths should be expanded + let built = config_builder.build_cloned()?; + config_builder = [ + "db_path", + "record_store_path", + "key_path", + "daemon.socket_path", + "daemon.pidfile_path", + "logs.dir", + "logs.search.file", + "logs.daemon.file", + ] + .iter() + .map(|key| (key, built.get_string(key).unwrap_or_default())) + .filter_map(|(key, value)| match Self::expand_path(&value) { + Ok(expanded) => Some((key, expanded)), + Err(e) => { + log::warn!("failed to expand path for {key}: {e}"); + None + } + }) + .fold(config_builder, |builder, (key, value)| { + builder + .set_override(key, value) + .unwrap_or_else(|_| panic!("failed to set absolute path override for {key}")) + }); + + config_builder.build().map_err(Into::into) + } + + pub(crate) fn new() -> Result<Self> { + let config = Self::build_config()?; + let settings: Self = config + .try_deserialize() + .map_err(|e| eyre!("failed to deserialize: {}", e))?; + + // Register meta store config for lazy initialization on first access + META_CONFIG + .set((settings.meta.db_path.clone(), settings.local_timeout)) + .ok(); + + Ok(settings) + } + + fn expand_path(path: &str) -> Result<String> { + shellexpand::full(&path) + .map(|p| p.to_string()) + .map_err(|e| eyre!("failed to expand path: {}", e)) + } + + pub(crate) fn paths_ok(&self) -> bool { + let mut paths: Vec<&str> = vec![ + &self.db_path, + &self.record_store_path, + &self.meta.db_path, + &self.daemon.socket_path, + ]; + + if let Some(path) = &self.sync.encryption_key_path { + paths.push(path.to_str().unwrap()); + } + if let Some(path) = &self.sync.user_id_path { + paths.push(path.to_str().unwrap()); + } + + paths.iter().all(|p| !utils::broken_symlink(p)) + } +} + +impl Default for Settings { + fn default() -> Self { + // if this panics something is very wrong, as the default config + // does not build or deserialize into the settings struct + Self::builder() + .expect("Could not build default") + .build() + .expect("Could not build config") + .try_deserialize() + .expect("Could not deserialize config") + } +} + +#[cfg(test)] +pub(crate) fn test_local_timeout() -> f64 { + std::env::var("ATUIN_TEST_LOCAL_TIMEOUT") + .ok() + .and_then(|x| x.parse().ok()) + // this hardcoded value should be replaced by a simple way to get the + // default local_timeout of Settings if possible + .unwrap_or(2.0) +} + +#[cfg(test)] +mod tests { + use eyre::Result; + + #[test] + fn builder_with_data_dir_uses_custom_paths() -> Result<()> { + use std::path::PathBuf; + + let custom_dir = PathBuf::from("/custom/data/dir"); + let builder = super::Settings::builder_with_data_dir(&custom_dir)?; + let config = builder.build()?; + + let db_path: String = config.get("db_path")?; + let key_path: String = config.get("key_path")?; + let record_store_path: String = config.get("record_store_path")?; + let kv_db_path: String = config.get("kv.db_path")?; + let scripts_db_path: String = config.get("scripts.db_path")?; + let meta_db_path: String = config.get("meta.db_path")?; + let daemon_socket_path: String = config.get("daemon.socket_path")?; + let daemon_pidfile_path: String = config.get("daemon.pidfile_path")?; + + assert_eq!(db_path, custom_dir.join("history.db").to_str().unwrap()); + assert_eq!(key_path, custom_dir.join("key").to_str().unwrap()); + assert_eq!( + record_store_path, + custom_dir.join("records.db").to_str().unwrap() + ); + assert_eq!(kv_db_path, custom_dir.join("kv.db").to_str().unwrap()); + assert_eq!( + scripts_db_path, + custom_dir.join("scripts.db").to_str().unwrap() + ); + assert_eq!(meta_db_path, custom_dir.join("meta.db").to_str().unwrap()); + assert_eq!( + daemon_pidfile_path, + custom_dir.join("atuin-daemon.pid").to_str().unwrap() + ); + + Ok(()) + } +} diff --git a/crates/turtle/src/atuin_client/utils.rs b/crates/daemon/src/aclient/utils.rs index 989f9fc1..18e732a0 100644 --- a/crates/turtle/src/atuin_client/utils.rs +++ b/crates/daemon/src/aclient/utils.rs @@ -1,18 +1,3 @@ -pub(crate) fn get_hostname() -> String { - std::env::var("ATUIN_HOST_NAME") - .unwrap_or_else(|_| whoami::hostname().unwrap_or_else(|_| "unknown-host".to_string())) -} - -pub(crate) fn get_username() -> String { - std::env::var("ATUIN_HOST_USER") - .unwrap_or_else(|_| whoami::username().unwrap_or_else(|_| "unknown-user".to_string())) -} - -/// Returns a pair of the hostname and username, separated by a colon. -pub(crate) fn get_host_user() -> String { - format!("{}:{}", get_hostname(), get_username()) -} - /// Setup a [`SQLite`] database. /// /// This takes care of correct locking, so that we avoid a race when setting up the database. @@ -28,7 +13,7 @@ macro_rules! setup_db { Ok(()) } - crate::atuin_client::utils::setup_db_inner($db_path, $a_timeout, $opts, migrate) + crate::aclient::utils::setup_db_inner($db_path, $a_timeout, $opts, migrate) }}; } pub(crate) use setup_db; diff --git a/crates/turtle/src/atuin_daemon/components/sync.rs b/crates/daemon/src/api/control.rs index 20d49839..16b4bd94 100644 --- a/crates/turtle/src/atuin_daemon/components/sync.rs +++ b/crates/daemon/src/api/control.rs @@ -1,29 +1,26 @@ -//! Sync component. -//! -//! Handles periodic synchronization with the Atuin cloud server. - use std::time::Duration; use eyre::Result; -use rand::Rng; -use tokio::sync::mpsc; +use rand::RngExt; use tokio::time::{self, MissedTickBehavior}; +use tonic::{Request, Response, Status}; +use tracing::{Level, instrument}; -use crate::atuin_client::{history::store::HistoryStore, record::sync, settings::Settings}; +use turtle_api::generated::{ + DAEMON_PROTOCOL_VERSION, + control::{ + ForceSyncReply, ForceSyncRequest, PathsReply, PathsRequest, StatusReply, StatusRequest, + control_server::{Control, ControlServer}, + }, +}; -use crate::atuin_daemon::{ - daemon::{Component, DaemonHandle}, +use crate::{ + DAEMON_VERSION, + aclient::{history::store::HistoryStore, record::sync, settings::Settings}, + daemon::DaemonHandle, events::DaemonEvent, }; -/// Commands that can be sent to the sync task. -enum SyncCommand { - /// Trigger an immediate sync. - ForceSync, - /// Stop the sync loop. - Stop, -} - /// Sync state - tracks whether we're in normal operation or retrying after failure. #[derive(Clone, Copy, PartialEq, Eq)] enum SyncState { @@ -34,77 +31,98 @@ enum SyncState { Retrying, } -/// Sync component - handles periodic cloud synchronization. +/// The Control gRPC service. /// -/// This component: -/// - Runs a background sync loop on a configurable interval -/// - Implements exponential backoff on sync failures -/// - Responds to [`ForceSync`] events for immediate sync -/// - Emits SyncCompleted/SyncFailed events -pub(crate) struct SyncComponent { - task_handle: Option<tokio::task::JoinHandle<()>>, - command_tx: Option<mpsc::Sender<SyncCommand>>, +/// This service is used by external processes to inject events into the daemon. +/// It's not a component - it's part of the daemon's core infrastructure. +pub(crate) struct ControlService { + handle: DaemonHandle, } -impl SyncComponent { - /// Create a new sync component. - pub(crate) fn new() -> Self { - Self { - task_handle: None, - command_tx: None, - } +impl ControlService { + /// Create a new control service with the given daemon handle. + pub(crate) fn new(handle: DaemonHandle) -> Self { + tokio::spawn(sync_loop(handle.clone())); + + Self { handle } } -} -impl Default for SyncComponent { - fn default() -> Self { - Self::new() + /// Get a tonic server for this service. + pub(crate) fn into_server(self) -> ControlServer<Self> { + ControlServer::new(self) } } #[tonic::async_trait] -impl Component for SyncComponent { - fn name(&self) -> &'static str { - "sync" - } +impl Control for ControlService { + #[instrument(skip_all, level = Level::INFO)] + async fn paths(&self, _request: Request<PathsRequest>) -> Result<Response<PathsReply>, Status> { + let settings = self.handle.settings().await; - async fn start(&mut self, handle: DaemonHandle) -> Result<()> { - let (cmd_tx, cmd_rx) = mpsc::channel(16); - self.command_tx = Some(cmd_tx); + let config = Settings::get_config_path() + .map_err(|e| Status::internal(format!("failed to get settings path: {e:?}")))?; - // Spawn the sync loop with its own copy of the handle - self.task_handle = Some(tokio::spawn(sync_loop(handle, cmd_rx))); + let reply = PathsReply { + config: config.to_string_lossy().to_string(), + db: settings.db_path.clone(), + socket: settings.daemon.socket_path.clone(), + }; - tracing::info!("sync component started"); - Ok(()) + Ok(Response::new(reply)) } - async fn handle_event(&mut self, event: &DaemonEvent) -> Result<()> { - match event { - DaemonEvent::ForceSync => { - tracing::info!("force sync requested"); - if let Some(tx) = &self.command_tx { - drop(tx.send(SyncCommand::ForceSync).await); - } - } - DaemonEvent::SyncFailed { error } => { - tracing::error!(?error, "Sync failed."); - } - _ => (), - } - Ok(()) + #[instrument(skip_all, level = Level::INFO)] + async fn status( + &self, + _request: Request<StatusRequest>, + ) -> Result<Response<StatusReply>, Status> { + let reply = StatusReply { + healthy: true, + version: DAEMON_VERSION.to_owned(), + pid: std::process::id(), + protocol: DAEMON_PROTOCOL_VERSION, + }; + + Ok(Response::new(reply)) } - async fn stop(&mut self) -> Result<()> { - if let Some(tx) = &self.command_tx { - drop(tx.send(SyncCommand::Stop).await); - } - if let Some(handle) = self.task_handle.take() { - // Give the task a moment to shut down gracefully - drop(time::timeout(Duration::from_secs(5), handle).await); - } - tracing::info!("sync component stopped"); - Ok(()) + #[instrument(skip_all, level = Level::INFO)] + async fn force_sync( + &self, + _request: Request<ForceSyncRequest>, + ) -> Result<Response<ForceSyncReply>, Status> { + self.handle.emit(DaemonEvent::ForceSync); + let event = self + .handle + .wait_for(|e| { + matches!( + e, + DaemonEvent::SyncFailed { .. } | DaemonEvent::SyncCompleted { .. } + ) + }) + .await + .map_err(|e| { + Status::internal(format!("failed to wait for sync response event: {e:?}")) + })?; + + let reply = match event { + DaemonEvent::SyncCompleted { + uploaded, + downloaded, + } => ForceSyncReply { + error: None, + uploaded: uploaded as u32, + downloaded: downloaded as u32, + }, + DaemonEvent::SyncFailed { error } => ForceSyncReply { + error: Some(error), + uploaded: 0, + downloaded: 0, + }, + _ => unreachable!(), + }; + + Ok(Response::new(reply)) } } @@ -113,7 +131,7 @@ impl Component for SyncComponent { /// This runs in a spawned task and handles periodic sync as well as /// force sync requests. #[expect(clippy::significant_drop_tightening, reason = "false positive")] -async fn sync_loop(handle: DaemonHandle, mut cmd_rx: mpsc::Receiver<SyncCommand>) { +async fn sync_loop(handle: DaemonHandle) { tracing::info!("sync loop starting"); // Clone settings since we need them across await points @@ -131,7 +149,7 @@ async fn sync_loop(handle: DaemonHandle, mut cmd_rx: mpsc::Receiver<SyncCommand> let history_store = HistoryStore::new(handle.store().clone(), host_id, encryption_key); // Don't backoff by more than 30 mins (with a random jitter of up to 1 min) - let max_interval: f64 = 60.0f64.mul_add(30.0, rand::thread_rng().gen_range(0.0..60.0)); + let max_interval: f64 = 60.0f64.mul_add(30.0, rand::rng().random_range(0.0..60.0)); let mut ticker = time::interval(Duration::from_secs(settings.daemon.sync_frequency)); @@ -141,6 +159,7 @@ async fn sync_loop(handle: DaemonHandle, mut cmd_rx: mpsc::Receiver<SyncCommand> let mut sync_state = SyncState::Idle; + let mut daemon_rx = handle.subscribe(); loop { tokio::select! { _ = ticker.tick() => { @@ -161,9 +180,9 @@ async fn sync_loop(handle: DaemonHandle, mut cmd_rx: mpsc::Receiver<SyncCommand> &settings, ).await; } - cmd = cmd_rx.recv() => { + cmd = daemon_rx.recv() => { match cmd { - Some(SyncCommand::ForceSync) => { + Ok(DaemonEvent::ForceSync) => { tracing::info!("executing force sync"); let settings = handle.settings().await; sync_state = do_sync_tick( @@ -173,11 +192,12 @@ async fn sync_loop(handle: DaemonHandle, mut cmd_rx: mpsc::Receiver<SyncCommand> max_interval, &settings, ).await; - } - Some(SyncCommand::Stop) | None => { + }, + Ok(DaemonEvent::ShutdownRequested) | Err(_) => { tracing::info!("sync loop stopping"); break; - } + }, + _ => () } } } @@ -217,14 +237,13 @@ async fn do_sync_tick( Err(e) => { tracing::error!("sync tick failed with {e}"); - // Emit failure event handle.emit(DaemonEvent::SyncFailed { error: e.to_string(), }); // Exponential backoff - let mut rng = rand::thread_rng(); - let mut new_interval = ticker.period().as_secs_f64() * rng.gen_range(2.0..2.2); + let mut rng = rand::rng(); + let mut new_interval = ticker.period().as_secs_f64() * rng.random_range(2.0..2.2); if new_interval > max_interval { new_interval = max_interval; @@ -256,9 +275,6 @@ async fn do_sync_tick( tracing::error!("failed to build history from downloaded records: {e}"); } - // Emit the records added event (for search indexing) - handle.emit(DaemonEvent::RecordsAdded(downloaded_records.clone())); - // Emit sync completed event handle.emit(DaemonEvent::SyncCompleted { uploaded: uploaded_count as usize, diff --git a/crates/daemon/src/api/history.rs b/crates/daemon/src/api/history.rs new file mode 100644 index 00000000..078e8c64 --- /dev/null +++ b/crates/daemon/src/api/history.rs @@ -0,0 +1,243 @@ +use std::{pin::Pin, time::Duration}; + +use dashmap::DashMap; +use eyre::Result; +use time::OffsetDateTime; +use tokio_stream::Stream; +use tonic::{Request, Response, Status}; +use tracing::{Level, instrument}; + +use crate::{ + aclient::{history::store::HistoryStore, settings::Settings}, + daemon::DaemonHandle, + events::DaemonEvent, +}; +use turtle_api::{ + client::{ + proto_duration_to_std, proto_timestamp_to_time, std_to_proto_duration, + time_to_proto_timestamp, + }, + generated::{ + DAEMON_PROTOCOL_VERSION, + history::{ + EndHistoryReply, EndHistoryRequest, HistoryEntry, HistoryEventKind, HistoryReply, + HistoryRequest, StartHistoryReply, StartHistoryRequest, TailHistoryReply, + TailHistoryRequest, + history_server::{History as HistorySvc, HistoryServer}, + }, + }, + history::{History, HistoryId}, +}; + +/// The gRPC service implementation. +/// +/// This is a thin wrapper that delegates to the component's shared state. +pub(crate) struct HistoryService { + /// Commands currently running (not yet completed). + running: DashMap<HistoryId, History>, + + /// Handle to the daemon (set during start). + handle: DaemonHandle, + + /// History store for pushing records + history_store: HistoryStore, +} + +impl HistoryService { + pub(crate) async fn new(handle: DaemonHandle) -> Result<Self> { + let host_id = Settings::host_id().await?; + let history_store = + HistoryStore::new(handle.store().clone(), host_id, *handle.encryption_key()); + + Ok(Self { + running: DashMap::new(), + handle, + history_store, + }) + } + + /// Get a tonic server for this service. + pub(crate) fn into_server(self) -> HistoryServer<Self> { + HistoryServer::new(self) + } +} + +fn history_to_reply(history: History) -> HistoryEntry { + HistoryEntry { + timestamp: time_to_proto_timestamp(history.timestamp), + id: history.id.to_string(), + command: history.command, + cwd: history.cwd, + session: history.session, + hostname: history.hostname, + author: history.author, + intent: history.intent.unwrap_or_default(), + exit: history.exit, + duration: std_to_proto_duration(history.duration), + } +} + +#[tonic::async_trait] +impl HistorySvc for HistoryService { + type TailHistoryStream = Pin<Box<dyn Stream<Item = Result<TailHistoryReply, Status>> + Send>>; + + #[instrument(skip_all, level = Level::INFO)] + async fn history( + &self, + request: Request<HistoryRequest>, + ) -> Result<Response<HistoryReply>, Status> { + let req = request.into_inner(); + + let entries = if let Some(range) = req.range { + let from = proto_timestamp_to_time(range.start); + let to = proto_timestamp_to_time(range.end); + + self.handle.history_db().range(from, to).await + } else { + self.handle.history_db().list(None, false, false).await + } + .map_err(|e| Status::internal(format!("failed to read db: {e:?}")))? + .into_iter() + .map(history_to_reply) + .collect(); + + Ok(Response::new(HistoryReply { entries })) + } + + #[instrument(skip_all, level = Level::INFO)] + async fn start_history( + &self, + request: Request<StartHistoryRequest>, + ) -> Result<Response<StartHistoryReply>, Status> { + let req = request.into_inner(); + + let timestamp = proto_timestamp_to_time(req.timestamp); + + let h: History = History::daemon() + .timestamp(timestamp) + .command(req.command) + .cwd(req.cwd) + .session(req.session) + .hostname(req.hostname) + .author(req.author) + .intent(req.intent.unwrap_or_default()) + .build() + .into(); + + self.handle.emit(DaemonEvent::HistoryStarted(h.clone())); + + let id = h.id; + tracing::info!(id = id.to_string(), "start history called"); + self.running.insert(id, h); + + let reply = StartHistoryReply { + id: id.to_string(), + version: env!("CARGO_PKG_VERSION").to_string(), + protocol: DAEMON_PROTOCOL_VERSION, + }; + + Ok(Response::new(reply)) + } + + #[instrument(skip_all, level = Level::INFO)] + async fn end_history( + &self, + request: Request<EndHistoryRequest>, + ) -> Result<Response<EndHistoryReply>, Status> { + let req = request.into_inner(); + let id = HistoryId::from(req.id); + + tracing::info!(id = id.to_string(), "end history called"); + + if let Some((_, mut history)) = self.running.remove(&id) { + history.exit = req.exit; + history.duration = match proto_duration_to_std(req.duration) { + Duration::ZERO => Duration::from_nanos_u128( + (OffsetDateTime::now_utc() - history.timestamp).whole_nanoseconds() as u128, + ), + value => value, + }; + + self.handle + .history_db() + .save(&history) + .await + .map_err(|e| Status::internal(format!("failed to write to db: {e:?}")))?; + + tracing::info!( + id = id.to_string(), + duration = history.duration.as_nanos(), + "end history" + ); + + let (record_id, idx) = self + .history_store + .push(history.clone()) + .await + .map_err(|e| Status::internal(format!("failed to push record to store: {e:?}")))?; + + self.handle.emit(DaemonEvent::HistoryEnded(history)); + + let reply = EndHistoryReply { + id: record_id.0.to_string(), + idx, + version: env!("CARGO_PKG_VERSION").to_string(), + protocol: DAEMON_PROTOCOL_VERSION, + }; + + return Ok(Response::new(reply)); + } + + Err(Status::not_found(format!( + "could not find history with id: {id}" + ))) + } + + #[instrument(skip_all, level = Level::INFO)] + async fn tail_history( + &self, + _request: Request<TailHistoryRequest>, + ) -> Result<Response<Self::TailHistoryStream>, Status> { + let mut rx = self.handle.subscribe(); + let (tx, out_rx) = tokio::sync::mpsc::channel::<Result<TailHistoryReply, Status>>(128); + + tokio::spawn(async move { + loop { + let event = match rx.recv().await { + Ok(event) => event, + Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { + drop( + tx.send(Err(Status::resource_exhausted(format!( + "tail stream lagged behind and dropped {skipped} events" + )))) + .await, + ); + break; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + }; + + let reply = match event { + DaemonEvent::HistoryStarted(history) => Some(TailHistoryReply { + kind: HistoryEventKind::Started.into(), + history: Some(history_to_reply(history)), + }), + DaemonEvent::HistoryEnded(history) => Some(TailHistoryReply { + kind: HistoryEventKind::Ended.into(), + history: Some(history_to_reply(history)), + }), + _ => None, + }; + + if let Some(reply) = reply + && tx.send(Ok(reply)).await.is_err() + { + break; + } + } + }); + + let stream = tokio_stream::wrappers::ReceiverStream::new(out_rx); + Ok(Response::new(Box::pin(stream))) + } +} diff --git a/crates/daemon/src/api/mod.rs b/crates/daemon/src/api/mod.rs new file mode 100644 index 00000000..8d475fe9 --- /dev/null +++ b/crates/daemon/src/api/mod.rs @@ -0,0 +1,2 @@ +pub(crate) mod control; +pub(crate) mod history; diff --git a/crates/turtle/src/atuin_daemon/daemon.rs b/crates/daemon/src/daemon.rs index 80aaeef8..70e65c1e 100644 --- a/crates/turtle/src/atuin_daemon/daemon.rs +++ b/crates/daemon/src/daemon.rs @@ -4,20 +4,19 @@ //! //! - [`DaemonState`]: Shared state owned by the daemon //! - [`DaemonHandle`]: A lightweight, cloneable handle for accessing daemon state -//! - [`Component`]: A trait for implementing daemon components //! - [`Daemon`]: The main daemon orchestrator //! - [`DaemonBuilder`]: Builder for constructing and configuring the daemon use std::sync::Arc; -use crate::atuin_client::{ +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::atuin_daemon::events::DaemonEvent; +use crate::events::DaemonEvent; // ============================================================================ // DaemonState @@ -25,7 +24,7 @@ use crate::atuin_daemon::events::DaemonEvent; /// Shared state owned by the daemon. /// -/// This contains all the resources that components and services need access to. +/// 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 @@ -48,7 +47,7 @@ pub(crate) struct DaemonState { /// A lightweight handle to the daemon's shared state. /// -/// This is the primary way for components, gRPC services, and spawned tasks to +/// This is the primary way for gRPC services, and spawned tasks to /// interact with the daemon. It provides access to: /// /// - Event emission and subscription @@ -88,12 +87,24 @@ impl DaemonHandle { 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. - /// Useful for components that need to listen for events outside of the - /// normal `handle_event` callback flow. pub(crate) fn subscribe(&self) -> broadcast::Receiver<DaemonEvent> { self.state.event_tx.subscribe() } @@ -113,16 +124,6 @@ impl DaemonHandle { 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 @@ -148,92 +149,12 @@ impl std::fmt::Debug for DaemonHandle { } // ============================================================================ -// Component Trait -// ============================================================================ - -/// A daemon component that handles a specific domain. -/// -/// Components are the building blocks of the daemon. Each component: -/// -/// - Has a unique name for logging and debugging -/// - Can optionally expose gRPC services -/// - Receives a [`DaemonHandle`] on startup for accessing daemon resources -/// - Handles events from the event bus -/// - Performs cleanup on shutdown -/// -/// # Lifecycle -/// -/// 1. **Construction**: Component is created (usually via `new()`) -/// 2. **Start**: `start()` is called with a [`DaemonHandle`] -/// 3. **Running**: `handle_event()` is called for each event on the bus -/// 4. **Shutdown**: `stop()` is called for cleanup -/// -/// # Example -/// -/// ```ignore -/// pub(crate) struct MyComponent { -/// handle: Option<DaemonHandle>, -/// } -/// -/// #[async_trait] -/// impl Component for MyComponent { -/// fn name(&self) -> &'static str { "my-component" } -/// -/// async fn start(&mut self, handle: DaemonHandle) -> Result<()> { -/// self.handle = Some(handle); -/// Ok(()) -/// } -/// -/// async fn handle_event(&mut self, event: &DaemonEvent) -> Result<()> { -/// match event { -/// DaemonEvent::SomeEvent => { -/// // Handle the event -/// if let Some(handle) = &self.handle { -/// handle.emit(DaemonEvent::ResponseEvent); -/// } -/// } -/// _ => {} -/// } -/// Ok(()) -/// } -/// -/// async fn stop(&mut self) -> Result<()> { -/// Ok(()) -/// } -/// } -/// ``` -#[tonic::async_trait] -pub(crate) trait Component: Send + Sync { - /// Human-readable name for logging and debugging. - fn name(&self) -> &'static str; - - /// Called once at startup. - /// - /// Store the handle if you need to emit events or access daemon resources - /// later. The handle is cheaply cloneable, so feel free to clone it for - /// spawned tasks. - async fn start(&mut self, handle: DaemonHandle) -> Result<()>; - - /// Handle an incoming event. - /// - /// Called for every event on the bus. To emit new events in response, - /// use the handle stored during `start()`. Events emitted here will be - /// processed in subsequent event loop iterations. - async fn handle_event(&mut self, event: &DaemonEvent) -> Result<()>; - - /// Called on graceful shutdown. - /// - /// Use this to clean up resources, abort spawned tasks, etc. - async fn stop(&mut self) -> Result<()>; -} - -// ============================================================================ // Daemon // ============================================================================ /// The main daemon orchestrator. /// -/// The daemon manages components, runs the event loop, and coordinates startup +/// The daemon runs the event loop, and coordinates startup /// and shutdown. It is constructed via [`DaemonBuilder`]. /// /// # Event Loop @@ -241,14 +162,11 @@ pub(crate) trait Component: Send + Sync { /// The daemon runs a simple event loop: /// /// 1. Wait for an event on the bus -/// 2. Dispatch the event to all components (in registration order) -/// 3. Components may emit new events in response /// 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 { - components: Vec<Box<dyn Component>>, handle: DaemonHandle, } @@ -265,26 +183,8 @@ impl Daemon { self.handle.clone() } - /// Start all components. - /// - /// This must be called before `run_event_loop()`. It initializes all - /// registered components with the daemon handle. - pub(crate) async fn start_components(&mut self) -> Result<()> { - for component in &mut self.components { - tracing::info!(component = component.name(), "starting component"); - component - .start(self.handle.clone()) - .await - .with_context(|| format!("failed to start component: {}", component.name()))?; - } - Ok(()) - } - /// Run the daemon event loop. - /// - /// This processes events until a [`ShutdownRequested`] event is received. - /// Components must be started first via `start_components()`. - pub(crate) async fn run_event_loop(&mut self) -> Result<()> { + pub(crate) async fn wait_for_shutdown(&mut self) -> Result<()> { let mut event_rx = self.handle.subscribe(); loop { match event_rx.recv().await { @@ -293,8 +193,7 @@ impl Daemon { break; } Ok(event) => { - tracing::debug!(?event, "processing event"); - self.dispatch_event(&event).await; + tracing::debug!(?event, "event received"); } Err(broadcast::error::RecvError::Lagged(n)) => { tracing::warn!( @@ -310,35 +209,6 @@ impl Daemon { } Ok(()) } - - /// Stop all components. - /// - /// This performs graceful shutdown of all components. - pub(crate) async fn stop_components(&mut self) { - for component in &mut self.components { - tracing::info!(component = component.name(), "stopping component"); - if let Err(e) = component.stop().await { - tracing::error!( - component = component.name(), - error = ?e, - "error stopping component" - ); - } - } - tracing::info!("all components stopped"); - } - - async fn dispatch_event(&mut self, event: &DaemonEvent) { - for component in &mut self.components { - if let Err(e) = component.handle_event(event).await { - tracing::error!( - component = component.name(), - error = ?e, - "error handling event" - ); - } - } - } } // ============================================================================ @@ -353,9 +223,6 @@ impl Daemon { /// let daemon = Daemon::builder(settings) /// .store(store) /// .history_db(history_db) -/// .component(HistoryComponent::new()) -/// .component(SearchComponent::new()) -/// .component(SyncComponent::new()) /// .build() /// .await?; /// @@ -365,7 +232,6 @@ pub(crate) struct DaemonBuilder { settings: Settings, store: Option<SqliteStore>, history_db: Option<HistoryDatabase>, - components: Vec<Box<dyn Component>>, } impl DaemonBuilder { @@ -375,7 +241,6 @@ impl DaemonBuilder { settings, store: None, history_db: None, - components: Vec::new(), } } @@ -391,14 +256,6 @@ impl DaemonBuilder { self } - /// Register a component. - /// - /// Components are started in registration order and stopped in reverse order. - pub(crate) fn component(mut self, component: impl Component + 'static) -> Self { - self.components.push(Box::new(component)); - self - } - /// Build the daemon. /// /// This loads the encryption key and creates the daemon state. @@ -428,9 +285,6 @@ impl DaemonBuilder { // Create the handle (just a reference to the state) let handle = DaemonHandle { state }; - Ok(Daemon { - components: self.components, - handle, - }) + Ok(Daemon { handle }) } } diff --git a/crates/turtle/src/atuin_daemon/events.rs b/crates/daemon/src/events.rs index d379277d..654e56cb 100644 --- a/crates/turtle/src/atuin_daemon/events.rs +++ b/crates/daemon/src/events.rs @@ -7,36 +7,26 @@ //! External processes (like CLI commands) can also inject events via the //! Control gRPC service. -use crate::atuin_client::history::{History, HistoryId}; -use crate::atuin_common::record::RecordId; +use turtle_api::history::History; /// Events that flow through the daemon's event bus. /// /// Events are broadcast to all components. Each component decides which /// events it cares about in its `handle_event` implementation. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum DaemonEvent { - // ---- History lifecycle ---- /// A command has started running. HistoryStarted(History), /// A command has finished running. HistoryEnded(History), - // ---- Sync ---- - /// Records were synced from the server. - /// - /// The search component uses this to update its index with new history. - RecordsAdded(Vec<RecordId>), - /// Sync completed successfully. SyncCompleted { /// Number of records uploaded. - #[expect(unused)] uploaded: usize, /// Number of records downloaded. - #[expect(unused)] downloaded: usize, }, @@ -49,29 +39,6 @@ pub(crate) enum DaemonEvent { /// Request an immediate sync (external trigger). ForceSync, - // ---- External commands ---- - /// History was pruned - search index needs a full rebuild. - /// - /// Emitted when the user runs `atuin history prune` or similar. - HistoryPruned, - - /// History was rebuilt - search index needs a full rebuild. - /// - /// Emitted when the user runs `atuin store rebuild history` or similar. - HistoryRebuilt, - - /// Specific history items were deleted. - /// - /// The search component should remove these from its index. - HistoryDeleted { - /// IDs of the deleted history entries. - ids: Vec<HistoryId>, - }, - - /// Settings have changed, components should reload if needed. - SettingsReloaded, - - // ---- Lifecycle ---- /// Request graceful shutdown of the daemon. ShutdownRequested, } diff --git a/crates/daemon/src/main.rs b/crates/daemon/src/main.rs new file mode 100644 index 00000000..50a41775 --- /dev/null +++ b/crates/daemon/src/main.rs @@ -0,0 +1,180 @@ +#![expect( + clippy::cast_possible_truncation, + clippy::cast_possible_wrap, + clippy::cast_sign_loss +)] + +use std::{ + fs::{self, File, OpenOptions}, + io::Write, + path::{Path, PathBuf}, +}; + +use clap::Parser; +use eyre::WrapErr; +use eyre::{Result, bail}; +use tracing_subscriber::EnvFilter; + +use crate::{ + aclient::{database::ClientSqlite, record::sqlite_store::SqliteStore, settings::Settings}, + api::{control::ControlService, history::HistoryService}, + daemon::Daemon, +}; + +pub(crate) mod aclient; +pub(crate) mod api; +pub(crate) mod daemon; +pub(crate) mod events; +pub(crate) mod server; + +const DAEMON_VERSION: &str = env!("CARGO_PKG_VERSION"); + +#[derive(Parser, Debug)] +#[command(infer_subcommands = true)] +enum Cmd { + /// Start the daemon server + Start { + /// Also write daemon logs to the console (useful for debugging) + #[arg(long)] + show_logs: bool, + }, +} + +#[tokio::main] +async fn main() -> Result<()> { + let settings = Settings::new().wrap_err("could not load client settings")?; + if !settings.paths_ok() { + bail!("Failed to verify all paths :("); + } + + let db_path = PathBuf::from(settings.db_path.as_str()); + let record_store_path = PathBuf::from(settings.record_store_path.as_str()); + + let history_db = ClientSqlite::new(db_path, settings.local_timeout).await?; + let sqlite_store = SqliteStore::new(record_store_path, settings.local_timeout).await?; + + match Cmd::parse() { + Cmd::Start { show_logs, .. } => { + if show_logs + && let Err(e) = tracing_subscriber::fmt() + .with_file(true) + .with_line_number(true) + .with_level(true) + .without_time() + .with_env_filter( + EnvFilter::builder() + .from_env_lossy() + .add_directive("turtle_daemon=debug".parse().unwrap()), + ) + .try_init() + { + eprintln!("failed to initialize logging: {e}"); + } + + boot(settings, sqlite_store, history_db).await + } + } +} + +/// Boot the daemon. +/// +/// This creates a daemon, +/// starts the gRPC server with services, and runs the event loop. +async fn boot(settings: Settings, store: SqliteStore, history_db: ClientSqlite) -> Result<()> { + let pidfile_path = PathBuf::from(&settings.daemon.pidfile_path); + let _pidfile_guard = PidfileGuard::acquire(&pidfile_path)?; + + let mut daemon = Daemon::builder(settings.clone()) + .store(store) + .history_db(history_db) + .build()?; + + let handle = { + let handle = daemon.handle(); + + // Spawn signal handler to emit ShutdownRequested on Ctrl+C/SIGTERM + let signal_handle = handle.clone(); + tokio::spawn(async move { + shutdown_signal().await; + tracing::info!("received shutdown signal"); + signal_handle.shutdown(); + }); + + handle + }; + + let history_service = HistoryService::new(handle.clone()).await?; + let control_service = ControlService::new(handle.clone()); + + server::run_grpc_server( + &settings, + history_service.into_server(), + control_service.into_server(), + handle, + )?; + + daemon.wait_for_shutdown().await?; + + tracing::info!("daemon shut down complete"); + Ok(()) +} + +/// Wait for a shutdown signal (Ctrl+C or SIGTERM). +#[cfg(unix)] +async fn shutdown_signal() { + let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .expect("failed to register sigterm handler"); + let mut int = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt()) + .expect("failed to register sigint handler"); + + tokio::select! { + _ = term.recv() => {}, + _ = int.recv() => {}, + } +} + +struct PidfileGuard { + file: File, +} + +impl PidfileGuard { + fn acquire(path: &Path) -> Result<Self> { + let mut file = open_lock_file(path)?; + + if let Err(fs::TryLockError::WouldBlock) = file.try_lock() { + bail!( + "daemon already running (pidfile lock busy at {})", + path.display() + ); + } + + file.set_len(0) + .wrap_err_with(|| format!("could not truncate daemon pidfile {}", path.display()))?; + writeln!(file, "{}", std::process::id()) + .and_then(|()| writeln!(file, "{DAEMON_VERSION}")) + .wrap_err_with(|| format!("could not write daemon pidfile {}", path.display()))?; + + Ok(Self { file }) + } +} + +impl Drop for PidfileGuard { + fn drop(&mut self) { + drop(self.file.unlock()); + } +} + +fn open_lock_file(path: &Path) -> Result<File> { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent) + .wrap_err_with(|| format!("could not create lock directory {}", parent.display()))?; + } + + OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(path) + .wrap_err_with(|| format!("could not open lock file {}", path.display())) +} diff --git a/crates/daemon/src/server.rs b/crates/daemon/src/server.rs new file mode 100644 index 00000000..d3427769 --- /dev/null +++ b/crates/daemon/src/server.rs @@ -0,0 +1,109 @@ +use std::{os::unix::net::SocketAddr, path::PathBuf}; + +use eyre::Result; +use eyre::{OptionExt, WrapErr}; +use turtle_api::generated::{ + control::control_server::ControlServer, history::history_server::HistoryServer, +}; + +use crate::api::{control::ControlService, history::HistoryService}; +use crate::{aclient::settings::Settings, daemon::DaemonHandle}; + +/// Run the gRPC server with the given services. +/// +/// This starts the gRPC server in the background and returns immediately. +/// The server will shut down when a [`ShutdownRequested`] event is received. +#[cfg(unix)] +pub(crate) fn run_grpc_server( + settings: &Settings, + history_service: HistoryServer<HistoryService>, + control_service: ControlServer<ControlService>, + handle: DaemonHandle, +) -> Result<()> { + use tokio::net::UnixListener; + use tokio_stream::wrappers::UnixListenerStream; + + let socket_path = settings.daemon.socket_path.clone(); + + let (uds, cleanup) = if settings.daemon.systemd_socket { + tracing::info!("getting systemd socket"); + let listener = listenfd::ListenFd::from_env() + .take_unix_listener(0)? + .ok_or_eyre("missing systemd socket")?; + listener.set_nonblocking(true)?; + let actual_path: Result<PathBuf, eyre::Report> = listener + .local_addr() + .context("getting systemd socket's path") + .and_then(|addr: SocketAddr| { + addr.as_pathname() + .ok_or_eyre("systemd socket missing path") + .map(|path: &std::path::Path| path.to_owned()) + }); + match actual_path { + Ok(actual_path) => { + tracing::info!("listening on systemd socket: {actual_path:?}"); + if actual_path != std::path::Path::new(&socket_path) { + tracing::warn!( + "systemd socket is not at configured client path: {socket_path:?}" + ); + } + } + Err(err) => { + tracing::warn!( + "could not detect systemd socket path, ensure that it's at the configured path: {socket_path:?}, error: {err:?}" + ); + } + } + (UnixListener::from_std(listener)?, false) + } else { + tracing::info!("listening on unix socket {socket_path:?}"); + ( + UnixListener::bind(socket_path.clone()).with_context(|| { + format!("Failed to bind to unix socket at: {socket_path}") + })?, + true, + ) + }; + + let uds_stream = UnixListenerStream::new(uds); + + // Create shutdown signal from daemon handle + let shutdown_signal = async move { + let mut rx = handle.subscribe(); + + loop { + use crate::events::DaemonEvent; + + match rx.recv().await { + Err(_) | Ok(DaemonEvent::ShutdownRequested) => break, + Ok(_) => (), + } + } + + if cleanup { + eprintln!("Removing socket..."); + if let Err(e) = std::fs::remove_file(&socket_path) + && e.kind() != std::io::ErrorKind::NotFound + { + eprintln!("failed to remove socket: {e}"); + } + } + eprintln!("Shutting down gRPC server..."); + }; + + // Spawn the server in the background + tokio::spawn(async move { + use tonic::transport::Server; + + if let Err(e) = Server::builder() + .add_service(history_service) + .add_service(control_service) + .serve_with_incoming_shutdown(uds_stream, shutdown_signal) + .await + { + tracing::error!("gRPC server error: {e}"); + } + }); + + Ok(()) +} |