omgbase_sync/
driver.rs

1//! The driver (`spec/sync/README.md` §6): the engine-side loop over a
2//! [`SyncSource`] — `reconcile_changes` (fetch, `observe_batch`, checkpoint)
3//! and `attach_source` (`ensure_repo`, then ingest every enumerated member).
4
5use omgbase_reconcile::Config;
6use omgbase_store::{BatchItem, Origin, Store};
7use serde_json::Value;
8
9use crate::checkpoint::{CheckpointResult, finish_checkpoint};
10use crate::error::Result;
11use crate::registry::ensure_repo;
12use crate::source::{SourceIdentity, SyncSource};
13
14/// Fetch every path through the source (`None` → gone), `observe_batch`,
15/// record the checkpoint exactly as §4.1.
16pub fn reconcile_changes(
17    store: &mut Store,
18    repo_id: &str,
19    source: &mut dyn SyncSource,
20    paths: &[String],
21    ts: &str,
22    git_head: Option<&str>,
23    config: &Config,
24) -> Result<CheckpointResult> {
25    let mut items = Vec::with_capacity(paths.len());
26    for path in paths {
27        items.push(BatchItem {
28            path: path.clone(),
29            source: source.fetch(path)?.map(|it| it.content),
30        });
31    }
32    let outcomes = store.observe_batch(repo_id, &items, ts, config)?;
33    finish_checkpoint(store, repo_id, &outcomes, ts, git_head)
34}
35
36/// What `attach_source` did.
37#[derive(Clone, Debug, PartialEq, Eq)]
38pub struct AttachResult {
39    pub repo_id: String,
40    pub file_count: usize,
41    pub all_converged: bool,
42}
43
44impl AttachResult {
45    #[must_use]
46    pub fn to_json(&self) -> Value {
47        serde_json::json!({
48            "repo_id": self.repo_id,
49            "file_count": self.file_count,
50            "all_converged": self.all_converged,
51        })
52    }
53}
54
55/// `ensure_repo`, then for every enumerated entry fetch and ingest — with
56/// the reconciling resolver when `identity` is `inferred`, the plain re-mint
57/// ingest otherwise (§9) — counting files and whether all converged. A
58/// member whose fetch returns `None` is skipped.
59pub fn attach_source(
60    store: &mut Store,
61    slug: &str,
62    root_path: Option<&str>,
63    source: &mut dyn SyncSource,
64    ts: &str,
65    config: &Config,
66) -> Result<AttachResult> {
67    let repo_id = ensure_repo(store, slug, root_path)?;
68    let inferred = source.capabilities().identity == SourceIdentity::Inferred;
69    let mut file_count = 0;
70    let mut all_converged = true;
71    for entry in source.enumerate()? {
72        let Some(item) = source.fetch(&entry.path)? else {
73            continue;
74        };
75        let committed = if inferred {
76            store.reconciling_ingest(
77                &repo_id,
78                &entry.path,
79                &item.content,
80                ts,
81                Origin::Observed,
82                None,
83                None,
84                config,
85            )?
86        } else {
87            store.fresh_ingest(&repo_id, &entry.path, &item.content, ts, Origin::Observed)?
88        };
89        file_count += 1;
90        if !committed.converged {
91            all_converged = false;
92        }
93    }
94    Ok(AttachResult {
95        repo_id,
96        file_count,
97        all_converged,
98    })
99}
100
101#[cfg(test)]
102mod tests {
103    use super::*;
104    use crate::source::{MemSource, SourceCapabilities};
105    use omgbase_store::SequentialMinter;
106    use rusqlite::params;
107
108    const TS: &str = "2026-09-26T10:00:00.000Z";
109
110    #[test]
111    fn attach_then_reconcile_changes() {
112        let mut store =
113            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
114        let cfg = Config::default();
115        let mut source = MemSource::with_files(&[("a.md", "# A\n\nOne.\n"), ("b.md", "# B\n")]);
116        let r = attach_source(
117            &mut store,
118            "vault",
119            Some("/data/vault"),
120            &mut source,
121            TS,
122            &cfg,
123        )
124        .unwrap();
125        assert_eq!(
126            r,
127            AttachResult {
128                repo_id: "rp_0".into(),
129                file_count: 2,
130                all_converged: true
131            }
132        );
133        assert_eq!(r.to_json()["file_count"], 2);
134        let fs_source: String = store
135            .conn()
136            .query_row("SELECT name FROM sources WHERE adapter = 'fs'", [], |r| {
137                r.get(0)
138            })
139            .unwrap();
140        assert_eq!(fs_source, "vault-fs");
141        let dispositions: i64 = store
142            .conn()
143            .query_row("SELECT count(*) FROM dispositions", [], |r| r.get(0))
144            .unwrap();
145        assert_eq!(
146            dispositions, 3,
147            "inferred identity: the reconciling resolver records dispositions"
148        );
149
150        // Attaching again re-ingests (no echo gate on attach) and reuses the repo.
151        let again = attach_source(&mut store, "vault", None, &mut source, TS, &cfg).unwrap();
152        assert_eq!(again.repo_id, "rp_0");
153        let commits: i64 = store
154            .conn()
155            .query_row("SELECT count(*) FROM commits", [], |r| r.get(0))
156            .unwrap();
157        assert_eq!(commits, 4);
158
159        source.set("a.md", "# A\n\nOne, edited.\n");
160        source.files.remove("b.md");
161        let paths: Vec<String> = ["a.md", "b.md", "nope.md"]
162            .iter()
163            .map(|s| (*s).to_owned())
164            .collect();
165        let cp = reconcile_changes(
166            &mut store,
167            "rp_0",
168            &mut source,
169            &paths,
170            TS,
171            Some("deadbeef"),
172            &cfg,
173        )
174        .unwrap();
175        assert_eq!(cp.ingested, ["a.md"]);
176        assert_eq!(cp.deleted, ["b.md"]);
177        assert_eq!(cp.checkpoint_id, "cp_0");
178        let head: Option<String> = store
179            .conn()
180            .query_row("SELECT git_head FROM checkpoints", [], |r| r.get(0))
181            .unwrap();
182        assert_eq!(head.as_deref(), Some("deadbeef"));
183        let cp = reconcile_changes(
184            &mut store,
185            "rp_0",
186            &mut source,
187            &["a.md".to_owned()],
188            TS,
189            None,
190            &cfg,
191        )
192        .unwrap();
193        assert_eq!(cp.suppressed, ["a.md"]);
194    }
195
196    #[test]
197    fn borne_sources_re_mint() {
198        let mut store =
199            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
200        let mut source = MemSource::new(SourceCapabilities {
201            identity: SourceIdentity::Borne,
202            write_through: false,
203            watch: false,
204        });
205        source.set("a.md", "# A\n\nOne.\n");
206        let r = attach_source(&mut store, "b", None, &mut source, TS, &Config::default()).unwrap();
207        assert_eq!(r.file_count, 1);
208        assert!(r.all_converged);
209        let dispositions: i64 = store
210            .conn()
211            .query_row("SELECT count(*) FROM dispositions", [], |r| r.get(0))
212            .unwrap();
213        assert_eq!(dispositions, 0, "the re-mint path records no dispositions");
214        let blocks: i64 = store
215            .conn()
216            .query_row(
217                "SELECT count(*) FROM blocks WHERE repo_id = ?1",
218                params![r.repo_id],
219                |r| r.get(0),
220            )
221            .unwrap();
222        assert_eq!(blocks, 2);
223        assert!(
224            store
225                .conn()
226                .query_row("SELECT 1 FROM sources", [], |_| Ok(()))
227                .is_err(),
228            "no root: no fs source"
229        );
230    }
231}
omgbase_sync/driver.rs

omgbase_sync/
driver.rs
