aboutsummaryrefslogtreecommitdiffstats
path: root/crates/daemon/src/aclient
diff options
context:
space:
mode:
Diffstat (limited to 'crates/daemon/src/aclient')
-rw-r--r--crates/daemon/src/aclient/api_client.rs201
-rw-r--r--crates/daemon/src/aclient/database/mod.rs214
-rw-r--r--crates/daemon/src/aclient/encryption.rs142
-rw-r--r--crates/daemon/src/aclient/history/mod.rs374
-rw-r--r--crates/daemon/src/aclient/history/store.rs296
-rw-r--r--crates/daemon/src/aclient/meta.rs182
-rw-r--r--crates/daemon/src/aclient/mod.rs9
-rw-r--r--crates/daemon/src/aclient/record/encryption.rs379
-rw-r--r--crates/daemon/src/aclient/record/mod.rs3
-rw-r--r--crates/daemon/src/aclient/record/sqlite_store.rs412
-rw-r--r--crates/daemon/src/aclient/record/sync.rs453
-rw-r--r--crates/daemon/src/aclient/settings/meta.rs17
-rw-r--r--crates/daemon/src/aclient/settings/mod.rs590
-rw-r--r--crates/daemon/src/aclient/utils.rs77
14 files changed, 3349 insertions, 0 deletions
diff --git a/crates/daemon/src/aclient/api_client.rs b/crates/daemon/src/aclient/api_client.rs
new file mode 100644
index 00000000..1eba51bd
--- /dev/null
+++ b/crates/daemon/src/aclient/api_client.rs
@@ -0,0 +1,201 @@
+use std::env;
+use std::time::Duration;
+
+use eyre::{Result, bail, eyre};
+use reqwest::{Response, StatusCode, Url, header::HeaderMap};
+use tracing::debug;
+use uuid::Uuid;
+
+use turtle_common::{api::ErrorResponse, record::RecordStatus};
+use turtle_common::{
+ api::{ATUIN_CARGO_VERSION, ATUIN_HEADER_VERSION, ATUIN_VERSION},
+ record::{EncryptedData, HostId, Record, RecordIdx},
+};
+
+use semver::Version;
+
+static APP_USER_AGENT: &str = concat!("atuin/", env!("CARGO_PKG_VERSION"),);
+
+pub(crate) struct Client<'a> {
+ sync_addr: &'a str,
+ user_id: Uuid,
+ inner: reqwest::Client,
+}
+
+fn make_url(address: &str, path: &str, user_id: Uuid) -> Result<String> {
+ let address = address.strip_suffix('/').unwrap_or(address);
+
+ // `join()` expects a trailing `/` in order to join paths
+ // e.g. it treats `http://host:port/subdir` as a file called `subdir`
+ let address = &format!("{address}/api/v0/{user_id}/");
+
+ // passing a path with a leading `/` will cause `join()` to replace the entire URL path
+ let path = path.strip_prefix("/").unwrap_or(path);
+
+ let url = Url::parse(address)
+ .map(|url| url.join(path))?
+ .map_err(|_| eyre!("invalid address"))?;
+
+ Ok(url.to_string())
+}
+
+fn ensure_version(response: &Response) -> Result<bool> {
+ let version = response.headers().get(ATUIN_HEADER_VERSION);
+
+ let version = if let Some(version) = version {
+ match version.to_str() {
+ Ok(v) => Version::parse(v),
+ Err(e) => bail!("failed to parse server version: {:?}", e),
+ }
+ } else {
+ bail!("Server not reporting its version: it is either too old or unhealthy");
+ }?;
+
+ // If the client is newer than the server
+ if version.major < ATUIN_VERSION.major {
+ println!(
+ "Atuin version mismatch! In order to successfully sync, the server needs to run a newer version of Atuin"
+ );
+ println!("Client: {ATUIN_CARGO_VERSION}");
+ println!("Server: {version}");
+
+ return Ok(false);
+ }
+
+ Ok(true)
+}
+
+async fn handle_resp_error(resp: Response) -> Result<Response> {
+ let status = resp.status();
+ let url = resp.url().to_string();
+
+ if status == StatusCode::SERVICE_UNAVAILABLE {
+ bail!(
+ "Service unavailable: check https://status.atuin.sh (or get in touch with your host)"
+ );
+ }
+
+ if status == StatusCode::TOO_MANY_REQUESTS {
+ bail!("Rate limited; please wait before doing that again");
+ }
+
+ if !status.is_success() {
+ if let Ok(error) = resp.json::<ErrorResponse<'_>>().await {
+ let reason = error.reason;
+
+ if status.is_client_error() {
+ bail!("Invalid request to the service at {url}, {status} - {reason}.")
+ }
+
+ bail!(
+ "There was an error with the atuin sync service at {url}, server error {status}: {reason}.\nIf the problem persists, contact the host"
+ )
+ }
+
+ bail!(
+ "There was an error with the atuin sync service at {url}, Status {status:?}.\nIf the problem persists, contact the host"
+ )
+ }
+
+ 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,
+ connect_timeout: u64,
+ timeout: u64,
+ user_id: Uuid,
+ ) -> Result<Self> {
+ ensure_crypto_provider();
+ let mut headers = HeaderMap::new();
+
+ // used for semver server check
+ headers.insert(ATUIN_HEADER_VERSION, ATUIN_CARGO_VERSION.parse()?);
+
+ Ok(Client {
+ user_id,
+ sync_addr,
+ inner: reqwest::Client::builder()
+ .user_agent(APP_USER_AGENT)
+ .default_headers(headers)
+ .connect_timeout(Duration::new(connect_timeout, 0))
+ .timeout(Duration::new(timeout, 0))
+ .build()?,
+ })
+ }
+
+ 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())?;
+
+ debug!("uploading {} records to {url}", records.len());
+
+ let resp = self.inner.post(url).json(records).send().await?;
+ handle_resp_error(resp).await?;
+
+ Ok(())
+ }
+
+ pub(crate) async fn next_records(
+ &self,
+ host: HostId,
+ tag: String,
+ start: RecordIdx,
+ count: u64,
+ ) -> Result<Vec<Record<EncryptedData>>> {
+ debug!("fetching record/s from host {}/{}/{}", host.0, tag, start);
+
+ let url = make_url(
+ self.sync_addr,
+ &format!(
+ "/record/next?host={}&tag={}&count={}&start={}",
+ host.0, tag, count, start
+ ),
+ self.user_id,
+ )?;
+
+ let url = Url::parse(url.as_str())?;
+
+ let resp = self.inner.get(url).send().await?;
+ let resp = handle_resp_error(resp).await?;
+
+ let records = resp.json::<Vec<Record<EncryptedData>>>().await?;
+
+ Ok(records)
+ }
+
+ pub(crate) async fn record_status(&self) -> Result<RecordStatus> {
+ let url = make_url(self.sync_addr, "/record", self.user_id)?;
+ let url = Url::parse(url.as_str())?;
+
+ let resp = self.inner.get(url).send().await?;
+ let resp = handle_resp_error(resp).await?;
+
+ if !ensure_version(&resp)? {
+ bail!("could not sync records due to version mismatch");
+ }
+
+ let index = resp.json().await?;
+
+ debug!("got remote index {index:?}");
+
+ Ok(index)
+ }
+}
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/daemon/src/aclient/encryption.rs b/crates/daemon/src/aclient/encryption.rs
new file mode 100644
index 00000000..45e82ab3
--- /dev/null
+++ b/crates/daemon/src/aclient/encryption.rs
@@ -0,0 +1,142 @@
+// The general idea is that we NEVER send cleartext history to the server
+// This way the odds of anything private ending up where it should not are
+// very low
+// The server authenticates via the usual username and password. This has
+// nothing to do with the encryption, and is purely authentication! The client
+// generates its own secret key, and encrypts all shell history with libsodium's
+// secretbox. The data is then sent to the server, where it is stored. All
+// clients must share the secret in order to be able to sync, as it is needed
+// to decrypt
+
+use std::io::prelude::Write;
+
+use base64::prelude::{BASE64_STANDARD, Engine};
+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::aclient::settings::Settings;
+
+fn generate_encoded_key() -> Result<(Key, String)> {
+ let key = XSalsa20Poly1305::generate_key(&mut OsRng);
+ let encoded = encode_key(&key)?;
+
+ Ok((key, encoded))
+}
+
+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() {
+ let (key, encoded) = generate_encoded_key()?;
+
+ let mut file = fs::File::create(path)?;
+ file.write_all(encoded.as_bytes())?;
+
+ Ok(key)
+ } else {
+ bail!("No key-path set, cannot generate key")
+ }
+}
+
+// Loads the secret key, will create + save if it doesn't exist
+pub(crate) fn load_key(settings: &Settings) -> Result<Key> {
+ if let Some(key) = settings.sync.encryption_key()? {
+ Ok(key)
+ } else {
+ Ok(new_key(settings)?)
+ }
+}
+
+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")?;
+ for b in key {
+ rmp::encode::write_uint(&mut buf, u64::from(*b))
+ .wrap_err("could not encode key to message pack")?;
+ }
+ let buf = BASE64_STANDARD.encode(buf);
+
+ Ok(buf)
+}
+
+pub(crate) fn decode_key(key: &str) -> Result<Key> {
+ use rmp::decode;
+
+ let buf = BASE64_STANDARD
+ .decode(key.trim_end())
+ .wrap_err("encryption key is not a valid base64 encoding")?;
+
+ // old code wrote the key as a fixed length array of 32 bytes
+ // new code writes the key with a length prefix
+ if let Ok(key) = <[u8; 32]>::try_from(&*buf) {
+ Ok(key.into())
+ } else {
+ let mut bytes = decode::Bytes::new(&buf);
+
+ match Marker::from_u8(buf[0]) {
+ Marker::Bin8 => {
+ let len = decode::read_bin_len(&mut bytes).map_err(|err| eyre!("{err:?}"))?;
+ ensure!(len == 32, "encryption key is not the correct size");
+ let key = <[u8; 32]>::try_from(bytes.remaining_slice())
+ .context("could not decode encryption key")?;
+ Ok(key.into())
+ }
+ Marker::Array16 => {
+ let len = decode::read_array_len(&mut bytes).map_err(|err| eyre!("{err:?}"))?;
+ ensure!(len == 32, "encryption key is not the correct size");
+
+ let mut key = Key::default();
+ for i in &mut key {
+ *i = decode::read_int(&mut bytes).map_err(|err| eyre!("{err:?}"))?;
+ }
+ Ok(key)
+ }
+ _ => bail!("could not decode encryption key"),
+ }
+ }
+}
+
+#[cfg(test)]
+mod test {
+ #[test]
+ fn key_encodings() {
+ use super::{Key, decode_key, encode_key};
+
+ // a history of our key encodings.
+ // v11.0.0 xCAbWypb0msJ2Kq+8j4GVEWUlDX7deKnrTRSIopuqXxc5Q==
+ // v12.0.0 xCAbWypb0msJ2Kq+8j4GVEWUlDX7deKnrTRSIopuqXxc5Q==
+ // v13.0.0 xCAbWypb0msJ2Kq+8j4GVEWUlDX7deKnrTRSIopuqXxc5Q==
+ // v13.0.1 xCAbWypb0msJ2Kq+8j4GVEWUlDX7deKnrTRSIopuqXxc5Q==
+ // v14.0.0 xCAbWypb0msJ2Kq+8j4GVEWUlDX7deKnrTRSIopuqXxc5Q==
+ // v14.0.1 xCAbWypb0msJ2Kq+8j4GVEWUlDX7deKnrTRSIopuqXxc5Q==
+ // c7d89c1 3AAgG1sqW8zSawnM2MyqzL7M8j4GVEXMlMyUNcz7dczizKfMrTRSIsyKbsypfFzM5Q== (https://github.com/ellie/atuin/pull/805)
+ // b53ca35 3AAgG1sqW8zSawnM2MyqzL7M8j4GVEXMlMyUNcz7dczizKfMrTRSIsyKbsypfFzM5Q== (https://github.com/ellie/atuin/pull/974)
+ // v15.0.0 3AAgG1sqW8zSawnM2MyqzL7M8j4GVEXMlMyUNcz7dczizKfMrTRSIsyKbsypfFzM5Q==
+ // b8b57c8 xCAbWypb0msJ2Kq+8j4GVEWUlDX7deKnrTRSIopuqXxc5Q== (https://github.com/ellie/atuin/pull/1057)
+ // 8c94d79 3AAgG1sqW8zSawnM2MyqzL7M8j4GVEXMlMyUNcz7dczizKfMrTRSIsyKbsypfFzM5Q== (https://github.com/ellie/atuin/pull/1089)
+
+ let key = Key::from([
+ 27, 91, 42, 91, 210, 107, 9, 216, 170, 190, 242, 62, 6, 84, 69, 148, 148, 53, 251, 117,
+ 226, 167, 173, 52, 82, 34, 138, 110, 169, 124, 92, 229,
+ ]);
+
+ assert_eq!(
+ encode_key(&key).unwrap(),
+ "3AAgG1sqW8zSawnM2MyqzL7M8j4GVEXMlMyUNcz7dczizKfMrTRSIsyKbsypfFzM5Q=="
+ );
+
+ // key encodings we have to support
+ let valid_encodings = [
+ "xCAbWypb0msJ2Kq+8j4GVEWUlDX7deKnrTRSIopuqXxc5Q==",
+ "3AAgG1sqW8zSawnM2MyqzL7M8j4GVEXMlMyUNcz7dczizKfMrTRSIsyKbsypfFzM5Q==",
+ ];
+
+ for k in valid_encodings {
+ assert_eq!(decode_key(k).expect(k), key);
+ }
+ }
+}
diff --git a/crates/daemon/src/aclient/history/mod.rs b/crates/daemon/src/aclient/history/mod.rs
new file mode 100644
index 00000000..0b5f4982
--- /dev/null
+++ b/crates/daemon/src/aclient/history/mod.rs
@@ -0,0 +1,374 @@
+use std::time::Duration;
+
+use rmp::decode::DecodeStringError;
+use rmp::decode::ValueReadError;
+use rmp::{Marker, decode::Bytes};
+use turtle_api::history::History;
+
+use turtle_common::record::DecryptedData;
+
+use eyre::{Result, bail, eyre};
+
+use time::OffsetDateTime;
+
+pub(crate) mod store;
+
+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;
+const HISTORY_VERSION: &str = HISTORY_VERSION_V1;
+const HISTORY_TAG: &str = "history";
+
+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 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.
+
+ use rmp::encode;
+
+ let mut output = vec![];
+
+ // write the version
+ encode::write_u16(&mut output, HISTORY_RECORD_VERSION_V1)?;
+ 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.to_string())?;
+ encode::write_u64(&mut output, self.timestamp.unix_timestamp_nanos() as u64)?;
+ 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)?;
+ encode::write_str(&mut output, &self.session)?;
+ encode::write_str(&mut output, &self.hostname)?;
+
+ match self.deleted_at {
+ Some(d) => encode::write_u64(&mut output, d.unix_timestamp_nanos() as u64)?,
+ None => encode::write_nil(&mut output)?,
+ }
+
+ encode::write_str(&mut output, self.author.as_str())?;
+ if let Some(intent) = &self.intent {
+ encode::write_str(&mut output, intent.as_str())?;
+ }
+
+ Ok(DecryptedData(output))
+ }
+
+ fn read_optional_string(bytes: &[u8]) -> Result<(Option<String>, &[u8])> {
+ use rmp::decode;
+
+ fn error_report<E: std::fmt::Debug>(err: E) -> eyre::Report {
+ eyre!("{err:?}")
+ }
+
+ match decode::read_str_from_slice(bytes) {
+ Ok((value, bytes)) => Ok((Some(value.to_owned()), bytes)),
+ Err(DecodeStringError::TypeMismatch(Marker::Null)) => {
+ let mut cursor = Bytes::new(bytes);
+ decode::read_nil(&mut cursor).map_err(error_report)?;
+
+ Ok((None, cursor.remaining_slice()))
+ }
+ Err(err) => Err(error_report(err)),
+ }
+ }
+
+ fn deserialize_v0(bytes: &[u8]) -> Result<Self> {
+ use rmp::decode;
+
+ fn error_report<E: std::fmt::Debug>(err: E) -> eyre::Report {
+ eyre!("{err:?}")
+ }
+
+ let mut bytes = Bytes::new(bytes);
+
+ let version = decode::read_u16(&mut bytes).map_err(error_report)?;
+
+ if version != HISTORY_RECORD_VERSION_V0 {
+ bail!("expected decoding v0 record, found v{version}");
+ }
+
+ let nfields = decode::read_array_len(&mut bytes).map_err(error_report)?;
+
+ if nfields != 9 {
+ bail!("cannot decrypt history from a different version of Atuin");
+ }
+
+ let bytes = bytes.remaining_slice();
+ let (id, bytes) = decode::read_str_from_slice(bytes).map_err(error_report)?;
+
+ 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(|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();
+ let (command, bytes) = decode::read_str_from_slice(bytes).map_err(error_report)?;
+ let (cwd, bytes) = decode::read_str_from_slice(bytes).map_err(error_report)?;
+ let (session, bytes) = decode::read_str_from_slice(bytes).map_err(error_report)?;
+ let (hostname, bytes) = decode::read_str_from_slice(bytes).map_err(error_report)?;
+
+ let mut bytes = Bytes::new(bytes);
+
+ let (deleted_at, bytes) = match decode::read_u64(&mut bytes) {
+ Ok(unix) => (Some(unix), bytes.remaining_slice()),
+ // we accept null here
+ Err(ValueReadError::TypeMismatch(Marker::Null)) => (None, bytes.remaining_slice()),
+ Err(err) => return Err(error_report(err)),
+ };
+ if !bytes.is_empty() {
+ bail!("trailing bytes in encoded history. malformed")
+ }
+
+ Ok(Self {
+ id: id.to_owned().into(),
+ timestamp: OffsetDateTime::from_unix_timestamp_nanos(i128::from(timestamp))?,
+ duration,
+ exit,
+ command: command.to_owned(),
+ cwd: cwd.to_owned(),
+ session: session.to_owned(),
+ hostname: hostname.to_owned(),
+ author: Self::author_from_hostname(hostname),
+ intent: None,
+ deleted_at: deleted_at
+ .map(|t| OffsetDateTime::from_unix_timestamp_nanos(i128::from(t)))
+ .transpose()?,
+ })
+ }
+
+ fn deserialize_v1(bytes: &[u8]) -> Result<Self> {
+ use rmp::decode;
+
+ fn error_report<E: std::fmt::Debug>(err: E) -> eyre::Report {
+ eyre!("{err:?}")
+ }
+
+ let mut bytes = Bytes::new(bytes);
+
+ let version = decode::read_u16(&mut bytes).map_err(error_report)?;
+
+ if version != HISTORY_RECORD_VERSION_V1 {
+ bail!("expected decoding v1 record, found v{version}");
+ }
+
+ let nfields = decode::read_array_len(&mut bytes).map_err(error_report)?;
+
+ if !(10..=11).contains(&nfields) {
+ bail!("cannot decrypt history from a different version of Atuin");
+ }
+
+ let bytes = bytes.remaining_slice();
+ let (id, bytes) = decode::read_str_from_slice(bytes).map_err(error_report)?;
+
+ 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(|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();
+ let (command, bytes) = decode::read_str_from_slice(bytes).map_err(error_report)?;
+ let (cwd, bytes) = decode::read_str_from_slice(bytes).map_err(error_report)?;
+ let (session, bytes) = decode::read_str_from_slice(bytes).map_err(error_report)?;
+ let (hostname, bytes) = decode::read_str_from_slice(bytes).map_err(error_report)?;
+
+ let mut bytes = Bytes::new(bytes);
+
+ let (deleted_at, bytes) = match decode::read_u64(&mut bytes) {
+ Ok(unix) => (Some(unix), bytes.remaining_slice()),
+ // we accept null here
+ Err(ValueReadError::TypeMismatch(Marker::Null)) => (None, bytes.remaining_slice()),
+ Err(err) => return Err(error_report(err)),
+ };
+ let (author, bytes) = Self::read_optional_string(bytes)?;
+ let (intent, bytes) = if nfields > 10 {
+ Self::read_optional_string(bytes)?
+ } else {
+ (None, bytes)
+ };
+
+ if !bytes.is_empty() {
+ bail!("trailing bytes in encoded history. malformed")
+ }
+
+ Ok(Self {
+ id: id.to_owned().into(),
+ timestamp: OffsetDateTime::from_unix_timestamp_nanos(i128::from(timestamp))?,
+ duration,
+ exit,
+ command: command.to_owned(),
+ cwd: cwd.to_owned(),
+ session: session.to_owned(),
+ hostname: hostname.to_owned(),
+ author: author.unwrap_or_else(|| Self::author_from_hostname(hostname)),
+ intent,
+ deleted_at: deleted_at
+ .map(|t| OffsetDateTime::from_unix_timestamp_nanos(i128::from(t)))
+ .transpose()?,
+ })
+ }
+
+ fn deserialize(bytes: &[u8], version: &str) -> Result<Self> {
+ match version {
+ HISTORY_VERSION_V0 => Self::deserialize_v0(bytes),
+ HISTORY_VERSION_V1 => Self::deserialize_v1(bytes),
+
+ _ => bail!("unknown version {version:?}"),
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::time::Duration;
+
+ use time::macros::datetime;
+
+ use crate::aclient::history::{HISTORY_VERSION, HistoryExt};
+
+ use super::History;
+
+ #[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: Duration::from_nanos(49_206_000),
+ exit: 0,
+ command: "git status".to_owned(),
+ cwd: "/Users/conrad.ludgate/Documents/code/atuin".to_owned(),
+ session: "b97d9a306f274473a203d2eba41f9457".to_owned(),
+ hostname: "fvfg936c0kpf:conrad.ludgate".to_owned(),
+ author: "conrad.ludgate".to_owned(),
+ intent: None,
+ deleted_at: None,
+ };
+
+ let serialized = history.serialize().expect("failed to serialize history");
+ assert_eq!(
+ &serialized.0[0..3],
+ [205, 0, 1],
+ "should encode as history v1"
+ );
+
+ let deserialized = History::deserialize(&serialized.0, HISTORY_VERSION)
+ .expect("failed to deserialize history");
+ assert_eq!(history, deserialized);
+ }
+
+ #[test]
+ fn test_serialize_deserialize_deleted() {
+ let history = History {
+ id: "66d16cbee7cd47538e5c5b8b44e9006e".to_owned().into(),
+ timestamp: datetime!(2023-05-28 18:35:40.633872 +00:00),
+ duration: Duration::from_nanos(49_206_000),
+ exit: 0,
+ command: "git status".to_owned(),
+ cwd: "/Users/conrad.ludgate/Documents/code/atuin".to_owned(),
+ session: "b97d9a306f274473a203d2eba41f9457".to_owned(),
+ hostname: "fvfg936c0kpf:conrad.ludgate".to_owned(),
+ author: "conrad.ludgate".to_owned(),
+ intent: None,
+ deleted_at: Some(datetime!(2023-11-19 20:18 +00:00)),
+ };
+
+ let serialized = history.serialize().expect("failed to serialize history");
+
+ let deserialized = History::deserialize(&serialized.0, HISTORY_VERSION)
+ .expect("failed to deserialize history");
+
+ assert_eq!(history, deserialized);
+ }
+
+ #[test]
+ fn test_serialize_deserialize_with_author_and_intent() {
+ let history = History {
+ id: "66d16cbee7cd47538e5c5b8b44e9006e".to_owned().into(),
+ timestamp: datetime!(2023-05-28 18:35:40.633872 +00:00),
+ duration: Duration::from_nanos(49_206_000),
+ exit: 0,
+ command: "git status".to_owned(),
+ cwd: "/Users/conrad.ludgate/Documents/code/atuin".to_owned(),
+ session: "b97d9a306f274473a203d2eba41f9457".to_owned(),
+ hostname: "fvfg936c0kpf:conrad.ludgate".to_owned(),
+ author: "claude".to_owned(),
+ intent: Some("check repository status".to_owned()),
+ deleted_at: None,
+ };
+
+ let serialized = history.serialize().expect("failed to serialize history");
+ let deserialized = History::deserialize(&serialized.0, HISTORY_VERSION)
+ .expect("failed to deserialize history");
+
+ assert_eq!(history, deserialized);
+ }
+
+ #[test]
+ fn test_serialize_deserialize_version() {
+ // v0
+ let bytes_v0 = [
+ 205, 0, 0, 153, 217, 32, 54, 54, 100, 49, 54, 99, 98, 101, 101, 55, 99, 100, 52, 55,
+ 53, 51, 56, 101, 53, 99, 53, 98, 56, 98, 52, 52, 101, 57, 48, 48, 54, 101, 207, 23, 99,
+ 98, 117, 24, 210, 246, 128, 206, 2, 238, 210, 240, 0, 170, 103, 105, 116, 32, 115, 116,
+ 97, 116, 117, 115, 217, 42, 47, 85, 115, 101, 114, 115, 47, 99, 111, 110, 114, 97, 100,
+ 46, 108, 117, 100, 103, 97, 116, 101, 47, 68, 111, 99, 117, 109, 101, 110, 116, 115,
+ 47, 99, 111, 100, 101, 47, 97, 116, 117, 105, 110, 217, 32, 98, 57, 55, 100, 57, 97,
+ 51, 48, 54, 102, 50, 55, 52, 52, 55, 51, 97, 50, 48, 51, 100, 50, 101, 98, 97, 52, 49,
+ 102, 57, 52, 53, 55, 187, 102, 118, 102, 103, 57, 51, 54, 99, 48, 107, 112, 102, 58,
+ 99, 111, 110, 114, 97, 100, 46, 108, 117, 100, 103, 97, 116, 101, 192,
+ ];
+
+ let deserialized = History::deserialize(&bytes_v0, "v0");
+ assert!(deserialized.is_ok());
+
+ let deserialized = History::deserialize(&bytes_v0, HISTORY_VERSION);
+ assert!(deserialized.is_err());
+
+ let current = History {
+ id: "66d16cbee7cd47538e5c5b8b44e9006e".to_owned().into(),
+ timestamp: datetime!(2023-05-28 18:35:40.633872 +00:00),
+ duration: Duration::from_nanos(49_206_000),
+ exit: 0,
+ command: "git status".to_owned(),
+ cwd: "/Users/conrad.ludgate/Documents/code/atuin".to_owned(),
+ session: "b97d9a306f274473a203d2eba41f9457".to_owned(),
+ hostname: "fvfg936c0kpf:conrad.ludgate".to_owned(),
+ author: "conrad.ludgate".to_owned(),
+ intent: None,
+ deleted_at: None,
+ };
+
+ let bytes_v1 = current.serialize().expect("failed to serialize history");
+ let deserialized = History::deserialize(&bytes_v1.0, HISTORY_VERSION);
+ assert!(deserialized.is_ok());
+
+ let deserialized = History::deserialize(&bytes_v1.0, "v0");
+ assert!(deserialized.is_err());
+ }
+}
diff --git a/crates/daemon/src/aclient/history/store.rs b/crates/daemon/src/aclient/history/store.rs
new file mode 100644
index 00000000..a6d6a627
--- /dev/null
+++ b/crates/daemon/src/aclient/history/store.rs
@@ -0,0 +1,296 @@
+use eyre::{Result, bail, eyre};
+use rmp::decode::Bytes;
+use turtle_api::history::{History, HistoryId};
+
+use crate::aclient::{
+ database::ClientSqlite,
+ history::HistoryExt,
+ record::{encryption::PASETO_V4, sqlite_store::SqliteStore},
+};
+use turtle_common::record::{DecryptedData, Host, HostId, Record, RecordId, RecordIdx};
+
+use super::{HISTORY_TAG, HISTORY_VERSION, HISTORY_VERSION_V0};
+
+#[derive(Debug, Clone)]
+pub(crate) struct HistoryStore {
+ store: SqliteStore,
+ host_id: HostId,
+ encryption_key: [u8; 32],
+}
+
+#[derive(Debug, Eq, PartialEq, Clone)]
+enum HistoryRecord {
+ Create(History), // Create a history record
+ Delete(HistoryId), // Delete a history record, identified by ID
+}
+
+impl HistoryRecord {
+ /// Serialize a history record, returning `DecryptedData`
+ /// The record will be of a certain type
+ /// We map those like so:
+ ///
+ /// `HistoryRecord::Create` -> 0
+ /// `HistoryRecord::Delete`-> 1
+ ///
+ /// This numeric identifier is then written as the first byte to the buffer. For history, we
+ /// append the serialized history right afterwards, to avoid having to handle serialization
+ /// twice.
+ ///
+ /// Deletion simply refers to the history by ID
+ 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;
+
+ let mut output = vec![];
+
+ match self {
+ Self::Create(history) => {
+ // 0 -> a history create
+ encode::write_u8(&mut output, 0)?;
+
+ let bytes = history.serialize()?;
+
+ encode::write_bin(&mut output, &bytes.0)?;
+ }
+ Self::Delete(id) => {
+ // 1 -> a history delete
+ encode::write_u8(&mut output, 1)?;
+ encode::write_str(&mut output, id.to_string().as_str())?;
+ }
+ }
+
+ Ok(DecryptedData(output))
+ }
+
+ fn deserialize(bytes: &DecryptedData, version: &str) -> Result<Self> {
+ use rmp::decode;
+
+ fn error_report<E: std::fmt::Debug>(err: E) -> eyre::Report {
+ eyre!("{err:?}")
+ }
+
+ let mut bytes = Bytes::new(&bytes.0);
+
+ let record_type = decode::read_u8(&mut bytes).map_err(error_report)?;
+
+ match record_type {
+ // 0 -> HistoryRecord::Create
+ 0 => {
+ // not super useful to us atm, but perhaps in the future
+ // written by write_bin above
+ decode::read_bin_len(&mut bytes).map_err(error_report)?;
+
+ let record = History::deserialize(bytes.remaining_slice(), version)?;
+
+ Ok(Self::Create(record))
+ }
+
+ // 1 -> HistoryRecord::Delete
+ 1 => {
+ let bytes = bytes.remaining_slice();
+ let (id, bytes) = decode::read_str_from_slice(bytes).map_err(error_report)?;
+
+ if !bytes.is_empty() {
+ bail!(
+ "trailing bytes decoding HistoryRecord::Delete - malformed? got {bytes:?}"
+ );
+ }
+
+ Ok(Self::Delete(id.to_string().into()))
+ }
+
+ n => {
+ bail!("unknown HistoryRecord type {n}")
+ }
+ }
+ }
+}
+
+impl HistoryStore {
+ pub(crate) fn new(store: SqliteStore, host_id: HostId, encryption_key: [u8; 32]) -> Self {
+ Self {
+ store,
+ host_id,
+ encryption_key,
+ }
+ }
+
+ async fn push_record(&self, record: HistoryRecord) -> Result<(RecordId, RecordIdx)> {
+ let bytes = record.serialize()?;
+ let idx = self
+ .store
+ .last(self.host_id, HISTORY_TAG)
+ .await?
+ .map_or(0, |p| p.idx + 1);
+
+ let record = Record::builder()
+ .host(Host::new(self.host_id))
+ .version(HISTORY_VERSION.to_string())
+ .tag(HISTORY_TAG.to_string())
+ .idx(idx)
+ .data(bytes)
+ .build();
+
+ let id = record.id;
+
+ self.store
+ .push(&record.encrypt::<PASETO_V4>(&self.encryption_key))
+ .await?;
+
+ Ok((id, idx))
+ }
+
+ 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
+ let record = HistoryRecord::Create(history);
+
+ self.push_record(record).await
+ }
+
+ // 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,
+ database: &ClientSqlite,
+ ids: &[RecordId],
+ ) -> Result<()> {
+ for id in ids {
+ let record = self.store.get(*id).await;
+
+ if let Ok(record) = record {
+ if record.tag != HISTORY_TAG {
+ continue;
+ }
+
+ let version = record.version.clone();
+ let decrypted = record.decrypt::<PASETO_V4>(&self.encryption_key)?;
+ let record = match version.as_str() {
+ HISTORY_VERSION_V0 | HISTORY_VERSION => {
+ HistoryRecord::deserialize(&decrypted.data, version.as_str())?
+ }
+ version => bail!("unknown history version {version:?}"),
+ };
+
+ match record {
+ HistoryRecord::Create(h) => {
+ // TODO: benchmark CPU time/memory tradeoff of batch commit vs one at a time
+ database.save(&h).await?;
+ }
+ HistoryRecord::Delete(id) => {
+ database.delete_rows(&[id]).await?;
+ }
+ }
+ }
+ }
+
+ Ok(())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::time::Duration;
+
+ use time::macros::datetime;
+ use turtle_common::record::DecryptedData;
+
+ use crate::aclient::history::{HISTORY_VERSION, store::HistoryRecord};
+
+ use super::History;
+
+ #[test]
+ fn test_serialize_deserialize_create() {
+ let bytes = [
+ 204, 0, 196, 147, 205, 0, 1, 154, 217, 32, 48, 49, 56, 99, 100, 52, 102, 101, 56, 49,
+ 55, 53, 55, 99, 100, 50, 97, 101, 101, 54, 53, 99, 100, 55, 56, 54, 49, 102, 57, 99,
+ 56, 49, 207, 23, 166, 251, 212, 181, 82, 0, 0, 100, 0, 162, 108, 115, 217, 41, 47, 85,
+ 115, 101, 114, 115, 47, 101, 108, 108, 105, 101, 47, 115, 114, 99, 47, 103, 105, 116,
+ 104, 117, 98, 46, 99, 111, 109, 47, 97, 116, 117, 105, 110, 115, 104, 47, 97, 116, 117,
+ 105, 110, 217, 32, 48, 49, 56, 99, 100, 52, 102, 101, 97, 100, 56, 57, 55, 53, 57, 55,
+ 56, 53, 50, 53, 50, 55, 97, 51, 49, 99, 57, 57, 56, 48, 53, 57, 170, 98, 111, 111, 112,
+ 58, 101, 108, 108, 105, 101, 192, 165, 101, 108, 108, 105, 101,
+ ];
+
+ let history = History {
+ id: "018cd4fe81757cd2aee65cd7861f9c81".to_owned().into(),
+ timestamp: datetime!(2024-01-04 00:00:00.000000 +00:00),
+ duration: Duration::from_nanos(100),
+ exit: 0,
+ command: "ls".to_owned(),
+ cwd: "/Users/ellie/src/github.com/atuinsh/atuin".to_owned(),
+ session: "018cd4fead897597852527a31c998059".to_owned(),
+ hostname: "boop:ellie".to_owned(),
+ author: "ellie".to_owned(),
+ intent: None,
+ deleted_at: None,
+ };
+
+ let record = HistoryRecord::Create(history);
+
+ let serialized = record.serialize().expect("failed to serialize history");
+ assert_eq!(serialized.0, bytes);
+
+ let deserialized = HistoryRecord::deserialize(&serialized, HISTORY_VERSION)
+ .expect("failed to deserialize HistoryRecord");
+ assert_eq!(deserialized, record);
+
+ // check the snapshot too
+ let deserialized =
+ HistoryRecord::deserialize(&DecryptedData(Vec::from(bytes)), HISTORY_VERSION)
+ .expect("failed to deserialize HistoryRecord");
+ assert_eq!(deserialized, record);
+ }
+
+ #[test]
+ fn test_serialize_deserialize_delete() {
+ let bytes = [
+ 204, 1, 217, 32, 48, 49, 56, 99, 100, 52, 102, 101, 56, 49, 55, 53, 55, 99, 100, 50,
+ 97, 101, 101, 54, 53, 99, 100, 55, 56, 54, 49, 102, 57, 99, 56, 49,
+ ];
+ let record = HistoryRecord::Delete("018cd4fe81757cd2aee65cd7861f9c81".to_string().into());
+
+ let serialized = record.serialize().expect("failed to serialize history");
+ assert_eq!(serialized.0, bytes);
+
+ let deserialized = HistoryRecord::deserialize(&serialized, HISTORY_VERSION)
+ .expect("failed to deserialize HistoryRecord");
+ assert_eq!(deserialized, record);
+
+ let deserialized =
+ HistoryRecord::deserialize(&DecryptedData(Vec::from(bytes)), HISTORY_VERSION)
+ .expect("failed to deserialize HistoryRecord");
+ assert_eq!(deserialized, record);
+ }
+}
diff --git a/crates/daemon/src/aclient/meta.rs b/crates/daemon/src/aclient/meta.rs
new file mode 100644
index 00000000..ea660745
--- /dev/null
+++ b/crates/daemon/src/aclient/meta.rs
@@ -0,0 +1,182 @@
+use std::path::Path;
+use std::str::FromStr;
+use std::time::Duration;
+
+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";
+const KEY_LAST_SYNC: &str = "last_sync_time";
+
+pub(crate) struct MetaStore {
+ pool: SqlitePool,
+ cached_host_id: OnceCell<HostId>,
+}
+
+impl MetaStore {
+ pub(crate) async fn new(path: impl AsRef<Path>, timeout: f64) -> Result<Self> {
+ let path = path.as_ref();
+ let path_str = path
+ .as_os_str()
+ .to_str()
+ .ok_or_else(|| eyre!("meta database path is not valid UTF-8: {path:?}"))?;
+ debug!("opening meta sqlite database at {path:?}");
+
+ let is_memory = path_str.contains(":memory:");
+
+ if !is_memory
+ && !path.exists()
+ && let Some(dir) = path.parent()
+ {
+ fs_err::create_dir_all(dir)?;
+ }
+
+ // Use DELETE journal mode instead of WAL. This is a small, infrequently-
+ // written KV store — WAL's concurrency benefits aren't needed, and DELETE
+ // mode avoids creating auxiliary -wal/-shm files that complicate
+ // permission handling.
+ let opts = SqliteConnectOptions::from_str(path_str)?
+ .journal_mode(SqliteJournalMode::Delete)
+ .optimize_on_close(true, None)
+ .create_if_missing(true);
+
+ let pool = SqlitePoolOptions::new()
+ .acquire_timeout(Duration::from_secs_f64(timeout))
+ .connect_with(opts)
+ .await?;
+
+ sqlx::migrate!("./db/client-meta-migrations")
+ .run(&pool)
+ .await?;
+
+ // Session tokens are stored in this database, so restrict permissions.
+ #[cfg(unix)]
+ if !is_memory {
+ use std::os::unix::fs::PermissionsExt;
+ std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?;
+ }
+
+ let store = Self {
+ pool,
+ cached_host_id: OnceCell::const_new(),
+ };
+
+ Ok(store)
+ }
+
+ // Generic key-value operations
+
+ 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)
+ .await?;
+
+ Ok(row.map(|r| r.0))
+ }
+
+ async fn set(&self, key: &str, value: &str) -> Result<()> {
+ sqlx::query(
+ "
+ INSERT INTO meta (key, value, updated_at)
+ VALUES (?1, ?2, strftime('%s', 'now'))
+ ON CONFLICT(key) DO UPDATE
+ SET value = ?2, updated_at = strftime('%s', 'now')
+ ",
+ )
+ .bind(key)
+ .bind(value)
+ .execute(&self.pool)
+ .await?;
+
+ Ok(())
+ }
+
+ // Typed accessors
+
+ pub(crate) async fn host_id(&self) -> Result<HostId> {
+ self.cached_host_id
+ .get_or_try_init(|| async {
+ if let Some(id) = self.get(KEY_HOST_ID).await? {
+ let parsed = Uuid::from_str(id.as_str())
+ .map_err(|e| eyre!("failed to parse host ID: {e}"))?;
+ return Ok(HostId(parsed));
+ }
+
+ let uuid = turtle_common::utils::uuid_v7();
+ self.set(KEY_HOST_ID, uuid.as_simple().to_string().as_ref())
+ .await?;
+
+ Ok(HostId(uuid))
+ })
+ .await
+ .copied()
+ }
+
+ pub(crate) async fn last_sync(&self) -> Result<OffsetDateTime> {
+ match self.get(KEY_LAST_SYNC).await? {
+ Some(v) => Ok(OffsetDateTime::parse(v.as_str(), &Rfc3339)?),
+ None => Ok(OffsetDateTime::UNIX_EPOCH),
+ }
+ }
+
+ pub(crate) async fn save_sync_time(&self) -> Result<()> {
+ self.set(
+ KEY_LAST_SYNC,
+ OffsetDateTime::now_utc().format(&Rfc3339)?.as_str(),
+ )
+ .await
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::{MetaStore, OffsetDateTime};
+
+ async fn new_test_store() -> MetaStore {
+ MetaStore::new("sqlite::memory:", 2.0).await.unwrap()
+ }
+
+ #[tokio::test]
+ async fn test_get_set_delete() {
+ let store = new_test_store().await;
+
+ assert_eq!(store.get("foo").await.unwrap(), None);
+
+ store.set("foo", "bar").await.unwrap();
+ assert_eq!(store.get("foo").await.unwrap(), Some("bar".to_string()));
+
+ 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);
+ }
+
+ #[tokio::test]
+ async fn test_host_id_generation_and_stability() {
+ let store = new_test_store().await;
+
+ let id1 = store.host_id().await.unwrap();
+ let id2 = store.host_id().await.unwrap();
+
+ assert_eq!(id1, id2, "host_id should be stable across calls");
+ }
+
+ #[tokio::test]
+ async fn test_sync_time() {
+ let store = new_test_store().await;
+
+ let t = store.last_sync().await.unwrap();
+ assert_eq!(t, OffsetDateTime::UNIX_EPOCH);
+
+ store.save_sync_time().await.unwrap();
+ let t = store.last_sync().await.unwrap();
+ assert!(t > OffsetDateTime::UNIX_EPOCH);
+ }
+}
diff --git a/crates/daemon/src/aclient/mod.rs b/crates/daemon/src/aclient/mod.rs
new file mode 100644
index 00000000..2c445945
--- /dev/null
+++ b/crates/daemon/src/aclient/mod.rs
@@ -0,0 +1,9 @@
+pub(crate) mod database;
+pub(crate) mod encryption;
+pub(crate) mod history;
+pub(crate) mod record;
+pub(crate) mod settings;
+
+mod api_client;
+mod meta;
+mod utils;
diff --git a/crates/daemon/src/aclient/record/encryption.rs b/crates/daemon/src/aclient/record/encryption.rs
new file mode 100644
index 00000000..11de96d5
--- /dev/null
+++ b/crates/daemon/src/aclient/record/encryption.rs
@@ -0,0 +1,379 @@
+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
+};
+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)]
+pub(crate) struct PASETO_V4;
+
+/*
+Why do we use a random content-encryption key?
+Originally I was planning on using a derived key for encryption based on additional data.
+This would be a lot more secure than using the master key directly.
+
+However, there's an established norm of using a random key. This scheme might be otherwise known as
+- client-side encryption
+- envelope encryption
+- key wrapping
+
+A HSM (Hardware Security Module) provider, eg: AWS, Azure, GCP, or even a physical device like a YubiKey
+will have some keys that they keep to themselves. These keys never leave their physical hardware.
+If they never leave the hardware, then encrypting large amounts of data means giving them the data and waiting.
+This is not a practical solution. Instead, generate a unique key for your data, encrypt that using your HSM
+and then store that with your data.
+
+See
+ - <https://docs.aws.amazon.com/wellarchitected/latest/financial-services-industry-lens/use-envelope-encryption-with-customer-master-keys.html>
+ - <https://cloud.google.com/kms/docs/envelope-encryption>
+ - <https://learn.microsoft.com/en-us/azure/storage/blobs/client-side-encryption?tabs=dotnet#encryption-and-decryption-via-the-envelope-technique>
+ - <https://www.yubico.com/gb/product/yubihsm-2-fips/>
+ - <https://cheatsheetseries.owasp.org/cheatsheets/Cryptographic_Storage_Cheat_Sheet.html#encrypting-stored-keys>
+
+Why would we care? In the past we have received some requests for company solutions. If in future we can configure a
+KMS service with little effort, then that would solve a lot of issues for their security team.
+
+Even for personal use, if a user is not comfortable with sharing keys between hosts,
+GCP HSM costs $1/month and $0.03 per 10,000 key operations. Assuming an active user runs
+1000 atuin records a day, that would only cost them $1 and 10 cent a month.
+
+Additionally, key rotations are much simpler using this scheme. Rotating a key is as simple as re-encrypting the CEK, and not the message contents.
+This makes it very fast to rotate a key in bulk.
+
+For future reference, with asymmetric encryption, you can encrypt the CEK without the HSM's involvement, but decrypting
+will need the HSM. This allows the encryption path to still be extremely fast (no network calls) but downloads/decryption
+that happens in the background can make the network calls to the HSM
+*/
+
+impl Encryption for PASETO_V4 {
+ fn re_encrypt(
+ mut data: EncryptedData,
+ _ad: AdditionalData<'_>,
+ old_key: &[u8; 32],
+ new_key: &[u8; 32],
+ ) -> Result<EncryptedData> {
+ let cek = Self::decrypt_cek(&data.content_encryption_key, old_key)?;
+ data.content_encryption_key = Self::encrypt_cek(cek, new_key);
+ Ok(data)
+ }
+
+ fn encrypt(data: DecryptedData, ad: AdditionalData<'_>, key: &[u8; 32]) -> EncryptedData {
+ // generate a random key for this entry
+ // aka content-encryption-key (CEK)
+ let random_key = Key::<V4, Local>::new_os_random();
+
+ let assertions = Assertions::from(ad).encode();
+
+ // build the payload and encrypt the token
+ let payload = serde_json::to_string(&AtuinPayload {
+ data: general_purpose::URL_SAFE_NO_PAD.encode(data.0),
+ })
+ .expect("json encoding can't fail");
+ let nonce = DataKey::<32>::try_new_random().expect("could not source from random");
+ let nonce = PasetoNonce::<V4, LocalPurpose>::from(&nonce);
+
+ let token = Paseto::<V4, LocalPurpose>::builder()
+ .set_payload(Payload::from(payload.as_str()))
+ .set_implicit_assertion(ImplicitAssertion::from(assertions.as_str()))
+ .try_encrypt(&random_key.into(), &nonce)
+ .expect("error encrypting atuin data");
+
+ EncryptedData {
+ data: token,
+ content_encryption_key: Self::encrypt_cek(random_key, key),
+ }
+ }
+
+ fn decrypt(
+ data: EncryptedData,
+ ad: AdditionalData<'_>,
+ key: &[u8; 32],
+ ) -> Result<DecryptedData> {
+ let token = data.data;
+ let cek = Self::decrypt_cek(&data.content_encryption_key, key)?;
+
+ // encode the implicit assertions
+ let assertions = Assertions::from(ad).encode();
+
+ // decrypt the payload with the footer and implicit assertions
+ let payload = Paseto::<V4, LocalPurpose>::try_decrypt(
+ &token,
+ &cek.into(),
+ None,
+ ImplicitAssertion::from(&*assertions),
+ )
+ .context("could not decrypt entry")?;
+
+ let payload: AtuinPayload = serde_json::from_str(&payload)?;
+ let data = general_purpose::URL_SAFE_NO_PAD.decode(payload.data)?;
+ Ok(DecryptedData(data))
+ }
+}
+
+impl PASETO_V4 {
+ fn decrypt_cek(wrapped_cek: &str, key: &[u8; 32]) -> Result<Key<V4, Local>> {
+ let wrapping_key = Key::<V4, Local>::from_bytes(*key);
+
+ // let wrapping_key = PasetoSymmetricKey::from(Key::from(key));
+
+ let AtuinFooter { kid, wpk } = serde_json::from_str(wrapped_cek)
+ .context("wrapped cek did not contain the correct contents")?;
+
+ // check that the wrapping key matches the required key to decrypt.
+ // In future, we could support multiple keys and use this key to
+ // look up the key rather than only allow one key.
+ // For now though we will only support the one key and key rotation will
+ // have to be a hard reset
+ let current_kid = wrapping_key.to_id();
+
+ ensure!(
+ current_kid == kid,
+ "attempting to decrypt with incorrect key. currently using {current_kid}, expecting {kid}"
+ );
+
+ // decrypt the random key
+ Ok(wpk.unwrap_key(&wrapping_key)?)
+ }
+
+ 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,
+ };
+ serde_json::to_string(&wrapped_cek).expect("could not serialize wrapped cek")
+ }
+}
+
+#[derive(Serialize, Deserialize)]
+struct AtuinPayload {
+ data: String,
+}
+
+#[derive(Serialize, Deserialize)]
+/// Well-known footer claims for decrypting. This is not encrypted but is stored in the record.
+/// <https://github.com/paseto-standard/paseto-spec/blob/master/docs/02-Implementation-Guide/04-Claims.md#optional-footer-claims>
+struct AtuinFooter {
+ /// Wrapped key
+ wpk: PieWrappedKey<V4, Local>,
+ /// ID of the key which was used to wrap
+ kid: KeyId<V4, Local>,
+}
+
+/// Used in the implicit assertions. This is not encrypted and not stored in the data blob.
+// This cannot be changed, otherwise it breaks the authenticated encryption.
+#[derive(Debug, Copy, Clone, Serialize)]
+struct Assertions<'a> {
+ id: &'a RecordId,
+ idx: &'a RecordIdx,
+ version: &'a str,
+ tag: &'a str,
+ host: &'a HostId,
+}
+
+impl<'a> From<AdditionalData<'a>> for Assertions<'a> {
+ fn from(ad: AdditionalData<'a>) -> Self {
+ Self {
+ id: ad.id,
+ version: ad.version,
+ tag: ad.tag,
+ host: ad.host,
+ idx: ad.idx,
+ }
+ }
+}
+
+impl Assertions<'_> {
+ fn encode(&self) -> String {
+ serde_json::to_string(self).expect("could not serialize implicit assertions")
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use turtle_common::{
+ record::{Host, Record},
+ utils::uuid_v7,
+ };
+
+ use super::{
+ AdditionalData, DecryptedData, Encryption, HostId, Key, Local, PASETO_V4, RecordId, V4,
+ };
+
+ #[test]
+ fn round_trip() {
+ let key = Key::<V4, Local>::new_os_random();
+
+ let ad = AdditionalData {
+ id: &RecordId(uuid_v7()),
+ version: "v0",
+ tag: "kv",
+ host: &HostId(uuid_v7()),
+ idx: &0,
+ };
+
+ let data = DecryptedData(vec![1, 2, 3, 4]);
+
+ let encrypted = PASETO_V4::encrypt(data.clone(), ad, &key.to_bytes());
+ let decrypted = PASETO_V4::decrypt(encrypted, ad, &key.to_bytes()).unwrap();
+ assert_eq!(decrypted, data);
+ }
+
+ #[test]
+ fn same_entry_different_output() {
+ let key = Key::<V4, Local>::new_os_random();
+
+ let ad = AdditionalData {
+ id: &RecordId(uuid_v7()),
+ version: "v0",
+ tag: "kv",
+ host: &HostId(uuid_v7()),
+ idx: &0,
+ };
+
+ let data = DecryptedData(vec![1, 2, 3, 4]);
+
+ let encrypted = PASETO_V4::encrypt(data.clone(), ad, &key.to_bytes());
+ let encrypted2 = PASETO_V4::encrypt(data, ad, &key.to_bytes());
+
+ assert_ne!(
+ encrypted.data, encrypted2.data,
+ "re-encrypting the same contents should have different output due to key randomization"
+ );
+ }
+
+ #[test]
+ fn cannot_decrypt_different_key() {
+ let key = Key::<V4, Local>::new_os_random();
+ let fake_key = Key::<V4, Local>::new_os_random();
+
+ let ad = AdditionalData {
+ id: &RecordId(uuid_v7()),
+ version: "v0",
+ tag: "kv",
+ host: &HostId(uuid_v7()),
+ idx: &0,
+ };
+
+ let data = DecryptedData(vec![1, 2, 3, 4]);
+
+ let encrypted = PASETO_V4::encrypt(data, ad, &key.to_bytes());
+ drop(PASETO_V4::decrypt(encrypted, ad, &fake_key.to_bytes()).unwrap_err());
+ }
+
+ #[test]
+ fn cannot_decrypt_different_id() {
+ let key = Key::<V4, Local>::new_os_random();
+
+ let ad = AdditionalData {
+ id: &RecordId(uuid_v7()),
+ version: "v0",
+ tag: "kv",
+ host: &HostId(uuid_v7()),
+ idx: &0,
+ };
+
+ let data = DecryptedData(vec![1, 2, 3, 4]);
+
+ let encrypted = PASETO_V4::encrypt(data, ad, &key.to_bytes());
+
+ let ad = AdditionalData {
+ id: &RecordId(uuid_v7()),
+ ..ad
+ };
+ drop(PASETO_V4::decrypt(encrypted, ad, &key.to_bytes()).unwrap_err());
+ }
+
+ #[test]
+ fn re_encrypt_round_trip() {
+ let key1 = Key::<V4, Local>::new_os_random();
+ let key2 = Key::<V4, Local>::new_os_random();
+
+ let ad = AdditionalData {
+ id: &RecordId(uuid_v7()),
+ version: "v0",
+ tag: "kv",
+ host: &HostId(uuid_v7()),
+ idx: &0,
+ };
+
+ let data = DecryptedData(vec![1, 2, 3, 4]);
+
+ let encrypted1 = PASETO_V4::encrypt(data.clone(), ad, &key1.to_bytes());
+ let encrypted2 =
+ PASETO_V4::re_encrypt(encrypted1.clone(), ad, &key1.to_bytes(), &key2.to_bytes())
+ .unwrap();
+
+ // we only re-encrypt the content keys
+ assert_eq!(encrypted1.data, encrypted2.data);
+ assert_ne!(
+ encrypted1.content_encryption_key,
+ encrypted2.content_encryption_key
+ );
+
+ let decrypted = PASETO_V4::decrypt(encrypted2, ad, &key2.to_bytes()).unwrap();
+
+ assert_eq!(decrypted, data);
+ }
+
+ #[test]
+ fn full_record_round_trip() {
+ let key = [0x55; 32];
+ let record = Record::builder()
+ .id(RecordId(uuid_v7()))
+ .version("v0".to_owned())
+ .tag("kv".to_owned())
+ .host(Host::new(HostId(uuid_v7())))
+ .timestamp(1_687_244_806_000_000)
+ .data(DecryptedData(vec![1, 2, 3, 4]))
+ .idx(0)
+ .build();
+
+ let encrypted = record.encrypt::<PASETO_V4>(&key);
+
+ assert!(!encrypted.data.data.is_empty());
+ assert!(!encrypted.data.content_encryption_key.is_empty());
+
+ let decrypted = encrypted.decrypt::<PASETO_V4>(&key).unwrap();
+
+ assert_eq!(decrypted.data.0, [1, 2, 3, 4]);
+ }
+
+ #[test]
+ fn full_record_round_trip_fail() {
+ let key = [0x55; 32];
+ let record = Record::builder()
+ .id(RecordId(uuid_v7()))
+ .version("v0".to_owned())
+ .tag("kv".to_owned())
+ .host(Host::new(HostId(uuid_v7())))
+ .timestamp(1_687_244_806_000_000)
+ .data(DecryptedData(vec![1, 2, 3, 4]))
+ .idx(0)
+ .build();
+
+ let encrypted = record.encrypt::<PASETO_V4>(&key);
+
+ let mut enc1 = encrypted.clone();
+ enc1.host = Host::new(HostId(uuid_v7()));
+ let _ = enc1
+ .decrypt::<PASETO_V4>(&key)
+ .expect_err("tampering with the host should result in auth failure");
+
+ let mut enc2 = encrypted;
+ enc2.id = RecordId(uuid_v7());
+ let _ = enc2
+ .decrypt::<PASETO_V4>(&key)
+ .expect_err("tampering with the id should result in auth failure");
+ }
+}
diff --git a/crates/daemon/src/aclient/record/mod.rs b/crates/daemon/src/aclient/record/mod.rs
new file mode 100644
index 00000000..4e5774ea
--- /dev/null
+++ b/crates/daemon/src/aclient/record/mod.rs
@@ -0,0 +1,3 @@
+pub(crate) mod encryption;
+pub(crate) mod sqlite_store;
+pub(crate) mod sync;
diff --git a/crates/daemon/src/aclient/record/sqlite_store.rs b/crates/daemon/src/aclient/record/sqlite_store.rs
new file mode 100644
index 00000000..0026690b
--- /dev/null
+++ b/crates/daemon/src/aclient/record/sqlite_store.rs
@@ -0,0 +1,412 @@
+// Here we are using sqlite as a pretty dumb store, and will not be running any complex queries.
+// Multiple stores of multiple types are all stored in one chonky table (for now), and we just index
+// by tag/host
+
+use std::path::Path;
+use std::str::FromStr;
+
+use eyre::{Result, eyre};
+use fs_err as fs;
+
+use sqlx::{
+ Row,
+ sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePool, SqliteRow},
+};
+use tracing::debug;
+
+use crate::aclient::utils::setup_db;
+use turtle_common::record::{
+ EncryptedData, Host, HostId, Record, RecordId, RecordIdx, RecordStatus,
+};
+use turtle_common::utils;
+use uuid::Uuid;
+
+#[derive(Debug, Clone)]
+pub(crate) struct SqliteStore {
+ pool: SqlitePool,
+}
+
+impl SqliteStore {
+ pub(crate) async fn new(path: impl AsRef<Path>, timeout: f64) -> Result<Self> {
+ fn mk_opts(path: &str) -> sqlx::Result<SqliteConnectOptions> {
+ let opts = SqliteConnectOptions::from_str(path)?
+ .journal_mode(SqliteJournalMode::Wal)
+ .foreign_keys(true)
+ .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-record-migrations").await?;
+
+ Ok(Self { pool })
+ }
+
+ async fn save_raw(
+ tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
+ r: &Record<EncryptedData>,
+ ) -> Result<()> {
+ // In sqlite, we are "limited" to i64. But that is still fine, until 2262.
+ sqlx::query(
+ "insert or ignore into store(id, idx, host, tag, timestamp, version, data, cek)
+ values(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
+ )
+ .bind(r.id.0.as_hyphenated().to_string())
+ .bind(r.idx as i64)
+ .bind(r.host.id.0.as_hyphenated().to_string())
+ .bind(r.tag.as_str())
+ .bind(r.timestamp as i64)
+ .bind(r.version.as_str())
+ .bind(r.data.data.as_str())
+ .bind(r.data.content_encryption_key.as_str())
+ .execute(&mut **tx)
+ .await?;
+
+ Ok(())
+ }
+
+ #[expect(
+ clippy::needless_pass_by_value,
+ reason = "this is used in a place with fixed function signature"
+ )]
+ fn query_row(row: SqliteRow) -> Record<EncryptedData> {
+ let idx: i64 = row.get("idx");
+ let timestamp: i64 = row.get("timestamp");
+
+ // tbh at this point things are pretty fucked so just panic
+ let id = Uuid::from_str(row.get("id")).expect("invalid id UUID format in sqlite DB");
+ let host = Uuid::from_str(row.get("host")).expect("invalid host UUID format in sqlite DB");
+
+ Record {
+ id: RecordId(id),
+ idx: idx as u64,
+ host: Host::new(HostId(host)),
+ timestamp: timestamp as u64,
+ tag: row.get("tag"),
+ version: row.get("version"),
+ data: EncryptedData {
+ data: row.get("data"),
+ content_encryption_key: row.get("cek"),
+ },
+ }
+ }
+}
+
+/// A record store stores records
+/// In more detail - we tend to need to process this into _another_ format to actually query it.
+/// As is, the record store is intended as the source of truth for arbitrary data, which could
+/// be shell history, kvs, etc.
+impl SqliteStore {
+ /// Push a record
+ pub(crate) async fn push(&self, record: &Record<EncryptedData>) -> Result<()> {
+ self.push_batch(std::iter::once(record)).await
+ }
+
+ /// Push a batch of records, all in one transaction
+ pub(crate) async fn push_batch(
+ &self,
+ records: impl Iterator<Item = &Record<EncryptedData>> + Send + Sync,
+ ) -> Result<()> {
+ let mut tx = self.pool.begin().await?;
+
+ for record in records {
+ Self::save_raw(&mut tx, record).await?;
+ }
+
+ tx.commit().await?;
+
+ Ok(())
+ }
+
+ pub(crate) async fn get(&self, id: RecordId) -> Result<Record<EncryptedData>> {
+ let res = sqlx::query("select * from store where store.id = ?1")
+ .bind(id.0.as_hyphenated().to_string())
+ .map(Self::query_row)
+ .fetch_one(&self.pool)
+ .await?;
+
+ Ok(res)
+ }
+
+ pub(crate) async fn last(
+ &self,
+ host: HostId,
+ tag: &str,
+ ) -> Result<Option<Record<EncryptedData>>> {
+ let res =
+ sqlx::query("select * from store where host=?1 and tag=?2 order by idx desc limit 1")
+ .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(record) => Ok(Some(record)),
+ }
+ }
+
+ /// Get the next `limit` records, after and including the given index
+ pub(crate) async fn next(
+ &self,
+ host: HostId,
+ tag: &str,
+ idx: RecordIdx,
+ limit: u64,
+ ) -> Result<Vec<Record<EncryptedData>>> {
+ let res = sqlx::query(
+ "select * from store where idx >= ?1 and host = ?2 and tag = ?3 order by idx asc limit ?4",
+ )
+ .bind(idx as i64)
+ .bind(host.0.as_hyphenated().to_string())
+ .bind(tag)
+ .bind(limit as i64)
+ .map(Self::query_row)
+ .fetch_all(&self.pool)
+ .await?;
+
+ Ok(res)
+ }
+
+ pub(crate) async fn status(&self) -> Result<RecordStatus> {
+ let mut status = RecordStatus::new();
+
+ let res: Result<Vec<(String, String, i64)>, sqlx::Error> =
+ sqlx::query_as("select host, tag, max(idx) from store group by host, tag")
+ .fetch_all(&self.pool)
+ .await;
+
+ let res = match res {
+ Err(e) => return Err(eyre!("failed to fetch local store status: {}", e)),
+ Ok(v) => v,
+ };
+
+ for i in res {
+ let host = HostId(
+ Uuid::from_str(i.0.as_str()).expect("failed to parse uuid for local store status"),
+ );
+
+ status.set_raw(host, i.1, i.2 as u64);
+ }
+
+ Ok(status)
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use crate::{
+ atuin_client::{
+ encryption::generate_encoded_key, record::encryption::PASETO_V4,
+ settings::test_local_timeout,
+ },
+ atuin_common::{
+ record::{DecryptedData, EncryptedData, Host, HostId, Record},
+ utils::uuid_v7,
+ },
+ };
+
+ use super::SqliteStore;
+
+ fn test_record() -> Record<EncryptedData> {
+ Record::builder()
+ .host(Host::new(HostId(uuid_v7())))
+ .version("v1".into())
+ .tag(uuid_v7().simple().to_string())
+ .data(EncryptedData {
+ data: "1234".into(),
+ content_encryption_key: "1234".into(),
+ })
+ .idx(0)
+ .build()
+ }
+
+ #[tokio::test]
+ async fn create_db() {
+ let db = SqliteStore::new(":memory:", test_local_timeout()).await;
+
+ assert!(
+ db.is_ok(),
+ "db could not be created, {:?}",
+ db.err().unwrap()
+ );
+ }
+
+ #[tokio::test]
+ async fn push_record() {
+ let db = SqliteStore::new(":memory:", test_local_timeout())
+ .await
+ .unwrap();
+ let record = test_record();
+
+ db.push(&record).await.expect("failed to insert record");
+ }
+
+ #[tokio::test]
+ async fn get_record() {
+ let db = SqliteStore::new(":memory:", test_local_timeout())
+ .await
+ .unwrap();
+ let record = test_record();
+ db.push(&record).await.unwrap();
+
+ let new_record = db.get(record.id).await.expect("failed to fetch record");
+
+ assert_eq!(record, new_record, "records are not equal");
+ }
+
+ #[tokio::test]
+ async fn last() {
+ let db = SqliteStore::new(":memory:", test_local_timeout())
+ .await
+ .unwrap();
+ let record = test_record();
+ db.push(&record).await.unwrap();
+
+ let last = db
+ .last(record.host.id, record.tag.as_str())
+ .await
+ .expect("failed to get store len");
+
+ assert_eq!(
+ last.unwrap().id,
+ record.id,
+ "expected to get back the same record that was inserted"
+ );
+ }
+
+ #[tokio::test]
+ async fn first() {
+ let db = SqliteStore::new(":memory:", test_local_timeout())
+ .await
+ .unwrap();
+ let record = test_record();
+ db.push(&record).await.unwrap();
+
+ let first = db
+ .first(record.host.id, record.tag.as_str())
+ .await
+ .expect("failed to get store len");
+
+ assert_eq!(
+ first.unwrap().id,
+ record.id,
+ "expected to get back the same record that was inserted"
+ );
+ }
+
+ #[tokio::test]
+ async fn len() {
+ let db = SqliteStore::new(":memory:", test_local_timeout())
+ .await
+ .unwrap();
+ let record = test_record();
+ db.push(&record).await.unwrap();
+
+ let len = db
+ .len(record.host.id, record.tag.as_str())
+ .await
+ .expect("failed to get store len");
+
+ assert_eq!(len, 1, "expected length of 1 after insert");
+ }
+
+ #[tokio::test]
+ async fn len_tag() {
+ let db = SqliteStore::new(":memory:", test_local_timeout())
+ .await
+ .unwrap();
+ let record = test_record();
+ db.push(&record).await.unwrap();
+
+ let len = db
+ .len_tag(record.tag.as_str())
+ .await
+ .expect("failed to get store len");
+
+ assert_eq!(len, 1, "expected length of 1 after insert");
+ }
+
+ #[tokio::test]
+ async fn re_encrypt() {
+ let store = SqliteStore::new(":memory:", test_local_timeout())
+ .await
+ .unwrap();
+ let (key, _) = generate_encoded_key().unwrap();
+ let data = vec![0u8, 1u8, 2u8, 3u8];
+ let host_id = HostId(uuid_v7());
+
+ for i in 0..10 {
+ let record = Record::builder()
+ .host(Host::new(host_id))
+ .version(String::from("test"))
+ .tag(String::from("test"))
+ .idx(i)
+ .data(DecryptedData(data.clone()))
+ .build();
+
+ let record = record.encrypt::<PASETO_V4>(&key.into());
+ store
+ .push(&record)
+ .await
+ .expect("failed to push encrypted record");
+ }
+
+ // first, check that we can decrypt the data with the current key
+ let all = store.all_tagged("test").await.unwrap();
+
+ assert_eq!(all.len(), 10, "failed to fetch all records");
+
+ for record in all {
+ let decrypted = record.decrypt::<PASETO_V4>(&key.into()).unwrap();
+ assert_eq!(decrypted.data.0, data);
+ }
+
+ // reencrypt the store, then check if
+ // 1) it cannot be decrypted with the old key
+ // 2) it can be decrypted with the new key
+
+ let (new_key, _) = generate_encoded_key().unwrap();
+ store
+ .re_encrypt(&key.into(), &new_key.into())
+ .await
+ .expect("failed to re-encrypt store");
+
+ let all = store.all_tagged("test").await.unwrap();
+
+ for record in all.iter() {
+ let decrypted = record.clone().decrypt::<PASETO_V4>(&key.into());
+ assert!(
+ decrypted.is_err(),
+ "did not get error decrypting with old key after re-encrypt"
+ )
+ }
+
+ for record in all {
+ let decrypted = record.decrypt::<PASETO_V4>(&new_key.into()).unwrap();
+ assert_eq!(decrypted.data.0, data);
+ }
+
+ assert_eq!(store.len(host_id, "test").await.unwrap(), 10);
+ }
+}
diff --git a/crates/daemon/src/aclient/record/sync.rs b/crates/daemon/src/aclient/record/sync.rs
new file mode 100644
index 00000000..79239b99
--- /dev/null
+++ b/crates/daemon/src/aclient/record/sync.rs
@@ -0,0 +1,453 @@
+// do a sync :O
+use std::{cmp::Ordering, fmt::Write};
+
+use eyre::{OptionExt, Result};
+use thiserror::Error;
+use tracing::error;
+
+use super::encryption::PASETO_V4;
+use crate::aclient::record::sqlite_store::SqliteStore;
+use crate::aclient::{api_client::Client, settings::Settings};
+
+use indicatif::{ProgressBar, ProgressState, ProgressStyle};
+use turtle_common::record::{Diff, HostId, RecordId, RecordIdx, RecordStatus};
+
+#[derive(Error, Debug)]
+pub(crate) enum SyncError {
+ #[error("an issue with the local database occurred: {msg:?}")]
+ LocalStoreError { msg: String },
+
+ #[error("something has gone wrong with the sync logic: {msg:?}")]
+ SyncLogicError { msg: String },
+
+ #[error("operational error: {msg:?}")]
+ OperationalError { msg: String },
+
+ #[error("a request to the sync server failed: {msg}")]
+ RemoteRequestError { msg: String },
+
+ #[error(
+ "the encryption key on this machine does not match the data on the server. \
+ this usually means a new machine was set up without copying the existing key. \
+ to fix: run `atuin key` on a machine that already syncs correctly, then run \
+ `atuin store rekey <key>` on this machine with the value from the other machine"
+ )]
+ WrongKey,
+}
+
+#[derive(Debug, Eq, PartialEq)]
+enum Operation {
+ // Either upload or download until the states matches the below
+ Upload {
+ local: RecordIdx,
+ remote: Option<RecordIdx>,
+ host: HostId,
+ tag: String,
+ },
+ Download {
+ local: Option<RecordIdx>,
+ remote: RecordIdx,
+ host: HostId,
+ tag: String,
+ },
+ Noop {
+ host: HostId,
+ tag: String,
+ },
+}
+
+fn build_client(settings: &Settings) -> Result<Client<'_>, SyncError> {
+ Client::new(
+ &settings.sync.address,
+ settings.network_connect_timeout,
+ settings.network_timeout,
+ settings
+ .sync
+ .user_id()
+ .map_err(|e| SyncError::RemoteRequestError { msg: e.to_string() })?
+ .ok_or_eyre("No sync user-id set")
+ .map_err(|e| SyncError::RemoteRequestError { msg: e.to_string() })?,
+ )
+ .map_err(|e| SyncError::OperationalError { msg: e.to_string() })
+}
+
+async fn diff(
+ client: &Client<'_>,
+ store: &SqliteStore,
+) -> Result<(Vec<Diff>, RecordStatus), SyncError> {
+ let local_index = store
+ .status()
+ .await
+ .map_err(|e| SyncError::LocalStoreError { msg: e.to_string() })?;
+
+ let remote_index = client
+ .record_status()
+ .await
+ .map_err(|e| SyncError::RemoteRequestError { msg: e.to_string() })?;
+
+ let diff = local_index.diff(&remote_index);
+
+ Ok((diff, remote_index))
+}
+
+// Take a diff, along with a local store, and resolve it into a set of operations.
+// 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
+fn operations(diffs: Vec<Diff>, _store: &SqliteStore) -> Result<Vec<Operation>, SyncError> {
+ let mut operations = Vec::with_capacity(diffs.len());
+
+ for diff in diffs {
+ let op = match (diff.local, diff.remote) {
+ // We both have it! Could be either. Compare.
+ (Some(local), Some(remote)) => match local.cmp(&remote) {
+ Ordering::Equal => Operation::Noop {
+ host: diff.host,
+ tag: diff.tag,
+ },
+ Ordering::Greater => Operation::Upload {
+ local,
+ remote: Some(remote),
+ host: diff.host,
+ tag: diff.tag,
+ },
+ Ordering::Less => Operation::Download {
+ local: Some(local),
+ remote,
+ host: diff.host,
+ tag: diff.tag,
+ },
+ },
+
+ // Remote has it, we don't. Gotta be download
+ (None, Some(remote)) => Operation::Download {
+ local: None,
+ remote,
+ host: diff.host,
+ tag: diff.tag,
+ },
+
+ // We have it, remote doesn't. Gotta be upload.
+ (Some(local), None) => Operation::Upload {
+ local,
+ remote: None,
+ host: diff.host,
+ tag: diff.tag,
+ },
+
+ // something is pretty fucked.
+ (None, None) => {
+ return Err(SyncError::SyncLogicError {
+ msg: String::from(
+ "diff has nothing for local or remote - (host, tag) does not exist",
+ ),
+ });
+ }
+ };
+
+ operations.push(op);
+ }
+
+ // sort them - purely so we have a stable testing order, and can rely on
+ // same input = same output
+ // We can sort by ID so long as we continue to use UUIDv7 or something
+ // with the same properties
+
+ operations.sort_by_key(|op| match op {
+ Operation::Noop { host, tag } => (0, *host, tag.clone()),
+
+ Operation::Upload { host, tag, .. } => (1, *host, tag.clone()),
+
+ Operation::Download { host, tag, .. } => (2, *host, tag.clone()),
+ });
+
+ Ok(operations)
+}
+
+async fn sync_upload(
+ store: &SqliteStore,
+ client: &Client<'_>,
+ host: HostId,
+ tag: String,
+ local: RecordIdx,
+ remote: Option<RecordIdx>,
+ page_size: u64,
+) -> Result<i64, SyncError> {
+ let remote = remote.unwrap_or(0);
+ let expected = local - remote;
+ let mut progress = 0;
+
+ let pb = ProgressBar::new(expected);
+ pb.set_style(ProgressStyle::with_template("{spinner:.green} [{elapsed_precise}] [{wide_bar:.cyan/blue}] {human_pos}/{human_len} ({eta})")
+ .unwrap()
+ .with_key("eta", |state: &ProgressState, w: &mut dyn Write| write!(w, "{:.1}s", state.eta().as_secs_f64()).unwrap())
+ .progress_chars("#>-"));
+
+ println!(
+ "Uploading {} records to {}/{}",
+ expected,
+ host.0.as_simple(),
+ tag
+ );
+
+ loop {
+ let page = store
+ .next(host, tag.as_str(), remote + progress, page_size)
+ .await
+ .map_err(|e| {
+ error!("failed to read upload page: {e:?}");
+
+ SyncError::LocalStoreError { msg: e.to_string() }
+ })?;
+
+ if page.is_empty() {
+ break;
+ }
+
+ client.post_records(&page).await.map_err(|e| {
+ error!("failed to post records: {e:?}");
+
+ SyncError::RemoteRequestError { msg: e.to_string() }
+ })?;
+
+ progress += page.len() as u64;
+ pb.set_position(progress);
+
+ if progress >= expected {
+ break;
+ }
+ }
+
+ pb.finish_with_message("Uploaded records");
+
+ Ok(progress as i64)
+}
+
+async fn sync_download(
+ store: &SqliteStore,
+ client: &Client<'_>,
+ host: HostId,
+ tag: String,
+ local: Option<RecordIdx>,
+ remote: RecordIdx,
+ page_size: u64,
+) -> Result<Vec<RecordId>, SyncError> {
+ let local = local.unwrap_or(0);
+ let expected = remote - local;
+ let mut progress = 0;
+ let mut ret = Vec::new();
+
+ println!(
+ "Downloading {} records from {}/{}",
+ expected,
+ host.0.as_simple(),
+ tag
+ );
+
+ let pb = ProgressBar::new(expected);
+ pb.set_style(ProgressStyle::with_template("{spinner:.green} [{elapsed_precise}] [{wide_bar:.cyan/blue}] {human_pos}/{human_len} ({eta})")
+ .unwrap()
+ .with_key("eta", |state: &ProgressState, w: &mut dyn Write| write!(w, "{:.1}s", state.eta().as_secs_f64()).unwrap())
+ .progress_chars("#>-"));
+
+ loop {
+ let page = client
+ .next_records(host, tag.clone(), local + progress, page_size)
+ .await
+ .map_err(|e| SyncError::RemoteRequestError { msg: e.to_string() })?;
+
+ if page.is_empty() {
+ break;
+ }
+
+ store
+ .push_batch(page.iter())
+ .await
+ .map_err(|e| SyncError::LocalStoreError { msg: e.to_string() })?;
+
+ ret.extend(page.iter().map(|f| f.id));
+
+ progress += page.len() as u64;
+ pb.set_position(progress);
+
+ if progress >= expected {
+ break;
+ }
+ }
+
+ pb.finish_with_message("Downloaded records");
+
+ Ok(ret)
+}
+
+async fn sync_remote(
+ client: &Client<'_>,
+ operations: Vec<Operation>,
+ local_store: &SqliteStore,
+ page_size: u64,
+) -> Result<(i64, Vec<RecordId>), SyncError> {
+ let mut uploaded = 0;
+ let mut downloaded = Vec::new();
+
+ // this can totally run in parallel, but lets get it working first
+ for i in operations {
+ match i {
+ Operation::Upload {
+ host,
+ tag,
+ local,
+ remote,
+ } => {
+ uploaded +=
+ sync_upload(local_store, client, host, tag, local, remote, page_size).await?;
+ }
+
+ Operation::Download {
+ host,
+ tag,
+ local,
+ remote,
+ } => {
+ let mut d =
+ sync_download(local_store, client, host, tag, local, remote, page_size).await?;
+ downloaded.append(&mut d);
+ }
+
+ Operation::Noop { .. } => (),
+ }
+ }
+
+ Ok((uploaded, downloaded))
+}
+
+async fn check_encryption_key(
+ client: &Client<'_>,
+ remote_index: &RecordStatus,
+ encryption_key: &[u8; 32],
+) -> Result<(), SyncError> {
+ let sample = remote_index
+ .hosts
+ .iter()
+ .flat_map(|(host, tags)| tags.keys().map(move |tag| (*host, tag.clone())))
+ .next();
+
+ let Some((host, tag)) = sample else {
+ return Ok(());
+ };
+
+ let records = client
+ .next_records(host, tag, 0, 1)
+ .await
+ .map_err(|e| SyncError::RemoteRequestError { msg: e.to_string() })?;
+
+ let Some(record) = records.into_iter().next() else {
+ return Ok(());
+ };
+
+ record.decrypt::<PASETO_V4>(encryption_key).map_err(|err| {
+ error!("Wrong key error: {err}");
+ SyncError::WrongKey
+ })?;
+
+ Ok(())
+}
+
+pub(crate) async fn sync(
+ settings: &Settings,
+ store: &SqliteStore,
+ encryption_key: &[u8; 32],
+) -> Result<(i64, Vec<RecordId>), SyncError> {
+ let client = build_client(settings)?;
+ let (diff, remote_index) = diff(&client, store).await?;
+
+ // Bail before mutating either side if the local key can't read the remote.
+ check_encryption_key(&client, &remote_index, encryption_key).await?;
+
+ let operations = operations(diff, store)?;
+ let (uploaded, downloaded) = sync_remote(&client, operations, store, 100).await?;
+
+ Ok((uploaded, downloaded))
+}
+
+#[cfg(test)]
+mod tests {
+ use crate::aclient::record::sync::Operation;
+ use turtle_common::record::{Diff, EncryptedData, HostId, Record};
+
+ use crate::aclient::{
+ record::{
+ sqlite_store::SqliteStore,
+ sync::{self},
+ },
+ settings::test_local_timeout,
+ };
+
+ fn test_record() -> Record<EncryptedData> {
+ Record::builder()
+ .host(turtle_common::record::Host::new(HostId(
+ turtle_common::utils::uuid_v7(),
+ )))
+ .version("v1".into())
+ .tag(turtle_common::utils::uuid_v7().simple().to_string())
+ .data(EncryptedData {
+ data: String::new(),
+ content_encryption_key: String::new(),
+ })
+ .idx(0)
+ .build()
+ }
+
+ // Take a list of local records, and a list of remote records.
+ // Return the local database, and a diff of local/remote, ready to build
+ // ops
+ async fn build_test_diff(
+ local_records: Vec<Record<EncryptedData>>,
+ remote_records: Vec<Record<EncryptedData>>,
+ ) -> (SqliteStore, Vec<Diff>) {
+ let local_store = SqliteStore::new(":memory:", test_local_timeout())
+ .await
+ .expect("failed to open in memory sqlite");
+ let remote_store = SqliteStore::new(":memory:", test_local_timeout())
+ .await
+ .expect("failed to open in memory sqlite"); // "remote"
+
+ for i in local_records {
+ local_store.push(&i).await.unwrap();
+ }
+
+ for i in remote_records {
+ remote_store.push(&i).await.unwrap();
+ }
+
+ let local_index = local_store.status().await.unwrap();
+ let remote_index = remote_store.status().await.unwrap();
+
+ let diff = local_index.diff(&remote_index);
+
+ (local_store, diff)
+ }
+
+ #[tokio::test]
+ async fn test_basic_diff() {
+ // a diff where local is ahead of remote. nothing else.
+
+ let record = test_record();
+ let (store, diff) = build_test_diff(vec![record.clone()], vec![]).await;
+
+ assert_eq!(diff.len(), 1);
+
+ let operations = sync::operations(diff, &store).unwrap();
+
+ assert_eq!(operations.len(), 1);
+
+ assert_eq!(
+ operations[0],
+ Operation::Upload {
+ host: record.host.id,
+ tag: record.tag,
+ local: record.idx,
+ remote: None,
+ }
+ );
+ }
+}
diff --git a/crates/daemon/src/aclient/settings/meta.rs b/crates/daemon/src/aclient/settings/meta.rs
new file mode 100644
index 00000000..1c9b9cd1
--- /dev/null
+++ b/crates/daemon/src/aclient/settings/meta.rs
@@ -0,0 +1,17 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub(crate) struct Settings {
+ pub(super) db_path: String,
+}
+
+impl Default for Settings {
+ fn default() -> Self {
+ let dir = turtle_common::utils::data_dir();
+ let path = dir.join("meta.db");
+
+ Self {
+ db_path: path.to_string_lossy().to_string(),
+ }
+ }
+}
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/daemon/src/aclient/utils.rs b/crates/daemon/src/aclient/utils.rs
new file mode 100644
index 00000000..18e732a0
--- /dev/null
+++ b/crates/daemon/src/aclient/utils.rs
@@ -0,0 +1,77 @@
+/// Setup a [`SQLite`] database.
+///
+/// This takes care of correct locking, so that we avoid a race when setting up the database.
+macro_rules! setup_db {
+ (
+ $db_path:expr,
+ $a_timeout:expr,
+ $opts:expr,
+ $m_name:literal $(,)?
+ ) => {{
+ async fn migrate(pool: &SqlitePool) -> sqlx::Result<()> {
+ { sqlx::sqlx_macros::migrate!($m_name) }.run(pool).await?;
+ Ok(())
+ }
+
+ crate::aclient::utils::setup_db_inner($db_path, $a_timeout, $opts, migrate)
+ }};
+}
+pub(crate) use setup_db;
+
+use std::{os::fd::AsRawFd, path::Path, time::Duration};
+
+use fs_err::OpenOptions;
+use sqlx::{
+ SqlitePool,
+ sqlite::{SqliteConnectOptions, SqlitePoolOptions},
+};
+use tracing::debug;
+
+/// Helper for `setup_db!`
+pub(crate) async fn setup_db_inner(
+ db_path: &Path,
+ acquire_timeout: f64,
+ mk_opts: fn(&str) -> sqlx::Result<SqliteConnectOptions>,
+ migrate: impl AsyncFn(&SqlitePool) -> sqlx::Result<()>,
+) -> sqlx::Result<SqlitePool> {
+ async fn open_db(timeout: f64, opts: SqliteConnectOptions) -> sqlx::Result<SqlitePool> {
+ let pool = SqlitePoolOptions::new()
+ .acquire_timeout(Duration::from_secs_f64(timeout))
+ .connect_with(opts)
+ .await?;
+
+ Ok(pool)
+ }
+
+ {
+ let file = OpenOptions::new()
+ .read(true)
+ .write(true)
+ .create(true)
+ .open(db_path)?;
+
+ // Lock the db file while we are running the migrations.
+ // Why? Because there is a small chance that we start running migrations (e.g. as the daemon)
+ // and then another process is started, which will also try to run migrations.
+ // Essentially, one of the processes will receive with a SQLite UNIQUE constraint failure.
+ // So let's avoid that possibility from the start.
+ file.lock()?;
+
+ let pool = open_db(
+ acquire_timeout,
+ mk_opts(format!("/proc/self/fd/{}", file.as_raw_fd()).as_str())?,
+ )
+ .await?;
+
+ debug!("running sqlite database setup");
+
+ migrate(&pool).await?;
+
+ file.unlock()?;
+ }
+
+ let real_opts = mk_opts(db_path.to_str().expect("Should be utf-8"))?;
+ let pool = open_db(acquire_timeout, real_opts).await?;
+
+ Ok(pool)
+}
generated by cgit v1.3.1 (git 2.54.0) at 2026-09-10 04:47:46 +0000