tauri_plugin_syncular/

core.rs

//! The transport-agnostic, Tauri-free CORE of the plugin.
//!
//! Everything here is plain Rust with no dependency on the `tauri` crate, so
//! it is unit-testable without a window or a mock runtime. The Tauri shell
//! (see `lib.rs`) is a thin layer that owns one [`SyncularCore`] on a
//! dedicated thread, forwards mutable commands and raw queries to it, and pumps
//! drained events onto the `syncular://event` Tauri channel. The shell's
//! file-backed atomic snapshot sidecar is intentionally outside this mutable
//! core and uses a separate read-only SQLite connection.
//!
//! This mirrors the `syncular-ffi` `Handle`: one owned [`SyncClient`], one
//! owned [`HostTransport`] (native HTTP+WS via the `native-transport` feature),
//! and an exact core-output event queue. The plugin is the THIRD consumer of the shared
//! `syncular-command` router (after the conformance shim and the FFI core), so
//! the command surface stays conformance-locked.
//!
//! ## Thread-safety, honestly
//!
//! [`SyncClient`] is synchronous and NOT `Sync` — it owns a rusqlite
//! connection. Exactly one thread owns the mutable core, and all mutable access
//! arrives through a command mailbox (an mpsc channel). Tauri commands never
//! touch that client directly. The background host loop (§8.4) runs on the same
//! owner, so the mutable connection is never accessed concurrently. The shell
//! may independently read the file database through SQLite's snapshot model;
//! it does not access this `SyncClient`.

use std::collections::VecDeque;

use serde_json::{json, Value};
use syncular_client::{
    ClientDiagnosticsRequest, ClientDiagnosticsSnapshot, QueryOwner, QueryReadFailure, SyncClient,
    SyncIntent,
};
use syncular_command::{dispatch, CreateEffects};

use crate::transport::{self, HostTransport, HostTransportPolicy};

/// One client-observable event (§8 realtime signals + §6 conflicts + §1.6
/// schema floor + §7.3 lease). JSON-able; delivered onto the Tauri channel.
/// The same event vocabulary the FFI `poll_event` surfaces.
#[derive(Debug, Clone)]
pub struct Event {
    pub json: Value,
}

/// The Tauri-free core: one client, its owned transport, explicit scheduling
/// state, and the pending exact-event queue. Lives on ONE owning thread.
pub struct SyncularCore {
    pub progress_listener: Option<syncular_client::SyncProgressListener>,
    progress_subscription: Option<syncular_client::ProgressSubscription>,
    client: Option<SyncClient>,
    transport: HostTransport,
    effects: CreateEffects,
    queue: VecDeque<Event>,
    last_diagnostics_snapshot: Option<ClientDiagnosticsSnapshot>,
    /// Diagnostics snapshots are computed only after a consumer registers —
    /// the analogue of the web client's `ClientDiagnosticsEmitter.observed`.
    /// Set by the explicit `enableDiagnostics` command and by the first
    /// `diagnosticsSnapshot` pull (the devtools attach signal).
    diagnostics_observed: bool,
    interactive_sync: bool,
    background_sync_ms: Option<u64>,
    round_pending: bool,
    controls: Vec<String>,
}

impl SyncularCore {
    /// Build a core from the plugin config JSON (`baseUrl`, `headers`, …). A
    /// `baseUrl` under the `native-transport` feature owns a real HTTP+WS
    /// transport; without it the core is client-local only (tests, offline).
    pub fn new(config: &Value) -> Result<Self, String> {
        Self::new_with_notify(config, None)
    }

    pub fn new_with_notify(
        config: &Value,
        notify: Option<std::sync::Arc<dyn Fn() + Send + Sync>>,
    ) -> Result<Self, String> {
        let transport = HostTransport::from_config_with_notify(config, notify)?;
        Ok(SyncularCore {
            progress_listener: None,
            progress_subscription: None,
            client: None,
            transport,
            effects: CreateEffects::default(),
            queue: VecDeque::new(),
            last_diagnostics_snapshot: None,
            diagnostics_observed: false,
            interactive_sync: false,
            background_sync_ms: None,
            round_pending: false,
            controls: Vec::new(),
        })
    }

    /// Test-only host seam: pin the client clock to the host's test time.
    /// `create`'s `nowMs` sets the initial clock; this sets it again so
    /// timestamps and expiries are deterministic without sleeps. Off by
    /// default; consumers opt in with the `test-clock` feature, and the
    /// plugin's own tests see it under `cfg(test)`.
    #[cfg(any(test, feature = "test-clock"))]
    pub fn set_now_ms(&mut self, now_ms: i64) -> Result<(), String> {
        let Some(client) = self.client.as_mut() else {
            return Err("client.not_created: create a client before setting the clock".to_owned());
        };
        client.set_now_ms(now_ms);
        Ok(())
    }

