feat: Finalize design for fish-shell integration

10 files changed, 157 insertions, 66 deletions

diff --git a/crates/daemon/src/aclient/database/mod.rs b/crates/daemon/src/aclient/database/mod.rs
index b112b076..f24eb777 100644
--- a/crates/daemon/src/aclient/database/mod.rs
+++ b/crates/daemon/src/aclient/database/mod.rs
@@ -1,4 +1,4 @@
-use std::{path::Path, str::FromStr};
+use std::{path::Path, str::FromStr, time::Duration};
 
 use fs_err::{self as fs};
 use sql_builder::{SqlBuilder, SqlName};
@@ -8,7 +8,7 @@ use sqlx::{
 };
 use time::OffsetDateTime;
 use tracing::debug;
-use turtle::history::{History, HistoryId};
+use turtle_api::history::{History, HistoryId};
 use turtle_common::utils;
 
 use crate::aclient::utils::setup_db;
@@ -59,9 +59,9 @@ impl ClientSqlite {
             "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.0.as_str())
+        .bind(h.id.to_string().as_str())
         .bind(h.timestamp.unix_timestamp_nanos() as i64)
-        .bind(h.duration)
+        .bind(h.duration.as_nanos() as i64)
         .bind(h.exit)
         .bind(h.command.as_str())
         .bind(h.cwd.as_str())
@@ -81,7 +81,7 @@ impl ClientSqlite {
         id: HistoryId,
     ) -> Result<()> {
         sqlx::query("delete from history where id = ?1")
-            .bind(id.0.as_str())
+            .bind(id.to_string().as_str())
             .execute(&mut **tx)
             .await?;
 
@@ -107,7 +107,9 @@ impl ClientSqlite {
                 ))
                 .unwrap(),
             )
-            .duration(row.get("duration"))
+            .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"))
diff --git a/crates/daemon/src/aclient/history/mod.rs b/crates/daemon/src/aclient/history/mod.rs
index ae7654a6..35abc89d 100644
--- a/crates/daemon/src/aclient/history/mod.rs
+++ b/crates/daemon/src/aclient/history/mod.rs
@@ -1,7 +1,9 @@
+use std::time::Duration;
+
 use rmp::decode::DecodeStringError;
 use rmp::decode::ValueReadError;
 use rmp::{Marker, decode::Bytes};
-use turtle::history::History;
+use turtle_api::history::History;
 
 use turtle_common::record::DecryptedData;
 
