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(|| {
159                workspaces.iter().find(|workspace| {
160                    workspace.id.as_str() == crate::workspaces::DEFAULT_WORKSPACE_ID
161                })
162            })
163            .cloned()
164            .ok_or_else(|| anyhow::anyhow!("no available workspace"))?;
165        set_meta(&mut tx, IOS_QUEUE_WORKSPACE_META_KEY, selected.id.as_str()).await?;
166        tx.commit().await?;
167        Ok(selected)
168    }
169
170    pub async fn select_queue_workspace(
171        &self,
172        workspace_id: &WorkspaceId,
173    ) -> Result<crate::workspaces::Workspace> {
174        let mut conn = self.acquire_writer().await?;
175        let mut tx = begin_immediate(&mut conn).await?;
176        let workspace = crate::workspaces::workspace_for_id(&mut tx, workspace_id).await?;
177        set_meta(&mut tx, IOS_QUEUE_WORKSPACE_META_KEY, workspace.id.as_str()).await?;
178        tx.commit().await?;
179        Ok(workspace)
180    }
181
182    pub async fn onboarding_status(&self) -> Result<OnboardingStatus> {
183        let mut conn = self.acquire_reader().await?;
184        let marker = get_meta(&mut conn, ONBOARDING_META_KEY).await?;
185        if marker_version(marker.as_deref())
186            .is_some_and(|version| version >= FIRST_LAUNCH_ONBOARDING_VERSION)
187        {
188            return Ok(OnboardingStatus::Complete);
189        }
190
191        let has_tasks: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM tasks)")
192            .fetch_one(&mut *conn)
193            .await?;
194        Ok(if has_tasks {
195            OnboardingStatus::Established
196        } else {
197            OnboardingStatus::Due
198        })
199    }
200
201    pub async fn complete_onboarding(&self) -> Result<()> {
202        let mut conn = self.acquire_writer().await?;
203        let mut tx = begin_immediate(&mut conn).await?;
204        let marker = get_meta(&mut tx, ONBOARDING_META_KEY).await?;
205        if marker_version(marker.as_deref())
206            .is_none_or(|version| version < FIRST_LAUNCH_ONBOARDING_VERSION)
207        {
208            set_meta(
209                &mut tx,
210                ONBOARDING_META_KEY,
211                &FIRST_LAUNCH_ONBOARDING_VERSION.to_string(),
212            )
213            .await?;
214        }
215        tx.commit().await?;
216        Ok(())
217    }
218}
219
220/// A reminder this device has handled, keyed by kind, task and trigger.
221/// Digest entries use an empty `task_id`.
222#[derive(Debug, Clone, PartialEq, Eq, sqlx::FromRow)]
223pub struct ReminderLogEntry {
224    pub kind: String,
225    pub task_id: String,
226    pub trigger: String,
227    pub outcome: String,
228}
229
230impl Database {
231    pub async fn reminder_log(&self) -> Result<Vec<ReminderLogEntry>> {
232        let mut conn = self.acquire_reader().await?;
233        Ok(sqlx::query_as::<_, ReminderLogEntry>(
234            "SELECT kind, task_id, trigger, outcome FROM local_reminder_log",
235        )
236        .fetch_all(&mut *conn)
237        .await?)
238    }
239
240    /// Records entries in one transaction, replacing the outcome of existing
241    /// keys.
242    pub async fn record_reminders(&self, entries: &[ReminderLogEntry]) -> Result<()> {
243        if entries.is_empty() {
244            return Ok(());
245        }
246        let recorded_at = crate::ids::now();
247        let mut conn = self.acquire_writer().await?;
248        let mut tx = begin_immediate(&mut conn).await?;
249        for entry in entries {
250            sqlx::query(
251                "INSERT INTO local_reminder_log(kind, task_id, trigger, outcome, recorded_at)
252                 VALUES (?, ?, ?, ?, ?)
253                 ON CONFLICT(kind, task_id, trigger)
254                 DO UPDATE SET outcome = excluded.outcome, recorded_at = excluded.recorded_at",
255            )
256            .bind(&entry.kind)
257            .bind(&entry.task_id)
258            .bind(&entry.trigger)
259            .bind(&entry.outcome)
260            .bind(&recorded_at)
261            .execute(&mut *tx)
262            .await?;
263        }
264        tx.commit().await?;
265        Ok(())
266    }
267}
268
269fn marker_version(value: Option<&str>) -> Option<u32> {
270    value?.trim().parse().ok()
271}
272
273/// Database-local sidebar preferences shared across workspaces and projects.
274#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
275pub enum SidebarSection {
276    Views,
277    Workspaces,
278    Projects,
279}
280
281impl SidebarSection {
282    pub fn label(self) -> &'static str {
283        match self {
284            Self::Views => "Views",
285            Self::Workspaces => "Workspaces",
286            Self::Projects => "Projects",
287        }
288    }
289
290    fn meta_key(self) -> &'static str {
291        match self {
292            Self::Views => "tui_sidebar_views_collapsed",
293            Self::Workspaces => "tui_sidebar_scope_collapsed",
294            Self::Projects => "tui_sidebar_projects_collapsed",
295        }
296    }
297}
298
299impl Database {
300    pub async fn collapsed_sidebar_sections(&self) -> Result<BTreeSet<SidebarSection>> {
301        let mut conn = self.acquire_reader().await?;
302        let mut sections = BTreeSet::new();
303        for section in [
304            SidebarSection::Views,
305            SidebarSection::Workspaces,
306            SidebarSection::Projects,
307        ] {
308            if get_meta(&mut conn, section.meta_key())
309                .await?
310                .as_deref()
311                .map(str::trim)
312                == Some("true")
313            {
314                sections.insert(section);
315            }
316        }
317        Ok(sections)
318    }
319
320    pub async fn set_sidebar_section_collapsed(
321        &self,
322        section: SidebarSection,
323        collapsed: bool,
324    ) -> Result<()> {
325        let mut conn = self.acquire_writer().await?;
326        let mut tx = begin_immediate(&mut conn).await?;
327        set_meta(
328            &mut tx,
329            section.meta_key(),
330            if collapsed { "true" } else { "false" },
331        )
332        .await?;
333        tx.commit().await?;
334        Ok(())
335    }
336}
337
338/// When each task was last kept as it stands in triage, in unix seconds. The
339/// marks are local to this database: they only quiet triage on this machine.
340pub type TriageReviews = BTreeMap<String, i64>;
341
342impl Database {
343    pub async fn triage_reviews(&self) -> Result<TriageReviews> {
344        let mut conn = self.acquire_reader().await?;
345        Ok(parse_triage_reviews(
346            get_meta(&mut conn, TRIAGE_REVIEWS_META_KEY)
347                .await?
348                .as_deref(),
349        ))
350    }
351
352    /// Sets or clears a task's review mark and returns the previous one. Marks
353    /// older than the longest stale window no longer affect triage and are
354    /// dropped.
355    pub async fn set_triage_review(
356        &self,
357        task_id: &str,
358        reviewed_at: Option<i64>,
359        now_seconds: i64,
360    ) -> Result<Option<i64>> {
361        let mut conn = self.acquire_writer().await?;
362        let mut tx = begin_immediate(&mut conn).await?;
363        let mut reviews =
364            parse_triage_reviews(get_meta(&mut tx, TRIAGE_REVIEWS_META_KEY).await?.as_deref());
365        let previous = match reviewed_at {
366            Some(at) => reviews.insert(task_id.to_string(), at),
367            None => reviews.remove(task_id),
368        };
369        let horizon = now_seconds - crate::queue::STALE_TODO_DAYS * 86_400;
370        reviews.retain(|_, at| *at > horizon);
371        set_meta(
372            &mut tx,
373            TRIAGE_REVIEWS_META_KEY,
374            &serde_json::to_string(&reviews)?,
375        )
376        .await?;
377        tx.commit().await?;
378        Ok(previous)
379    }
380}
381
382fn parse_triage_reviews(value: Option<&str>) -> TriageReviews {
383    value
384        .and_then(|value| serde_json::from_str(value).ok())
385        .unwrap_or_default()
386}
387
388#[cfg(test)]
389mod tests {
390    use super::*;
391
392    #[tokio::test]
393    async fn daemon_health_preserves_first_seen_and_clears_on_success_and_reset() {
394        let dir = tempfile::tempdir().unwrap();
395        let path = dir.path().join("health.sqlite");
396        let database = Database::open(&path).await.unwrap();
397        assert_eq!(
398            database.daemon_sync_state().await.unwrap(),
399            DaemonSyncState::default()
400        );
401        database.record_daemon_sync_round(100, None).await.unwrap();
402        database
403            .record_daemon_sync_round(110, Some(("network", "retry", 111)))
404            .await
405            .unwrap();
406        database
407            .record_daemon_sync_round(120, Some(("network", "retry", 122)))
408            .await
409            .unwrap();
410        let state = database.daemon_sync_state().await.unwrap();
411        assert_eq!(state.last_success_at, Some(100));
412        assert_eq!(state.error.as_ref().unwrap().since, 110);
413        assert_eq!(state.next_retry_at, Some(122));
414        sqlx::raw_sql(
415            "CREATE TRIGGER reject_health_write BEFORE UPDATE ON meta
416            WHEN NEW.key = 'sync_daemon_health'
417            BEGIN SELECT RAISE(ABORT, 'redundant health write'); END;",
418        )
419        .execute(&mut *database.acquire_writer().await.unwrap())
420        .await
421        .unwrap();
422        database
423            .record_daemon_sync_round(121, Some(("network", "retry", 122)))
424            .await
425            .unwrap();
426        sqlx::raw_sql("DROP TRIGGER reject_health_write")
427            .execute(&mut *database.acquire_writer().await.unwrap())
428            .await
429            .unwrap();
430        assert_eq!(
431            Database::open(&path)
432                .await
433                .unwrap()
434                .daemon_sync_state()
435                .await
436                .unwrap(),
437            state
438        );
439        database
440            .record_daemon_sync_round(130, Some(("keys", "unlock", 140)))
441            .await
442            .unwrap();
443        assert_eq!(
444            database
445                .daemon_sync_state()
446                .await
447                .unwrap()
448                .error
449                .unwrap()
450                .since,
451            130
452        );
453        database.record_daemon_sync_round(150, None).await.unwrap();
454        assert_eq!(
455            database.daemon_sync_state().await.unwrap(),
456            DaemonSyncState {
457                last_success_at: Some(150),
458                error: None,
459                next_retry_at: None,
460            }
461        );
462        database.clear_sync_state().await.unwrap();
463        assert_eq!(
464            database.daemon_sync_state().await.unwrap(),
465            DaemonSyncState::default()
466        );
467    }
468
469    #[tokio::test]
470    async fn deleted_queue_workspace_falls_back_to_the_renamed_default_workspace() {
471        let dir = tempfile::tempdir().unwrap();
472        let path = dir.path().join("queue.sqlite");
473        let database = Database::open(&path).await.unwrap();
474        database.create_workspace("Alpha").await.unwrap();
475        database.rename_workspace("default", "Home").await.unwrap();
476        let personal = database.create_workspace("Personal").await.unwrap();
477        database.select_queue_workspace(&personal.id).await.unwrap();
478        crate::test_support::set_established_protocol(&database, 19)
479            .await
480            .unwrap();
481        database
482            .delete_workspace(&personal.id, &[], true)
483            .await
484            .unwrap();
485        assert_eq!(
486            database.restore_queue_workspace().await.unwrap().key,
487            "home"
488        );
489        database.close().await;
490        let reopened = Database::open(&path).await.unwrap();
491        assert_eq!(
492            reopened.restore_queue_workspace().await.unwrap().key,
493            "home"
494        );
495    }
496
497    #[tokio::test]
498    async fn sync_access_refusal_persists_until_cleared() {
499        let database = Database::open(std::path::Path::new(":memory:"))
500            .await
501            .unwrap();
502        assert_eq!(database.sync_access_refusal().await.unwrap(), None);
503
504        let recorded = database.record_sync_access_refusal().await.unwrap();
505        assert!(!recorded.at.is_empty());
506        assert_eq!(
507            database.sync_access_refusal().await.unwrap(),
508            Some(recorded)
509        );
510
511        database.clear_sync_access_refusal().await.unwrap();
512        assert_eq!(database.sync_access_refusal().await.unwrap(), None);
513    }
514
515    #[tokio::test]
516    async fn reminder_outcome_replaces_existing_key() {
517        let database = Database::open(std::path::Path::new(":memory:"))
518            .await
519            .unwrap();
520        let entry = |outcome: &str| ReminderLogEntry {
521            kind: "available".into(),
522            task_id: "T1".into(),
523            trigger: "2026-10-14T09:00:00Z".into(),
524            outcome: outcome.into(),
525        };
526
527        database
528            .record_reminders(&[entry("waiting")])
529            .await
530            .unwrap();
531        database.record_reminders(&[entry("sent")]).await.unwrap();
532
533        assert_eq!(database.reminder_log().await.unwrap(), vec![entry("sent")]);
534    }
535
536    #[tokio::test]
537    async fn sync_vault_deletion_persists_apart_from_access_refusal_until_cleared() {
538        let database = Database::open(std::path::Path::new(":memory:"))
539            .await
540            .unwrap();
541        assert_eq!(database.sync_vault_deletion().await.unwrap(), None);
542
543        let recorded = database.record_sync_vault_deletion().await.unwrap();
544        assert_eq!(
545            database.sync_vault_deletion().await.unwrap(),
546            Some(recorded)
547        );
548        assert_eq!(database.sync_access_refusal().await.unwrap(), None);
549
550        database.clear_sync_vault_deletion().await.unwrap();
551        assert_eq!(database.sync_vault_deletion().await.unwrap(), None);
552    }
553
554    #[tokio::test]
555    async fn sidebar_preferences_are_local_and_invalid_values_default_to_expanded() {
556        let database = Database::open(std::path::Path::new(":memory:"))
557            .await
558            .unwrap();
559        let before: i64 = {
560            let mut conn = database.acquire_reader().await.unwrap();
561            sqlx::query_scalar("SELECT COUNT(*) FROM changes")
562                .fetch_one(&mut *conn)
563                .await
564                .unwrap()
565        };
566        {
567            let mut conn = database.acquire_writer().await.unwrap();
568            set_meta(&mut conn, "tui_sidebar_scope_collapsed", "true")
569                .await
570                .unwrap();
571        }
572        {
573            let mut conn = database.acquire_writer().await.unwrap();
574            set_meta(&mut conn, SidebarSection::Views.meta_key(), "invalid")
575                .await
576                .unwrap();
577        }
578        assert_eq!(
579            database.collapsed_sidebar_sections().await.unwrap(),
580            BTreeSet::from([SidebarSection::Workspaces])
581        );
582        let mut conn = database.acquire_reader().await.unwrap();
583        let after: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM changes")
584            .fetch_one(&mut *conn)
585            .await
586            .unwrap();
587        assert_eq!(before, after);
588    }
589
590    #[tokio::test]
591    async fn triage_reviews_round_trip_and_drop_expired_marks() {
592        let database = Database::open(std::path::Path::new(":memory:"))
593            .await
594            .unwrap();
595        let now = 100 * 86_400;
596        database
597            .set_triage_review("OLD", Some(now - 31 * 86_400), now)
598            .await
599            .unwrap();
600        assert_eq!(
601            database
602                .set_triage_review("NEW", Some(now), now)
603                .await
604                .unwrap(),
605            None
606        );
607
608        let reviews = database.triage_reviews().await.unwrap();
609        assert_eq!(reviews.get("NEW"), Some(&now));
610        assert!(!reviews.contains_key("OLD"));
611
612        assert_eq!(
613            database.set_triage_review("NEW", None, now).await.unwrap(),
614            Some(now)
615        );
616        assert!(database.triage_reviews().await.unwrap().is_empty());
617    }
618}
aven_core/local_state.rs

aven_core/
local_state.rs
