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/// What a commit that wrote no revision did: a document deletion (observed
79/// or api) names the tombstoned paths, a move its `move <from> -> <to>`
80/// reason.
81#[derive(Clone, Debug, Default, PartialEq, Eq)]
82struct SummaryExtra {
83    deleted: Vec<String>,
84    moved: Option<(String, String)>,
85}
86
87/// The reference's `renderSummary`: `<origin>(<actor ?? ?>): <paths> — <n
88/// kind>, …` for `api`/`import`, `observed: <paths> — …` otherwise (`same`
89/// dispositions omitted). A commit with no revision says `deleted <paths>`
90/// or `moved <from> → <to>` instead (`spec/cli` §6 `log`).
91fn render_summary(
92    origin: &str,
93    actor: Option<&str>,
94    paths: &[&str],
95    dispositions: &[(String, i64)],
96    extra: &SummaryExtra,
97) -> String {
98    let mut paths = paths.join(", ");
99    if paths.is_empty() {
100        if !extra.deleted.is_empty() {
101            paths = format!("deleted {}", extra.deleted.join(", "));
102        } else if let Some((from, to)) = &extra.moved {
103            paths = format!("moved {from} → {to}");
104        }
105    }
106    let parts: Vec<String> = dispositions
107        .iter()
108        .filter(|(kind, _)| kind != "same")
109        .map(|(kind, n)| format!("{n} {kind}"))
110        .collect();
111    let detail = if parts.is_empty() {
112        String::new()
113    } else {
114        format!(" — {}", parts.join(", "))
115    };
116    if origin == "api" || origin == "import" {
117        format!("{origin}({}): {paths}{detail}", actor.unwrap_or("?"))
118    } else {
119        format!("observed: {paths}{detail}")
120    }
121}
122
123/// `move <from> -> <to>` (a `docs_move` commit's reason) → `(from, to)`.
124fn parse_move(reason: &str) -> Option<(String, String)> {
125    let rest = reason.strip_prefix("move ")?;
126    let (from, to) = rest.split_once(" -> ")?;
127    if from.is_empty() || to.is_empty() {
128        return None;
129    }
130    Some((from.to_owned(), to.to_owned()))
131}
132
133/// The repo's commits with `seq > cursor` (optionally of one `origin`) in
134/// `seq` order, `limit + 1` fetched to set `truncated`; each digest's
135/// revisions are the commit's `revisions` rows in row order.
136pub fn changes_since(
137    conn: &Connection,
138    repo_id: &str,
139    cursor: i64,
140    limit: usize,
141    origin: Option<&str>,
142) -> Result<ChangesPage> {
143    let head: i64 = conn.query_row(
144        "SELECT COALESCE(MAX(seq), 0) FROM commits WHERE repo_id = ?1",
145        params![repo_id],
146        |r| r.get(0),
147    )?;
148    let fetch = i64::try_from(limit).unwrap_or(i64::MAX).saturating_add(1);
149    type Row = (String, i64, String, String, Option<String>, Option<String>);
150    let mut commits: Vec<Row> = match origin {
151        Some(o) => {
152            let mut stmt = conn.prepare(
153                "SELECT commit_id, seq, ts, origin, actor, reason FROM commits
154                 WHERE repo_id = ?1 AND seq > ?2 AND origin = ?3 ORDER BY seq LIMIT ?4",
155            )?;
156            let rows = stmt.query_map(params![repo_id, cursor, o, fetch], |r| {
157                Ok((
158                    r.get(0)?,
159                    r.get(1)?,
160                    r.get(2)?,
161                    r.get(3)?,
162                    r.get(4)?,
163                    r.get(5)?,
164                ))
165            })?;
166            rows.collect::<std::result::Result<Vec<_>, _>>()?
167        }
168        None => {
169            let mut stmt = conn.prepare(
170                "SELECT commit_id, seq, ts, origin, actor, reason FROM commits
171                 WHERE repo_id = ?1 AND seq > ?2 ORDER BY seq LIMIT ?3",
172            )?;
173            let rows = stmt.query_map(params![repo_id, cursor, fetch], |r| {
174                Ok((
175                    r.get(0)?,
176                    r.get(1)?,
177                    r.get(2)?,
178                    r.get(3)?,
179                    r.get(4)?,
180                    r.get(5)?,
181                ))
182            })?;
183            rows.collect::<std::result::Result<Vec<_>, _>>()?
184        }
185    };
186    let truncated = commits.len() > limit;
187    commits.truncate(limit);
188
189    let mut revs_stmt = conn.prepare(
190        "SELECT doc_id, path, rendered_hash FROM revisions WHERE commit_id = ?1 ORDER BY rowid",
191    )?;
192    let mut disp_stmt =
193        conn.prepare("SELECT kind, count(*) FROM dispositions WHERE commit_id = ?1 GROUP BY kind")?;
194    // A deletion or a move writes no revision: the tombstone names the commit
195    // (`docs.deleted_commit`), a move only its reason (`move <from> -> <to>`).
196    let mut deleted_stmt =
197        conn.prepare("SELECT path FROM docs WHERE deleted_commit = ?1 ORDER BY path")?;
198    let mut digests = Vec::with_capacity(commits.len());
199    for (commit_id, seq, ts, origin, actor, reason) in commits {
200        let revisions: Vec<DigestRevision> = revs_stmt
201            .query_map(params![commit_id], |r| {
202                Ok(DigestRevision {
203                    doc: r.get(0)?,
204                    path: r.get(1)?,
205                    content_hash: hex(&r.get::<_, Vec<u8>>(2)?),
206                })
207            })?
208            .collect::<std::result::Result<_, _>>()?;
209        let dispositions: Vec<(String, i64)> = disp_stmt
210            .query_map(params![commit_id], |r| Ok((r.get(0)?, r.get(1)?)))?
211            .collect::<std::result::Result<_, _>>()?;
212        let paths: Vec<&str> = revisions.iter().map(|r| r.path.as_str()).collect();
213        let extra = if revisions.is_empty() {
214            SummaryExtra {
215                deleted: deleted_stmt
216                    .query_map(params![commit_id], |r| r.get::<_, String>(0))?
217                    .collect::<std::result::Result<_, _>>()?,
218                moved: reason.as_deref().and_then(parse_move),
219            }
220        } else {
221            SummaryExtra::default()
222        };
223        let summary = render_summary(&origin, actor.as_deref(), &paths, &dispositions, &extra);
224        digests.push(CommitDigest {
225            commit: commit_id,
226            seq,
227            ts,
228            origin,
229            actor,
230            summary,
231            revisions,
232        });
233    }
234    let next = digests.last().map_or(cursor, |d| d.seq);
235    Ok(ChangesPage {
236        digests,
237        cursor: next,
238        truncated,
239        head,
240    })
241}
242
243#[cfg(test)]
244mod tests {
245    use super::*;
246    use crate::{BatchItem, SequentialMinter, Store};
247    use omgbase_reconcile::Config;
248
249    fn store_with_commits(n: usize) -> (Store, String) {
250        let mut store =
251            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
252        let repo = store.create_repo("r").unwrap();
253        for i in 0..n {
254            let items = [BatchItem::observed("a.md", &format!("# T\n\nv{i}\n"))];
255            store
256                .observe_batch(
257                    &repo,
258                    &items,
259                    "2026-09-26T10:00:00.000Z",
260                    &Config::default(),
261                )
262                .unwrap();
263        }
264        (store, repo)
265    }
266
267    #[test]
268    fn pages_in_seq_order_with_truncation_and_head() {
269        let (store, repo) = store_with_commits(3);
270        let page = store.changes_since(&repo, 0, 2, None).unwrap();
271        assert_eq!(page.digests.len(), 2);
272        assert!(page.truncated);
273        assert_eq!(page.cursor, 2);
274        assert_eq!(page.head, 3);
275        assert_eq!(page.digests[0].seq, 1);
276        assert_eq!(page.digests[0].commit, "c_0");
277        assert_eq!(page.digests[0].origin, "observed");
278        assert_eq!(page.digests[0].revisions.len(), 1);
279        assert_eq!(page.digests[0].revisions[0].path, "a.md");
280        assert_eq!(page.digests[0].revisions[0].doc, "d_0");
281        assert_eq!(page.digests[0].revisions[0].content_hash.len(), 64);
282        assert!(page.digests[0].summary.starts_with("observed: a.md"));
283
284        let rest = store.changes_since(&repo, page.cursor, 2, None).unwrap();
285        assert_eq!(rest.digests.len(), 1);
286        assert!(!rest.truncated);
287        assert_eq!(rest.cursor, 3);
288
289        let empty = store.changes_since(&repo, 3, 50, None).unwrap();
290        assert!(empty.digests.is_empty());
291        assert_eq!(empty.cursor, 3, "the input cursor when the page is empty");
292        assert_eq!(empty.head, 3);
293    }
294
295    #[test]
296    fn filters_by_origin() {
297        let (store, repo) = store_with_commits(2);
298        let api = store.changes_since(&repo, 0, 50, Some("api")).unwrap();
299        assert!(api.digests.is_empty());
300        assert_eq!(api.cursor, 0);
301        assert_eq!(api.head, 2);
302        let observed = store.changes_since(&repo, 0, 50, Some("observed")).unwrap();
303        assert_eq!(observed.digests.len(), 2);
304    }
305
306    #[test]
307    fn summary_follows_the_reference_shape() {
308        assert_eq!(
309            render_summary(
310                "api",
311                Some("agent"),
312                &["a.md"],
313                &[("same".into(), 3), ("edited".into(), 1)],
314                &SummaryExtra::default(),
315            ),
316            "api(agent): a.md — 1 edited"
317        );
318        assert_eq!(
319            render_summary(
320                "import",
321                None,
322                &["a.md", "b.md"],
323                &[],
324                &SummaryExtra::default()
325            ),
326            "import(?): a.md, b.md"
327        );
328        assert_eq!(
329            render_summary(
330                "observed",
331                None,
332                &[],
333                &[("inserted".into(), 2)],
334                &SummaryExtra::default()
335            ),
336            "observed:  — 2 inserted"
337        );
338        let deleted = SummaryExtra {
339            deleted: vec!["a.md".into()],
340            moved: None,
341        };
342        assert_eq!(
343            render_summary("observed", None, &[], &[], &deleted),
344            "observed: deleted a.md"
345        );
346        let moved = SummaryExtra {
347            deleted: vec![],
348            moved: parse_move("move a.md -> b/a.md"),
349        };
350        assert_eq!(
351            render_summary("api", Some("human:spec"), &[], &[], &moved),
352            "api(human:spec): moved a.md → b/a.md"
353        );
354        assert_eq!(parse_move("nope"), None);
355        let (store, repo) = store_with_commits(1);
356        let page = store.changes_since(&repo, 0, 50, None).unwrap();
357        let json = page.to_json();
358        assert_eq!(
359            json["digests"][0]["revisions"][0]["contentHash"],
360            page.digests[0].revisions[0].content_hash
361        );
362        assert_eq!(json["head"], 1);
363    }
364}
omgbase_store/history.rs

omgbase_store/
history.rs
