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

omgbase_store/
history.rs
