omgbase_sync/
source.rs

1//! The sync source contract (`spec/sync/README.md` §5): the in-engine view
2//! of an adapter — capabilities, `enumerate`/`fetch`, `write`/`remove` when
3//! it writes through, a `watch` stream when it can watch. Every call crosses
4//! a pipe in production ([`crate::external::ExternalSource`]); an in-memory
5//! source serves tests.
6
7use std::collections::BTreeMap;
8use std::sync::mpsc::Receiver;
9
10use serde_json::Value;
11
12use crate::error::{Error, Result};
13
14/// How a source's identity relates to omgbase block identity.
15#[derive(Clone, Copy, Debug, PartialEq, Eq)]
16pub enum SourceIdentity {
17    /// Anonymous bytes; the reconciler infers continuity.
18    Inferred,
19    /// Stable per-member ids upstream (no adapter exists; §9).
20    Borne,
21}
22
23impl SourceIdentity {
24    #[must_use]
25    pub const fn as_str(&self) -> &'static str {
26        match self {
27            SourceIdentity::Inferred => "inferred",
28            SourceIdentity::Borne => "borne",
29        }
30    }
31}
32
33/// The handshake's capabilities.
34#[derive(Clone, Copy, Debug, PartialEq, Eq)]
35pub struct SourceCapabilities {
36    pub identity: SourceIdentity,
37    pub write_through: bool,
38    pub watch: bool,
39}
40
41impl Default for SourceCapabilities {
42    fn default() -> Self {
43        Self {
44            identity: SourceIdentity::Inferred,
45            write_through: false,
46            watch: false,
47        }
48    }
49}
50
51/// JavaScript truthiness (the reference's `Boolean(c.writeThrough)`).
52fn js_truthy(v: Option<&Value>) -> bool {
53    match v {
54        None | Some(Value::Null) => false,
55        Some(Value::Bool(b)) => *b,
56        Some(Value::Number(n)) => n.as_f64().is_some_and(|f| f != 0.0 && !f.is_nan()),
57        Some(Value::String(s)) => !s.is_empty(),
58        Some(Value::Array(_) | Value::Object(_)) => true,
59    }
60}
61
62impl SourceCapabilities {
63    /// From the handshake's `capabilities` object: `identity` is `borne`
64    /// only when it says so; the booleans by JavaScript truthiness.
65    #[must_use]
66    pub fn from_json(v: Option<&Value>) -> Self {
67        let obj = v.and_then(Value::as_object);
68        Self {
69            identity: match obj.and_then(|o| o.get("identity")).and_then(Value::as_str) {
70                Some("borne") => SourceIdentity::Borne,
71                _ => SourceIdentity::Inferred,
72            },
73            write_through: js_truthy(obj.and_then(|o| o.get("writeThrough"))),
74            watch: js_truthy(obj.and_then(|o| o.get("watch"))),
75        }
76    }
77
78    #[must_use]
79    pub fn to_json(&self) -> Value {
80        serde_json::json!({
81            "identity": self.identity.as_str(),
82            "writeThrough": self.write_through,
83            "watch": self.watch,
84        })
85    }
86}
87
88/// A member of the scope: its storage key and cheap change token.
89#[derive(Clone, Debug, PartialEq, Eq)]
90pub struct SourceEntry {
91    /// The repo-relative canonical path (`docs.path`).
92    pub path: String,
93    /// The source's change token (the fs adapter: `"<mtime_ns>:<size>"`).
94    pub revision: String,
95    /// The source locator when it differs from `path`.
96    pub source_id: Option<String>,
97}
98
99impl SourceEntry {
100    /// From a protocol `entries[]` element; `None` when `path` is missing.
101    #[must_use]
102    pub fn from_json(v: &Value) -> Option<Self> {
103        Some(Self {
104            path: v.get("path")?.as_str()?.to_owned(),
105            revision: match v.get("revision") {
106                Some(Value::String(s)) => s.clone(),
107                Some(other) if !other.is_null() => other.to_string(),
108                _ => String::new(),
109            },
110            source_id: v.get("sourceId").and_then(Value::as_str).map(str::to_owned),
111        })
112    }
113}
114
115impl SourceEntry {
116    /// `{ path, revision, sourceId? }` as the protocol carries it.
117    #[must_use]
118    pub fn to_json(&self) -> Value {
119        let mut v = serde_json::json!({ "path": self.path, "revision": self.revision });
120        if let Some(id) = &self.source_id {
121            v["sourceId"] = Value::String(id.clone());
122        }
123        v
124    }
125}
126
127/// A hydrated member: the entry plus its bytes.
128#[derive(Clone, Debug, PartialEq, Eq)]
129pub struct SourceItem {
130    pub entry: SourceEntry,
131    /// The engine hashes this itself (§5).
132    pub content: String,
133}
134
135impl SourceItem {
136    /// From a protocol `item`; `None` for `null`/missing or no `path`.
137    #[must_use]
138    pub fn from_json(v: Option<&Value>) -> Option<Self> {
139        let v = v?;
140        if v.is_null() {
141            return None;
142        }
143        let entry = SourceEntry::from_json(v)?;
144        let content = match v.get("content") {
145            Some(Value::String(s)) => s.clone(),
146            _ => String::new(),
147        };
148        Some(Self { entry, content })
149    }
150
151    /// `{ path, revision, sourceId?, content }`.
152    #[must_use]
153    pub fn to_json(&self) -> Value {
154        let mut v = self.entry.to_json();
155        v["content"] = Value::String(self.content.clone());
156        v
157    }
158}
159
160/// The in-engine handle to a source.
161pub trait SyncSource {
162    fn capabilities(&self) -> SourceCapabilities;
163
164    /// The full current scope.
165    fn enumerate(&mut self) -> Result<Vec<SourceEntry>>;
166
167    /// One member's current state, or `None` when it left the scope.
168    fn fetch(&mut self, path: &str) -> Result<Option<SourceItem>>;
169
170    /// Persist engine-authored bytes (write-through sources only).
171    fn write(&mut self, _path: &str, _content: &str) -> Result<()> {
172        Err(Error::Unsupported("write".to_owned()))
173    }
174
175    /// Remove a member (write-through sources only).
176    fn remove(&mut self, _path: &str) -> Result<()> {
177        Err(Error::Unsupported("remove".to_owned()))
178    }
179
180    /// Subscribe to change batches: each received value is one batch of
181    /// changed paths (watching sources only).
182    fn watch(&mut self) -> Result<Receiver<Vec<String>>> {
183        Err(Error::Unsupported("watch".to_owned()))
184    }
185
186    /// Stop the batches.
187    fn unwatch(&mut self) -> Result<()> {
188        Ok(())
189    }
190
191    /// Release the process / resources.
192    fn close(&mut self) -> Result<()> {
193        Ok(())
194    }
195}
196
197/// An in-memory source for tests: a `path → content` map with a configurable
198/// capability set; writes and removes are applied to the map and logged.
199#[derive(Debug, Default)]
200pub struct MemSource {
201    pub caps: SourceCapabilities,
202    pub files: BTreeMap<String, String>,
203    /// `("write", path, content)` / `("remove", path, "")` in call order.
204    pub log: Vec<(&'static str, String, String)>,
205    /// Revision counter per path (bumped on every write).
206    revisions: BTreeMap<String, u64>,
207    batches: Option<std::sync::mpsc::Sender<Vec<String>>>,
208}
209
210impl MemSource {
211    #[must_use]
212    pub fn new(caps: SourceCapabilities) -> Self {
213        Self {
214            caps,
215            ..Self::default()
216        }
217    }
218
219    /// A source with `path → content` files and every capability.
220    #[must_use]
221    pub fn with_files(files: &[(&str, &str)]) -> Self {
222        let mut s = Self::new(SourceCapabilities {
223            identity: SourceIdentity::Inferred,
224            write_through: true,
225            watch: true,
226        });
227        for (p, c) in files {
228            s.files.insert((*p).to_owned(), (*c).to_owned());
229        }
230        s
231    }
232
233    /// Set a file from "outside" (as if the human edited it).
234    pub fn set(&mut self, path: &str, content: &str) {
235        self.files.insert(path.to_owned(), content.to_owned());
236        *self.revisions.entry(path.to_owned()).or_insert(0) += 1;
237    }
238
239    /// Emit a watch batch (a no-op when nothing watches).
240    pub fn emit(&self, paths: &[&str]) {
241        if let Some(tx) = &self.batches {
242            let _ = tx.send(paths.iter().map(|p| (*p).to_owned()).collect());
243        }
244    }
245
246    fn entry(&self, path: &str) -> SourceEntry {
247        SourceEntry {
248            path: path.to_owned(),
249            revision: self.revisions.get(path).copied().unwrap_or(0).to_string(),
250            source_id: None,
251        }
252    }
253}
254
255impl SyncSource for MemSource {
256    fn capabilities(&self) -> SourceCapabilities {
257        self.caps
258    }
259
260    fn enumerate(&mut self) -> Result<Vec<SourceEntry>> {
261        Ok(self.files.keys().map(|p| self.entry(p)).collect())
262    }
263
264    fn fetch(&mut self, path: &str) -> Result<Option<SourceItem>> {
265        Ok(self.files.get(path).map(|content| SourceItem {
266            entry: self.entry(path),
267            content: content.clone(),
268        }))
269    }
270
271    fn write(&mut self, path: &str, content: &str) -> Result<()> {
272        if !self.caps.write_through {
273            return Err(Error::Unsupported("write".to_owned()));
274        }
275        self.set(path, content);
276        self.log
277            .push(("write", path.to_owned(), content.to_owned()));
278        Ok(())
279    }
280
281    fn remove(&mut self, path: &str) -> Result<()> {
282        if !self.caps.write_through {
283            return Err(Error::Unsupported("remove".to_owned()));
284        }
285        self.files.remove(path);
286        self.log.push(("remove", path.to_owned(), String::new()));
287        Ok(())
288    }
289
290    fn watch(&mut self) -> Result<Receiver<Vec<String>>> {
291        if !self.caps.watch {
292            return Err(Error::Unsupported("watch".to_owned()));
293        }
294        let (tx, rx) = std::sync::mpsc::channel();
295        self.batches = Some(tx);
296        Ok(rx)
297    }
298
299    fn unwatch(&mut self) -> Result<()> {
300        self.batches = None;
301        Ok(())
302    }
303}
304
305#[cfg(test)]
306mod tests {
307    use super::*;
308    use serde_json::json;
309
310    #[test]
311    fn capabilities_parse_with_js_truthiness() {
312        let c = SourceCapabilities::from_json(Some(
313            &json!({"identity": "borne", "writeThrough": "yes", "watch": 0}),
314        ));
315        assert_eq!(c.identity, SourceIdentity::Borne);
316        assert!(c.write_through);
317        assert!(!c.watch);
318        let c = SourceCapabilities::from_json(Some(&json!({"identity": "weird", "watch": true})));
319        assert_eq!(c.identity, SourceIdentity::Inferred);
320        assert!(c.watch && !c.write_through);
321        assert_eq!(
322            SourceCapabilities::from_json(None),
323            SourceCapabilities::default()
324        );
325        assert_eq!(
326            SourceCapabilities::from_json(Some(&json!(null))),
327            SourceCapabilities::default()
328        );
329        assert_eq!(
330            c.to_json(),
331            json!({"identity": "inferred", "writeThrough": false, "watch": true})
332        );
333    }
334
335    #[test]
336    fn entries_and_items_parse() {
337        let e =
338            SourceEntry::from_json(&json!({"path": "a.md", "revision": "1:2", "sourceId": "x"}))
339                .unwrap();
340        assert_eq!(
341            (e.path.as_str(), e.revision.as_str(), e.source_id.as_deref()),
342            ("a.md", "1:2", Some("x"))
343        );
344        assert_eq!(
345            SourceEntry::from_json(&json!({"path": "a.md", "revision": 7}))
346                .unwrap()
347                .revision,
348            "7"
349        );
350        assert_eq!(SourceEntry::from_json(&json!({"revision": "1"})), None);
351        let it = SourceItem::from_json(Some(
352            &json!({"path": "a.md", "revision": "r", "content": "# A\n"}),
353        ))
354        .unwrap();
355        assert_eq!(it.content, "# A\n");
356        assert_eq!(
357            it.to_json(),
358            json!({"path": "a.md", "revision": "r", "content": "# A\n"})
359        );
360        assert_eq!(
361            e.to_json(),
362            json!({"path": "a.md", "revision": "1:2", "sourceId": "x"})
363        );
364        assert_eq!(SourceItem::from_json(Some(&json!(null))), None);
365        assert_eq!(SourceItem::from_json(None), None);
366    }
367
368    #[test]
369    fn mem_source_behaves() {
370        let mut s = MemSource::with_files(&[("a.md", "A")]);
371        assert_eq!(s.enumerate().unwrap()[0].path, "a.md");
372        assert_eq!(s.fetch("a.md").unwrap().unwrap().content, "A");
373        assert!(s.fetch("b.md").unwrap().is_none());
374        s.write("b.md", "B").unwrap();
375        s.remove("a.md").unwrap();
376        assert_eq!(s.log.len(), 2);
377        let rx = s.watch().unwrap();
378        s.emit(&["b.md"]);
379        assert_eq!(rx.recv().unwrap(), ["b.md"]);
380        s.unwatch().unwrap();
381        s.close().unwrap();
382        let mut ro = MemSource::new(SourceCapabilities::default());
383        assert!(matches!(ro.write("x", "y"), Err(Error::Unsupported(_))));
384        assert!(matches!(ro.remove("x"), Err(Error::Unsupported(_))));
385        assert!(matches!(ro.watch(), Err(Error::Unsupported(_))));
386    }
387}
omgbase_sync/source.rs

omgbase_sync/
source.rs
