aven_core/
local_state.rs

1use std::collections::{BTreeMap, BTreeSet};
2
3use anyhow::Result;
4
5use crate::db::{Database, begin_immediate, get_meta, set_meta};
6use crate::ids::WorkspaceId;
7
8const IOS_QUEUE_WORKSPACE_META_KEY: &str = "ios_queue_workspace_id";
9const ONBOARDING_META_KEY: &str = "tui_onboarding_version";
10const SYNC_ACCESS_REFUSED_AT_META_KEY: &str = "sync_access_refused_at";
11/// The `sync_` prefix places it in the sync state that reset clears.
12const SYNC_VAULT_DELETED_AT_META_KEY: &str = "sync_vault_deleted_at";
13const FIRST_LAUNCH_ONBOARDING_VERSION: u32 = 1;
14const TRIAGE_REVIEWS_META_KEY: &str = "tui_triage_reviews";
15
16#[derive(Debug, Clone, PartialEq, Eq)]
17pub struct SyncAccessRefusal {
18    pub at: String,
19}
20
21#[derive(Debug, Clone, PartialEq, Eq)]
22pub struct SyncVaultDeletion {
23    pub at: String,
24}
25
26/// Local daemon health, cleared with the other sync metadata on reset.
27#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
28pub struct DaemonSyncState {
29    /// Unix seconds of the last completed daemon round.
30    pub last_success_at: Option<i64>,
31    pub error: Option<DaemonSyncError>,
32    /// Unix seconds of the backoff deadline, absent after success.
33    pub next_retry_at: Option<i64>,
34}
35
36#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
37pub struct DaemonSyncError {
38    pub code: String,
39    pub message: String,
40    /// Unix seconds when this error code first appeared without a success.
41    pub since: i64,
42}
43
44#[derive(Debug, Clone, Copy, PartialEq, Eq)]
45pub enum OnboardingStatus {
46    Due,
47    Complete,
48    Established,
49}
50
51impl Database {
52    pub async fn daemon_sync_state(&self) -> Result<DaemonSyncState> {
53        self.meta("sync_daemon_health")
54            .await?
55            .map(|value| serde_json::from_str(&value).map_err(Into::into))
56            .unwrap_or_else(|| Ok(DaemonSyncState::default()))
57    }
58
59    /// Records an actual round, not idle polling or lock contention. Repeated
60    /// errors retain their first-seen time; unchanged snapshots cause no write.
61    pub async fn record_daemon_sync_round(
62        &self,
63        at: i64,
64        error: Option<(&str, &str, i64)>,
65    ) -> Result<()> {
66        let previous = self.daemon_sync_state().await?;
67        let mut state = previous.clone();
68        if let Some((code, message, retry_at)) = error {
69            let since = previous
70                .error
71                .as_ref()
72                .filter(|error| error.code == code)
73                .map_or(at, |error| error.since);
74            state.error = Some(DaemonSyncError {
75                code: code.to_string(),
76                message: message.to_string(),
77                since,
78            });
79            state.next_retry_at = Some(retry_at);
80        } else {
81            state.last_success_at = Some(at);
82            state.error = None;
83            state.next_retry_at = None;
84        }
85        if state != previous {
86            let mut conn = self.acquire_writer().await?;
87            set_meta(
88                &mut conn,
89                "sync_daemon_health",
90                &serde_json::to_string(&state)?,
91            )
92            .await?;
93        }
94        Ok(())
95    }
96
97    /// Returns the last time this installation's sync credentials were refused.
98    pub async fn sync_access_refusal(&self) -> Result<Option<SyncAccessRefusal>> {
99        Ok(self
100            .meta(SYNC_ACCESS_REFUSED_AT_META_KEY)
101            .await?
102            .map(|at| SyncAccessRefusal { at }))
103    }
104
105    /// Remembers an authentication refusal until a later sync succeeds.
106    pub async fn record_sync_access_refusal(&self) -> Result<SyncAccessRefusal> {
107        let at = crate::ids::now();
108        let mut conn = self.acquire_writer().await?;
109        set_meta(&mut conn, SYNC_ACCESS_REFUSED_AT_META_KEY, &at).await?;
110        Ok(SyncAccessRefusal { at })
111    }
112
113    /// Clears a previous authentication refusal after a successful sync.
114    pub async fn clear_sync_access_refusal(&self) -> Result<()> {
115        let mut conn = self.acquire_writer().await?;
116        sqlx::query("DELETE FROM meta WHERE key = ?")
117            .bind(SYNC_ACCESS_REFUSED_AT_META_KEY)
118            .execute(&mut *conn)
119            .await?;
120        Ok(())
121    }
122
123    /// Returns when the server last reported this sync's vault deleted.
124    pub async fn sync_vault_deletion(&self) -> Result<Option<SyncVaultDeletion>> {
125        Ok(self
126            .meta(SYNC_VAULT_DELETED_AT_META_KEY)
127            .await?
128            .map(|at| SyncVaultDeletion { at }))
129    }
130
131    /// Remembers a vault deletion until a user-started sync succeeds or sync
132    /// is reset.
133    pub async fn record_sync_vault_deletion(&self) -> Result<SyncVaultDeletion> {
134        let at = crate::ids::now();
135        let mut conn = self.acquire_writer().await?;
136        set_meta(&mut conn, SYNC_VAULT_DELETED_AT_META_KEY, &at).await?;
137        Ok(SyncVaultDeletion { at })
138    }
139
140    pub async fn clear_sync_vault_deletion(&self) -> Result<()> {
141        let mut conn = self.acquire_writer().await?;
142        sqlx::query("DELETE FROM meta WHERE key = ?")
143            .bind(SYNC_VAULT_DELETED_AT_META_KEY)
144            .execute(&mut *conn)
145            .await?;
146        Ok(())
147    }
148
149    pub async fn restore_queue_workspace(&self) -> Result<crate::workspaces::Workspace> {
150        let mut conn = self.acquire_writer().await?;
151        let mut tx = begin_immediate(&mut conn).await?;
152        let workspaces = crate::workspaces::list_workspaces(&mut tx).await?;
153        let stored_id = get_meta(&mut tx, IOS_QUEUE_WORKSPACE_META_KEY)
154            .await?
155            .and_then(|value| value.parse::<WorkspaceId>().ok());
156        let selected = stored_id
157            .and_then(|id| workspaces.iter().find(|workspace| workspace.id == id))
158            .or_else(|| workspaces.first())
159            .cloned()
160            .ok_or_else(|| anyhow::anyhow!("no available workspace"))?;
161        set_meta(&mut tx, IOS_QUEUE_WORKSPACE_META_KEY, selected.id.as_str()).await?;
162        tx.commit().await?;
163        Ok(selected)
164    }
165
166    pub async fn select_queue_workspace(
167        &self,
168        workspace_id: &WorkspaceId,
169    ) -> Result<crate::workspaces::Workspace> {
170        let mut conn = self.acquire_writer().await?;
171        let mut tx = begin_immediate(&mut conn).await?;
172        let workspace = crate::workspaces::workspace_for_id(&mut tx, workspace_id).await?;
173        set_meta(&mut tx, IOS_QUEUE_WORKSPACE_META_KEY, workspace.id.as_str()).await?;
174        tx.commit().await?;
175        Ok(workspace)
176    }
177
178    pub async fn onboarding_status(&self) -> Result<OnboardingStatus> {
179        let mut conn = self.acquire_reader().await?;
180        let marker = get_meta(&mut conn, ONBOARDING_META_KEY).await?;
181        if marker_version(marker.as_deref())
182            .is_some_and(|version| version >= FIRST_LAUNCH_ONBOARDING_VERSION)
183        {
184            return Ok(OnboardingStatus::Complete);
185        }
186
187        let has_tasks: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM tasks)")
188            .fetch_one(&mut *conn)
189            .await?;
190        Ok(if has_tasks {
191            OnboardingStatus::Established
192        } else {
193            OnboardingStatus::Due
194        })
195    }
196
197    pub async fn complete_onboarding(&self) -> Result<()> {
198        let mut conn = self.acquire_writer().await?;
199        let mut tx = begin_immediate(&mut conn).await?;
200        let marker = get_meta(&mut tx, ONBOARDING_META_KEY).await?;
201        if marker_version(marker.as_deref())
202            .is_none_or(|version| version < FIRST_LAUNCH_ONBOARDING_VERSION)
203        {
204            set_meta(
205                &mut tx,
206                ONBOARDING_META_KEY,
207                &FIRST_LAUNCH_ONBOARDING_VERSION.to_string(),
208            )
209            .await?;
210        }
211        tx.commit().await?;
212        Ok(())
213    }
214}
215
216/// A reminder this device has handled, keyed by kind, task and trigger.
217/// Digest entries use an empty `task_id`.
218#[derive(Debug, Clone, PartialEq, Eq, sqlx::FromRow)]
219pub struct ReminderLogEntry {
220    pub kind: String,
221    pub task_id: String,
222    pub trigger: String,
223    pub outcome: String,
224}
225
226impl Database {
227    pub async fn reminder_log(&self) -> Result<Vec<ReminderLogEntry>> {
228        let mut conn = self.acquire_reader().await?;
229        Ok(sqlx::query_as::<_, ReminderLogEntry>(
230            "SELECT kind, task_id, trigger, outcome FROM local_reminder_log",
231        )
232        .fetch_all(&mut *conn)
233        .await?)
234    }
235
236    /// Records entries in one transaction, replacing the outcome of existing
237    /// keys.
238    pub async fn record_reminders(&self, entries: &[ReminderLogEntry]) -> Result<()> {
239        if entries.is_empty() {
240            return Ok(());
241        }
242        let recorded_at = crate::ids::now();
243        let mut conn = self.acquire_writer().await?;
244        let mut tx = begin_immediate(&mut conn).await?;
245        for entry in entries {
246            sqlx::query(
247                "INSERT INTO local_reminder_log(kind, task_id, trigger, outcome, recorded_at)
248                 VALUES (?, ?, ?, ?, ?)
249                 ON CONFLICT(kind, task_id, trigger)
250                 DO UPDATE SET outcome = excluded.outcome, recorded_at = excluded.recorded_at",
251            )
252            .bind(&entry.kind)
253            .bind(&entry.task_id)
254            .bind(&entry.trigger)
255            .bind(&entry.outcome)
256            .bind(&recorded_at)
257            .execute(&mut *tx)
258            .await?;
259        }
260        tx.commit().await?;
261        Ok(())
262    }
263}
264
265fn marker_version(value: Option<&str>) -> Option<u32> {
266    value?.trim().parse().ok()
267}
268
269/// Database-local sidebar preferences shared across workspaces and projects.
270#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
271pub enum SidebarSection {
272    Views,
273    Scope,
274    Projects,
275}
276
277impl SidebarSection {
278    fn meta_key(self) -> &'static str {
279        match self {
280            Self::Views => "tui_sidebar_views_collapsed",
281            Self::Scope => "tui_sidebar_scope_collapsed",
282            Self::Projects => "tui_sidebar_projects_collapsed",
283        }
284    }
285}
286
287impl Database {
288    pub async fn collapsed_sidebar_sections(&self) -> Result<BTreeSet<SidebarSection>> {
289        let mut conn = self.acquire_reader().await?;
290        let mut sections = BTreeSet::new();
291        for section in [
292            SidebarSection::Views,
293            SidebarSection::Scope,
294            SidebarSection::Projects,
295        ] {
296            if get_meta(&mut conn, section.meta_key())
297                .await?
298                .as_deref()
299                .map(str::trim)
300                == Some("true")
301            {
302                sections.insert(section);
303            }
304        }
305        Ok(sections)
306    }
307
308    pub async fn set_sidebar_section_collapsed(
309        &self,
310        section: SidebarSection,
311        collapsed: bool,
312    ) -> Result<()> {
313        let mut conn = self.acquire_writer().await?;
314        let mut tx = begin_immediate(&mut conn).await?;
315        set_meta(
316            &mut tx,
317            section.meta_key(),
318            if collapsed { "true" } else { "false" },
319        )
320        .await?;
321        tx.commit().await?;
322        Ok(())
323    }
324}
325
326/// When each task was last kept as it stands in triage, in unix seconds. The
327/// marks are local to this database: they only quiet triage on this machine.
328pub type TriageReviews = BTreeMap<String, i64>;
329
330impl Database {
331    pub async fn triage_reviews(&self) -> Result<TriageReviews> {
332        let mut conn = self.acquire_reader().await?;
333        Ok(parse_triage_reviews(
334            get_meta(&mut conn, TRIAGE_REVIEWS_META_KEY)
335                .await?
336                .as_deref(),
337        ))
338    }
339
340    /// Sets or clears a task's review mark and returns the previous one. Marks
341    /// older than the longest stale window no longer affect triage and are
342    /// dropped.
343    pub async fn set_triage_review(
344        &self,
345        task_id: &str,
346        reviewed_at: Option<i64>,
347        now_seconds: i64,
348    ) -> Result<Option<i64>> {
349        let mut conn = self.acquire_writer().await?;
350        let mut tx = begin_immediate(&mut conn).await?;
351        let mut reviews =
352            parse_triage_reviews(get_meta(&mut tx, TRIAGE_REVIEWS_META_KEY).await?.as_deref());
353        let previous = match reviewed_at {
354            Some(at) => reviews.insert(task_id.to_string(), at),
355            None => reviews.remove(task_id),
356        };
357        let horizon = now_seconds - crate::queue::STALE_TODO_DAYS * 86_400;
358        reviews.retain(|_, at| *at > horizon);
359        set_meta(
360            &mut tx,
361            TRIAGE_REVIEWS_META_KEY,
362            &serde_json::to_string(&reviews)?,
363        )
364        .await?;
365        tx.commit().await?;
366        Ok(previous)
367    }
368}
369
370fn parse_triage_reviews(value: Option<&str>) -> TriageReviews {
371    value
372        .and_then(|value| serde_json::from_str(value).ok())
373        .unwrap_or_default()
374}
375
376#[cfg(test)]
377mod tests {
378    use super::*;
379
380    #[tokio::test]
381    async fn daemon_health_preserves_first_seen_and_clears_on_success_and_reset() {
382        let dir = tempfile::tempdir().unwrap();
383        let path = dir.path().join("health.sqlite");
384        let database = Database::open(&path).await.unwrap();
385        assert_eq!(
386            database.daemon_sync_state().await.unwrap(),
387            DaemonSyncState::default()
388        );
389        database.record_daemon_sync_round(100, None).await.unwrap();
390        database
391            .record_daemon_sync_round(110, Some(("network", "retry", 111)))
392            .await
393            .unwrap();
394        database
395            .record_daemon_sync_round(120, Some(("network", "retry", 122)))
396            .await
397            .unwrap();
398        let state = database.daemon_sync_state().await.unwrap();
399        assert_eq!(state.last_success_at, Some(100));
400        assert_eq!(state.error.as_ref().unwrap().since, 110);
401        assert_eq!(state.next_retry_at, Some(122));
402        sqlx::raw_sql(
403            "CREATE TRIGGER reject_health_write BEFORE UPDATE ON meta
404            WHEN NEW.key = 'sync_daemon_health'
405            BEGIN SELECT RAISE(ABORT, 'redundant health write'); END;",
406        )
407        .execute(&mut *database.acquire_writer().await.unwrap())
408        .await
409        .unwrap();
410        database
411            .record_daemon_sync_round(121, Some(("network", "retry", 122)))
412            .await
413            .unwrap();
414        sqlx::raw_sql("DROP TRIGGER reject_health_write")
415            .execute(&mut *database.acquire_writer().await.unwrap())
416            .await
417            .unwrap();
418        assert_eq!(
419            Database::open(&path)
420                .await
421                .unwrap()
422                .daemon_sync_state()
423                .await
424                .unwrap(),
425            state
426        );
427        database
428            .record_daemon_sync_round(130, Some(("keys", "unlock", 140)))
429            .await
430            .unwrap();
431        assert_eq!(
432            database
433                .daemon_sync_state()
434                .await
435                .unwrap()
436                .error
437                .unwrap()
438                .since,
439            130
440        );
441        database.record_daemon_sync_round(150, None).await.unwrap();
442        assert_eq!(
443            database.daemon_sync_state().await.unwrap(),
444            DaemonSyncState {
445                last_success_at: Some(150),
446                error: None,
447                next_retry_at: None,
448            }
449        );
450        database.clear_sync_state().await.unwrap();
451        assert_eq!(
452            database.daemon_sync_state().await.unwrap(),
453            DaemonSyncState::default()
454        );
455    }
456
457    #[tokio::test]
458    async fn sync_access_refusal_persists_until_cleared() {
459        let database = Database::open(std::path::Path::new(":memory:"))
460            .await
461            .unwrap();
462        assert_eq!(database.sync_access_refusal().await.unwrap(), None);
463
464        let recorded = database.record_sync_access_refusal().await.unwrap();
465        assert!(!recorded.at.is_empty());
466        assert_eq!(
467            database.sync_access_refusal().await.unwrap(),
468            Some(recorded)
469        );
470
471        database.clear_sync_access_refusal().await.unwrap();
472        assert_eq!(database.sync_access_refusal().await.unwrap(), None);
473    }
474
475    #[tokio::test]
476    async fn reminder_outcome_replaces_existing_key() {
477        let database = Database::open(std::path::Path::new(":memory:"))
478            .await
479            .unwrap();
480        let entry = |outcome: &str| ReminderLogEntry {
481            kind: "available".into(),
482            task_id: "T1".into(),
483            trigger: "2026-10-14T09:00:00Z".into(),
484            outcome: outcome.into(),
485        };
486
487        database
488            .record_reminders(&[entry("waiting")])
489            .await
490            .unwrap();
491        database.record_reminders(&[entry("sent")]).await.unwrap();
492
493        assert_eq!(database.reminder_log().await.unwrap(), vec![entry("sent")]);
494    }
495
496    #[tokio::test]
497    async fn sync_vault_deletion_persists_apart_from_access_refusal_until_cleared() {
498        let database = Database::open(std::path::Path::new(":memory:"))
499            .await
500            .unwrap();
501        assert_eq!(database.sync_vault_deletion().await.unwrap(), None);
502
503        let recorded = database.record_sync_vault_deletion().await.unwrap();
504        assert_eq!(
505            database.sync_vault_deletion().await.unwrap(),
506            Some(recorded)
507        );
508        assert_eq!(database.sync_access_refusal().await.unwrap(), None);
509
510        database.clear_sync_vault_deletion().await.unwrap();
511        assert_eq!(database.sync_vault_deletion().await.unwrap(), None);
512    }
513
514    #[tokio::test]
515    async fn sidebar_preferences_are_local_and_invalid_values_default_to_expanded() {
516        let database = Database::open(std::path::Path::new(":memory:"))
517            .await
518            .unwrap();
519        let before: i64 = {
520            let mut conn = database.acquire_reader().await.unwrap();
521            sqlx::query_scalar("SELECT COUNT(*) FROM changes")
522                .fetch_one(&mut *conn)
523                .await
524                .unwrap()
525        };
526        database
527            .set_sidebar_section_collapsed(SidebarSection::Scope, true)
528            .await
529            .unwrap();
530        {
531            let mut conn = database.acquire_writer().await.unwrap();
532            set_meta(&mut conn, SidebarSection::Views.meta_key(), "invalid")
533                .await
534                .unwrap();
535        }
536        assert_eq!(
537            database.collapsed_sidebar_sections().await.unwrap(),
538            BTreeSet::from([SidebarSection::Scope])
539        );
540        let mut conn = database.acquire_reader().await.unwrap();
541        let after: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM changes")
542            .fetch_one(&mut *conn)
543            .await
544            .unwrap();
545        assert_eq!(before, after);
546    }
547
548    #[tokio::test]
549    async fn triage_reviews_round_trip_and_drop_expired_marks() {
550        let database = Database::open(std::path::Path::new(":memory:"))
551            .await
552            .unwrap();
553        let now = 100 * 86_400;
554        database
555            .set_triage_review("OLD", Some(now - 31 * 86_400), now)
556            .await
557            .unwrap();
558        assert_eq!(
559            database
560                .set_triage_review("NEW", Some(now), now)
561                .await
562                .unwrap(),
563            None
564        );
565
566        let reviews = database.triage_reviews().await.unwrap();
567        assert_eq!(reviews.get("NEW"), Some(&now));
568        assert!(!reviews.contains_key("OLD"));
569
570        assert_eq!(
571            database.set_triage_review("NEW", None, now).await.unwrap(),
572            Some(now)
573        );
574        assert!(database.triage_reviews().await.unwrap().is_empty());
575    }
576}
aven_core/local_state.rs

aven_core/
local_state.rs