@@ -40,9 +42,12 @@ impl HistoryExt for History {
         let include_intent = self.intent.is_some();
         encode::write_array_len(&mut output, 10 + u32::from(include_intent))?;
 
-        encode::write_str(&mut output, &self.id.0)?;
+        encode::write_str(&mut output, &self.id.to_string())?;
         encode::write_u64(&mut output, self.timestamp.unix_timestamp_nanos() as u64)?;
-        encode::write_sint(&mut output, self.duration)?;
+        encode::write_sint(
+            &mut output,
+            i64::try_from(self.duration.as_nanos()).expect("should be small enough"),
+        )?;
         encode::write_sint(&mut output, self.exit)?;
         encode::write_str(&mut output, &self.command)?;
         encode::write_str(&mut output, &self.cwd)?;
@@ -107,7 +112,11 @@ impl HistoryExt for History {
 
         let mut bytes = Bytes::new(bytes);
         let timestamp = decode::read_u64(&mut bytes).map_err(error_report)?;
-        let duration = decode::read_int(&mut bytes).map_err(error_report)?;
+        let duration = decode::read_int(&mut bytes)
+            .map(|int: i64| {
+                Duration::from_nanos(u64::try_from(int).expect("should be small enough"))
+            })
+            .map_err(error_report)?;
         let exit = decode::read_int(&mut bytes).map_err(error_report)?;
 
         let bytes = bytes.remaining_slice();
@@ -171,7 +180,9 @@ impl HistoryExt for History {
 
         let mut bytes = Bytes::new(bytes);
         let timestamp = decode::read_u64(&mut bytes).map_err(error_report)?;
-        let duration = decode::read_int(&mut bytes).map_err(error_report)?;
+        let duration = decode::read_int(&mut bytes)
+            .map(|int: i64| Duration::from_nanos(u64::try_from(int).expect("to be small enough")))
+            .map_err(error_report)?;
         let exit = decode::read_int(&mut bytes).map_err(error_report)?;
 
         let bytes = bytes.remaining_slice();
@@ -228,6 +239,8 @@ impl HistoryExt for History {
 
 #[cfg(test)]
 mod tests {
+    use std::time::Duration;
+
     use time::macros::datetime;
 
     use crate::aclient::history::{HISTORY_VERSION, HistoryExt};
@@ -239,7 +252,7 @@ mod tests {
         let history = History {
             id: "66d16cbee7cd47538e5c5b8b44e9006e".to_owned().into(),
             timestamp: datetime!(2023-05-28 18:35:40.633872 +00:00),
-            duration: 49_206_000,
+            duration: Duration::from_nanos(49_206_000),
             exit: 0,
             command: "git status".to_owned(),
             cwd: "/Users/conrad.ludgate/Documents/code/atuin".to_owned(),
@@ -267,7 +280,7 @@ mod tests {
         let history = History {
             id: "66d16cbee7cd47538e5c5b8b44e9006e".to_owned().into(),
             timestamp: datetime!(2023-05-28 18:35:40.633872 +00:00),
-            duration: 49_206_000,
+            duration: Duration::from_nanos(49_206_000),
             exit: 0,
             command: "git status".to_owned(),
             cwd: "/Users/conrad.ludgate/Documents/code/atuin".to_owned(),
@@ -291,7 +304,7 @@ mod tests {
         let history = History {
             id: "66d16cbee7cd47538e5c5b8b44e9006e".to_owned().into(),
             timestamp: datetime!(2023-05-28 18:35:40.633872 +00:00),
-            duration: 49_206_000,
+            duration: Duration::from_nanos(49_206_000),
             exit: 0,
             command: "git status".to_owned(),
             cwd: "/Users/conrad.ludgate/Documents/code/atuin".to_owned(),
@@ -333,7 +346,7 @@ mod tests {
         let current = History {
             id: "66d16cbee7cd47538e5c5b8b44e9006e".to_owned().into(),
             timestamp: datetime!(2023-05-28 18:35:40.633872 +00:00),
-            duration: 49_206_000,
+            duration: Duration::from_nanos(49_206_000),
             exit: 0,
             command: "git status".to_owned(),
             cwd: "/Users/conrad.ludgate/Documents/code/atuin".to_owned(),
diff --git a/crates/daemon/src/aclient/history/store.rs b/crates/daemon/src/aclient/history/store.rs
index 952f2070..a6d6a627 100644
--- a/crates/daemon/src/aclient/history/store.rs
+++ b/crates/daemon/src/aclient/history/store.rs
@@ -1,6 +1,6 @@
 use eyre::{Result, bail, eyre};
 use rmp::decode::Bytes;
-use turtle::history::{History, HistoryId};
+use turtle_api::history::{History, HistoryId};
 
 use crate::aclient::{
     database::ClientSqlite,
@@ -56,7 +56,7 @@ impl HistoryRecord {
             Self::Delete(id) => {
                 // 1 -> a history delete
                 encode::write_u8(&mut output, 1)?;
-                encode::write_str(&mut output, id.0.as_str())?;
+                encode::write_str(&mut output, id.to_string().as_str())?;
             }
         }
 
@@ -221,6 +221,8 @@ impl HistoryStore {
 
 #[cfg(test)]
 mod tests {
+    use std::time::Duration;
+
     use time::macros::datetime;
     use turtle_common::record::DecryptedData;
 
@@ -244,7 +246,7 @@ mod tests {
         let history = History {
             id: "018cd4fe81757cd2aee65cd7861f9c81".to_owned().into(),
             timestamp: datetime!(2024-01-04 00:00:00.000000 +00:00),
-            duration: 100,
+            duration: Duration::from_nanos(100),
             exit: 0,
             command: "ls".to_owned(),
             cwd: "/Users/ellie/src/github.com/atuinsh/atuin".to_owned(),
diff --git a/crates/daemon/src/aclient/settings/mod.rs b/crates/daemon/src/aclient/settings/mod.rs
index ef3f1dd0..379ee563 100644
--- a/crates/daemon/src/aclient/settings/mod.rs
+++ b/crates/daemon/src/aclient/settings/mod.rs
@@ -252,9 +252,8 @@ impl Settings {
         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::runtime_dir().join("atuin.sock");
+        let socket_path = utils::daemon_socket_path();
         let pidfile_path = data_dir.join("atuin-daemon.pid");
-        let logs_dir = utils::logs_dir();
 
         let key_path = data_dir.join("key");
         let meta_path = data_dir.join("meta.db");
@@ -320,7 +319,6 @@ impl Settings {
             .set_default("daemon.systemd_socket", false)?
             .set_default("daemon.tcp_port", 8889)?
             .set_default("logs.enabled", true)?
-            .set_default("logs.dir", logs_dir.to_str())?
             .set_default("logs.level", "info")?
             .set_default("logs.search.file", "search.log")?
             .set_default("logs.daemon.file", "daemon.log")?
@@ -583,13 +581,6 @@ mod tests {
         );
         assert_eq!(meta_db_path, custom_dir.join("meta.db").to_str().unwrap());
         assert_eq!(
-            daemon_socket_path,
-            turtle_common::utils::runtime_dir()
-                .join("atuin.sock")
-                .to_str()
-                .unwrap()
-        );
-        assert_eq!(
             daemon_pidfile_path,
             custom_dir.join("atuin-daemon.pid").to_str().unwrap()
         );
diff --git a/crates/daemon/src/api/control.rs b/crates/daemon/src/api/control.rs
index ff19f593..8277f434 100644
--- a/crates/daemon/src/api/control.rs
+++ b/crates/daemon/src/api/control.rs
@@ -6,7 +6,7 @@ use tokio::time::{self, MissedTickBehavior};
 use tonic::{Request, Response, Status};
 use tracing::{Level, instrument};
 
-use turtle::generated::{
+use turtle_api::generated::{
     DAEMON_PROTOCOL_VERSION,
     control::{
         ForceSyncReply, ForceSyncRequest, PathsReply, PathsRequest, StatusReply, StatusRequest,
@@ -91,9 +91,36 @@ impl Control for ControlService {
         &self,
         _request: Request<ForceSyncRequest>,
     ) -> Result<Response<ForceSyncReply>, Status> {
-        let reply = ForceSyncReply { accepted: true };
-
         self.handle.emit(DaemonEvent::ForceSync);
+        let event = self
+            .handle
+            .wait_for(|e| {
+                matches!(
+                    e,
+                    DaemonEvent::SyncFailed { .. } | DaemonEvent::SyncCompleted { .. }
+                )
+            })
+            .await
+            .map_err(|e| {
+                Status::internal(format!("failed to wait for sync response event: {e:?}"))
+            })?;
+
+        let reply = match event {
+            DaemonEvent::SyncCompleted {
+                uploaded,
+                downloaded,
+            } => ForceSyncReply {
+                error: None,
+                uploaded: uploaded as u32,
+                downloaded: downloaded as u32,
+            },
+            DaemonEvent::SyncFailed { error } => ForceSyncReply {
+                error: Some(error),
+                uploaded: 0,
+                downloaded: 0,
+            },
+            _ => unreachable!(),
+        };
 
         Ok(Response::new(reply))
     }
diff --git a/crates/daemon/src/api/history.rs b/crates/daemon/src/api/history.rs
index 6165464d..0b373604 100644
--- a/crates/daemon/src/api/history.rs
+++ b/crates/daemon/src/api/history.rs
@@ -1,10 +1,10 @@
-use std::pin::Pin;
+use std::{pin::Pin, time::Duration};
 
 use dashmap::DashMap;
 use eyre::Result;
 use time::OffsetDateTime;
 use tokio_stream::Stream;
-use tonic::{Request, Response, Status};
+use tonic::{IntoRequest, Request, Response, Status};
 use tracing::{Level, instrument};
 
 use crate::{
@@ -12,12 +12,16 @@ use crate::{
     daemon::DaemonHandle,
     events::DaemonEvent,
 };
-use turtle::{
+use turtle_api::{
+    client::{
+        proto_duration_to_std, proto_timestamp_to_time, std_to_proto_duration,
+        time_to_proto_timestamp,
+    },
     generated::{
         DAEMON_PROTOCOL_VERSION,
         history::{
-            EndHistoryReply, EndHistoryRequest, HistoryEntry, HistoryEventKind, HistoryReply,
-            HistoryRequest, StartHistoryReply, StartHistoryRequest, TailHistoryReply,
+            AddHistoryRequest, EndHistoryReply, EndHistoryRequest, HistoryEntry, HistoryEventKind,
+            HistoryReply, HistoryRequest, StartHistoryReply, StartHistoryRequest, TailHistoryReply,
             TailHistoryRequest,
             history_server::{History as HistorySvc, HistoryServer},
         },
@@ -60,8 +64,8 @@ impl HistoryService {
 
 fn history_to_reply(history: History) -> HistoryEntry {
     HistoryEntry {
-        timestamp: history.timestamp.unix_timestamp_nanos() as u64,
-        id: history.id.0,
+        timestamp: time_to_proto_timestamp(history.timestamp),
+        id: history.id.to_string(),
         command: history.command,
         cwd: history.cwd,
         session: history.session,
@@ -69,7 +73,7 @@ fn history_to_reply(history: History) -> HistoryEntry {
         author: history.author,
         intent: history.intent.unwrap_or_default(),
         exit: history.exit,
-        duration: history.duration,
+        duration: std_to_proto_duration(history.duration),
     }
 }
 
@@ -85,8 +89,8 @@ impl HistorySvc for HistoryService {
         let req = request.into_inner();
 
         let entries = if let Some(range) = req.range {
-            let from = OffsetDateTime::from_unix_timestamp(range.start as i64).unwrap();
-            let to = OffsetDateTime::from_unix_timestamp(range.end as i64).unwrap();
+            let from = proto_timestamp_to_time(range.start);
+            let to = proto_timestamp_to_time(range.end);
 
             self.handle.history_db().range(from, to).await
         } else {
@@ -101,18 +105,41 @@ impl HistorySvc for HistoryService {
     }
 
     #[instrument(skip_all, level = Level::INFO)]
+    async fn add_history(
+        &self,
+        request: Request<AddHistoryRequest>,
+    ) -> Result<Response<EndHistoryReply>, Status> {
+        let req = request.into_inner();
+        let start_req = req.start.expect("is some");
+
+        let start_response = self
+            .start_history(start_req.into_request())
+            .await?
+            .into_inner();
+        let end_responnse = self
+            .end_history(
+                EndHistoryRequest {
+                    id: start_response.id,
+                    exit: req.exit,
+                    duration: req.duration,
+                }
+                .into_request(),
+            )
+            .await?;
+
+        Ok(end_responnse)
+    }
+
+    #[instrument(skip_all, level = Level::INFO)]
     async fn start_history(
         &self,
         request: Request<StartHistoryRequest>,
     ) -> Result<Response<StartHistoryReply>, Status> {
+        tokio::time::sleep(Duration::from_secs(5)).await;
+
         let req = request.into_inner();
 
-        let timestamp = OffsetDateTime::from_unix_timestamp_nanos(i128::from(req.timestamp))
-            .map_err(|_| {
-                Status::invalid_argument(
-                    "failed to parse timestamp as unix time (expected nanos since epoch)",
-                )
-            })?;
+        let timestamp = proto_timestamp_to_time(req.timestamp);
 
         let h: History = History::daemon()
             .timestamp(timestamp)
@@ -121,7 +148,7 @@ impl HistorySvc for HistoryService {
             .session(req.session)
             .hostname(req.hostname)
             .author(req.author)
-            .intent(req.intent)
+            .intent(req.intent.unwrap_or_default())
             .build()
             .into();
 
@@ -146,16 +173,15 @@ impl HistorySvc for HistoryService {
         request: Request<EndHistoryRequest>,
     ) -> Result<Response<EndHistoryReply>, Status> {
         let req = request.into_inner();
-        let id = HistoryId(req.id);
+        let id = HistoryId::from(req.id);
 
         if let Some((_, mut history)) = self.running.remove(&id) {
             history.exit = req.exit;
-            history.duration = match req.duration {
-                0 => i64::try_from(
-                    (OffsetDateTime::now_utc() - history.timestamp).whole_nanoseconds(),
-                )
-                .expect("failed to convert calculated duration to i64"),
-                value => i64::try_from(value).expect("failed to get i64 duration"),
+            history.duration = match proto_duration_to_std(req.duration) {
+                Duration::ZERO => Duration::from_nanos_u128(
+                    (OffsetDateTime::now_utc() - history.timestamp).whole_nanoseconds() as u128,
+                ),
+                value => value,
             };
 
             self.handle
@@ -164,7 +190,11 @@ impl HistorySvc for HistoryService {
                 .await
                 .map_err(|e| Status::internal(format!("failed to write to db: {e:?}")))?;
 
-            tracing::info!(id = id.0, duration = history.duration, "end history");
+            tracing::info!(
+                id = id.to_string(),
+                duration = history.duration.as_nanos(),
+                "end history"
+            );
 
             let (record_id, idx) = self
                 .history_store
diff --git a/crates/daemon/src/daemon.rs b/crates/daemon/src/daemon.rs
index 1c3afcde..70e65c1e 100644
--- a/crates/daemon/src/daemon.rs
+++ b/crates/daemon/src/daemon.rs
@@ -87,6 +87,20 @@ impl DaemonHandle {
             tracing::warn!("failed to emit event (no receivers?): {e}");
         }
     }
+    pub(crate) async fn wait_for(&self, matches: fn(&DaemonEvent) -> bool) -> Result<DaemonEvent> {
+        let mut rx = self.subscribe();
+        loop {
+            match rx.recv().await {
+                Ok(e) if matches(&e) => {
+                    return Ok(e);
+                }
+                Err(err) => {
+                    return Err(err).context("while waiting for events");
+                }
+                Ok(_) => (),
+            }
+        }
+    }
 
     /// Subscribe to the event bus.
     ///
diff --git a/crates/daemon/src/events.rs b/crates/daemon/src/events.rs
index 3b6fa5d8..654e56cb 100644
--- a/crates/daemon/src/events.rs
+++ b/crates/daemon/src/events.rs
@@ -7,15 +7,14 @@
 //! External processes (like CLI commands) can also inject events via the
 //! Control gRPC service.
 
-use turtle::history::History;
+use turtle_api::history::History;
 
 /// Events that flow through the daemon's event bus.
 ///
 /// Events are broadcast to all components. Each component decides which
 /// events it cares about in its `handle_event` implementation.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) enum DaemonEvent {
-    // ---- History lifecycle ----
     /// A command has started running.
     HistoryStarted(History),
 
@@ -25,11 +24,9 @@ pub(crate) enum DaemonEvent {
     /// Sync completed successfully.
     SyncCompleted {
         /// Number of records uploaded.
-        #[expect(unused)]
         uploaded: usize,
 
         /// Number of records downloaded.
-        #[expect(unused)]
         downloaded: usize,
     },
 
@@ -42,7 +39,6 @@ pub(crate) enum DaemonEvent {
     /// Request an immediate sync (external trigger).
     ForceSync,
 
-    // ---- Lifecycle ----
     /// Request graceful shutdown of the daemon.
     ShutdownRequested,
 }
diff --git a/crates/daemon/src/main.rs b/crates/daemon/src/main.rs
index 9d5a5333..8fd3c119 100644
--- a/crates/daemon/src/main.rs
+++ b/crates/daemon/src/main.rs
@@ -14,7 +14,7 @@ use clap::Parser;
 use eyre::WrapErr;
 use eyre::{Result, bail};
 use fs4::fs_std::FileExt;
-use tracing_subscriber::util::SubscriberInitExt;
+use tracing_subscriber::EnvFilter;
 
 use crate::{
     aclient::{database::ClientSqlite, record::sqlite_store::SqliteStore, settings::Settings},
@@ -43,7 +43,18 @@ enum Cmd {
 
 #[tokio::main]
 async fn main() -> Result<()> {
-    if let Err(e) = tracing_subscriber::registry().try_init() {
+    if let Err(e) = tracing_subscriber::fmt()
+        .with_file(true)
+        .with_line_number(true)
+        .with_level(true)
+        .without_time()
+        .with_env_filter(
+            EnvFilter::builder()
+                .from_env_lossy()
+                .add_directive("turtle_daemon=debug".parse().unwrap()),
+        )
+        .try_init()
+    {
         eprintln!("failed to initialize logging: {e}");
     }
 
diff --git a/crates/daemon/src/server.rs b/crates/daemon/src/server.rs
index 6747f276..d3427769 100644
--- a/crates/daemon/src/server.rs
+++ b/crates/daemon/src/server.rs
@@ -2,7 +2,7 @@ use std::{os::unix::net::SocketAddr, path::PathBuf};
 
 use eyre::Result;
 use eyre::{OptionExt, WrapErr};
-use turtle::generated::{
+use turtle_api::generated::{
     control::control_server::ControlServer, history::history_server::HistoryServer,
 };
 
@@ -57,7 +57,12 @@ pub(crate) fn run_grpc_server(
         (UnixListener::from_std(listener)?, false)
     } else {
         tracing::info!("listening on unix socket {socket_path:?}");
-        (UnixListener::bind(socket_path.clone())?, true)
+        (
+            UnixListener::bind(socket_path.clone()).with_context(|| {
+                format!("Failed to bind to unix socket at: {socket_path}")
+            })?,
+            true,
+        )
     };
 
     let uds_stream = UnixListenerStream::new(uds);