    /// Run one JSON command (`{"method","params"}`) through the shared router,
    /// then drain inbound realtime traffic and exact core events. Returns the
    /// driver-protocol `{"result"|"error"}` reply.
    pub fn command(&mut self, command: &Value) -> Value {
        let method = command.get("method").and_then(Value::as_str).unwrap_or("");
        let params = command.get("params").cloned().unwrap_or(Value::Null);
        if method == "enableDiagnostics" {
            // Host-local registration signal (the Tauri event channel carries
            // no listener count). Emission starts on this drain: the reset
            // fingerprint guarantees the observer receives a first snapshot.
            self.diagnostics_observed = true;
            self.last_diagnostics_snapshot = None;
            self.drain_realtime();
            self.drain_core_outputs();
            self.emit_diagnostics_if_changed();
            return json!({ "result": {} });
        }
        if method == "diagnosticsSnapshot" {
            // A direct pull proves a diagnostics consumer exists; keep the
            // pushed snapshots flowing for it from here on.
            self.diagnostics_observed = true;
        }
        let result = dispatch(
            &mut self.transport,
            &mut self.client,
            &mut self.effects,
            method,
            &params,
        );
        if result.is_ok() && matches!(method, "activateSecurity" | "setHeaders") {
            if let Some(headers) = params.get("headers") {
                self.set_headers(
                    syncular_command::parse_headers(headers).expect("router validated headers"),
                );
            }
        }
        if method == "create" && result.is_ok() {
            self.progress_subscription = self.client.as_ref().and_then(|client| {
                self.progress_listener.clone().map(|listener| {
                    client
                        .progress()
                        .subscribe(move |progress| listener(progress))
                })
            });
        }
        if method == "shutdown" {
            self.progress_subscription = None;
        }
        if method == "create" {
            self.last_diagnostics_snapshot = None;
            self.transport.set_signed_urls(self.effects.signed_urls);
        }
        if method == "beginSecurityPreflight"
            || method == "shutdown"
            || (method == "create"
                && params
                    .get("securityPreflight")
                    .and_then(Value::as_bool)
                    .unwrap_or(false))
        {
            self.interactive_sync = false;
            self.background_sync_ms = None;
        }
        if let Ok(value) = &result {
            if value.pointer("/effects/sync/kind").and_then(Value::as_str) == Some("interactive") {
                self.interactive_sync = true;
            }
        }
        self.drain_realtime();
        self.drain_core_outputs();
        self.emit_diagnostics_if_changed();
        match result {
            Ok(mut value) => {
                // The router's effects are consumed by this native host above;
                // they are not part of the public command acknowledgement.
                if let Some(object) = value.as_object_mut() {
                    object.remove("effects");
                }
                json!({ "result": value })
            }
            Err(error) => {
                let mut envelope = json!({
                    "code": error.code,
                    "message": error.message,
                    "retryable": error.retryable
                });
                if let Some(details) = error.details {
                    envelope["details"] = details;
                }
                json!({ "error": envelope })
            }
        }
    }

    /// The `syncular_query` fast path: arbitrary read-only SQL over the local
    /// database. Routed through the same `query` command so there is one
    /// implementation (the router owns it); this wrapper spares the JS bridge
    /// from wrapping the method/params envelope for the hot live-query path.
    pub fn query(&mut self, sql: &str, params: Value) -> Value {
        let bind = match params {
            Value::Null => Value::Array(Vec::new()),
            other => other,
        };
        self.command(&json!({ "method": "query", "params": { "sql": sql, "params": bind } }))
    }

    /// Consume the next coalesced host schedule. Interactive work preempts a
    /// pending retry; background work keeps the earliest real deadline.
    pub fn take_sync_intent(&mut self) -> SyncIntent {
        if !self.transport_enabled() {
            return SyncIntent::None;
        }
        if std::mem::take(&mut self.interactive_sync) {
            self.background_sync_ms = None;
            SyncIntent::Interactive
        } else if let Some(delay_ms) = self.background_sync_ms.take() {
            SyncIntent::Background { delay_ms }
        } else {
            SyncIntent::None
        }
    }

    pub fn transport_enabled(&self) -> bool {
        self.client
            .as_ref()
            .is_some_and(SyncClient::transport_enabled)
    }

