omgbase_sync/
engine.rs

1//! The engine client (`spec/sync/README.md` §6): the coordinator's view of
2//! "the omgbase side" — `observe_many`, `observe_delete`, `changes_since`,
3//! `read_doc` — behind one trait so the same coordinator runs against an
4//! in-process store or a remote server. The in-process client is here; a
5//! remote client belongs to the binary that speaks MCP.
6
7use omgbase_format::hash::hex;
8use omgbase_reconcile::Config;
9use omgbase_store::{BatchItem, BatchOutcome, ChangesPage, DeleteOutcome, ObserveOutcome, Store};
10use rusqlite::{OptionalExtension, params};
11
12use crate::error::{Error, Result};
13
14/// Bytes to observe at a path.
15#[derive(Clone, Debug, PartialEq, Eq)]
16pub struct FileBytes {
17    pub path: String,
18    pub content: String,
19}
20
21impl FileBytes {
22    #[must_use]
23    pub fn new(path: &str, content: &str) -> Self {
24        Self {
25            path: path.to_owned(),
26            content: content.to_owned(),
27        }
28    }
29}
30
31/// A document's current bytes and hash.
32#[derive(Clone, Debug, PartialEq, Eq)]
33pub struct DocBytes {
34    pub content: String,
35    /// Hex of the doc's `file_hash` (`""` when null).
36    pub content_hash: String,
37}
38
39/// The default `changes_since` page size.
40pub const DEFAULT_LIMIT: usize = 50;
41
42/// Everything the reconcile loop needs from the engine.
43pub trait EngineClient {
44    /// Observe whole-file bytes as observed commits (echo-suppressed
45    /// engine-side), one batch, then sweep the pool.
46    fn observe_many(&mut self, files: &[FileBytes]) -> Result<Vec<ObserveOutcome>>;
47    /// Mirror a source-side deletion (tombstone, then sweep).
48    fn observe_delete(&mut self, path: &str) -> Result<DeleteOutcome>;
49    /// The repo's change feed after `cursor` (§6).
50    fn changes_since(
51        &mut self,
52        cursor: i64,
53        limit: Option<usize>,
54        origin: Option<&str>,
55    ) -> Result<ChangesPage>;
56    /// The current bytes at `path`, or `None` when no live doc reads there.
57    fn read_doc(&mut self, path: &str) -> Result<Option<DocBytes>>;
58    /// Release the connection (the store's lifetime is the caller's).
59    fn close(&mut self) -> Result<()> {
60        Ok(())
61    }
62}
63
64/// The in-process client over an open store: each call is timestamped by
65/// `clock` (the wall clock by default; a fixture pins it).
66pub struct InProcessEngineClient<'a> {
67    store: &'a mut Store,
68    repo_id: String,
69    config: Config,
70    clock: Box<dyn FnMut() -> String + 'a>,
71}
72
73impl std::fmt::Debug for InProcessEngineClient<'_> {
74    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
75        f.debug_struct("InProcessEngineClient")
76            .field("repo_id", &self.repo_id)
77            .finish_non_exhaustive()
78    }
79}
80
81impl<'a> InProcessEngineClient<'a> {
82    #[must_use]
83    pub fn new(store: &'a mut Store, repo_id: &str) -> Self {
84        Self {
85            store,
86            repo_id: repo_id.to_owned(),
87            config: Config::default(),
88            clock: Box::new(crate::now_ts),
89        }
90    }
91
92    /// The matcher thresholds every observe uses.
93    #[must_use]
94    pub fn with_config(mut self, config: Config) -> Self {
95        self.config = config;
96        self
97    }
98
99    /// Replace the wall clock.
100    #[must_use]
101    pub fn with_clock(mut self, clock: impl FnMut() -> String + 'a) -> Self {
102        self.clock = Box::new(clock);
103        self
104    }
105
106    /// Every call at one fixed timestamp.
107    #[must_use]
108    pub fn at(self, ts: &str) -> Self {
109        let ts = ts.to_owned();
110        self.with_clock(move || ts.clone())
111    }
112
113    #[must_use]
114    pub fn store(&self) -> &Store {
115        self.store
116    }
117
118    #[must_use]
119    pub fn repo_id(&self) -> &str {
120        &self.repo_id
121    }
122}
123
124impl EngineClient for InProcessEngineClient<'_> {
125    fn observe_many(&mut self, files: &[FileBytes]) -> Result<Vec<ObserveOutcome>> {
126        let ts = (self.clock)();
127        let items: Vec<BatchItem> = files
128            .iter()
129            .map(|f| BatchItem::observed(&f.path, &f.content))
130            .collect();
131        let outcomes = self
132            .store
133            .observe_batch(&self.repo_id, &items, &ts, &self.config)?;
134        let mut out = Vec::with_capacity(outcomes.len());
135        for o in outcomes {
136            match o {
137                BatchOutcome::Observed(ob) => out.push(ob),
138                BatchOutcome::Deleted(d) => {
139                    return Err(Error::Other(format!(
140                        "observe_many: unexpected outcome for {}",
141                        d.path
142                    )));
143                }
144            }
145        }
146        self.store.sweep_pool(&ts)?;
147        Ok(out)
148    }
149
150    fn observe_delete(&mut self, path: &str) -> Result<DeleteOutcome> {
151        let ts = (self.clock)();
152        Ok(self.store.observe_delete(&self.repo_id, path, &ts)?)
153    }
154
155    fn changes_since(
156        &mut self,
157        cursor: i64,
158        limit: Option<usize>,
159        origin: Option<&str>,
160    ) -> Result<ChangesPage> {
161        Ok(self.store.changes_since(
162            &self.repo_id,
163            cursor,
164            limit.unwrap_or(DEFAULT_LIMIT),
165            origin,
166        )?)
167    }
168
169    fn read_doc(&mut self, path: &str) -> Result<Option<DocBytes>> {
170        let row: Option<(String, Option<Vec<u8>>)> = self
171            .store
172            .conn()
173            .query_row(
174                "SELECT doc_id, file_hash FROM docs WHERE repo_id = ?1 AND path = ?2 AND deleted_commit IS NULL",
175                params![self.repo_id, path],
176                |r| Ok((r.get(0)?, r.get(1)?)),
177            )
178            .optional()?;
179        let Some((doc_id, file_hash)) = row else {
180            return Ok(None);
181        };
182        let Some(content) = self.store.reconstruct(&doc_id)? else {
183            return Ok(None);
184        };
185        Ok(Some(DocBytes {
186            content,
187            content_hash: file_hash.as_deref().map(hex).unwrap_or_default(),
188        }))
189    }
190}
191
192#[cfg(test)]
193mod tests {
194    use super::*;
195    use omgbase_store::SequentialMinter;
196
197    const TS: &str = "2026-09-26T10:00:00.000Z";
198
199    #[test]
200    fn in_process_client_round_trip() {
201        let mut store =
202            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
203        let repo = store.create_repo("fixture").unwrap();
204        let mut client = InProcessEngineClient::new(&mut store, &repo).at(TS);
205        assert_eq!(client.repo_id(), "rp_0");
206        let out = client
207            .observe_many(&[
208                FileBytes::new("a.md", "# A\n"),
209                FileBytes::new("b.md", "# B\n"),
210            ])
211            .unwrap();
212        assert_eq!(out.len(), 2);
213        assert!(!out[0].echo && out[0].converged);
214        let again = client
215            .observe_many(&[FileBytes::new("a.md", "# A\n")])
216            .unwrap();
217        assert!(again[0].echo);
218        let doc = client.read_doc("a.md").unwrap().unwrap();
219        assert_eq!(doc.content, "# A\n");
220        assert_eq!(
221            doc.content_hash,
222            hex(&omgbase_format::hash::sha256(b"# A\n"))
223        );
224        assert!(client.read_doc("zzz.md").unwrap().is_none());
225        let d = client.observe_delete("b.md").unwrap();
226        assert!(d.deleted());
227        assert!(!client.observe_delete("b.md").unwrap().deleted());
228        assert!(client.read_doc("b.md").unwrap().is_none());
229        let page = client.changes_since(0, None, None).unwrap();
230        assert_eq!(page.digests.len(), 3);
231        assert_eq!(page.head, 3);
232        assert_eq!(page.digests[2].origin, "observed");
233        assert!(
234            page.digests[2].revisions.is_empty(),
235            "a tombstone writes no revision"
236        );
237        let one = client.changes_since(0, Some(1), None).unwrap();
238        assert!(one.truncated);
239        assert_eq!(client.store().user_version().unwrap(), 13);
240        client.close().unwrap();
241        drop(client);
242        let commits: Vec<String> = {
243            let mut stmt = store
244                .conn()
245                .prepare("SELECT ts FROM commits ORDER BY seq")
246                .unwrap();
247            stmt.query_map([], |r| r.get(0))
248                .unwrap()
249                .map(|r| r.unwrap())
250                .collect()
251        };
252        assert!(
253            commits.iter().all(|t| t == TS),
254            "the pinned clock stamps every commit"
255        );
256    }
257}
omgbase_sync/engine.rs

omgbase_sync/
engine.rs
