omgbase_sync/
admin.rs

1//! Status (`spec/sync/README.md` §4.4): `repos_status` and `sync_status`.
2//! Disk agreement is only ever reported from a read-only scan the caller
3//! asked for; without one `checked` is false and nothing is ever green.
4
5use std::path::Path;
6
7use omgbase_store::Store;
8use rusqlite::{OptionalExtension, params};
9use serde_json::Value;
10
11use crate::error::Result;
12use crate::freshness::{DiskDrift, detect_disk_drift};
13use crate::fs::FileSystem;
14
15/// The drift counts plus whether a scan ran.
16#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
17pub struct DiskStatus {
18    pub drift: DiskDrift,
19    /// `true` iff a root was supplied and the read-only scan ran.
20    pub checked: bool,
21}
22
23impl DiskStatus {
24    #[must_use]
25    pub fn to_json(&self) -> Value {
26        serde_json::json!({
27            "changed": self.drift.changed,
28            "deleted": self.drift.deleted,
29            "untracked": self.drift.untracked,
30            "checked": self.checked,
31        })
32    }
33}
34
35/// `repos_status`.
36#[derive(Clone, Debug, PartialEq, Eq)]
37pub struct RepoStatus {
38    pub repo_id: String,
39    pub slug: String,
40    /// §9: the caller's argument or `""`, never the derived root.
41    pub root_path: String,
42    pub docs: i64,
43    pub blocks: i64,
44    pub commits: i64,
45    pub open_edges: i64,
46    /// Live docs whose `file_hash` differs from their current revision's
47    /// `rendered_hash`.
48    pub unconverged: i64,
49    pub disk: DiskStatus,
50}
51
52impl RepoStatus {
53    /// The MCP shape (`camelCase`).
54    #[must_use]
55    pub fn to_json(&self) -> Value {
56        serde_json::json!({
57            "repoId": self.repo_id,
58            "slug": self.slug,
59            "rootPath": self.root_path,
60            "docs": self.docs,
61            "blocks": self.blocks,
62            "commits": self.commits,
63            "openEdges": self.open_edges,
64            "unconverged": self.unconverged,
65            "disk": self.disk.to_json(),
66        })
67    }
68}
69
70/// §4.4 `repos_status`: counts, `unconverged`, and the drift with `checked:
71/// true` when a root (and filesystem) was supplied.
72pub fn repos_status(
73    store: &Store,
74    repo_id: &str,
75    disk: Option<(&dyn FileSystem, &Path)>,
76) -> Result<RepoStatus> {
77    let conn = store.conn();
78    let slug: Option<String> = conn
79        .query_row(
80            "SELECT slug FROM repos WHERE repo_id = ?1",
81            params![repo_id],
82            |r| r.get(0),
83        )
84        .optional()?;
85    let count =
86        |sql: &str| -> Result<i64> { Ok(conn.query_row(sql, params![repo_id], |r| r.get(0))?) };
87    let unconverged = count(
88        "SELECT count(*) FROM docs d JOIN revisions r ON r.rev_id = d.current_rev
89         WHERE d.repo_id = ?1 AND d.deleted_commit IS NULL AND d.file_hash IS NOT r.rendered_hash",
90    )?;
91    let (root_path, disk) = match disk {
92        Some((fs, root)) => (
93            root.to_string_lossy().into_owned(),
94            DiskStatus {
95                drift: detect_disk_drift(store, repo_id, fs, root)?,
96                checked: true,
97            },
98        ),
99        None => (String::new(), DiskStatus::default()),
100    };
101    Ok(RepoStatus {
102        repo_id: repo_id.to_owned(),
103        slug: slug.unwrap_or_default(),
104        root_path,
105        docs: count("SELECT count(*) FROM docs WHERE repo_id = ?1 AND deleted_commit IS NULL")?,
106        blocks: count("SELECT count(*) FROM blocks WHERE repo_id = ?1 AND deleted_commit IS NULL")?,
107        commits: count("SELECT count(*) FROM commits WHERE repo_id = ?1")?,
108        open_edges: count("SELECT count(*) FROM edges WHERE repo_id = ?1 AND to_commit IS NULL")?,
109        unconverged,
110        disk,
111    })
112}
113
114/// `sync_status`.
115#[derive(Clone, Debug, PartialEq, Eq)]
116pub struct SyncStatus {
117    pub last_commit_seq: i64,
118    /// The latest checkpoint by `ts`.
119    pub last_checkpoint: Option<String>,
120    /// `unconverged == 0 && checked && no drift`.
121    pub convergent: bool,
122    pub disk: DiskStatus,
123}
124
125impl SyncStatus {
126    #[must_use]
127    pub fn to_json(&self) -> Value {
128        serde_json::json!({
129            "lastCommitSeq": self.last_commit_seq,
130            "lastCheckpoint": self.last_checkpoint,
131            "convergent": self.convergent,
132            "diskChecked": self.disk.checked,
133            "disk": self.disk.to_json(),
134        })
135    }
136}
137
138/// §4.4 `sync_status`.
139pub fn sync_status(
140    store: &Store,
141    repo_id: &str,
142    disk: Option<(&dyn FileSystem, &Path)>,
143) -> Result<SyncStatus> {
144    let conn = store.conn();
145    let last_commit_seq: i64 = conn.query_row(
146        "SELECT COALESCE(MAX(seq), 0) FROM commits WHERE repo_id = ?1",
147        params![repo_id],
148        |r| r.get(0),
149    )?;
150    let last_checkpoint: Option<String> = conn
151        .query_row(
152            "SELECT id FROM checkpoints WHERE repo_id = ?1 ORDER BY ts DESC LIMIT 1",
153            params![repo_id],
154            |r| r.get(0),
155        )
156        .optional()?;
157    let status = repos_status(store, repo_id, disk)?;
158    Ok(SyncStatus {
159        last_commit_seq,
160        last_checkpoint,
161        convergent: status.unconverged == 0 && status.disk.checked && status.disk.drift.is_clean(),
162        disk: status.disk,
163    })
164}
165
166#[cfg(test)]
167mod tests {
168    use super::*;
169    use crate::checkpoint::process_checkpoint;
170    use crate::fs::MemFileSystem;
171    use omgbase_reconcile::Config;
172    use omgbase_store::SequentialMinter;
173
174    const TS: &str = "2026-09-26T10:00:00.000Z";
175
176    #[test]
177    fn status_counts_and_never_green_unverified() {
178        let mut store =
179            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
180        let repo = store.create_repo("fixture").unwrap();
181        let s = sync_status(&store, &repo, None).unwrap();
182        assert_eq!(s.last_commit_seq, 0);
183        assert_eq!(s.last_checkpoint, None);
184        assert!(!s.convergent, "an unverified disk is never convergent");
185        let rs = repos_status(&store, &repo, None).unwrap();
186        assert_eq!(
187            (
188                rs.docs,
189                rs.blocks,
190                rs.commits,
191                rs.open_edges,
192                rs.unconverged
193            ),
194            (0, 0, 0, 0, 0)
195        );
196        assert_eq!(rs.slug, "fixture");
197        assert_eq!(rs.root_path, "");
198        assert!(!rs.disk.checked);
199
200        let mut fs = MemFileSystem::new();
201        fs.set("a.md", "# A\n\nSee [b](b.md).\n", 1);
202        let root = Path::new("/r");
203        process_checkpoint(
204            &mut store,
205            &repo,
206            &fs,
207            root,
208            &["a.md".to_owned()],
209            TS,
210            None,
211            &Config::default(),
212        )
213        .unwrap();
214        let rs = repos_status(&store, &repo, Some((&fs, root))).unwrap();
215        assert_eq!(
216            (
217                rs.docs,
218                rs.blocks,
219                rs.commits,
220                rs.open_edges,
221                rs.unconverged
222            ),
223            (1, 2, 1, 1, 0)
224        );
225        assert_eq!(rs.root_path, "/r");
226        assert!(rs.disk.checked);
227        // The checkpoint did not record file_stats, so a.md is a candidate
228        // but its bytes match the doc: no drift.
229        assert!(rs.disk.drift.is_clean());
230        let s = sync_status(&store, &repo, Some((&fs, root))).unwrap();
231        assert_eq!(s.last_commit_seq, 1);
232        assert_eq!(s.last_checkpoint.as_deref(), Some("cp_0"));
233        assert!(s.convergent);
234        assert_eq!(s.to_json()["diskChecked"], true);
235        assert_eq!(rs.to_json()["openEdges"], 1);
236
237        fs.set("a.md", "# A\n\nEdited.\n", 2);
238        let s = sync_status(&store, &repo, Some((&fs, root))).unwrap();
239        assert!(!s.convergent);
240        assert_eq!(s.disk.drift.changed, 1);
241
242        // unconverged: a doc whose file_hash no longer matches its revision.
243        store
244            .conn()
245            .execute("UPDATE docs SET file_hash = x'00'", [])
246            .unwrap();
247        assert_eq!(repos_status(&store, &repo, None).unwrap().unconverged, 1);
248        assert_eq!(repos_status(&store, "rp_nope", None).unwrap().slug, "");
249    }
250}
omgbase_sync/admin.rs

omgbase_sync/
admin.rs