    /// Run one `syncUntilIdle` round for the background host loop, deriving
    /// events afterwards. A no-op (empty reply) before `create`.
    pub fn sync_until_idle(&mut self) -> Value {
        if self.client.is_none() {
            return json!({ "result": null });
        }
        self.command(&json!({ "method": "syncUntilIdle", "params": {} }))
    }

    /// Prepare on the owner; the returned transport view belongs to the I/O
    /// executor. Local commands continue using the owned client connection.
    pub fn prepare_round(
        &mut self,
    ) -> Result<
        (syncular_client::PreparedSyncRound, HostTransport),
        Box<syncular_client::SyncOutcome>,
    > {
        let Some(client) = self.client.as_mut() else {
            return Err(Box::new(syncular_client::SyncOutcome::Failed {
                error_code: "client.not_created".into(),
                message: "create a client before syncing".into(),
                details: None,
            }));
        };
        use syncular_client::Transport;
        let result = client.prepare_sync_round(self.transport.supports_url_fetch());
        self.round_pending = result.is_ok();
        self.drain_core_outputs();
        self.emit_diagnostics_if_changed();
        result.map(|prepared| (prepared, self.transport.fork_round()))
    }

    pub fn apply_round(
        &mut self,
        completed: syncular_client::CompletedSyncRound,
    ) -> syncular_client::AppliedSyncRound {
        self.round_pending = false;
        let mut applied = match self.client.as_mut() {
            Some(client) => client.apply_sync_round(completed),
            None => syncular_client::AppliedSyncRound::Complete {
                outcome: syncular_client::SyncOutcome::Failed {
                    error_code: "client.round_cancelled".into(),
                    message: "sync round belongs to a released client context".into(),
                    details: None,
                },
                controls: Vec::new(),
                more: false,
                bootstrap_advanced: false,
            },
        };
        self.round_pending = matches!(applied, syncular_client::AppliedSyncRound::Continue(_));
        if !self.round_pending && !self.transport_enabled() {
            if let Some(client) = self.client.as_mut() {
                client.set_transport_enabled(&mut self.transport, false);
            }
        }
        if let syncular_client::AppliedSyncRound::Complete { controls, .. } = &mut applied {
            self.controls.append(controls);
        }
        // Buffered socket deltas follow the captured response, preventing a
        // delayed reply from overwriting a newer unsolicited delta.
        self.drain_realtime();
        self.drain_core_outputs();
        self.emit_diagnostics_if_changed();
        applied
    }

    /// Replace the native transport's bounds for subsequent network calls
    /// (per-round deadlines, request/response byte limits, redirect policy).
    /// JSON config reaches the same policy through `new`; this is the
    /// programmatic seam for a host that builds its configuration in Rust. A
    /// round past its first exchange keeps its anchored deadline; every other
    /// bound takes effect on the next call. An invalid policy is rejected and
    /// the current policy stays in force.
    pub fn set_transport_policy(&mut self, policy: HostTransportPolicy) -> Result<(), String> {
        self.transport.set_policy(policy)
    }

    pub fn take_controls(&mut self) -> Option<(Vec<String>, HostTransport)> {
        if !self.transport_enabled() {
            self.controls.clear();
            return None;
        }
        if self.controls.is_empty() {
            return None;
        }
        Some((
            std::mem::take(&mut self.controls),
            self.transport.fork_round(),
        ))
    }

    /// §7.6: record an owned snapshot read the read sidecar ran.
    pub fn record_query_read(
        &mut self,
        id: &str,
        tables: &[String],
        failure: Option<&QueryReadFailure>,
    ) {
        let Some(client) = self.client.as_mut() else {
            return;
        };
        let tables: Vec<&str> = tables.iter().map(String::as_str).collect();
        client.record_query_read(
            QueryOwner {
                id,
                tables: &tables,
            },
            failure,
        );
        self.emit_diagnostics_if_changed();
    }

    /// Owner-mailbox wake from the native realtime reader.
    pub fn poll_transport(&mut self) {
        self.drain_realtime();
        self.drain_core_outputs();
        self.emit_diagnostics_if_changed();
    }

    /// Drain every event queued since the last call (the host thread pushes
    /// them onto the Tauri channel). Mirrors the FFI `poll_event`, batched.
    pub fn drain_events(&mut self) -> Vec<Event> {
        self.queue.drain(..).collect()
    }

    /// Replace the transport's request headers. Rotating
    /// auth without tearing the plugin down). See
    /// `HostTransport::set_headers` for the HTTP/WS pickup semantics.
    pub fn set_headers(&mut self, headers: Vec<(String, String)>) {
        self.transport.set_headers(headers);
    }

