launcher/src-tauri/src/sync.rs

//! 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. `version` is reported
/// so the DHP's install list can say which version each PC is on.
pub async fn run(pg: &Pg, dhp: &Dhp<'_>, install_id: &str, version: &str, pull: bool) -> Result<Report, SyncError> {
  let mut report = Report::default();
  // Say which DHP database these edits were made against. A PC carried over from another
  // one (a test copy, say) would otherwise apply its edits, deletions included, here.
  let joined = blocking(pg, joined_db_id).await?.unwrap_or_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, "version": version, "db_id": joined, "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, "install_id": install_id, "version": version })),
      )
      .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}");
  }
}