diff --git a/crates/bridge/src/bridge.rs b/crates/bridge/src/bridge.rs index 19d2a2e..ae9bd20 100644 --- a/crates/bridge/src/bridge.rs +++ b/crates/bridge/src/bridge.rs @@ -84,6 +84,10 @@ pub struct BridgeHandle { status: Arc>, shutdown_tx: Option>, log_tx: broadcast::Sender, + /// Fires whenever something `stats()` would surface has changed — + /// mail count, ws state, uptime tick on start/stop. Lets the UI replace + /// its 1s poll with a push subscription. + stats_dirty_tx: broadcast::Sender<()>, started_at: Option, store: Option>, task: Option>, @@ -94,10 +98,12 @@ pub struct BridgeHandle { impl BridgeHandle { pub fn new() -> Self { let (log_tx, _) = broadcast::channel(256); + let (stats_dirty_tx, _) = broadcast::channel(16); Self { status: Arc::new(RwLock::new(BridgeStatus::Stopped)), shutdown_tx: None, log_tx, + stats_dirty_tx, started_at: None, store: None, task: None, @@ -113,6 +119,13 @@ impl BridgeHandle { self.log_tx.clone() } + /// Subscribe to "stats dirty" pulses. The receiver gets a `()` every + /// time something that `stats()` would surface has changed; the UI calls + /// `stats()` on each pulse instead of polling on a timer. + pub fn subscribe_stats(&self) -> broadcast::Receiver<()> { + self.stats_dirty_tx.subscribe() + } + pub async fn status(&self) -> BridgeStatus { self.status.read().await.clone() } @@ -268,6 +281,28 @@ impl BridgeHandle { let bus_ids_for_handler = bus_client.last_batch_ids(); self.ws_state_rx = Some(bus_client.state()); + // Spawn a tiny watcher BEFORE the outer task swallows the + // observables: it turns MailStore / WsState changes into + // `stats_dirty_tx` pulses so the UI can replace its 1s poll with + // an event subscription. Aborts when shutdown fires. + { + let stats_dirty = self.stats_dirty_tx.clone(); + let mut store_watch = store.subscribe(); + let mut ws_watch = bus_client.state(); + let mut shutdown_watch = shutdown_sync_rx.clone(); + tokio::spawn(async move { + // Initial pulse so a subscriber sees the freshly-started state. + let _ = stats_dirty.send(()); + loop { + tokio::select! { + _ = store_watch.changed() => { let _ = stats_dirty.send(()); } + _ = ws_watch.changed() => { let _ = stats_dirty.send(()); } + _ = shutdown_watch.changed() => return, + } + } + }); + } + let task = tokio::spawn(async move { let imap_tls = tls_acceptor.clone(); let smtp_tls = tls_acceptor; @@ -362,8 +397,10 @@ impl BridgeHandle { }); self.task = Some(task); + *self.status.write().await = BridgeStatus::Running; self.emit_log("Bridge is running"); + let _ = self.stats_dirty_tx.send(()); // status transition Ok(()) } @@ -376,6 +413,9 @@ impl BridgeHandle { } self.started_at = None; self.ws_state_rx = None; + // Final pulse so the UI re-reads `stats()` (now reporting Stopped / + // zero uptime / zero mails) without waiting for a poll. + let _ = self.stats_dirty_tx.send(()); } fn emit_log(&self, msg: &str) { diff --git a/crates/bridge/src/sync.rs b/crates/bridge/src/sync.rs index 6af26ac..090ad08 100644 --- a/crates/bridge/src/sync.rs +++ b/crates/bridge/src/sync.rs @@ -12,8 +12,6 @@ use crate::tuta::{FolderInfo, MailBackend}; const INTER_REQUEST_DELAY: Duration = Duration::from_millis(150); const INTER_FOLDER_DELAY: Duration = Duration::from_millis(300); -/// Pause between background body-prefetch sweeps. -const PREFETCH_INTERVAL: Duration = Duration::from_secs(30); const MAX_RETRIES: u32 = 3; #[derive(Clone)] @@ -384,14 +382,19 @@ pub async fn run_syncer( info!("Mail syncer shutting down"); } -/// Slow loop: progressively prefetch mail bodies in the background, on its own -/// cadence so it never delays `list_sync_loop`. +/// Event-driven background prefetch. Wakes whenever the `MailStore`'s +/// generation counter changes (a new mail arrived, a sync finished, …) and +/// catches up any folder that still has bodies missing. When the store is +/// quiet — every Mail has its `.eml` cached, no folder mutations in flight — +/// the loop sleeps indefinitely on the watch channel; no timer-driven +/// polling. async fn prefetch_loop( store: &MailStore, local_store: &LocalStore, backend: &dyn MailBackend, mut shutdown: watch::Receiver, ) { + let mut store_watch = store.subscribe(); loop { if *shutdown.borrow() { return; @@ -403,8 +406,11 @@ async fn prefetch_loop( prefetch_details(store, local_store, backend, &folder).await; } + // Wait for the store to change again (or shutdown). `changed()` on a + // freshly-borrowed receiver returns immediately if a bump happened + // during the prefetch pass above, so we never miss a generation. tokio::select! { - _ = tokio::time::sleep(PREFETCH_INTERVAL) => {} + _ = store_watch.changed() => {} _ = shutdown.changed() => return, } } diff --git a/src-tauri/src/main.rs b/src-tauri/src/main.rs index 411b557..74119c7 100644 --- a/src-tauri/src/main.rs +++ b/src-tauri/src/main.rs @@ -18,6 +18,7 @@ fn main() { let handle = BridgeHandle::new(); let log_rx = handle.subscribe_logs(); + let stats_rx = handle.subscribe_stats(); let shared = Arc::new(Mutex::new(handle)); tauri::Builder::default() @@ -40,6 +41,12 @@ fn main() { stream_logs(app_handle, log_rx).await; }); + let app_handle = app.handle().clone(); + let stats_state = app.state::().inner().clone(); + tauri::async_runtime::spawn(async move { + stream_stats(app_handle, stats_rx, stats_state).await; + }); + let state = app.state::().inner().clone(); tauri::async_runtime::spawn(async move { auto_start(state).await; @@ -75,6 +82,36 @@ async fn auto_start(state: Arc>) { } } +async fn stream_stats( + app: tauri::AppHandle, + mut rx: tokio::sync::broadcast::Receiver<()>, + state: BridgeState, +) { + use tauri::Emitter; + // Emit a single snapshot covering both bridge status and stats. Status + // transitions (start / stop) and stats changes (new mail, ws state) all + // pulse the same channel, so the UI replaces its periodic poll with one + // listen per topic. + async fn emit_snapshot(app: &tauri::AppHandle, state: &BridgeState) { + let handle = state.lock().await; + let status = handle.status().await; + let stats = handle.stats().await; + drop(handle); + let _ = app.emit("bridge://stats", &stats); + let _ = app.emit("bridge://status", &status); + } + + emit_snapshot(&app, &state).await; + loop { + match rx.recv().await { + Ok(()) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { + emit_snapshot(&app, &state).await; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } +} + async fn stream_logs( app: tauri::AppHandle, mut rx: tokio::sync::broadcast::Receiver, diff --git a/ui/src/hooks/useBridge.ts b/ui/src/hooks/useBridge.ts index 65d0125..bbe848a 100644 --- a/ui/src/hooks/useBridge.ts +++ b/ui/src/hooks/useBridge.ts @@ -1,10 +1,9 @@ -import { useState, useEffect, useCallback, useRef } from "react"; +import { useState, useEffect, useCallback } from "react"; import { invoke } from "@tauri-apps/api/core"; import { listen } from "@tauri-apps/api/event"; import type { Config, BridgeStatus, BridgeStats } from "../types"; const MAX_LOG_LINES = 500; -const POLL_INTERVAL = 1000; export function useBridge() { const [config, setConfig] = useState(null); @@ -18,7 +17,6 @@ export function useBridge() { const [bridgePassword, setBridgePassword] = useState(null); const [logs, setLogs] = useState([]); const [loading, setLoading] = useState(false); - const pollRef = useRef | null>(null); const refresh = useCallback(() => { invoke("get_status").then(setStatus); @@ -31,9 +29,13 @@ export function useBridge() { invoke("get_bridge_password").then(setBridgePassword); refresh(); - pollRef.current = setInterval(refresh, POLL_INTERVAL); + // The bridge pushes `bridge://stats` and `bridge://status` whenever + // anything changes (mail count, ws state, start/stop). No setInterval. + const unlistenStats = listen("bridge://stats", (e) => setStats(e.payload)); + const unlistenStatus = listen("bridge://status", (e) => setStatus(e.payload)); return () => { - if (pollRef.current) clearInterval(pollRef.current); + unlistenStats.then((fn) => fn()); + unlistenStatus.then((fn) => fn()); }; }, [refresh]);