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

omgbase_store/
history.rs
