Skip to main content

tauri_plugin_syncular/
core.rs

1//! The transport-agnostic, Tauri-free CORE of the plugin.
2//!
3//! Everything here is plain Rust with no dependency on the `tauri` crate, so
4//! it is unit-testable without a window or a mock runtime. The Tauri shell
5//! (see `lib.rs`) is a thin layer that owns one [`SyncularCore`] on a
6//! dedicated thread, forwards mutable commands and raw queries to it, and pumps
7//! drained events onto the `syncular://event` Tauri channel. The shell's
8//! file-backed atomic snapshot sidecar is intentionally outside this mutable
9//! core and uses a separate read-only SQLite connection.
10//!
11//! This mirrors the `syncular-ffi` `Handle`: one owned [`SyncClient`], one
12//! owned [`HostTransport`] (native HTTP+WS via the `native-transport` feature),
13//! and an exact core-output event queue. The plugin is the THIRD consumer of the shared
14//! `syncular-command` router (after the conformance shim and the FFI core), so
15//! the command surface stays conformance-locked.
16//!
17//! ## Thread-safety, honestly
18//!
19//! [`SyncClient`] is synchronous and NOT `Sync` — it owns a rusqlite
20//! connection. Exactly one thread owns the mutable core, and all mutable access
21//! arrives through a command mailbox (an mpsc channel). Tauri commands never
22//! touch that client directly. The background host loop (§8.4) runs on the same
23//! owner, so the mutable connection is never accessed concurrently. The shell
24//! may independently read the file database through SQLite's snapshot model;
25//! it does not access this `SyncClient`.
26
27use std::collections::VecDeque;
28
29use serde_json::{json, Value};
30use syncular_client::{
31    ClientDiagnosticsRequest, ClientDiagnosticsSnapshot, SyncClient, SyncIntent,
32};
33use syncular_command::{dispatch, CreateEffects};
34
35use crate::transport::{self, HostTransport};
36
37/// One client-observable event (§8 realtime signals + §6 conflicts + §1.6
38/// schema floor + §7.3 lease). JSON-able; delivered onto the Tauri channel.
39/// The same event vocabulary the FFI `poll_event` surfaces.
40#[derive(Debug, Clone)]
41pub struct Event {
42    pub json: Value,
43}
44
45/// The Tauri-free core: one client, its owned transport, explicit scheduling
46/// state, and the pending exact-event queue. Lives on ONE owning thread.
47pub struct SyncularCore {
48    client: Option<SyncClient>,
49    transport: HostTransport,
50    effects: CreateEffects,
51    queue: VecDeque<Event>,
52    last_diagnostics_snapshot: Option<ClientDiagnosticsSnapshot>,
53    /// Diagnostics snapshots are computed only after a consumer registers —
54    /// the analogue of the web client's `ClientDiagnosticsEmitter.observed`.
55    /// Set by the explicit `enableDiagnostics` command and by the first
56    /// `diagnosticsSnapshot` pull (the devtools attach signal).
57    diagnostics_observed: bool,
58    interactive_sync: bool,
59    background_sync_ms: Option<u64>,
60}
61
62impl SyncularCore {
63    /// Build a core from the plugin config JSON (`baseUrl`, `headers`, …). A
64    /// `baseUrl` under the `native-transport` feature owns a real HTTP+WS
65    /// transport; without it the core is client-local only (tests, offline).
66    pub fn new(config: &Value) -> Result<Self, String> {
67        Self::new_with_notify(config, None)
68    }
69
70    pub fn new_with_notify(
71        config: &Value,
72        notify: Option<std::sync::Arc<dyn Fn() + Send + Sync>>,
73    ) -> Result<Self, String> {
74        let transport = HostTransport::from_config_with_notify(config, notify)?;
75        Ok(SyncularCore {
76            client: None,
77            transport,
78            effects: CreateEffects::default(),
79            queue: VecDeque::new(),
80            last_diagnostics_snapshot: None,
81            diagnostics_observed: false,
82            interactive_sync: false,
83            background_sync_ms: None,
84        })
85    }
86
87    /// Run one JSON command (`{"method","params"}`) through the shared router,
88    /// then drain inbound realtime traffic and exact core events. Returns the
89    /// driver-protocol `{"result"|"error"}` reply.
90    pub fn command(&mut self, command: &Value) -> Value {
91        let method = command.get("method").and_then(Value::as_str).unwrap_or("");
92        let params = command.get("params").cloned().unwrap_or(Value::Null);
93        if method == "enableDiagnostics" {
94            // Host-local registration signal (the Tauri event channel carries
95            // no listener count). Emission starts on this drain: the reset
96            // fingerprint guarantees the observer receives a first snapshot.
97            self.diagnostics_observed = true;
98            self.last_diagnostics_snapshot = None;
99            self.drain_realtime();
100            self.drain_core_outputs();
101            self.emit_diagnostics_if_changed();
102            return json!({ "result": {} });
103        }
104        if method == "diagnosticsSnapshot" {
105            // A direct pull proves a diagnostics consumer exists; keep the
106            // pushed snapshots flowing for it from here on.
107            self.diagnostics_observed = true;
108        }
109        let result = dispatch(
110            &mut self.transport,
111            &mut self.client,
112            &mut self.effects,
113            method,
114            &params,
115        );
116        if method == "create" {
117            self.last_diagnostics_snapshot = None;
118            self.transport.set_signed_urls(self.effects.signed_urls);
119        }
120        if method == "beginSecurityPreflight"
121            || method == "shutdown"
122            || (method == "create"
123                && params
124                    .get("securityPreflight")
125                    .and_then(Value::as_bool)
126                    .unwrap_or(false))
127        {
128            self.interactive_sync = false;
129            self.background_sync_ms = None;
130        }
131        if let Ok(value) = &result {
132            if value.pointer("/effects/sync/kind").and_then(Value::as_str) == Some("interactive") {
133                self.interactive_sync = true;
134            }
135        }
136        self.drain_realtime();
137        self.drain_core_outputs();
138        self.emit_diagnostics_if_changed();
139        match result {
140            Ok(mut value) => {
141                // The router's effects are consumed by this native host above;
142                // they are not part of the public command acknowledgement.
143                if let Some(object) = value.as_object_mut() {
144                    object.remove("effects");
145                }
146                json!({ "result": value })
147            }
148            Err((code, message)) => json!({ "error": { "code": code, "message": message } }),
149        }
150    }
151
152    /// The `syncular_query` fast path: arbitrary read-only SQL over the local
153    /// database. Routed through the same `query` command so there is one
154    /// implementation (the router owns it); this wrapper spares the JS bridge
155    /// from wrapping the method/params envelope for the hot live-query path.
156    pub fn query(&mut self, sql: &str, params: Value) -> Value {
157        let bind = match params {
158            Value::Null => Value::Array(Vec::new()),
159            other => other,
160        };
161        self.command(&json!({ "method": "query", "params": { "sql": sql, "params": bind } }))
162    }
163
164    /// Consume the next coalesced host schedule. Interactive work preempts a
165    /// pending retry; background work keeps the earliest real deadline.
166    pub fn take_sync_intent(&mut self) -> SyncIntent {
167        if std::mem::take(&mut self.interactive_sync) {
168            self.background_sync_ms = None;
169            SyncIntent::Interactive
170        } else if let Some(delay_ms) = self.background_sync_ms.take() {
171            SyncIntent::Background { delay_ms }
172        } else {
173            SyncIntent::None
174        }
175    }
176
177    /// Run one `syncUntilIdle` round for the background host loop, deriving
178    /// events afterwards. A no-op (empty reply) before `create`.
179    pub fn sync_until_idle(&mut self) -> Value {
180        if self.client.is_none() {
181            return json!({ "result": null });
182        }
183        self.command(&json!({ "method": "syncUntilIdle", "params": {} }))
184    }
185
186    /// Owner-mailbox wake from the native realtime reader.
187    pub fn poll_transport(&mut self) {
188        self.drain_realtime();
189        self.drain_core_outputs();
190        self.emit_diagnostics_if_changed();
191    }
192
193    /// Drain every event queued since the last call (the host thread pushes
194    /// them onto the Tauri channel). Mirrors the FFI `poll_event`, batched.
195    pub fn drain_events(&mut self) -> Vec<Event> {
196        self.queue.drain(..).collect()
197    }
198
199    /// Replace the transport's request headers. Rotating
200    /// auth without tearing the plugin down). See
201    /// `HostTransport::set_headers` for the HTTP/WS pickup semantics.
202    pub fn set_headers(&mut self, headers: Vec<(String, String)>) {
203        self.transport.set_headers(headers);
204    }
205
206    /// Release the socket/reader thread. Idempotent.
207    pub fn shutdown(&mut self) {
208        if let Some(mut client) = self.client.take() {
209            client.disconnect_realtime(&mut self.transport);
210            // Teardown barrier, not a quarantine: persisting the gate here
211            // would refuse every later plain create against this replica.
212            client.seal_security_on_teardown();
213        }
214        self.interactive_sync = false;
215        self.background_sync_ms = None;
216        // The TS driver clears its diagnostics listeners on dispose; require
217        // a fresh registration after any restart.
218        self.diagnostics_observed = false;
219        self.transport.shutdown();
220    }
221
222    fn push(&mut self, json: Value) {
223        self.queue.push_back(Event { json });
224    }
225
226    /// Feed buffered inbound WS frames to the client (which may ack back through
227    /// the same transport). A no-op without a native socket.
228    fn drain_realtime(&mut self) {
229        if self.client.is_none() {
230            return;
231        }
232        let frames = self.transport.take_inbound();
233        for frame in frames {
234            match frame {
235                transport::Inbound::Text(text) => {
236                    if is_presence_control(&text) {
237                        self.push(json!({ "type": "presence" }));
238                    }
239                    if let Some(client) = self.client.as_mut() {
240                        client.on_realtime_text(&text);
241                    }
242                }
243                transport::Inbound::Binary(bytes) => {
244                    if let Some(client) = self.client.as_mut() {
245                        client.on_realtime_binary(&mut self.transport, &bytes);
246                    }
247                }
248            }
249        }
250    }
251
252    /// Drain exact observer batches and sync intents produced by the Rust core.
253    fn drain_core_outputs(&mut self) {
254        let Some(client) = self.client.as_mut() else {
255            return;
256        };
257        let batches = client.drain_change_batches();
258        let intents = client.drain_sync_intents();
259        for batch in batches {
260            self.push(json!({ "type": "change", "batch": batch }));
261        }
262        for intent in intents {
263            match intent {
264                SyncIntent::Interactive => self.interactive_sync = true,
265                SyncIntent::Background { delay_ms } => {
266                    self.background_sync_ms = Some(
267                        self.background_sync_ms
268                            .map_or(delay_ms, |current| current.min(delay_ms)),
269                    );
270                }
271                SyncIntent::None => {}
272            }
273        }
274    }
275
276    /// Emit one privacy-safe snapshot only when a diagnostics consumer has
277    /// registered (`diagnostics_observed`) AND durable/client status changed.
278    /// The observer gate keeps the per-command snapshot work (subscription
279    /// scan, PRAGMA page counts, SUM aggregates, serialization) off every
280    /// unobserved command and transport poll.
281    /// `capturedAtMs` is excluded from the fingerprint so polling and read-only
282    /// commands do not create event noise. Expected-but-unregistered intent is
283    /// request-local and is obtained through `diagnosticsSnapshot` directly.
284    fn emit_diagnostics_if_changed(&mut self) {
285        if !self.diagnostics_observed {
286            return;
287        }
288        let Some(client) = self.client.as_ref() else {
289            return;
290        };
291        if client.security_preflight() {
292            return;
293        }
294        let Ok(snapshot) = client.diagnostics_snapshot(&ClientDiagnosticsRequest::default()) else {
295            return;
296        };
297        if let Some(previous) = &mut self.last_diagnostics_snapshot {
298            // Capture time is observation metadata, excluded from evidence equality.
299            previous.captured_at_ms = snapshot.captured_at_ms;
300            if previous == &snapshot {
301                return;
302            }
303        }
304        let event = json!({ "type": "diagnostics", "snapshot": snapshot });
305        self.last_diagnostics_snapshot = Some(snapshot);
306        self.push(event);
307    }
308}
309
310/// A presence fanout control frame (§8.6.2) — `{"event":"presence",...}`, the
311/// one inbound realtime event a native host surfaces directly.
312fn is_presence_control(text: &str) -> bool {
313    serde_json::from_str::<Value>(text)
314        .ok()
315        .and_then(|v| {
316            v.get("event")
317                .and_then(Value::as_str)
318                .map(|e| e == "presence")
319        })
320        .unwrap_or(false)
321}
322
323#[cfg(test)]
324mod tests {
325    use super::*;
326
327    fn simple_schema() -> Value {
328        json!({
329            "version": 1,
330            "tables": [{
331                "name": "todo",
332                "primaryKey": "id",
333                "columns": [
334                    { "name": "id", "type": "string", "nullable": false },
335                    { "name": "title", "type": "string", "nullable": false },
336                    { "name": "done", "type": "boolean", "nullable": false }
337                ],
338                "scopes": []
339            }]
340        })
341    }
342
343    fn create(core: &mut SyncularCore) {
344        let reply = core.command(&json!({
345            "method": "create",
346            "params": { "clientId": "c1", "schema": simple_schema() }
347        }));
348        assert_eq!(reply["result"], json!({}), "create ok: {reply}");
349    }
350
351    #[test]
352    fn command_round_trip_create_mutate_query() {
353        let mut core = SyncularCore::new(&json!({})).unwrap();
354        create(&mut core);
355
356        let sub = core.command(&json!({
357            "method": "subscribe",
358            "params": { "id": "s1", "table": "todo", "scopes": {} }
359        }));
360        assert_eq!(sub["result"], json!({}));
361
362        let mutate = core.command(&json!({
363            "method": "mutate",
364            "params": { "mutations": [{
365                "op": "upsert", "table": "todo",
366                "values": { "id": "t1", "title": "hello", "done": false }
367            }] }
368        }));
369        assert!(mutate["result"]["clientCommitId"].is_string(), "{mutate}");
370
371        // The query fast path sees the optimistic overlay immediately.
372        let rows = core.query("SELECT id, title FROM todo ORDER BY id", Value::Null);
373        let list = rows["result"]["rows"].as_array().expect("rows");
374        assert_eq!(list.len(), 1);
375        assert_eq!(list[0]["title"], "hello");
376        assert_eq!(list[0]["id"], "t1");
377    }
378
379    #[test]
380    fn query_binds_params() {
381        let mut core = SyncularCore::new(&json!({})).unwrap();
382        create(&mut core);
383        core.command(&json!({
384            "method": "mutate",
385            "params": { "mutations": [
386                { "op": "upsert", "table": "todo", "values": { "id": "a", "title": "A", "done": false } },
387                { "op": "upsert", "table": "todo", "values": { "id": "b", "title": "B", "done": true } }
388            ] }
389        }));
390        let rows = core.query("SELECT id FROM todo WHERE done = ?", json!([true]));
391        let list = rows["result"]["rows"].as_array().expect("rows");
392        assert_eq!(list.len(), 1);
393        assert_eq!(list[0]["id"], "b");
394    }
395
396    #[test]
397    fn events_derived_after_mutate() {
398        let mut core = SyncularCore::new(&json!({})).unwrap();
399        // Before create, no events.
400        create(&mut core);
401        // Register a diagnostics consumer so mutate-driven snapshots flow.
402        let enabled = core.command(&json!({ "method": "enableDiagnostics", "params": {} }));
403        assert_eq!(enabled["result"], json!({}));
404        let _ = core.drain_events();
405        core.command(&json!({
406            "method": "mutate",
407            "params": { "mutations": [{
408                "op": "upsert", "table": "todo",
409                "values": { "id": "t1", "title": "x", "done": false }
410            }] }
411        }));
412        let events = core.drain_events();
413        // A local mutate writes the optimistic overlay and revision in one
414        // transaction, then emits the exact committed change batch.
415        let kinds: Vec<&str> = events
416            .iter()
417            .filter_map(|e| e.json.get("type").and_then(Value::as_str))
418            .collect();
419        assert!(kinds.contains(&"change"), "kinds: {kinds:?}");
420        assert!(kinds.contains(&"diagnostics"), "kinds: {kinds:?}");
421        let change = events
422            .iter()
423            .find(|event| event.json["type"] == "change")
424            .expect("change event");
425        assert_eq!(change.json["batch"]["revision"], "1");
426        assert_eq!(change.json["batch"]["tables"][0]["table"], "todo");
427        assert_eq!(change.json["batch"]["status"]["outbox"], 1);
428        // `syncNeeded` is an inbound pull/catch-up signal. Local push work is
429        // represented exactly by `outbox` and by the interactive sync intent.
430        assert_eq!(change.json["batch"]["status"]["syncNeeded"], false);
431        assert!(matches!(core.take_sync_intent(), SyncIntent::Interactive));
432        // Draining is exhaustive.
433        assert!(core.drain_events().is_empty());
434    }
435
436    #[test]
437    fn diagnostics_keep_fresh_capture_times_and_revision_independent_changes() {
438        let mut core = SyncularCore::new(&json!({})).unwrap();
439        let created = core.command(&json!({"method": "create", "params": {
440            "clientId": "diagnostics-comparison", "schema": simple_schema(), "nowMs": 1000
441        }}));
442        assert!(created.get("error").is_none(), "{created}");
443        assert!(core.drain_events().is_empty());
444        core.command(&json!({"method": "enableDiagnostics", "params": {}}));
445        let initial = core.drain_events();
446        assert_eq!(initial.len(), 1);
447        assert_eq!(initial[0].json["snapshot"]["capturedAtMs"], 1000);
448
449        core.client.as_mut().unwrap().set_now_ms(2000);
450        let read = json!({"method": "query", "params": {"sql": "SELECT 1 AS id"}});
451        assert_eq!(core.command(&read)["result"]["rows"], json!([{"id": 1}]));
452        assert!(core.drain_events().is_empty());
453        let changed = core.command(&json!({"method": "mutate", "params": {
454            "mutations": [{"op": "upsert", "table": "todo", "values": {
455                "id": "first", "title": "private value", "done": false
456            }}]
457        }}));
458        assert!(changed.get("error").is_none(), "{changed}");
459        let expected = serde_json::to_value(
460            core.client
461                .as_ref()
462                .unwrap()
463                .diagnostics_snapshot(&Default::default())
464                .unwrap(),
465        )
466        .unwrap();
467        let diagnostics: Vec<Value> = core
468            .drain_events()
469            .into_iter()
470            .filter(|event| event.json["type"] == "diagnostics")
471            .map(|event| event.json)
472            .collect();
473        assert_eq!(
474            diagnostics,
475            vec![json!({"type": "diagnostics", "snapshot": expected})]
476        );
477        assert_eq!(diagnostics[0]["snapshot"]["capturedAtMs"], 2000);
478
479        core.client.as_mut().unwrap().set_now_ms(3000);
480        assert!(core.command(&read).get("error").is_none());
481        assert!(core.drain_events().is_empty());
482        let revision = core.client.as_ref().unwrap().local_revision();
483        core.command(&json!({"method": "sync", "params": {}}));
484        assert_eq!(core.client.as_ref().unwrap().local_revision(), revision);
485        let expected = serde_json::to_value(
486            core.client
487                .as_ref()
488                .unwrap()
489                .diagnostics_snapshot(&Default::default())
490                .unwrap(),
491        )
492        .unwrap();
493        let diagnostics: Vec<Value> = core
494            .drain_events()
495            .into_iter()
496            .filter(|event| event.json["type"] == "diagnostics")
497            .map(|event| event.json)
498            .collect();
499        assert_eq!(
500            diagnostics,
501            vec![json!({"type": "diagnostics", "snapshot": expected})]
502        );
503        assert_eq!(diagnostics[0]["snapshot"]["capturedAtMs"], 3000);
504        assert_eq!(diagnostics[0]["snapshot"]["lastRound"]["status"], "failed");
505
506        let preflight = core.command(&json!({"method": "beginSecurityPreflight", "params": {}}));
507        assert!(preflight.get("error").is_none(), "{preflight}");
508        assert_eq!(
509            core.command(&read)["error"]["code"],
510            "client.security_preflight_required"
511        );
512        assert!(core
513            .drain_events()
514            .iter()
515            .all(|event| event.json["type"] != "diagnostics"));
516    }
517
518    #[test]
519    fn diagnostics_events_wait_for_a_registered_consumer() {
520        let mut core = SyncularCore::new(&json!({})).unwrap();
521        create(&mut core);
522        let _ = core.drain_events();
523        core.command(&json!({
524            "method": "mutate",
525            "params": { "mutations": [{
526                "op": "upsert", "table": "todo",
527                "values": { "id": "t1", "title": "x", "done": false }
528            }] }
529        }));
530        let kinds: Vec<String> = core
531            .drain_events()
532            .iter()
533            .filter_map(|e| e.json.get("type").and_then(Value::as_str))
534            .map(str::to_owned)
535            .collect();
536        assert!(kinds.contains(&"change".to_owned()), "kinds: {kinds:?}");
537        // Without a registered consumer the snapshot is never computed.
538        assert!(
539            !kinds.contains(&"diagnostics".to_owned()),
540            "kinds: {kinds:?}"
541        );
542
543        // Registration delivers a first snapshot on the same drain.
544        let enabled = core.command(&json!({ "method": "enableDiagnostics", "params": {} }));
545        assert_eq!(enabled["result"], json!({}));
546        let events = core.drain_events();
547        assert!(
548            events.iter().any(|e| e.json["type"] == "diagnostics"),
549            "events: {events:?}"
550        );
551
552        // Subsequent status changes keep flowing through the fingerprint gate.
553        core.command(&json!({
554            "method": "mutate",
555            "params": { "mutations": [{
556                "op": "upsert", "table": "todo",
557                "values": { "id": "t2", "title": "y", "done": false }
558            }] }
559        }));
560        let events = core.drain_events();
561        assert!(
562            events.iter().any(|e| e.json["type"] == "diagnostics"),
563            "events: {events:?}"
564        );
565    }
566
567    #[test]
568    fn a_snapshot_pull_registers_the_diagnostics_consumer() {
569        let mut core = SyncularCore::new(&json!({})).unwrap();
570        create(&mut core);
571        let _ = core.drain_events();
572        let reply = core.command(&json!({ "method": "diagnosticsSnapshot", "params": {} }));
573        assert_eq!(reply["result"]["version"], 1);
574        let _ = core.drain_events();
575        core.command(&json!({
576            "method": "mutate",
577            "params": { "mutations": [{
578                "op": "upsert", "table": "todo",
579                "values": { "id": "t1", "title": "x", "done": false }
580            }] }
581        }));
582        let events = core.drain_events();
583        assert!(
584            events.iter().any(|e| e.json["type"] == "diagnostics"),
585            "events: {events:?}"
586        );
587    }
588
589    #[test]
590    fn diagnostics_are_versioned_bounded_and_payload_free() {
591        let mut core = SyncularCore::new(&json!({})).unwrap();
592        create(&mut core);
593        let reply = core.command(&json!({
594            "method": "diagnosticsSnapshot",
595            "params": {
596                "expectedSubscriptions": [{ "id": "membership", "table": "todo" }]
597            }
598        }));
599        assert_eq!(reply["result"]["version"], 1);
600        assert_eq!(reply["result"]["subscriptions"][0]["state"], "unregistered");
601        let encoded = reply.to_string();
602        assert!(!encoded.contains("clientId"));
603        assert!(!encoded.contains("dbPath"));
604        assert!(!encoded.contains("operations"));
605    }
606
607    #[test]
608    fn sync_without_native_transport_fails_loud() {
609        let mut core = SyncularCore::new(&json!({})).unwrap();
610        create(&mut core);
611        let outcome = core.command(&json!({ "method": "sync", "params": {} }));
612        assert_eq!(outcome["result"]["ok"], json!(false), "{outcome}");
613        assert_eq!(outcome["result"]["errorCode"], "transport.unavailable");
614    }
615
616    #[test]
617    fn file_db_persists_across_reopen() {
618        let dir = std::env::temp_dir();
619        let path = dir.join(format!("syncular-tauri-test-{}.db", std::process::id()));
620        let path_str = path.to_string_lossy().to_string();
621        let _ = std::fs::remove_file(&path);
622
623        {
624            let mut core = SyncularCore::new(&json!({})).unwrap();
625            let reply = core.command(&json!({
626                "method": "create",
627                "params": { "clientId": "c1", "schema": simple_schema(), "dbPath": path_str }
628            }));
629            assert_eq!(reply["result"], json!({}), "create with dbPath: {reply}");
630            core.command(&json!({
631                "method": "mutate",
632                "params": { "mutations": [{
633                    "op": "upsert", "table": "todo",
634                    "values": { "id": "persisted", "title": "kept", "done": false }
635                }] }
636            }));
637            let revision = core.command(&json!({
638                "method": "localRevision", "params": {}
639            }));
640            assert_eq!(revision["result"]["revision"], "1");
641        }
642        // Reopen without supplying an id: identity, revision, outbox/status,
643        // and the optimistic visible row all come from the durable database.
644        {
645            let mut core = SyncularCore::new(&json!({})).unwrap();
646            let reopened = core.command(&json!({
647                "method": "create",
648                "params": { "schema": simple_schema(), "dbPath": path_str }
649            }));
650            assert_eq!(reopened["result"], json!({}), "reopen: {reopened}");
651            let rows = core.query("SELECT title FROM todo", Value::Null);
652            let list = rows["result"]["rows"].as_array().expect("rows");
653            assert_eq!(list.len(), 1, "reopened db: {rows}");
654            assert_eq!(list[0]["title"], "kept");
655            let revision = core.command(&json!({
656                "method": "localRevision", "params": {}
657            }));
658            assert_eq!(revision["result"]["revision"], "1");
659            let pending = core.command(&json!({
660                "method": "pendingCommitIds", "params": {}
661            }));
662            assert_eq!(pending["result"]["ids"].as_array().map(Vec::len), Some(1));
663            let status = core.command(&json!({
664                "method": "statusSnapshot", "params": {}
665            }));
666            assert_eq!(status["result"]["outbox"], 1);
667            assert_eq!(status["result"]["syncNeeded"], true);
668            assert!(matches!(core.take_sync_intent(), SyncIntent::Interactive));
669        }
670        {
671            let mut core = SyncularCore::new(&json!({})).unwrap();
672            let mismatch = core.command(&json!({
673                "method": "create",
674                "params": { "clientId": "different", "schema": simple_schema(), "dbPath": path_str }
675            }));
676            assert_eq!(mismatch["error"]["code"], "client.identity_mismatch");
677        }
678        let _ = std::fs::remove_file(&path);
679    }
680
681    #[test]
682    fn config_validation_rejects_baseurl_without_native_feature() {
683        let result = SyncularCore::new(&json!({ "baseUrl": "http://localhost:9/sync" }));
684        #[cfg(not(feature = "native-transport"))]
685        assert!(
686            result.is_err(),
687            "baseUrl must be refused without native-transport"
688        );
689        #[cfg(feature = "native-transport")]
690        assert!(result.is_ok(), "baseUrl builds with native-transport");
691    }
692}