omgbase_store/
history.rs

1//! The change feed (`spec/sync/README.md` §6, the reference's
2//! `graph/history.ts` `changesSince`): the repo's commits after a cursor as
3//! digests, each with the revisions it wrote. `seq` is a dense per-repo
4//! total order, so a cursor is only meaningful against the repo it came from;
5//! `head` (the repo's current max `seq`) tells "no new changes" from "cursor
6//! beyond this repo's feed".
7
8use omgbase_format::hash::hex;
9use rusqlite::{Connection, params};
10
11use crate::error::Result;
12
13/// One revision a commit wrote.
14#[derive(Clone, Debug, PartialEq, Eq)]
15pub struct DigestRevision {
16    pub doc: String,
17    pub path: String,
18    /// Hex of the revision's `rendered_hash`.
19    pub content_hash: String,
20}
21
22/// One commit of the feed.
23#[derive(Clone, Debug, PartialEq, Eq)]
24pub struct CommitDigest {
25    pub commit: String,
26    pub seq: i64,
27    pub ts: String,
28    pub origin: String,
29    pub actor: Option<String>,
30    /// A human line (the fixtures do not pin it).
31    pub summary: String,
32    pub revisions: Vec<DigestRevision>,
33}
34
35/// A page of the feed.
36#[derive(Clone, Debug, PartialEq, Eq)]
37pub struct ChangesPage {
38    pub digests: Vec<CommitDigest>,
39    /// The last digest's `seq`, or the input cursor when the page is empty.
40    pub cursor: i64,
41    /// More commits follow.
42    pub truncated: bool,
43    /// The repo's max `seq` (0 when it has no commits).
44    pub head: i64,
45}
46
47impl ChangesPage {
48    /// The page as the MCP surface renders it (`camelCase` as the reference).
49    #[must_use]
50    pub fn to_json(&self) -> serde_json::Value {
51        serde_json::json!({
52            "digests": self.digests.iter().map(CommitDigest::to_json).collect::<Vec<_>>(),
53            "cursor": self.cursor,
54            "truncated": self.truncated,
55            "head": self.head,
56        })
57    }
58}
59
60impl CommitDigest {
61    /// The digest as the MCP surface renders it.
62    #[must_use]
63    pub fn to_json(&self) -> serde_json::Value {
64        serde_json::json!({
65            "commit": self.commit,
66            "seq": self.seq,
67            "ts": self.ts,
68            "origin": self.origin,
69            "actor": self.actor,
70            "summary": self.summary,
71            "revisions": self.revisions.iter().map(|r| serde_json::json!({
72                "doc": r.doc, "path": r.path, "contentHash": r.content_hash,
73            })).collect::<Vec<_>>(),
74        })
75    }
76}
77
78/// The reference's `renderSummary`: `<origin>(<actor ?? ?>): <paths> — <n
79/// kind>, …` for `api`/`import`, `observed: <paths> — …` otherwise (`same`
80/// dispositions omitted).
81fn render_summary(
82    origin: &str,
83    actor: Option<&str>,
84    paths: &[&str],
85    dispositions: &[(String, i64)],
86) -> String {
87    let paths = paths.join(", ");
88    let parts: Vec<String> = dispositions
89        .iter()
90        .filter(|(kind, _)| kind != "same")
91        .map(|(kind, n)| format!("{n} {kind}"))
92        .collect();
93    let detail = if parts.is_empty() {
94        String::new()
95    } else {
96        format!(" — {}", parts.join(", "))
97    };
98    if origin == "api" || origin == "import" {
99        format!("{origin}({}): {paths}{detail}", actor.unwrap_or("?"))
100    } else {
101        format!("observed: {paths}{detail}")
102    }
103}
104
105/// The repo's commits with `seq > cursor` (optionally of one `origin`) in
106/// `seq` order, `limit + 1` fetched to set `truncated`; each digest's
107/// revisions are the commit's `revisions` rows in row order.
108pub fn changes_since(
109    conn: &Connection,
110    repo_id: &str,
111    cursor: i64,
112    limit: usize,
113    origin: Option<&str>,
114) -> Result<ChangesPage> {
115    let head: i64 = conn.query_row(
116        "SELECT COALESCE(MAX(seq), 0) FROM commits WHERE repo_id = ?1",
117        params![repo_id],
118        |r| r.get(0),
119    )?;
120    let fetch = i64::try_from(limit).unwrap_or(i64::MAX).saturating_add(1);
121    type Row = (String, i64, String, String, Option<String>);
122    let mut commits: Vec<Row> = match origin {
123        Some(o) => {
124            let mut stmt = conn.prepare(
125                "SELECT commit_id, seq, ts, origin, actor FROM commits
126                 WHERE repo_id = ?1 AND seq > ?2 AND origin = ?3 ORDER BY seq LIMIT ?4",
127            )?;
128            let rows = stmt.query_map(params![repo_id, cursor, o, fetch], |r| {
129                Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?))
130            })?;
131            rows.collect::<std::result::Result<Vec<_>, _>>()?
132        }
133        None => {
134            let mut stmt = conn.prepare(
135                "SELECT commit_id, seq, ts, origin, actor FROM commits
136                 WHERE repo_id = ?1 AND seq > ?2 ORDER BY seq LIMIT ?3",
137            )?;
138            let rows = stmt.query_map(params![repo_id, cursor, fetch], |r| {
139                Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?))
140            })?;
141            rows.collect::<std::result::Result<Vec<_>, _>>()?
142        }
143    };
144    let truncated = commits.len() > limit;
145    commits.truncate(limit);
146
147    let mut revs_stmt = conn.prepare(
148        "SELECT doc_id, path, rendered_hash FROM revisions WHERE commit_id = ?1 ORDER BY rowid",
149    )?;
150    let mut disp_stmt =
151        conn.prepare("SELECT kind, count(*) FROM dispositions WHERE commit_id = ?1 GROUP BY kind")?;
152    let mut digests = Vec::with_capacity(commits.len());
153    for (commit_id, seq, ts, origin, actor) in commits {
154        let revisions: Vec<DigestRevision> = revs_stmt
155            .query_map(params![commit_id], |r| {
156                Ok(DigestRevision {
157                    doc: r.get(0)?,
158                    path: r.get(1)?,
159                    content_hash: hex(&r.get::<_, Vec<u8>>(2)?),
160                })
161            })?
162            .collect::<std::result::Result<_, _>>()?;
163        let dispositions: Vec<(String, i64)> = disp_stmt
164            .query_map(params![commit_id], |r| Ok((r.get(0)?, r.get(1)?)))?
165            .collect::<std::result::Result<_, _>>()?;
166        let paths: Vec<&str> = revisions.iter().map(|r| r.path.as_str()).collect();
167        let summary = render_summary(&origin, actor.as_deref(), &paths, &dispositions);
168        digests.push(CommitDigest {
169            commit: commit_id,
170            seq,
171            ts,
172            origin,
173            actor,
174            summary,
175            revisions,
176        });
177    }
178    let next = digests.last().map_or(cursor, |d| d.seq);
179    Ok(ChangesPage {
180        digests,
181        cursor: next,
182        truncated,
183        head,
184    })
185}
186
187#[cfg(test)]
188mod tests {
189    use super::*;
190    use crate::{BatchItem, SequentialMinter, Store};
191    use omgbase_reconcile::Config;
192
193    fn store_with_commits(n: usize) -> (Store, String) {
194        let mut store =
195            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
196        let repo = store.create_repo("r").unwrap();
197        for i in 0..n {
198            let items = [BatchItem::observed("a.md", &format!("# T\n\nv{i}\n"))];
199            store
200                .observe_batch(
201                    &repo,
202                    &items,
203                    "2026-09-26T10:00:00.000Z",
204                    &Config::default(),
205                )
206                .unwrap();
207        }
208        (store, repo)
209    }
210
211    #[test]
212    fn pages_in_seq_order_with_truncation_and_head() {
213        let (store, repo) = store_with_commits(3);
214        let page = store.changes_since(&repo, 0, 2, None).unwrap();
215        assert_eq!(page.digests.len(), 2);
216        assert!(page.truncated);
217        assert_eq!(page.cursor, 2);
218        assert_eq!(page.head, 3);
219        assert_eq!(page.digests[0].seq, 1);
220        assert_eq!(page.digests[0].commit, "c_0");
221        assert_eq!(page.digests[0].origin, "observed");
222        assert_eq!(page.digests[0].revisions.len(), 1);
223        assert_eq!(page.digests[0].revisions[0].path, "a.md");
224        assert_eq!(page.digests[0].revisions[0].doc, "d_0");
225        assert_eq!(page.digests[0].revisions[0].content_hash.len(), 64);
226        assert!(page.digests[0].summary.starts_with("observed: a.md"));
227
228        let rest = store.changes_since(&repo, page.cursor, 2, None).unwrap();
229        assert_eq!(rest.digests.len(), 1);
230        assert!(!rest.truncated);
231        assert_eq!(rest.cursor, 3);
232
233        let empty = store.changes_since(&repo, 3, 50, None).unwrap();
234        assert!(empty.digests.is_empty());
235        assert_eq!(empty.cursor, 3, "the input cursor when the page is empty");
236        assert_eq!(empty.head, 3);
237    }
238
239    #[test]
240    fn filters_by_origin() {
241        let (store, repo) = store_with_commits(2);
242        let api = store.changes_since(&repo, 0, 50, Some("api")).unwrap();
243        assert!(api.digests.is_empty());
244        assert_eq!(api.cursor, 0);
245        assert_eq!(api.head, 2);
246        let observed = store.changes_since(&repo, 0, 50, Some("observed")).unwrap();
247        assert_eq!(observed.digests.len(), 2);
248    }
249
250    #[test]
251    fn summary_follows_the_reference_shape() {
252        assert_eq!(
253            render_summary(
254                "api",
255                Some("agent"),
256                &["a.md"],
257                &[("same".into(), 3), ("edited".into(), 1)]
258            ),
259            "api(agent): a.md — 1 edited"
260        );
261        assert_eq!(
262            render_summary("import", None, &["a.md", "b.md"], &[]),
263            "import(?): a.md, b.md"
264        );
265        assert_eq!(
266            render_summary("observed", None, &[], &[("inserted".into(), 2)]),
267            "observed:  — 2 inserted"
268        );
269        let (store, repo) = store_with_commits(1);
270        let page = store.changes_since(&repo, 0, 50, None).unwrap();
271        let json = page.to_json();
272        assert_eq!(
273            json["digests"][0]["revisions"][0]["contentHash"],
274            page.digests[0].revisions[0].content_hash
275        );
276        assert_eq!(json["head"], 1);
277    }
278}
omgbase_store/history.rs

omgbase_store/
history.rs
