diff options
Diffstat (limited to 'crates/daemon/src/api')
| -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 |
3 files changed, 151 insertions, 39 deletions
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, }; |