diff options
Diffstat (limited to 'crates/daemon')
| -rw-r--r-- | crates/daemon/proto/control.proto | 10 | ||||
| -rw-r--r-- | crates/daemon/proto/history.proto | 17 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/database/mod.rs | 7 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/history/builder.rs | 2 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/history/mod.rs | 42 | ||||
| -rw-r--r-- | crates/daemon/src/aclient/history/store.rs | 90 | ||||
| -rw-r--r-- | crates/daemon/src/api/client/mod.rs | 84 | ||||
| -rw-r--r-- | crates/daemon/src/api/server/control.rs | 18 | ||||
| -rw-r--r-- | crates/daemon/src/api/server/history.rs | 88 | ||||
| -rw-r--r-- | crates/daemon/src/lib.rs | 15 |
10 files changed, 257 insertions, 116 deletions
diff --git a/crates/daemon/proto/control.proto b/crates/daemon/proto/control.proto index 47a7b8b2..a8026cb8 100644 --- a/crates/daemon/proto/control.proto +++ b/crates/daemon/proto/control.proto @@ -7,6 +7,9 @@ service Control { // Query the daemon for it's status. rpc Status(StatusRequest) returns (StatusReply); + + // Query the daemon about used paths. + rpc Paths(PathsRequest) returns (PathsReply); } message ForceSyncRequest {} @@ -21,3 +24,10 @@ message StatusReply { uint32 pid = 3; uint32 protocol = 4; } + +message PathsRequest {} +message PathsReply { + string config = 1; + string db = 2; + string socket = 3; +} diff --git a/crates/daemon/proto/history.proto b/crates/daemon/proto/history.proto index 46a41a5e..850b16b9 100644 --- a/crates/daemon/proto/history.proto +++ b/crates/daemon/proto/history.proto @@ -4,7 +4,11 @@ package history; service History { rpc StartHistory(StartHistoryRequest) returns (StartHistoryReply); rpc EndHistory(EndHistoryRequest) returns (EndHistoryReply); + rpc TailHistory(TailHistoryRequest) returns (stream TailHistoryReply); + + // Request history from the daemon + rpc History(HistoryRequest) returns (HistoryReply); } message StartHistoryRequest { @@ -59,3 +63,16 @@ message HistoryEntry { int64 exit = 9; int64 duration = 10; } + +message Range { + uint64 start = 1; + uint64 end = 2; +} + +message HistoryRequest { + string session = 1; + optional Range range = 2; +} +message HistoryReply { + repeated HistoryEntry entries = 1; +} diff --git a/crates/daemon/src/aclient/database/mod.rs b/crates/daemon/src/aclient/database/mod.rs index 36049f80..5e25a0e9 100644 --- a/crates/daemon/src/aclient/database/mod.rs +++ b/crates/daemon/src/aclient/database/mod.rs @@ -51,10 +51,9 @@ pub(crate) struct OptFilters { pub(crate) include_duplicates: bool, } -pub(crate) async fn current_context() -> eyre::Result<Context> { - let session = env::var("ATUIN_SESSION").map_err(|_| { - eyre::eyre!("Failed to find $ATUIN_SESSION in the environment. Check that you have correctly set up your shell.") - })?; +pub(crate) async fn current_context(session: String) -> eyre::Result<Context> { + // TODO(@bpeetz): More of this needs to be moved to the client <2026-07-20> + let hostname = get_host_user(); let cwd = utils::get_current_dir(); let host_id = Settings::host_id().await?; diff --git a/crates/daemon/src/aclient/history/builder.rs b/crates/daemon/src/aclient/history/builder.rs index daa4ef49..ef52637b 100644 --- a/crates/daemon/src/aclient/history/builder.rs +++ b/crates/daemon/src/aclient/history/builder.rs @@ -49,7 +49,7 @@ impl From<HistoryImported> for History { /// so it doesn't have any fields which are known only after /// the command is finished, such as `exit` or `duration`. #[derive(Debug, Clone, TypedBuilder)] -pub(crate) struct HistoryCaptured { +pub struct HistoryCaptured { timestamp: time::OffsetDateTime, #[builder(setter(into))] command: String, diff --git a/crates/daemon/src/aclient/history/mod.rs b/crates/daemon/src/aclient/history/mod.rs index 09d24169..d61c95c5 100644 --- a/crates/daemon/src/aclient/history/mod.rs +++ b/crates/daemon/src/aclient/history/mod.rs @@ -1,4 +1,5 @@ use core::fmt::Formatter; +use regex::RegexSet; use rmp::decode::DecodeStringError; use rmp::decode::ValueReadError; use rmp::{Marker, decode::Bytes}; @@ -28,7 +29,7 @@ const HISTORY_AUTHOR_ENV: &str = "ATUIN_HISTORY_AUTHOR"; const HISTORY_INTENT_ENV: &str = "ATUIN_HISTORY_INTENT"; #[derive(Clone, Debug, Eq, PartialEq, Hash)] -pub struct HistoryId(pub(crate) String); +pub struct HistoryId(pub String); impl Display for HistoryId { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { @@ -60,37 +61,37 @@ pub struct History { /// A client-generated ID, used to identify the entry when syncing. /// /// Stored as `client_id` in the database. - pub(crate) id: HistoryId, + pub id: HistoryId, /// When the command was run. - pub(crate) timestamp: OffsetDateTime, + pub timestamp: OffsetDateTime, /// How long the command took to run. - pub(crate) duration: i64, + pub duration: i64, /// The exit code of the command. - pub(crate) exit: i64, + pub exit: i64, /// The command that was run. - pub(crate) command: String, + pub command: String, /// The current working directory when the command was run. - pub(crate) cwd: String, + pub cwd: String, /// The session ID, associated with a terminal session. - pub(crate) session: String, + pub session: String, /// The hostname of the machine the command was run on. - pub(crate) hostname: String, + pub hostname: String, /// Who wrote this command (human user or automation/agent identity). - pub(crate) author: String, + pub author: String, /// Optional rationale for why the command was executed. - pub(crate) intent: Option<String>, + pub intent: Option<String>, /// Timestamp, which is set when the entry is deleted, allowing a soft delete. - pub(crate) deleted_at: Option<OffsetDateTime>, + pub deleted_at: Option<OffsetDateTime>, } #[derive(Debug, Clone, PartialEq, Eq, sqlx::FromRow)] @@ -397,7 +398,7 @@ impl History { /// .build() /// .into(); /// ``` - pub(crate) fn capture() -> builder::HistoryCapturedBuilder { + pub fn capture() -> builder::HistoryCapturedBuilder { builder::HistoryCaptured::builder() } @@ -473,14 +474,21 @@ impl History { self.exit == 0 || self.duration == -1 } - pub(crate) fn should_save(&self, settings: &Settings) -> bool { + pub fn should_save(&self, filter: SettingsFilter<'_>) -> bool { !(self.command.is_empty() - || settings.history_filter.is_match(&self.command) - || settings.cwd_filter.is_match(&self.cwd) - || (settings.secrets_filter && SECRET_PATTERNS_RE.is_match(&self.command))) + || filter.history.is_match(&self.command) + || filter.cwd.is_match(&self.cwd) + || (filter.secrets && SECRET_PATTERNS_RE.is_match(&self.command))) } } +#[derive(Debug, Copy, Clone)] +pub struct SettingsFilter<'a> { + pub history: &'a RegexSet, + pub cwd: &'a RegexSet, + pub secrets: bool, +} + #[cfg(test)] mod tests { use regex::RegexSet; diff --git a/crates/daemon/src/aclient/history/store.rs b/crates/daemon/src/aclient/history/store.rs index db692590..a749a1a8 100644 --- a/crates/daemon/src/aclient/history/store.rs +++ b/crates/daemon/src/aclient/history/store.rs @@ -313,50 +313,52 @@ impl HistoryStore { } pub(crate) async fn init_store(&self, db: &ClientSqlite) -> Result<()> { - let pb = ProgressBar::new_spinner(); - pb.set_style( - ProgressStyle::with_template("{spinner:.blue} {msg}") - .unwrap() - .with_key("eta", |state: &ProgressState, w: &mut dyn Write| { - write!(w, "{:.1}s", state.eta().as_secs_f64()).unwrap(); - }) - .progress_chars("#>-"), - ); - pb.enable_steady_tick(Duration::from_millis(500)); + todo!(); - pb.set_message("Fetching history from old database"); - - let context = current_context().await?; - let history = db.list(&[], &context, None, false, true).await?; - - pb.set_message("Fetching history already in store"); - let store_ids = self.history_ids().await?; - - pb.set_message("Converting old history to new store"); - let mut records = Vec::new(); - - for i in history { - debug!("loaded {}", i.id); - - if store_ids.contains(&i.id) { - debug!("skipping {} - already exists", i.id); - continue; - } - - if i.deleted_at.is_some() { - records.push(HistoryRecord::Delete(i.id)); - } else { - records.push(HistoryRecord::Create(i)); - } - } - - pb.set_message("Writing to db"); - - if !records.is_empty() { - self.push_batch(records.into_iter()).await?; - } - - pb.finish_with_message("Import complete"); + // let pb = ProgressBar::new_spinner(); + // pb.set_style( + // ProgressStyle::with_template("{spinner:.blue} {msg}") + // .unwrap() + // .with_key("eta", |state: &ProgressState, w: &mut dyn Write| { + // write!(w, "{:.1}s", state.eta().as_secs_f64()).unwrap(); + // }) + // .progress_chars("#>-"), + // ); + // pb.enable_steady_tick(Duration::from_millis(500)); + // + // pb.set_message("Fetching history from old database"); + // + // let context = current_context().await?; + // let history = db.list(&[], &context, None, false, true).await?; + // + // pb.set_message("Fetching history already in store"); + // let store_ids = self.history_ids().await?; + // + // pb.set_message("Converting old history to new store"); + // let mut records = Vec::new(); + // + // for i in history { + // debug!("loaded {}", i.id); + // + // if store_ids.contains(&i.id) { + // debug!("skipping {} - already exists", i.id); + // continue; + // } + // + // if i.deleted_at.is_some() { + // records.push(HistoryRecord::Delete(i.id)); + // } else { + // records.push(HistoryRecord::Create(i)); + // } + // } + // + // pb.set_message("Writing to db"); + // + // if !records.is_empty() { + // self.push_batch(records.into_iter()).await?; + // } + // + // pb.finish_with_message("Import complete"); Ok(()) } @@ -364,8 +366,8 @@ impl HistoryStore { #[cfg(test)] mod tests { - use turtle_common::record::DecryptedData; use time::macros::datetime; + use turtle_common::record::DecryptedData; use crate::aclient::history::{HISTORY_VERSION, store::HistoryRecord}; diff --git a/crates/daemon/src/api/client/mod.rs b/crates/daemon/src/api/client/mod.rs index 71fa0e37..c588fb09 100644 --- a/crates/daemon/src/api/client/mod.rs +++ b/crates/daemon/src/api/client/mod.rs @@ -1,4 +1,5 @@ use eyre::{Context as EyreContext, Result}; +use time::OffsetDateTime; use tonic::Code; use tonic::transport::{Channel, Endpoint, Uri}; use tower::service_fn; @@ -8,8 +9,11 @@ use hyper_util::rt::TokioIo; #[cfg(unix)] use tokio::net::UnixStream; +use crate::api::generated; +use crate::api::generated::control::{ForceSyncReply, ForceSyncRequest, PathsReply, PathsRequest}; +use crate::api::generated::history::{HistoryEntry, HistoryRequest}; use crate::{ - aclient::{history::History, settings::Settings}, + aclient::history::History, api::{ DAEMON_PROTOCOL_VERSION, DAEMON_VERSION, generated::{ @@ -18,18 +22,49 @@ use crate::{ }, history::{ EndHistoryReply, EndHistoryRequest, StartHistoryReply, StartHistoryRequest, - TailHistoryReply, TailHistoryRequest, - history_client::HistoryClient as HistoryServiceClient, + TailHistoryRequest, history_client::HistoryClient as HistoryServiceClient, }, }, }, }; -fn daemon_matches_expected(version: &str, protocol: u32) -> bool { +pub use crate::api::generated::history::{HistoryEventKind, TailHistoryReply}; + +fn normalize_optional_field(value: &str) -> Option<String> { + let trimmed = value.trim(); + if trimmed.is_empty() { + None + } else { + Some(trimmed.to_owned()) + } +} + +pub fn history_entry_to_history(entry: HistoryEntry) -> History { + let timestamp = OffsetDateTime::from_unix_timestamp_nanos(i128::from(entry.timestamp)) + .expect("Daemon history timestamp should always be valid"); + + History { + id: entry.id.into(), + timestamp, + duration: entry.duration, + exit: entry.exit, + command: entry.command, + cwd: entry.cwd, + session: entry.session, + hostname: entry.hostname, + author: entry.author, + intent: normalize_optional_field(&entry.intent), + deleted_at: None, + } +} + +#[must_use] +pub fn daemon_matches_expected(version: &str, protocol: u32) -> bool { version == DAEMON_VERSION && protocol == DAEMON_PROTOCOL_VERSION } -fn daemon_mismatch_message(version: &str, protocol: u32) -> String { +#[must_use] +pub fn daemon_mismatch_message(version: &str, protocol: u32) -> String { if protocol == DAEMON_PROTOCOL_VERSION { format!("daemon is out of date: expected {DAEMON_VERSION}, got {version}") } else { @@ -99,6 +134,11 @@ pub struct HistoryClient { client: HistoryServiceClient<Channel>, } +pub struct Range { + pub start: OffsetDateTime, + pub end: OffsetDateTime, +} + // Wrap the grpc client impl HistoryClient { #[cfg(unix)] @@ -141,6 +181,24 @@ impl HistoryClient { Ok(self.client.start_history(req).await?.into_inner()) } + pub async fn history(&mut self, session: String, range: Option<Range>) -> Result<Vec<History>> { + let req = HistoryRequest { + session, + range: range.map(|r| generated::history::Range { + start: r.start.unix_timestamp() as u64, + end: r.end.unix_timestamp() as u64, + }), + }; + + let reply = self.client.history(req).await?.into_inner(); + + Ok(reply + .entries + .into_iter() + .map(history_entry_to_history) + .collect()) + } + pub async fn end_history( &mut self, id: String, @@ -152,7 +210,7 @@ impl HistoryClient { Ok(self.client.end_history(req).await?.into_inner()) } - pub(crate) async fn tail_history(&mut self) -> Result<tonic::Streaming<TailHistoryReply>> { + pub async fn tail_history(&mut self) -> Result<tonic::Streaming<TailHistoryReply>> { Ok(self .client .tail_history(TailHistoryRequest {}) @@ -196,10 +254,16 @@ impl ControlClient { Ok(Self { client }) } - /// Connect using settings. - #[cfg(unix)] - pub async fn from_settings(settings: &Settings) -> Result<Self> { - Self::new(settings.daemon.socket_path.clone()).await + pub async fn paths(&mut self) -> Result<PathsReply> { + Ok(self.client.paths(PathsRequest {}).await?.into_inner()) + } + + pub async fn force_sync(&mut self) -> Result<ForceSyncReply> { + Ok(self + .client + .force_sync(ForceSyncRequest {}) + .await? + .into_inner()) } pub async fn status(&mut self) -> Result<StatusReply> { diff --git a/crates/daemon/src/api/server/control.rs b/crates/daemon/src/api/server/control.rs index 8d1ec7b8..a5e26355 100644 --- a/crates/daemon/src/api/server/control.rs +++ b/crates/daemon/src/api/server/control.rs @@ -11,7 +11,7 @@ use crate::{ api::{ DAEMON_PROTOCOL_VERSION, DAEMON_VERSION, generated::control::{ - ForceSyncReply, ForceSyncRequest, StatusReply, StatusRequest, + ForceSyncReply, ForceSyncRequest, PathsReply, PathsRequest, StatusReply, StatusRequest, control_server::{Control, ControlServer}, }, }, @@ -58,6 +58,22 @@ impl ControlService { #[tonic::async_trait] impl Control for ControlService { #[instrument(skip_all, level = Level::INFO)] + async fn paths(&self, _request: Request<PathsRequest>) -> Result<Response<PathsReply>, Status> { + let settings = self.handle.settings().await; + + let config = Settings::get_config_path() + .map_err(|e| Status::internal(format!("failed to get settings path: {e:?}")))?; + + let reply = PathsReply { + config: config.to_string_lossy().to_string(), + db: settings.db_path.clone(), + socket: settings.daemon.socket_path.clone(), + }; + + Ok(Response::new(reply)) + } + + #[instrument(skip_all, level = Level::INFO)] async fn status( &self, _request: Request<StatusRequest>, diff --git a/crates/daemon/src/api/server/history.rs b/crates/daemon/src/api/server/history.rs index 989c7895..0edf3b94 100644 --- a/crates/daemon/src/api/server/history.rs +++ b/crates/daemon/src/api/server/history.rs @@ -9,14 +9,16 @@ use tracing::{Level, instrument}; use crate::{ aclient::{ + database::{ClientSqlite, current_context}, history::{History, HistoryId, store::HistoryStore}, settings::Settings, }, api::{ DAEMON_PROTOCOL_VERSION, generated::history::{ - EndHistoryReply, EndHistoryRequest, HistoryEntry, HistoryEventKind, StartHistoryReply, - StartHistoryRequest, TailHistoryReply, TailHistoryRequest, + EndHistoryReply, EndHistoryRequest, HistoryEntry, HistoryEventKind, HistoryReply, + HistoryRequest, StartHistoryReply, StartHistoryRequest, TailHistoryReply, + TailHistoryRequest, history_server::{History as HistorySvc, HistoryServer}, }, }, @@ -32,14 +34,16 @@ pub(crate) struct HistoryService { running: DashMap<HistoryId, History>, /// Handle to the daemon (set during start). - pub(crate) handle: DaemonHandle, + handle: DaemonHandle, - /// History store for pushing records (set during start). - pub(crate) history_store: HistoryStore, + /// History store for pushing records + history_store: HistoryStore, + + history_db: ClientSqlite, } impl HistoryService { - pub(crate) async fn new(handle: DaemonHandle) -> Result<Self> { + pub(crate) async fn new(handle: DaemonHandle, history_db: ClientSqlite) -> Result<Self> { let host_id = Settings::host_id().await?; let history_store = HistoryStore::new(handle.store().clone(), host_id, *handle.encryption_key()); @@ -48,6 +52,7 @@ impl HistoryService { running: DashMap::new(), handle, history_store, + history_db, }) } @@ -57,21 +62,18 @@ impl HistoryService { } } -fn history_to_tail_reply(kind: HistoryEventKind, history: History) -> TailHistoryReply { - TailHistoryReply { - kind: kind as i32, - history: Some(HistoryEntry { - timestamp: history.timestamp.unix_timestamp_nanos() as u64, - id: history.id.0, - command: history.command, - cwd: history.cwd, - session: history.session, - hostname: history.hostname, - author: history.author, - intent: history.intent.unwrap_or_default(), - exit: history.exit, - duration: history.duration, - }), +fn history_to_reply(history: History) -> HistoryEntry { + HistoryEntry { + timestamp: history.timestamp.unix_timestamp_nanos() as u64, + id: history.id.0, + command: history.command, + cwd: history.cwd, + session: history.session, + hostname: history.hostname, + author: history.author, + intent: history.intent.unwrap_or_default(), + exit: history.exit, + duration: history.duration, } } @@ -80,6 +82,35 @@ impl HistorySvc for HistoryService { type TailHistoryStream = Pin<Box<dyn Stream<Item = Result<TailHistoryReply, Status>> + Send>>; #[instrument(skip_all, level = Level::INFO)] + async fn history( + &self, + request: Request<HistoryRequest>, + ) -> Result<Response<HistoryReply>, Status> { + let req = request.into_inner(); + + let context = current_context(req.session) + .await + .map_err(|e| Status::internal(format!("failed to aquire context: {e:?}")))?; + + 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(); + + self.history_db.range(from, to).await + } else { + self.history_db + .list(&[], &context, None, false, false) + .await + } + .map_err(|e| Status::internal(format!("failed to read db: {e:?}")))? + .into_iter() + .map(history_to_reply) + .collect(); + + Ok(Response::new(HistoryReply { entries })) + } + + #[instrument(skip_all, level = Level::INFO)] async fn start_history( &self, request: Request<StartHistoryRequest>, @@ -120,7 +151,6 @@ impl HistorySvc for HistoryService { } #[instrument(skip_all, level = Level::INFO)] - #[expect(clippy::significant_drop_tightening, reason = "Would be a logic-bug")] async fn end_history( &self, request: Request<EndHistoryRequest>, @@ -195,12 +225,14 @@ impl HistorySvc for HistoryService { }; let reply = match event { - DaemonEvent::HistoryStarted(history) => { - Some(history_to_tail_reply(HistoryEventKind::Started, history)) - } - DaemonEvent::HistoryEnded(history) => { - Some(history_to_tail_reply(HistoryEventKind::Ended, history)) - } + DaemonEvent::HistoryStarted(history) => Some(TailHistoryReply { + kind: HistoryEventKind::Started.into(), + history: Some(history_to_reply(history)), + }), + DaemonEvent::HistoryEnded(history) => Some(TailHistoryReply { + kind: HistoryEventKind::Ended.into(), + history: Some(history_to_reply(history)), + }), _ => None, }; diff --git a/crates/daemon/src/lib.rs b/crates/daemon/src/lib.rs index 4f96e410..c6877096 100644 --- a/crates/daemon/src/lib.rs +++ b/crates/daemon/src/lib.rs @@ -12,10 +12,7 @@ use fs4::fs_std::FileExt; use tokio::time::sleep; use crate::{ - aclient::{ - database::ClientSqlite as HistoryDatabase, record::sqlite_store::SqliteStore, - settings::Settings, - }, + aclient::{database::ClientSqlite, record::sqlite_store::SqliteStore, settings::Settings}, api::{ DAEMON_VERSION, server::{control::ControlService, history::HistoryService}, @@ -34,17 +31,13 @@ pub(crate) mod server; /// /// This creates a daemon, /// starts the gRPC server with services, and runs the event loop. -pub async fn boot( - settings: Settings, - store: SqliteStore, - history_db: HistoryDatabase, -) -> Result<()> { +pub async fn boot(settings: Settings, store: SqliteStore, history_db: ClientSqlite) -> Result<()> { let pidfile_path = PathBuf::from(&settings.daemon.pidfile_path); let _pidfile_guard = PidfileGuard::acquire(&pidfile_path)?; let mut daemon = Daemon::builder(settings.clone()) .store(store) - .history_db(history_db) + .history_db(history_db.clone()) .build()?; let handle = { @@ -61,7 +54,7 @@ pub async fn boot( handle }; - let history_service = HistoryService::new(handle.clone()).await?; + let history_service = HistoryService::new(handle.clone(), history_db).await?; let control_service = ControlService::new(handle.clone()); server::run_grpc_server( |