    /// Release the socket/reader thread. Idempotent.
    pub fn shutdown(&mut self) {
        if let Some(mut client) = self.client.take() {
            client.disconnect_realtime(&mut self.transport);
            // Teardown barrier, not a quarantine: persisting the gate here
            // would refuse every later plain create against this replica.
            client.seal_security_on_teardown();
        }
        self.interactive_sync = false;
        self.background_sync_ms = None;
        // The TS driver clears its diagnostics listeners on dispose; require
        // a fresh registration after any restart.
        self.diagnostics_observed = false;
        self.transport.shutdown();
    }

    fn push(&mut self, json: Value) {
        self.queue.push_back(Event { json });
    }

    /// Feed buffered inbound WS frames to the client (which may ack back through
    /// the same transport). A no-op without a native socket.
    fn drain_realtime(&mut self) {
        if self.round_pending || !self.transport_enabled() {
            return;
        }
        let frames = self.transport.take_inbound();
        for frame in frames {
            match frame {
                transport::Inbound::Text(text) => {
                    if let Some(scope_key) = presence_scope_key(&text) {
                        self.push(json!({ "type": "presence", "scopeKey": scope_key }));
                    }
                    if let Some(client) = self.client.as_mut() {
                        client.on_realtime_text(&text);
                    }
                }
                transport::Inbound::Binary(bytes) => {
                    if let Some(client) = self.client.as_mut() {
                        if !client.security_preflight() {
                            self.controls
                                .extend(client.on_realtime_binary_queued(&bytes));
                        }
                    }
                }
            }
        }
    }

    /// Drain exact observer batches and sync intents produced by the Rust core.
    fn drain_core_outputs(&mut self) {
        let Some(client) = self.client.as_mut() else {
            return;
        };
        let batches = client.drain_change_batches();
        let intents = client.drain_sync_intents();
        for batch in batches {
            self.push(json!({ "type": "change", "batch": batch }));
        }
        for intent in intents {
            match intent {
                SyncIntent::Interactive => self.interactive_sync = true,
                SyncIntent::Background { delay_ms } => {
                    self.background_sync_ms = Some(
                        self.background_sync_ms
                            .map_or(delay_ms, |current| current.min(delay_ms)),
                    );
                }
                SyncIntent::None => {}
            }
        }
    }

    /// Emit one privacy-safe snapshot only when a diagnostics consumer has
    /// registered (`diagnostics_observed`) AND durable/client status changed.
    /// The observer gate keeps the per-command snapshot work (subscription
    /// scan, PRAGMA page counts, SUM aggregates, serialization) off every
    /// unobserved command and transport poll.
    /// `capturedAtMs` is excluded from the fingerprint so polling and read-only
    /// commands do not create event noise. Expected-but-unregistered intent is
    /// request-local and is obtained through `diagnosticsSnapshot` directly.
    fn emit_diagnostics_if_changed(&mut self) {
        if !self.diagnostics_observed {
            return;
        }
        let Some(client) = self.client.as_ref() else {
            return;
        };
        if client.security_preflight() {
            return;
        }
        let Ok(snapshot) = client.diagnostics_snapshot(&ClientDiagnosticsRequest::default()) else {
            return;
        };
        if let Some(previous) = &mut self.last_diagnostics_snapshot {
            // Capture time is observation metadata, excluded from evidence equality.
            previous.captured_at_ms = snapshot.captured_at_ms;
            if previous == &snapshot {
                return;
            }
        }
        let event = json!({ "type": "diagnostics", "snapshot": snapshot });
        self.last_diagnostics_snapshot = Some(snapshot);
        self.push(event);
    }
}

