1use 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#[derive(Debug, Clone)]
41pub struct Event {
42 pub json: Value,
43}
44
45pub struct SyncularCore {
48 client: Option<SyncClient>,
49 transport: HostTransport,
50 effects: CreateEffects,
51 queue: VecDeque<Event>,
52 last_diagnostics_snapshot: Option<ClientDiagnosticsSnapshot>,
53 diagnostics_observed: bool,
58 interactive_sync: bool,
59 background_sync_ms: Option<u64>,
60}
61
62impl SyncularCore {
63 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 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 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 self.diagnostics_observed = true;
108 }
109 let result = dispatch(
110 &mut self.transport,
111 &mut self.client,
112 &mut self.effects,
113 method,
114 ¶ms,
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 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 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 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 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 pub fn poll_transport(&mut self) {
188 self.drain_realtime();
189 self.drain_core_outputs();
190 self.emit_diagnostics_if_changed();
191 }
192
193 pub fn drain_events(&mut self) -> Vec<Event> {
196 self.queue.drain(..).collect()
197 }
198
199 pub fn set_headers(&mut self, headers: Vec<(String, String)>) {
203 self.transport.set_headers(headers);
204 }
205
206 pub fn shutdown(&mut self) {
208 if let Some(mut client) = self.client.take() {
209 client.disconnect_realtime(&mut self.transport);
210 client.seal_security_on_teardown();
213 }
214 self.interactive_sync = false;
215 self.background_sync_ms = None;
216 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 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 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 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 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
310fn 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 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 create(&mut core);
401 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 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 assert_eq!(change.json["batch"]["status"]["syncNeeded"], false);
431 assert!(matches!(core.take_sync_intent(), SyncIntent::Interactive));
432 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 assert!(
539 !kinds.contains(&"diagnostics".to_owned()),
540 "kinds: {kinds:?}"
541 );
542
543 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 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 {
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}