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, SyncClient, SyncIntent};
use syncular_command::{dispatch, CreateEffects};

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

/// 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 {
    client: Option<SyncClient>,
    transport: HostTransport,
    effects: CreateEffects,
    queue: VecDeque<Event>,
    last_diagnostics_fingerprint: Option<Value>,
    /// 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>,
}

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 {
            client: None,
            transport,
            effects: CreateEffects::default(),
            queue: VecDeque::new(),
            last_diagnostics_fingerprint: None,
            diagnostics_observed: false,
            interactive_sync: false,
            background_sync_ms: None,
        })
    }

    /// 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_fingerprint = 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 method == "create" {
            self.last_diagnostics_fingerprint = 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((code, message)) => json!({ "error": { "code": code, "message": message } }),
        }
    }

    /// 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 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
        }
    }

    /// 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": {} }))
    }

    /// 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.client.is_none() {
            return;
        }
        let frames = self.transport.take_inbound();
        for frame in frames {
            match frame {
                transport::Inbound::Text(text) => {
                    if is_presence_control(&text) {
                        self.push(json!({ "type": "presence" }));
                    }
                    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() {
                        client.on_realtime_binary(&mut self.transport, &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;
        };
        let Ok(mut fingerprint) = serde_json::to_value(&snapshot) else {
            return;
        };
        if let Some(object) = fingerprint.as_object_mut() {
            object.remove("capturedAtMs");
        }
        if self.last_diagnostics_fingerprint.as_ref() == Some(&fingerprint) {
            return;
        }
        self.last_diagnostics_fingerprint = Some(fingerprint);
        self.push(json!({ "type": "diagnostics", "snapshot": snapshot }));
    }
}

/// A presence fanout control frame (§8.6.2) — `{"event":"presence",...}`, the
/// one inbound realtime event a native host surfaces directly.
fn is_presence_control(text: &str) -> bool {
    serde_json::from_str::<Value>(text)
        .ok()
        .and_then(|v| {
            v.get("event")
                .and_then(Value::as_str)
                .map(|e| e == "presence")
        })
        .unwrap_or(false)
}

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

    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 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_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");
    }
}