chore: Commit

2 files changed, 232 insertions, 42 deletions

diff --git a/crates/daemon/src/api/server/control.rs b/crates/daemon/src/api/server/control.rs
index 63fef340..8d1ec7b8 100644
--- a/crates/daemon/src/api/server/control.rs
+++ b/crates/daemon/src/api/server/control.rs
@@ -1,7 +1,13 @@
+use std::time::Duration;
+
+use eyre::Result;
+use rand::Rng;
+use tokio::time::{self, MissedTickBehavior};
 use tonic::{Request, Response, Status};
 use tracing::{Level, instrument};
 
 use crate::{
+    aclient::{history::store::HistoryStore, record::sync, settings::Settings},
     api::{
         DAEMON_PROTOCOL_VERSION, DAEMON_VERSION,
         generated::control::{
@@ -10,20 +16,37 @@ use crate::{
         },
     },
     daemon::DaemonHandle,
+    events::DaemonEvent,
 };
 
+/// Sync state - tracks whether we're in normal operation or retrying after failure.
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum SyncState {
+    /// Normal operation. Periodic syncs only run if [`auto_sync`] is enabled.
+    Idle,
+    /// Retrying after a sync failure. Retries continue regardless of [`auto_sync`]
+    /// until the sync succeeds.
+    Retrying,
+}
+
 /// The Control gRPC service.
 ///
 /// This service is used by external processes to inject events into the daemon.
 /// It's not a component - it's part of the daemon's core infrastructure.
 pub(crate) struct ControlService {
     handle: DaemonHandle,
+    task_handle: tokio::task::JoinHandle<()>,
 }
 
 impl ControlService {
     /// Create a new control service with the given daemon handle.
     pub(crate) fn new(handle: DaemonHandle) -> Self {
-        Self { handle }
+        let task_handle = tokio::spawn(sync_loop(handle.clone()));
+
+        Self {
+            handle,
+            task_handle,
+        }
     }
 
     /// Get a tonic server for this service.
@@ -59,3 +82,181 @@ impl Control for ControlService {
         Ok(Response::new(reply))
     }
 }
+
+/// The main sync loop.
+///
+/// This runs in a spawned task and handles periodic sync as well as
+/// force sync requests.
+#[expect(clippy::significant_drop_tightening, reason = "false positive")]
+async fn sync_loop(handle: DaemonHandle) {
+    tracing::info!("sync loop starting");
+
+    // Clone settings since we need them across await points
+    let settings = handle.settings().await.clone();
+    let host_id = match Settings::host_id().await {
+        Ok(id) => id,
+        Err(e) => {
+            tracing::error!("failed to get host id, sync disabled: {e}");
+            return;
+        }
+    };
+
+    // Create the stores we need
+    let encryption_key = *handle.encryption_key();
+    let history_store = HistoryStore::new(handle.store().clone(), host_id, encryption_key);
+
+    // Don't backoff by more than 30 mins (with a random jitter of up to 1 min)
+    let max_interval: f64 = 60.0f64.mul_add(30.0, rand::thread_rng().gen_range(0.0..60.0));
+
+    let mut ticker = time::interval(Duration::from_secs(settings.daemon.sync_frequency));
+
+    // IMPORTANT: without this, if we miss ticks because a sync takes ages or is otherwise delayed,
+    // we may end up running a lot of syncs in a hot loop.
+    ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
+
+    let mut sync_state = SyncState::Idle;
+
+    let mut daemon_rx = handle.subscribe();
+    loop {
+        tokio::select! {
+            _ = ticker.tick() => {
+                let settings = handle.settings().await;
+
+                // Skip periodic ticks if auto_sync is disabled AND we're not retrying
+                // a previous failure. Retries must continue regardless of auto_sync.
+                if !settings.sync.auto && sync_state == SyncState::Idle {
+                    tracing::debug!("auto_sync disabled, skipping periodic sync tick");
+                    continue;
+                }
+
+                sync_state = do_sync_tick(
+                    &handle,
+                    &history_store,
+                    &mut ticker,
+                    max_interval,
+                    &settings,
+                ).await;
+            }
+            cmd = daemon_rx.recv() => {
+                match cmd {
+                    Ok(DaemonEvent::ForceSync) => {
+                        tracing::info!("executing force sync");
+                        let settings = handle.settings().await;
+                        sync_state = do_sync_tick(
+                            &handle,
+                            &history_store,
+                            &mut ticker,
+                            max_interval,
+                            &settings,
+                        ).await;
+                    },
+                    Ok(DaemonEvent::ShutdownRequested) | Err(_) => {
+                        tracing::info!("sync loop stopping");
+                        break;
+                    },
+                    _ => ()
+                }
+            }
+        }
+    }
+}
+
+/// Execute a single sync tick.
+///
+/// Returns the new sync state: `Idle` on success, `Retrying` on failure.
+async fn do_sync_tick(
+    handle: &DaemonHandle,
+    history_store: &HistoryStore,
+    ticker: &mut time::Interval,
+    max_interval: f64,
+    settings: &Settings,
+) -> SyncState {
+    tracing::info!("sync tick");
+
+    // Check if logged in
+    let logged_in = match settings.sync.have_sync_user() {
+        Ok(v) => v,
+        Err(e) => {
+            tracing::warn!("failed to check login status, skipping sync tick: {e}");
+            return SyncState::Idle;
+        }
+    };
+
+    if !logged_in {
+        tracing::debug!("not logged in, skipping sync tick");
+        return SyncState::Idle;
+    }
+
+    // Perform the sync
+    let res = sync::sync(settings, handle.store(), handle.encryption_key()).await;
+
+    match res {
+        Err(e) => {
+            tracing::error!("sync tick failed with {e}");
+
+            // Emit failure event
+            handle.emit(DaemonEvent::SyncFailed {
+                error: e.to_string(),
+            });
+
+            // Exponential backoff
+            let mut rng = rand::thread_rng();
+            let mut new_interval = ticker.period().as_secs_f64() * rng.gen_range(2.0..2.2);
+
+            if new_interval > max_interval {
+                new_interval = max_interval;
+            }
+
+            *ticker = time::interval_at(
+                time::Instant::now() + Duration::from_secs(new_interval as u64),
+                Duration::from_secs(new_interval as u64),
+            );
+            ticker.reset_after(Duration::from_secs(new_interval as u64));
+            ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
+
+            tracing::error!("backing off, next sync tick in {new_interval}");
+
+            SyncState::Retrying
+        }
+        Ok((uploaded_count, downloaded_records)) => {
+            tracing::info!(
+                uploaded = uploaded_count,
+                downloaded = downloaded_records.len(),
+                "sync complete"
+            );
+
+            // Build history from downloaded records
+            if let Err(e) = history_store
+                .incremental_build(handle.history_db(), &downloaded_records)
+                .await
+            {
+                tracing::error!("failed to build history from downloaded records: {e}");
+            }
+
+            // Emit the records added event (for search indexing)
+            handle.emit(DaemonEvent::RecordsAdded(downloaded_records.clone()));
+
+            // Emit sync completed event
+            handle.emit(DaemonEvent::SyncCompleted {
+                uploaded: uploaded_count as usize,
+                downloaded: downloaded_records.len(),
+            });
+
+            // Reset backoff on success
+            if ticker.period().as_secs() != settings.daemon.sync_frequency {
+                *ticker = time::interval_at(
+                    time::Instant::now() + Duration::from_secs(settings.daemon.sync_frequency),
+                    Duration::from_secs(settings.daemon.sync_frequency),
+                );
+                ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
+            }
+
+            // Store sync time
+            if let Err(e) = Settings::save_sync_time().await {
+                tracing::error!("failed to save sync time: {e}");
+            }
+
+            SyncState::Idle
+        }
+    }
+}
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 {