/// A presence fanout control frame (§8.6.2) — `{"event":"presence",...}`, the
/// one inbound realtime event a native host surfaces directly.
fn presence_scope_key(text: &str) -> Option<String> {
    serde_json::from_str::<Value>(text).ok().and_then(|v| {
        if v.get("event").and_then(Value::as_str) != Some("presence") {
            return None;
        }
        v.get("data")?
            .get("scopeKey")
            .and_then(Value::as_str)
            .map(str::to_owned)
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn presence_scope_key_reads_the_fanout_data() {
        assert_eq!(
            presence_scope_key(
                r#"{"event":"presence","data":{"scopeKey":"room:42","kind":"leave","actorId":"a","clientId":"c","doc":null,"timestamp":1}}"#
            )
            .as_deref(),
            Some("room:42")
        );
        assert_eq!(
            presence_scope_key(r#"{"event":"presence","data":{}}"#),
            None
        );
        assert_eq!(presence_scope_key(r#"{"event":"sync","data":{}}"#), None);
    }

    fn simple_schema() -> Value {
        json!({
            "version": 1,
            "tables": [{
                "name": "todo",
                "primaryKey": "id",
                "columns": [
                    { "name": "id", "type": "string", "nullable": false },
                    { "name": "title", "type": "string", "nullable": false },
                    { "name": "done", "type": "boolean", "nullable": false }
                ],
                "scopes": []
            }]
        })
    }

    fn create(core: &mut SyncularCore) {
        let reply = core.command(&json!({
            "method": "create",
            "params": { "clientId": "c1", "schema": simple_schema() }
        }));
        assert_eq!(reply["result"], json!({}), "create ok: {reply}");
    }

    #[test]
    fn transport_gate_survives_security_activation_and_owner_rounds() {
        let mut core = SyncularCore::new(&json!({})).unwrap();
        let result = core.command(&json!({ "method": "create", "params": {
            "schema": simple_schema(), "transportEnabled": false, "securityPreflight": true
        } }));
        assert!(result.get("error").is_none());
        assert!(core
            .command(&json!({ "method": "activateSecurity", "params": {} }))
            .get("error")
            .is_none());
        for title in ["first", "second"] {
            assert!(core.command(&json!({ "method": "mutate", "params": { "mutations": [{
                "op": "upsert", "table": "todo", "values": { "id": "one", "title": title, "done": false }
            }] } })).get("error").is_none());
            assert!(matches!(core.take_sync_intent(), SyncIntent::None));
            assert!(
                matches!(*core.prepare_round().err().expect("closed transport"), syncular_client::SyncOutcome::Failed { ref error_code, .. } if error_code == "sync.offline")
            );
        }
        assert_eq!(
            core.query("SELECT title FROM todo", Value::Null)["result"]["rows"][0]["title"],
            "second"
        );
        assert!(!core.transport_enabled());
        core.command(&json!({ "method": "setTransportEnabled", "params": { "enabled": true } }));
        assert!(matches!(core.take_sync_intent(), SyncIntent::Interactive));
        core.command(&json!({ "method": "setTransportEnabled", "params": { "enabled": true } }));
        assert!(matches!(core.take_sync_intent(), SyncIntent::None));
        core.command(&json!({ "method": "setTransportEnabled", "params": { "enabled": false } }));
        core.command(&json!({ "method": "beginSecurityPreflight" }));
        assert_eq!(
            core.query("SELECT title FROM todo", Value::Null)["error"]["code"],
            "client.security_preflight_required"
        );
        core.command(&json!({ "method": "activateSecurity", "params": {} }));
        assert!(!core.transport_enabled());
    }

    #[test]
    fn progress_sink_runs_before_the_sync_command_returns() {
        use std::sync::{
            atomic::{AtomicBool, Ordering},
            Arc, Mutex,
        };
        let mut core = SyncularCore::new(&json!({})).unwrap();
        let running = Arc::new(AtomicBool::new(false));
        let observed = Arc::new(Mutex::new(Vec::new()));
        let in_command = running.clone();
        let events = observed.clone();
        core.progress_listener = Some(Arc::new(move |progress| {
            events
                .lock()
                .unwrap()
                .push((in_command.load(Ordering::SeqCst), progress.state));
        }));
        create(&mut core);
        running.store(true, Ordering::SeqCst);
        core.command(&json!({ "method": "sync" }));
        running.store(false, Ordering::SeqCst);
        assert_eq!(
            *observed.lock().unwrap(),
            vec![
                (true, syncular_client::ProgressState::Running),
                (true, syncular_client::ProgressState::Failed)
            ]
        );
    }

    #[test]
    fn command_round_trip_create_mutate_query() {
        let mut core = SyncularCore::new(&json!({})).unwrap();
        create(&mut core);

        let sub = core.command(&json!({
            "method": "subscribe",
            "params": { "id": "s1", "table": "todo", "scopes": {} }
        }));
        assert_eq!(sub["result"], json!({}));

        let mutate = core.command(&json!({
            "method": "mutate",
            "params": { "mutations": [{
                "op": "upsert", "table": "todo",
                "values": { "id": "t1", "title": "hello", "done": false }
            }] }
        }));
        assert!(mutate["result"]["clientCommitId"].is_string(), "{mutate}");

        // The query fast path sees the optimistic overlay immediately.
        let rows = core.query("SELECT id, title FROM todo ORDER BY id", Value::Null);
        let list = rows["result"]["rows"].as_array().expect("rows");
        assert_eq!(list.len(), 1);
        assert_eq!(list[0]["title"], "hello");
        assert_eq!(list[0]["id"], "t1");
    }

    #[test]
    fn query_binds_params() {
        let mut core = SyncularCore::new(&json!({})).unwrap();
        create(&mut core);
        core.command(&json!({
            "method": "mutate",
            "params": { "mutations": [
                { "op": "upsert", "table": "todo", "values": { "id": "a", "title": "A", "done": false } },
                { "op": "upsert", "table": "todo", "values": { "id": "b", "title": "B", "done": true } }
            ] }
        }));
        let rows = core.query("SELECT id FROM todo WHERE done = ?", json!([true]));
        let list = rows["result"]["rows"].as_array().expect("rows");
        assert_eq!(list.len(), 1);
        assert_eq!(list[0]["id"], "b");
    }

    #[test]
    fn events_derived_after_mutate() {
        let mut core = SyncularCore::new(&json!({})).unwrap();
        // Before create, no events.
        create(&mut core);
        // Register a diagnostics consumer so mutate-driven snapshots flow.
        let enabled = core.command(&json!({ "method": "enableDiagnostics", "params": {} }));
        assert_eq!(enabled["result"], json!({}));
        let _ = core.drain_events();
        core.command(&json!({
            "method": "mutate",
            "params": { "mutations": [{
                "op": "upsert", "table": "todo",
                "values": { "id": "t1", "title": "x", "done": false }
            }] }
        }));
        let events = core.drain_events();
        // A local mutate writes the optimistic overlay and revision in one
        // transaction, then emits the exact committed change batch.
        let kinds: Vec<&str> = events
            .iter()
            .filter_map(|e| e.json.get("type").and_then(Value::as_str))
            .collect();
        assert!(kinds.contains(&"change"), "kinds: {kinds:?}");
        assert!(kinds.contains(&"diagnostics"), "kinds: {kinds:?}");
        let change = events
            .iter()
            .find(|event| event.json["type"] == "change")
            .expect("change event");
        assert_eq!(change.json["batch"]["revision"], "1");
        assert_eq!(change.json["batch"]["tables"][0]["table"], "todo");
        assert_eq!(change.json["batch"]["status"]["outbox"], 1);
        // `syncNeeded` is an inbound pull/catch-up signal. Local push work is
        // represented exactly by `outbox` and by the interactive sync intent.
        assert_eq!(change.json["batch"]["status"]["syncNeeded"], false);
        assert!(matches!(core.take_sync_intent(), SyncIntent::Interactive));
        // Draining is exhaustive.
        assert!(core.drain_events().is_empty());
    }

    #[test]
    fn diagnostics_keep_fresh_capture_times_and_revision_independent_changes() {
        let mut core = SyncularCore::new(&json!({})).unwrap();
        let created = core.command(&json!({"method": "create", "params": {
            "clientId": "diagnostics-comparison", "schema": simple_schema(), "nowMs": 1000
        }}));
        assert!(created.get("error").is_none(), "{created}");
        assert!(core.drain_events().is_empty());
        core.command(&json!({"method": "enableDiagnostics", "params": {}}));
        let initial = core.drain_events();
        assert_eq!(initial.len(), 1);
        assert_eq!(initial[0].json["snapshot"]["capturedAtMs"], 1000);

        core.set_now_ms(2000).unwrap();
        let read = json!({"method": "query", "params": {"sql": "SELECT 1 AS id"}});
        assert_eq!(core.command(&read)["result"]["rows"], json!([{"id": 1}]));
        assert!(core.drain_events().is_empty());
        let changed = core.command(&json!({"method": "mutate", "params": {
            "mutations": [{"op": "upsert", "table": "todo", "values": {
                "id": "first", "title": "private value", "done": false
            }}]
        }}));
        assert!(changed.get("error").is_none(), "{changed}");
        let expected = serde_json::to_value(
            core.client
                .as_ref()
                .unwrap()
                .diagnostics_snapshot(&Default::default())
                .unwrap(),
        )
        .unwrap();
        let diagnostics: Vec<Value> = core
            .drain_events()
            .into_iter()
            .filter(|event| event.json["type"] == "diagnostics")
            .map(|event| event.json)
            .collect();
        assert_eq!(
            diagnostics,
            vec![json!({"type": "diagnostics", "snapshot": expected})]
        );
        assert_eq!(diagnostics[0]["snapshot"]["capturedAtMs"], 2000);

        core.set_now_ms(3000).unwrap();
        assert!(core.command(&read).get("error").is_none());
        assert!(core.drain_events().is_empty());
        let revision = core.client.as_ref().unwrap().local_revision();
        core.command(&json!({"method": "sync", "params": {}}));
        assert_eq!(core.client.as_ref().unwrap().local_revision(), revision);
        let expected = serde_json::to_value(
            core.client
                .as_ref()
                .unwrap()
                .diagnostics_snapshot(&Default::default())
                .unwrap(),
        )
        .unwrap();
        let diagnostics: Vec<Value> = core
            .drain_events()
            .into_iter()
            .filter(|event| event.json["type"] == "diagnostics")
            .map(|event| event.json)
            .collect();
        assert_eq!(
            diagnostics,
            vec![json!({"type": "diagnostics", "snapshot": expected})]
        );
        assert_eq!(diagnostics[0]["snapshot"]["capturedAtMs"], 3000);
        assert_eq!(diagnostics[0]["snapshot"]["lastRound"]["status"], "failed");

        let preflight = core.command(&json!({"method": "beginSecurityPreflight", "params": {}}));
        assert!(preflight.get("error").is_none(), "{preflight}");
        assert_eq!(
            core.command(&read)["error"]["code"],
            "client.security_preflight_required"
        );
        assert!(core
            .drain_events()
            .iter()
            .all(|event| event.json["type"] != "diagnostics"));
    }

    #[test]
    fn set_now_ms_sets_observable_capture_times_and_requires_a_client() {
        let mut core = SyncularCore::new(&json!({})).unwrap();
        // The plugin's `client.not_created` identity (also used by
        // `prepare_round`) covers a missing client.
        assert_eq!(
            core.set_now_ms(500).unwrap_err(),
            "client.not_created: create a client before setting the clock"
        );
        let created = core.command(&json!({"method": "create", "params": {
            "clientId": "clock-seam", "schema": simple_schema(), "nowMs": 1000
        }}));
        assert!(created.get("error").is_none(), "{created}");
        core.command(&json!({"method": "enableDiagnostics", "params": {}}));
        let initial = core.drain_events();
        assert_eq!(
            initial[0].json["snapshot"]["capturedAtMs"], 1000,
            "create.nowMs pins the client clock"
        );

        core.set_now_ms(2500).unwrap();
        let read = json!({"method": "query", "params": {"sql": "SELECT 1 AS id"}});
        assert!(core.command(&read).get("error").is_none());
        core.command(&json!({"method": "mutate", "params": {
            "mutations": [{"op": "upsert", "table": "todo", "values": {
                "id": "clocked", "title": "x", "done": false
            }}]
        }}));
        let advanced = core
            .drain_events()
            .into_iter()
            .find(|event| event.json["type"] == "diagnostics")
            .expect("advanced diagnostics snapshot");
        assert_eq!(
            advanced.json["snapshot"]["capturedAtMs"], 2500,
            "the seam sets the observable clock"
        );
    }

    #[test]
    fn diagnostics_events_wait_for_a_registered_consumer() {
        let mut core = SyncularCore::new(&json!({})).unwrap();
        create(&mut core);
        let _ = core.drain_events();
        core.command(&json!({
            "method": "mutate",
            "params": { "mutations": [{
                "op": "upsert", "table": "todo",
                "values": { "id": "t1", "title": "x", "done": false }
            }] }
        }));
        let kinds: Vec<String> = core
            .drain_events()
            .iter()
            .filter_map(|e| e.json.get("type").and_then(Value::as_str))
            .map(str::to_owned)
            .collect();
        assert!(kinds.contains(&"change".to_owned()), "kinds: {kinds:?}");
        // Without a registered consumer the snapshot is never computed.
        assert!(
            !kinds.contains(&"diagnostics".to_owned()),
            "kinds: {kinds:?}"
        );

        // Registration delivers a first snapshot on the same drain.
        let enabled = core.command(&json!({ "method": "enableDiagnostics", "params": {} }));
        assert_eq!(enabled["result"], json!({}));
        let events = core.drain_events();
        assert!(
            events.iter().any(|e| e.json["type"] == "diagnostics"),
            "events: {events:?}"
        );

        // Subsequent status changes keep flowing through the fingerprint gate.
        core.command(&json!({
            "method": "mutate",
            "params": { "mutations": [{
                "op": "upsert", "table": "todo",
                "values": { "id": "t2", "title": "y", "done": false }
            }] }
        }));
        let events = core.drain_events();
        assert!(
            events.iter().any(|e| e.json["type"] == "diagnostics"),
            "events: {events:?}"
        );
    }

    #[test]
    fn a_snapshot_pull_registers_the_diagnostics_consumer() {
        let mut core = SyncularCore::new(&json!({})).unwrap();
        create(&mut core);
        let _ = core.drain_events();
        let reply = core.command(&json!({ "method": "diagnosticsSnapshot", "params": {} }));
        assert_eq!(reply["result"]["version"], 1);
        let _ = core.drain_events();
        core.command(&json!({
            "method": "mutate",
            "params": { "mutations": [{
                "op": "upsert", "table": "todo",
                "values": { "id": "t1", "title": "x", "done": false }
            }] }
        }));
        let events = core.drain_events();
        assert!(
            events.iter().any(|e| e.json["type"] == "diagnostics"),
            "events: {events:?}"
        );
    }

    #[test]
    fn diagnostics_are_versioned_bounded_and_payload_free() {
        let mut core = SyncularCore::new(&json!({})).unwrap();
        create(&mut core);
        let reply = core.command(&json!({
            "method": "diagnosticsSnapshot",
            "params": {
                "expectedSubscriptions": [{ "id": "membership", "table": "todo" }]
            }
        }));
        assert_eq!(reply["result"]["version"], 1);
        assert_eq!(reply["result"]["subscriptions"][0]["state"], "unregistered");
        let encoded = reply.to_string();
        assert!(!encoded.contains("clientId"));
        assert!(!encoded.contains("dbPath"));
        assert!(!encoded.contains("operations"));
    }

    #[test]
    fn sync_without_native_transport_fails_loud() {
        let mut core = SyncularCore::new(&json!({})).unwrap();
        create(&mut core);
        let outcome = core.command(&json!({ "method": "sync", "params": {} }));
        assert_eq!(outcome["result"]["ok"], json!(false), "{outcome}");
        assert_eq!(outcome["result"]["errorCode"], "transport.unavailable");
    }

    #[test]
    fn file_db_persists_across_reopen() {
        let dir = std::env::temp_dir();
        let path = dir.join(format!("syncular-tauri-test-{}.db", std::process::id()));
        let path_str = path.to_string_lossy().to_string();
        let _ = std::fs::remove_file(&path);

        {
            let mut core = SyncularCore::new(&json!({})).unwrap();
            let reply = core.command(&json!({
                "method": "create",
                "params": { "clientId": "c1", "schema": simple_schema(), "dbPath": path_str }
            }));
            assert_eq!(reply["result"], json!({}), "create with dbPath: {reply}");
            core.command(&json!({
                "method": "mutate",
                "params": { "mutations": [{
                    "op": "upsert", "table": "todo",
                    "values": { "id": "persisted", "title": "kept", "done": false }
                }] }
            }));
            let revision = core.command(&json!({
                "method": "localRevision", "params": {}
            }));
            assert_eq!(revision["result"]["revision"], "1");
        }
        // Reopen without supplying an id: identity, revision, outbox/status,
        // and the optimistic visible row all come from the durable database.
        {
            let mut core = SyncularCore::new(&json!({})).unwrap();
            let reopened = core.command(&json!({
                "method": "create",
                "params": { "schema": simple_schema(), "dbPath": path_str }
            }));
            assert_eq!(reopened["result"], json!({}), "reopen: {reopened}");
            let rows = core.query("SELECT title FROM todo", Value::Null);
            let list = rows["result"]["rows"].as_array().expect("rows");
            assert_eq!(list.len(), 1, "reopened db: {rows}");
            assert_eq!(list[0]["title"], "kept");
            let revision = core.command(&json!({
                "method": "localRevision", "params": {}
            }));
            assert_eq!(revision["result"]["revision"], "1");
            let pending = core.command(&json!({
                "method": "pendingCommitIds", "params": {}
            }));
            assert_eq!(pending["result"]["ids"].as_array().map(Vec::len), Some(1));
            let status = core.command(&json!({
                "method": "statusSnapshot", "params": {}
            }));
            assert_eq!(status["result"]["outbox"], 1);
            assert_eq!(status["result"]["syncNeeded"], true);
            assert!(matches!(core.take_sync_intent(), SyncIntent::Interactive));
        }
        {
            let mut core = SyncularCore::new(&json!({})).unwrap();
            let mismatch = core.command(&json!({
                "method": "create",
                "params": { "clientId": "different", "schema": simple_schema(), "dbPath": path_str }
            }));
            assert_eq!(mismatch["error"]["code"], "client.identity_mismatch");
        }
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn config_validation_rejects_baseurl_without_native_feature() {
        let result = SyncularCore::new(&json!({ "baseUrl": "http://localhost:9/sync" }));
        #[cfg(not(feature = "native-transport"))]
        assert!(
            result.is_err(),
            "baseUrl must be refused without native-transport"
        );
        #[cfg(feature = "native-transport")]
        assert!(result.is_ok(), "baseUrl builds with native-transport");
    }
}