kimun_notes/rag/sync.rs
1//! Background sync loop (P4). When a server URL is configured, a spawned task
2//! keeps the current vault in sync and reports connection status to the UI.
3
4use std::sync::Arc;
5use std::time::Duration;
6
7use kimun_core::NoteVault;
8use kimun_server_client::{
9 RagClient,
10 sync::{RagSync, ServerCapability},
11};
12use tokio::task::JoinHandle;
13
14use super::RagStatus;
15use super::client::server_config;
16use crate::components::events::{AppEvent, AppTx};
17use crate::settings::SharedSettings;
18
19/// How often the background task flushes pending changes and refreshes status.
20const SYNC_INTERVAL: Duration = Duration::from_secs(10);
21
22/// Run a full reconcile (index-wide read + full-collection hash fetch) only
23/// every Nth interval — the drain fast path handles the common case, and a
24/// reconnect forces a reconcile immediately. At 10s × 30 that's ~5 min.
25const RECONCILE_EVERY_N_TICKS: u32 = 30;
26
27/// Spawns the background sync loop for `vault` if a RAG server is configured.
28/// Returns the task handle (abort it when the vault is rebuilt), or `None` when
29/// the feature is off. Status is delivered to the UI via [`AppEvent::RagStatus`].
30pub fn spawn_rag_sync(
31 vault: Arc<NoteVault>,
32 settings: &SharedSettings,
33 tx: AppTx,
34) -> Option<JoinHandle<()>> {
35 let (url, token) = server_config(settings)?;
36
37 Some(tokio::spawn(async move {
38 let mut interval = tokio::time::interval(SYNC_INTERVAL);
39 // Don't stack missed ticks into a back-to-back burst if a slow tick
40 // overruns the interval (large vault / slow server).
41 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
42
43 // Resolve the vault id (which registers the observer) lazily so a
44 // transient failure just retries next tick instead of killing sync for
45 // the whole session.
46 let mut sync: Option<RagSync> = None;
47 // Force a reconcile on the first successful tick and after any offline
48 // gap; drain-only in between.
49 let mut ticks_since_reconcile = RECONCILE_EVERY_N_TICKS;
50 // Sticky across ticks: the last sync call was rejected with 401/403.
51 // Suppresses the per-tick "syncing" flash while the token stays wrong.
52 let mut auth_failed = false;
53
54 loop {
55 interval.tick().await;
56
57 if sync.is_none() {
58 match vault.vault_id().await {
59 Ok(id) => {
60 let client = RagClient::new(url.clone(), token.clone(), id.to_string());
61 sync = Some(RagSync::new(vault.clone(), client));
62 }
63 Err(e) => {
64 log::warn!("RAG: cannot read vault id (will retry): {e}");
65 let _ = tx.send(AppEvent::RagStatus(RagStatus::Offline));
66 continue;
67 }
68 }
69 }
70 let sync = sync.as_ref().expect("sync established above");
71
72 // One probe drives reachability, capability, and auth (adr/0024):
73 // offline, unconfigured (skip sync — the server rejects
74 // everything), or semantic-only/full (llm_available gates Ask).
75 let probe = match sync.probe().await {
76 Some(p) => p,
77 None => {
78 let _ = tx.send(AppEvent::RagStatus(RagStatus::Offline));
79 // Re-establish full consistency on the next successful tick.
80 ticks_since_reconcile = RECONCILE_EVERY_N_TICKS;
81 auth_failed = false;
82 continue;
83 }
84 };
85 if probe.capability == ServerCapability::Unconfigured {
86 let _ = tx.send(AppEvent::RagStatus(RagStatus::NotConfigured));
87 // When an embedder appears, start with a full reconcile.
88 ticks_since_reconcile = RECONCILE_EVERY_N_TICKS;
89 continue;
90 }
91 // The server gates its API behind a token and none is configured:
92 // every sync call would 401 (`/health` itself is un-gated, which
93 // is why the probe still succeeded). Say so up front instead of
94 // rediscovering it as a failure burst every tick.
95 if probe.auth_required && token.is_none() {
96 let _ = tx.send(AppEvent::RagStatus(RagStatus::Unauthorized));
97 ticks_since_reconcile = RECONCILE_EVERY_N_TICKS;
98 continue;
99 }
100 let llm_available = probe.capability.llm_available();
101
102 // The local index is empty while it (re)builds — a healed schema
103 // on first launch, an upgrade, or a manual reindex. Syncing from
104 // that snapshot is destructive (a reconcile reads "no notes" and
105 // would wipe the server collection), so wait, and run a full
106 // reconcile first thing once the index is filled.
107 // While a wrong token keeps failing, skip the transient "syncing"
108 // flash so the footer doesn't flicker syncing ↔ unauthorized.
109 if !auth_failed {
110 let _ = tx.send(AppEvent::RagStatus(RagStatus::Syncing { llm_available }));
111 }
112 if !sync.index_ready() {
113 ticks_since_reconcile = RECONCILE_EVERY_N_TICKS;
114 continue;
115 }
116
117 let result = if ticks_since_reconcile >= RECONCILE_EVERY_N_TICKS {
118 ticks_since_reconcile = 0;
119 sync.tick().await // drain + reconcile
120 } else {
121 ticks_since_reconcile += 1;
122 sync.drain().await // fast path
123 };
124 let status = match result {
125 Ok(true) => {
126 auth_failed = false;
127 RagStatus::Online { llm_available }
128 }
129 // Pass skipped: the index flipped to rebuilding between the
130 // gate above and the call. Nothing was synced, so keep
131 // reporting Syncing (not Online) and force a full reconcile
132 // once the index is ready again.
133 Ok(false) => {
134 ticks_since_reconcile = RECONCILE_EVERY_N_TICKS;
135 auth_failed = false;
136 RagStatus::Syncing { llm_available }
137 }
138 // The server rejected our token (401/403): a credentials
139 // problem, not an unreachable server.
140 Err(e) if e.is_auth() => {
141 log::warn!("RAG server rejected the configured token: {e}");
142 auth_failed = true;
143 RagStatus::Unauthorized
144 }
145 Err(e) => {
146 log::debug!("RAG sync failed: {e}");
147 auth_failed = false;
148 RagStatus::Offline
149 }
150 };
151 let _ = tx.send(AppEvent::RagStatus(status));
152 }
153 }))
154}