diff options
| author | Benedikt Peetz <benedikt.peetz@b-peetz.de> | 2026-07-20 15:13:12 +0200 |
|---|---|---|
| committer | Benedikt Peetz <benedikt.peetz@b-peetz.de> | 2026-07-20 15:13:12 +0200 |
| commit | 1594307d9cd819b1ed739fa1ec048c6b27e6f4b0 (patch) | |
| tree | 39b427e3cd6390bc218136454b809450016e0037 /crates/daemon/src/api/server/history.rs | |
| parent | chore: Commit (diff) | |
| download | atuin-1594307d9cd819b1ed739fa1ec048c6b27e6f4b0.zip | |
chore: Commit
Diffstat (limited to 'crates/daemon/src/api/server/history.rs')
| -rw-r--r-- | crates/daemon/src/api/server/history.rs | 71 |
1 files changed, 30 insertions, 41 deletions
diff --git a/crates/daemon/src/api/server/history.rs b/crates/daemon/src/api/server/history.rs index 032876a6..989c7895 100644 --- a/crates/daemon/src/api/server/history.rs +++ b/crates/daemon/src/api/server/history.rs @@ -8,13 +8,16 @@ use tonic::{Request, Response, Status}; use tracing::{Level, instrument}; use crate::{ - aclient::history::{History, HistoryId, store::HistoryStore}, + aclient::{ + history::{History, HistoryId, store::HistoryStore}, + settings::Settings, + }, api::{ DAEMON_PROTOCOL_VERSION, generated::history::{ EndHistoryReply, EndHistoryRequest, HistoryEntry, HistoryEventKind, StartHistoryReply, StartHistoryRequest, TailHistoryReply, TailHistoryRequest, - history_server::History as HistorySvc, + history_server::{History as HistorySvc, HistoryServer}, }, }, daemon::DaemonHandle, @@ -24,25 +27,33 @@ use crate::{ /// The gRPC service implementation. /// /// This is a thin wrapper that delegates to the component's shared state. -pub(crate) struct HistoryGrpcService { +pub(crate) struct HistoryService { /// Commands currently running (not yet completed). running: DashMap<HistoryId, History>, /// Handle to the daemon (set during start). - pub(crate) handle: tokio::sync::RwLock<Option<DaemonHandle>>, + pub(crate) handle: DaemonHandle, /// History store for pushing records (set during start). - pub(crate) history_store: tokio::sync::RwLock<Option<HistoryStore>>, + pub(crate) history_store: HistoryStore, } -impl HistoryGrpcService { - /// Create a new history component. - pub(crate) fn new() -> Self { - Self { +impl HistoryService { + pub(crate) async fn new(handle: DaemonHandle) -> Result<Self> { + let host_id = Settings::host_id().await?; + let history_store = + HistoryStore::new(handle.store().clone(), host_id, *handle.encryption_key()); + + Ok(Self { running: DashMap::new(), - handle: tokio::sync::RwLock::new(None), - history_store: tokio::sync::RwLock::new(None), - } + handle, + history_store, + }) + } + + /// Get a tonic server for this service. + pub(crate) fn into_server(self) -> HistoryServer<Self> { + HistoryServer::new(self) } } @@ -65,7 +76,7 @@ fn history_to_tail_reply(kind: HistoryEventKind, history: History) -> TailHistor } #[tonic::async_trait] -impl HistorySvc for HistoryGrpcService { +impl HistorySvc for HistoryService { type TailHistoryStream = Pin<Box<dyn Stream<Item = Result<TailHistoryReply, Status>> + Send>>; #[instrument(skip_all, level = Level::INFO)] @@ -93,10 +104,7 @@ impl HistorySvc for HistoryGrpcService { .build() .into(); - // Emit the event - if let Some(handle) = self.handle.read().await.as_ref() { - handle.emit(DaemonEvent::HistoryStarted(h.clone())); - } + self.handle.emit(DaemonEvent::HistoryStarted(h.clone())); let id = h.id.clone(); tracing::info!(id = id.to_string(), "start history"); @@ -130,19 +138,7 @@ impl HistorySvc for HistoryGrpcService { value => i64::try_from(value).expect("failed to get i64 duration"), }; - // Get the handle and store to save the history - let handle_guard = self.handle.read().await; - let handle = handle_guard - .as_ref() - .ok_or_else(|| Status::internal("component not initialized"))?; - - let store_guard = self.history_store.read().await; - let history_store = store_guard - .as_ref() - .ok_or_else(|| Status::internal("component not initialized"))?; - - // Save to database - handle + self.handle .history_db() .save(&history) .await @@ -150,14 +146,13 @@ impl HistorySvc for HistoryGrpcService { tracing::info!(id = id.0, duration = history.duration, "end history"); - // Push to record store - let (record_id, idx) = history_store + let (record_id, idx) = self + .history_store .push(history.clone()) .await .map_err(|e| Status::internal(format!("failed to push record to store: {e:?}")))?; - // Emit the event - handle.emit(DaemonEvent::HistoryEnded(history)); + self.handle.emit(DaemonEvent::HistoryEnded(history)); let reply = EndHistoryReply { id: record_id.0.to_string(), @@ -180,13 +175,7 @@ impl HistorySvc for HistoryGrpcService { &self, _request: Request<TailHistoryRequest>, ) -> Result<Response<Self::TailHistoryStream>, Status> { - let handle_guard = self.handle.read().await; - let handle = handle_guard - .as_ref() - .cloned() - .ok_or_else(|| Status::internal("component not initialized"))?; - - let mut rx = handle.subscribe(); + let mut rx = self.handle.subscribe(); let (tx, out_rx) = tokio::sync::mpsc::channel::<Result<TailHistoryReply, Status>>(128); tokio::spawn(async move { |
