M4

src-tauri/src/sync.rs

@@ -0,0 +1,292 @@
+//! Operator-data sync with the DHP (M3), this PC's side.
+//!
+//! The merge rules live in SQL that ships inside each signed version
+//! (`versions\<v>\sync\*.sql`, the same files the DHP runs). The launcher only
+//! carries JSON between the two databases, while the local Engine API is stopped
+//! so nothing it caches goes stale:
+//!
+//!   sync.outbox → POST /v1/sync/push → sync.pushed
+//!   POST /v1/sync/pull (every page) → sync.apply_pull
+//!
+//! Conflicts wait in the local `sync.conflicts` table until the operator decides.
+
+use std::path::Path;
+use std::time::Duration;
+
+use reqwest::StatusCode;
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+
+use crate::localdb::Pg;
+
+/// Load order of the shipped SQL (the DHP loads the same list: odian_sync.schema.SQL_FILES).
+pub const SQL_FILES: [&str; 5] = ["01_core.sql", "02_mark.sql", "03_dhp.sql", "04_install.sql", "05_registry.sql"];
+/// Records per push; the DHP takes at most 2000.
+const PUSH_BATCH: usize = 1000;
+/// Changes per pull page; the DHP serves at most 2000.
+const PULL_PAGE: u32 = 500;
+const HTTP_TIMEOUT: Duration = Duration::from_secs(300);
+
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct SyncStatus {
+  /// Local edits the DHP has not taken yet.
+  pub pending: u64,
+  /// Records both sides changed: the operator decides.
+  pub conflicts: u64,
+  /// Edits the DHP refused (another operator's record, ...).
+  pub rejected: u64,
+  /// DHP changes this PC could not apply yet.
+  pub waiting: u64,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct Conflict {
+  pub tbl: String,
+  pub key: Value,
+  /// "conflict" or "rejected".
+  pub kind: String,
+  /// Clashing fields; "*exists" when one side deleted the record, "*value" for a whole-value clash.
+  pub fields: Vec<String>,
+  pub base: Option<Value>,
+  pub mine: Option<Value>,
+  pub theirs: Option<Value>,
+  pub reason: Option<String>,
+}
+
+#[derive(Debug)]
+pub enum SyncError {
+  /// The DHP refused the Engine token: renew it and try again.
+  Unauthorized,
+  /// The DHP cannot sync right now (unreachable, sync not set up, an Engine that does not
+  /// name operators). Edits stay on this PC for the next connected run.
+  Unavailable(String),
+  Failed(String),
+}
+
+impl From<String> for SyncError {
+  fn from(message: String) -> Self {
+    SyncError::Failed(message)
+  }
+}
+
+impl std::fmt::Display for SyncError {
+  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+    match self {
+      SyncError::Unauthorized => write!(f, "the DHP refused this PC's sign-in"),
+      SyncError::Unavailable(message) | SyncError::Failed(message) => write!(f, "{message}"),
+    }
+  }
+}
+
+/// The DHP's update service and the token to talk to it with.
+pub struct Dhp<'a> {
+  pub http: &'a reqwest::Client,
+  pub site_url: &'a str,
+  pub token: &'a str,
+}
+
+#[derive(Debug, Default)]
+pub struct Report {
+  /// Records sent (conflicts waiting on the operator are sent again each run).
+  pub sent: usize,
+  /// Records the DHP took.
+  pub accepted: u64,
+  /// DHP changes received.
+  pub received: usize,
+  pub status: SyncStatus,
+}
+
+/// Does this version ship the sync schema?
+pub fn schema_shipped(dir: &Path) -> bool {
+  SQL_FILES.iter().all(|name| dir.join(name).is_file())
+}
+
+/// `text` as an SQL string literal.
+fn literal(text: &str) -> String {
+  format!("'{}'", text.replace('\'', "''"))
+}
+
+/// A JSON value as a jsonb literal, dollar-quoted with a tag its text cannot contain.
+fn jsonb(value: &Value) -> String {
+  let text = value.to_string();
+  let tag = (0u32..).map(|n| format!("$j{n}$")).find(|tag| !text.contains(tag.as_str())).unwrap_or_default();
+  format!("{tag}{text}{tag}::jsonb")
+}
+
+/// One statement whose single result is JSON (NULL when it printed nothing).
+fn query(pg: &Pg, sql: &str) -> Result<Value, String> {
+  let out = pg.script(&format!("{sql};\n"))?;
+  let text = out.trim();
+  if text.is_empty() {
+    return Ok(Value::Null);
+  }
+  serde_json::from_str(text).map_err(|e| format!("unexpected answer from this PC's database: {e}"))
+}
+
+async fn blocking<T: Send + 'static>(
+  pg: &Pg,
+  work: impl FnOnce(&Pg) -> Result<T, String> + Send + 'static,
+) -> Result<T, String> {
+  let pg = pg.clone();
+  tauri::async_runtime::spawn_blocking(move || work(&pg)).await.map_err(|e| format!("sync worker failed: {e}"))?
+}
+
+/// Load (or upgrade) the version's sync schema in one transaction. Every file is
+/// idempotent, so this runs before each sync and a new version's rules apply at once.
+pub fn load_schema(pg: &Pg, dir: &Path) -> Result<(), String> {
+  let mut script = String::from("BEGIN;\n");
+  for name in SQL_FILES {
+    let path = dir.join(name);
+    script += &std::fs::read_to_string(&path).map_err(|e| format!("cannot read {}: {e}", path.display()))?;
+    script.push('\n');
+  }
+  script.push_str("COMMIT;\n");
+  pg.script(&script).map(|_| ())
+}
+
+/// The DHP database this one syncs with, or None before its first sync.
+pub fn joined_db_id(pg: &Pg) -> Result<Option<String>, String> {
+  if query(pg, "SELECT to_jsonb(to_regclass('sync.state') IS NOT NULL)")? != Value::Bool(true) {
+    return Ok(None);
+  }
+  let id = query(pg, "SELECT to_jsonb((SELECT value FROM sync.state WHERE name = 'db_id'))")?;
+  Ok(id.as_str().map(str::to_string))
+}
+
+/// Make this database an install of the DHP database `db_id`, pulling from transaction `since`.
+pub fn join(pg: &Pg, db_id: &str, since: &str) -> Result<(), String> {
+  query(pg, &format!("SELECT sync.install('install', {}, {})", literal(db_id), literal(since))).map(|_| ())
+}
+
+pub fn status(pg: &Pg) -> Result<SyncStatus, String> {
+  serde_json::from_value(query(pg, "SELECT sync.status()")?).map_err(|e| format!("unreadable sync status: {e}"))
+}
+
+/// What the operator has to decide (links held with a record are resolved with it).
+pub fn conflicts(pg: &Pg) -> Result<Vec<Conflict>, String> {
+  let rows = query(
+    pg,
+    "SELECT coalesce(jsonb_agg(to_jsonb(c) - 'detected_at' - 'depends' ORDER BY c.detected_at), '[]') \
+     FROM sync.conflicts c WHERE c.depends IS NULL",
+  )?;
+  serde_json::from_value(rows).map_err(|e| format!("unreadable sync conflicts: {e}"))
+}
+
+/// Record the operator's decision: "theirs", "mine", "value" (with `value`) or "dismiss".
+pub fn resolve(pg: &Pg, tbl: &str, key: &Value, choice: &str, value: Option<&Value>) -> Result<(), String> {
+  if !matches!(choice, "theirs" | "mine" | "value" | "dismiss") {
+    return Err(format!("{choice:?} is not a sync decision"));
+  }
+  let value = value.map(jsonb).unwrap_or_else(|| "NULL".to_string());
+  query(pg, &format!("SELECT sync.resolve({}, {}, {}, {value})", literal(tbl), jsonb(key), literal(choice))).map(|_| ())
+}
+
+async fn call(dhp: &Dhp<'_>, path: &str, body: Option<&Value>) -> Result<Value, SyncError> {
+  let url = format!("{}{path}", dhp.site_url);
+  let request = match body {
+    Some(body) => dhp.http.post(url).json(body),
+    None => dhp.http.get(url),
+  };
+  let response = request
+    .bearer_auth(dhp.token)
+    .timeout(HTTP_TIMEOUT)
+    .send()
+    .await
+    .map_err(|e| SyncError::Unavailable(format!("the DHP did not answer: {e}")))?;
+  let status = response.status();
+  if status.is_success() {
+    return response.json().await.map_err(|e| SyncError::Failed(format!("the DHP's answer was unreadable: {e}")));
+  }
+  if status == StatusCode::UNAUTHORIZED {
+    return Err(SyncError::Unauthorized);
+  }
+  let detail = response.json::<Value>().await.ok().and_then(|b| b.get("error").and_then(Value::as_str).map(str::to_string));
+  Err(match status {
+    StatusCode::NOT_FOUND => SyncError::Unavailable("this DHP does not sync operator data".to_string()),
+    StatusCode::SERVICE_UNAVAILABLE => {
+      SyncError::Unavailable(detail.unwrap_or_else(|| "the DHP cannot sync right now".to_string()))
+    }
+    other => SyncError::Failed(detail.unwrap_or_else(|| format!("the DHP answered HTTP {other}"))),
+  })
+}
+
+/// The DHP database's sync id and its current horizon.
+pub async fn hello(dhp: &Dhp<'_>) -> Result<(String, String), SyncError> {
+  let answer = call(dhp, "/v1/sync/hello", None).await?;
+  match (answer.get("db_id").and_then(Value::as_str), answer.get("horizon").and_then(Value::as_str)) {
+    (Some(db_id), Some(horizon)) => Ok((db_id.to_string(), horizon.to_string())),
+    _ => Err(SyncError::Failed("the DHP's sync greeting is incomplete".to_string())),
+  }
+}
+
+/// Push this PC's edits; then, when `pull`, apply everything new on the DHP.
+/// Postgres must be running and the local Engine API stopped.
+pub async fn run(pg: &Pg, dhp: &Dhp<'_>, install_id: &str, pull: bool) -> Result<Report, SyncError> {
+  let mut report = Report::default();
+  loop {
+    let items = blocking(pg, |pg| query(pg, &format!("SELECT sync.outbox({PUSH_BATCH})"))).await?;
+    let count = items.as_array().map_or(0, Vec::len);
+    if count == 0 {
+      break;
+    }
+    let answer = call(dhp, "/v1/sync/push", Some(&json!({ "install_id": install_id, "changes": items }))).await?;
+    let recorded =
+      blocking(pg, move |pg| query(pg, &format!("SELECT sync.pushed({}, {})", jsonb(&items), jsonb(&answer)))).await?;
+    let accepted = recorded.get("accepted").and_then(Value::as_u64).unwrap_or(0);
+    report.sent += count;
+    report.accepted += accepted;
+    // Records waiting on the operator go out every time: stop once a batch moves nothing.
+    if count < PUSH_BATCH || accepted == 0 {
+      break;
+    }
+  }
+
+  if pull {
+    let since = blocking(pg, |pg| query(pg, "SELECT to_jsonb((SELECT value FROM sync.state WHERE name = 'since'))")).await?;
+    let since = since.as_str().ok_or_else(|| SyncError::Failed("this PC's sync cursor is missing".to_string()))?;
+    let mut cursor = json!({ "since": since });
+    let mut changes: Vec<Value> = Vec::new();
+    let db_id = loop {
+      let page = call(dhp, "/v1/sync/pull", Some(&json!({ "cursor": cursor, "limit": PULL_PAGE }))).await?;
+      if let Some(list) = page.get("changes").and_then(Value::as_array) {
+        changes.extend(list.iter().cloned());
+      }
+      cursor = page.get("next").cloned().unwrap_or(Value::Null);
+      if page.get("done").and_then(Value::as_bool).unwrap_or(true) {
+        break page.get("db_id").and_then(Value::as_str).unwrap_or_default().to_string();
+      }
+    };
+    let next = cursor.get("since").and_then(Value::as_str).unwrap_or_default().to_string();
+    if next.is_empty() {
+      return Err(SyncError::Failed("the DHP ended a pull without a cursor".to_string()));
+    }
+    report.received = changes.len();
+    let batch = Value::Array(changes);
+    blocking(pg, move |pg| {
+      query(pg, &format!("SELECT sync.apply_pull({}, {}, {})", jsonb(&batch), literal(&next), literal(&db_id)))
+    })
+    .await?;
+  }
+
+  report.status = blocking(pg, status).await?;
+  Ok(report)
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  #[test]
+  fn literals_double_single_quotes() {
+    assert_eq!(literal("it's"), "'it''s'");
+  }
+
+  #[test]
+  fn jsonb_literals_pick_a_tag_the_payload_does_not_contain() {
+    let plain = json!({ "name": "Alpha" });
+    assert_eq!(jsonb(&plain), "$j0${\"name\":\"Alpha\"}$j0$::jsonb");
+    let sneaky = json!({ "name": "$j0$; DROP TABLE x; --" });
+    let quoted = jsonb(&sneaky);
+    assert!(quoted.starts_with("$j1$") && quoted.ends_with("$j1$::jsonb"), "{quoted}");
+  }
+}