diff --git a/crates/odian-packs/src/manifest.rs b/crates/odian-packs/src/manifest.rs index 3a2f37f..f5ab27a 100644 --- a/crates/odian-packs/src/manifest.rs +++ b/crates/odian-packs/src/manifest.rs @@ -142,6 +142,20 @@ fn is_name(s: &str) -> bool { (1..=128).contains(&s.len()) && s.bytes().all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'.' | b'-')) } +/// Lower-case hyphenated uuid, as Postgres prints gen_random_uuid(). +fn is_uuid(s: &str) -> bool { + s.len() == 36 + && s.bytes().enumerate().all(|(i, b)| match i { + 8 | 13 | 18 | 23 => b == b'-', + _ => b.is_ascii_digit() || (b'a'..=b'f').contains(&b), + }) +} + +/// A Postgres xid8 in decimal. +fn is_xid(s: &str) -> bool { + (1..=20).contains(&s.len()) && s.bytes().all(|b| b.is_ascii_digit()) +} + fn is_snapshot(s: &str) -> bool { let b = s.as_bytes(); b.len() == 16 @@ -216,6 +230,13 @@ pub fn validate(manifest: &Value) -> Result<()> { .ok_or_else(|| Error::manifest("db.snapshot must look like 20260915T101500Z"))?; let files = entries(db.get("files"), "db.files")?; require(!files.is_empty(), || "db.files must not be empty".into())?; + if let Some(sync) = db.get("sync") { + let db_id = sync.get("db_id").and_then(Value::as_str).is_some_and(is_uuid); + let horizon = sync.get("horizon").and_then(Value::as_str).is_some_and(is_xid); + require(db_id && horizon, || { + "db.sync must hold the DHP's db_id (a uuid) and a horizon (a transaction id)".into() + })?; + } let prefix = format!("db/{snapshot}/"); let mut seen: HashSet = HashSet::new(); for (i, entry) in files.iter().enumerate() { @@ -328,6 +349,18 @@ pub struct DbSection { pub schema_version: u64, pub snapshot: String, pub files: Vec, + /// Present when the snapshot comes from a DHP running operator-data sync. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub sync: Option, +} + +/// Where an install restored from this snapshot starts syncing. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct DbSync { + /// The DHP database's sync id: an install only ever syncs with the database it came from. + pub db_id: String, + /// Oldest transaction still running when the snapshot was taken; the first pull starts there. + pub horizon: String, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] diff --git a/docs/CONTRACT.md b/docs/CONTRACT.md index 5fad931..b5615c4 100644 --- a/docs/CONTRACT.md +++ b/docs/CONTRACT.md @@ -16,6 +16,7 @@ type Phase = | "signing_in" // browser sign-in in progress | "installing" // downloading + verifying a version | "restoring" // loading the database snapshot into local Postgres + | "syncing" // exchanging operator data with the DHP (ODIAN is not running) | "launching" // starting Postgres → API → tile server → Odian.exe | "running" // Odian.exe is open | "error"; // the last long operation failed; see `error` @@ -34,6 +35,28 @@ interface Status { phase: Phase; error: string | null; // set with phase "error"; cleared by the next command or `refresh` install_dir: string; // e.g. "D:\\ODIAN" + sync: SyncStatus | null; // counts from the last sync this session; null before one + live: boolean | null; // while ODIAN runs: true = its data comes from the DHP, + // false = from this PC. null when it is not running, or + // the DHP offers no Engine API to forward to +} + +interface SyncStatus { + pending: number; // local edits the DHP has not taken yet + conflicts: number; // records both sides changed: the operator decides + rejected: number; // edits the DHP refused (another operator's record, ...) + waiting: number; // DHP changes this PC could not apply yet +} + +interface SyncConflict { + tbl: string; // database table, e.g. "db_ground_poi" + key: Record; // the record's key columns + kind: "conflict" | "rejected"; + fields: string[]; // clashing fields; "*exists": one side deleted it; "*value": a whole-value clash + base: unknown; // the DHP value before this PC's edit + mine: unknown; // this PC's value (null = deleted here) + theirs: unknown; // the DHP's value (null = deleted there) + reason: string | null; // why the DHP refused it (rejected only) } interface Settings { @@ -83,12 +106,34 @@ interface LogEvent { level: "info" | "warn" | "error"; message: string; at: numb | `get_settings` | — | `Settings` | | | `set_settings` | `{ settings: Settings }` | `Status` | | | `pick_folder` | — | `string \| null` | native folder picker | +| `sync_now` | — | `void` | pushes and pulls operator data now; only while ODIAN is stopped (phase `syncing` meanwhile) | +| `get_sync_conflicts` | — | `SyncConflict[]` | what waits for a decision, oldest first; starts the local Postgres just for the read | +| `resolve_sync_conflict` | `{ tbl, key, choice: "theirs" \| "mine" \| "value" \| "dismiss", value?: unknown }` | `SyncConflict[]` | records one decision and returns what is left. `theirs` takes the DHP's version; `mine` keeps this PC's (the next sync sends it); `value` keeps a field-by-field pick; `dismiss` keeps a refused edit on this PC only. Links and folder placements held with a record follow its decision | Errors reject the promise with a string message. Planned, not yet available: the size of `available` for the Install button, and inspecting a drive (its version and size) before installing from it. +## Live mode + +When the DHP advertises an Engine API (`engine_api_url` in its `/v1/health`), the +launcher starts the local API pointed at it and keeps a token file fresh for it. ODIAN +always talks to the local API; while the DHP answers, that API forwards the shared-data +routes there, so operators see each other's work and tracks are live. The moment the DHP +stops answering, the same API serves this PC's own database instead — no restart, and the +request that discovers the outage is already served locally. Coming back, the launcher +first sends what was edited offline, then hands the token back and ODIAN is live again. +`status.live` says which of the two is happening. + +## Operator-data sync + +When the installed version ships the sync schema (`versions\\sync\*.sql`) and the +DHP is reachable, `launch` first syncs (push, then pull), then starts the stack. When +ODIAN closes, the launcher pushes that session's work. Offline, edits stay on this PC. +A sync that cannot finish is logged and never blocks the launch. Conflicts and refused +edits wait in the local database until `resolve_sync_conflict`. + ## Events (`listen`) | Event | Payload | When | diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs index deed6dd..9f0b6b3 100644 --- a/src-tauri/src/commands.rs +++ b/src-tauri/src/commands.rs @@ -7,6 +7,7 @@ use tauri_plugin_opener::OpenerExt; use crate::launcher::{Launcher, Status}; use crate::settings::Settings; use crate::stack::StackEvent; +use crate::sync::Conflict; type CmdResult = Result; @@ -89,6 +90,29 @@ pub async fn get_stack(launcher: State<'_, Launcher>) -> CmdResult) -> CmdResult<()> { + launcher.sync_now(&app).await +} + +#[tauri::command] +pub async fn get_sync_conflicts(app: AppHandle, launcher: State<'_, Launcher>) -> CmdResult> { + launcher.sync_conflicts(&app).await +} + +/// `choice`: "theirs", "mine", "value" (with `value`) or "dismiss" (refused edits only). +#[tauri::command] +pub async fn resolve_sync_conflict( + app: AppHandle, + launcher: State<'_, Launcher>, + tbl: String, + key: serde_json::Value, + choice: String, + value: Option, +) -> CmdResult> { + launcher.resolve_sync_conflict(&app, tbl, key, choice, value).await +} + #[tauri::command] pub async fn get_settings(launcher: State<'_, Launcher>) -> CmdResult { Ok(launcher.settings().await) diff --git a/src-tauri/src/launcher.rs b/src-tauri/src/launcher.rs index 8bf9e78..a3b6352 100644 --- a/src-tauri/src/launcher.rs +++ b/src-tauri/src/launcher.rs @@ -1,6 +1,7 @@ //! The launcher's live state behind every command, and the `status` event. use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::Arc; @@ -15,7 +16,8 @@ use crate::localdb::{self, Pg}; use crate::settings::{normalize_url, Settings}; use crate::site::{self, SessionRefresh, TokenCheck}; use crate::stack::{self, Layout, RunningStack, StackConfig, StackEvent}; -use crate::state::{self, LauncherState}; +use crate::state::{self, LauncherState, SyncOrigin}; +use crate::sync::{self, SyncError, SyncStatus}; use crate::tokens::{self, SiteToken}; #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] @@ -26,6 +28,8 @@ pub enum Phase { SigningIn, Installing, Restoring, + /// Exchanging operator data with the DHP (ODIAN is not running meanwhile). + Syncing, Launching, Running, Error, @@ -50,6 +54,11 @@ pub struct Status { pub phase: Phase, pub error: Option, pub install_dir: String, + /// Operator-data sync counts from the last sync this session; None before one. + pub sync: Option, + /// While ODIAN runs: true when its data is coming from the DHP, false when from + /// this PC. None when ODIAN is not running, or the DHP offers no Engine API. + pub live: Option, } #[derive(Clone, Serialize)] @@ -90,6 +99,11 @@ struct Inner { token: Option, phase: Phase, error: Option, + sync: Option, + /// The DHP's Engine API, learned from its health answer; kept once seen so a launch + /// during a blip still knows where live mode will forward. + engine_url: Option, + live: Option, } impl Inner { @@ -111,6 +125,8 @@ impl Inner { phase: self.phase, error: self.error.clone(), install_dir: self.settings.install_dir.clone(), + sync: self.sync.clone(), + live: self.live, } } } @@ -140,7 +156,7 @@ impl Launcher { .map_err(|e| format!("cannot start the HTTP client: {e}"))?; Ok(Launcher { http, - inner: Mutex::new(Inner { settings, state, online: false, available: None, token, phase: Phase::Checking, error: None }), + inner: Mutex::new(Inner { settings, state, online: false, available: None, token, phase: Phase::Checking, error: None, sync: None, engine_url: None, live: None }), stack: Mutex::new(None), stack_states: std::sync::Mutex::new(Vec::new()), sign_in_cancel: Notify::new(), @@ -253,6 +269,10 @@ impl Launcher { let mut inner = self.inner.lock().await; inner.online = probe.online; inner.available = probe.version; + // Keep the last known address while offline: it is where live mode resumes. + if probe.engine_url.is_some() { + inner.engine_url = probe.engine_url; + } } // Offline nothing is checked or forgotten: the remembered operator still launches. if let (true, Some(token)) = (probe.online, token) { @@ -604,6 +624,8 @@ impl Launcher { restored.map_err(|e| format!("The database restore failed: {e}"))?; let mut inner = self.inner.lock().await; inner.state.db_snapshot = Some(db.snapshot.clone()); + inner.state.sync_origin = + db.sync.as_ref().map(|origin| SyncOrigin { db_id: origin.db_id.clone(), horizon: origin.horizon.clone() }); inner.state.save(install_dir)?; } @@ -682,6 +704,7 @@ impl Launcher { pg_password: inner.state.pg_password.clone(), install_id: inner.state.install_id.clone(), tile_sets, + upstream_url: inner.engine_url.clone(), } }; if let Ok(mut states) = self.stack_states.lock() { @@ -697,6 +720,19 @@ impl Launcher { } async fn run_stack(&self, app: &AppHandle, config: StackConfig) { + // Take in the other operators' work before the API loads it. Never blocks the launch. + self.set_phase(app, Phase::Syncing).await; + self.sync_run(app, &config.layout, &config.pg_password, &config.install_id, true).await; + self.set_phase(app, Phase::Launching).await; + + // Live mode: hand the API the DHP token before it starts, so a launch made online + // comes up on the DHP's data instead of switching a moment later. + let live_mode = config.upstream_url.is_some(); + if live_mode { + let token = self.site_access_token(app).await.ok(); + self.write_upstream_token(&config.layout, token.as_deref()); + } + let emit = |event: StackEvent| self.record_stack(app, event); let running = match stack::start(&self.http, &config, &emit).await { Ok(running) => running, @@ -709,14 +745,31 @@ impl Launcher { *self.stack.lock().await = Some(running); self.set_phase(app, Phase::Running).await; + let live_cancel = CancellationToken::new(); + if live_mode { + let handle = app.clone(); + let cancel = live_cancel.clone(); + let (layout, password, install_id) = + (config.layout.clone(), config.pg_password.clone(), config.install_id.clone()); + tauri::async_runtime::spawn(async move { + handle.state::().live_loop(&handle, layout, password, install_id, cancel).await; + }); + } + if let Some(odian) = odian { stack::wait_exit(&odian).await; } // Odian.exe closed. If Stop already took the stack, it stopped the rest too. + live_cancel.cancel(); + self.write_upstream_token(&config.layout, None); + self.inner.lock().await.live = None; let taken = self.stack.lock().await.take(); if let Some(mut running) = taken { running.stop(&emit).await; } + // Send this session's work while it is fresh; pulling waits for the next launch. + self.set_phase(app, Phase::Syncing).await; + self.sync_run(app, &config.layout, &config.pg_password, &config.install_id, false).await; self.finish(app, None).await; } @@ -729,6 +782,307 @@ impl Launcher { Ok(()) } + // ── operator-data sync ────────────────────────────────────────────────── + + /// Sync with the DHP while the local Engine API is not running. Never fails the caller: + /// offline, signed out or refused, the edits stay on this PC for the next connected run. + async fn sync_run(&self, app: &AppHandle, layout: &Layout, pg_password: &str, install_id: &str, pull: bool) { + let sql_dir = layout.sync_sql_dir(); + if !sync::schema_shipped(&sql_dir) { + return; // a version from before sync + } + let (online, site_url, origin) = { + let inner = self.inner.lock().await; + (inner.online, inner.settings.site_url.clone(), inner.state.sync_origin.clone()) + }; + if !online { + self.log(app, "info", "Offline: this PC's changes stay here until the next connected launch."); + return; + } + + let pg = Pg::new(layout, pg_password); + let db = pg.clone(); + let started = match tauri::async_runtime::spawn_blocking(move || -> Result { + let started = !db.running(); + if started { + db.start()?; + } + Ok(started) + }) + .await + { + Ok(Ok(started)) => started, + Ok(Err(message)) => { + self.log(app, "warn", format!("Sync skipped: {message}")); + return; + } + Err(e) => { + self.log(app, "warn", format!("Sync skipped: {e}")); + return; + } + }; + + match self.sync_with_dhp(app, &pg, &sql_dir, &site_url, install_id, origin, pull).await { + Ok(report) => { + let s = &report.status; + self.log( + app, + "info", + format!( + "Synced with the DHP: {} change(s) sent, {} received. {} waiting to send, {} conflict(s), {} refused.", + report.accepted, report.received, s.pending, s.conflicts, s.rejected + ), + ); + if s.conflicts + s.rejected > 0 { + self.log(app, "warn", "Some changes need your decision: review them under Sync."); + } + self.inner.lock().await.sync = Some(report.status); + } + Err(message) => self.log(app, "warn", format!("Sync did not complete: {message}")), + } + + if started { + let _ = tauri::async_runtime::spawn_blocking(move || pg.stop()).await; + } + self.emit(app).await; + } + + #[allow(clippy::too_many_arguments)] + async fn sync_with_dhp( + &self, + app: &AppHandle, + pg: &Pg, + sql_dir: &Path, + site_url: &str, + install_id: &str, + origin: Option, + pull: bool, + ) -> Result { + let token = self.site_access_token(app).await?; + let (db, dir) = (pg.clone(), sql_dir.to_path_buf()); + let joined = tauri::async_runtime::spawn_blocking(move || { + sync::load_schema(&db, &dir)?; + sync::joined_db_id(&db) + }) + .await + .map_err(|e| e.to_string())??; + + if joined.is_none() { + let (db_id, since) = match origin { + Some(origin) => (origin.db_id, origin.horizon), + None => { + let dhp = sync::Dhp { http: &self.http, site_url, token: &token }; + let greeting = sync::hello(&dhp).await.map_err(|e| e.to_string())?; + self.log(app, "warn", "This PC's database predates sync: it takes the DHP's changes from now on."); + greeting + } + }; + let db = pg.clone(); + tauri::async_runtime::spawn_blocking(move || sync::join(&db, &db_id, &since)).await.map_err(|e| e.to_string())??; + } + + let dhp = sync::Dhp { http: &self.http, site_url, token: &token }; + match sync::run(pg, &dhp, install_id, pull).await { + Err(SyncError::Unauthorized) => { + let token = self.renew(app).await?; + let dhp = sync::Dhp { http: &self.http, site_url, token: &token }; + sync::run(pg, &dhp, install_id, pull).await.map_err(|e| e.to_string()) + } + other => other.map_err(|e| e.to_string()), + } + } + + /// Give the local API the DHP token it forwards with — or take it away, which sends + /// the API back to this PC's data without restarting anything. + fn write_upstream_token(&self, layout: &Layout, token: Option<&str>) { + let path = stack::upstream_token_file(layout); + match token { + Some(token) => { + if let Err(e) = crate::fsutil::write_atomic(&path, token.as_bytes()) { + log::warn!("cannot hand the local API its DHP token: {e}"); + } + } + None => { + let _ = std::fs::remove_file(&path); + } + } + } + + /// Ask the local API where ODIAN's data is coming from right now. + async fn read_live_state(&self) { + let url = format!("http://127.0.0.1:{}/offline/status", stack::API_PORT); + let live = match self.http.get(url).timeout(Duration::from_secs(3)).send().await { + Ok(response) => response + .json::() + .await + .ok() + .and_then(|body| body.get("mode").and_then(|mode| mode.as_str()).map(|mode| mode == "live")), + Err(_) => None, + }; + self.inner.lock().await.live = live; + } + + /// While ODIAN runs against a DHP: keep its token fresh, notice the connection coming + /// and going, send this PC's offline work the moment it returns, and keep this PC's + /// own copy warm so an abrupt drop lands on fresh data rather than an old snapshot. + async fn live_loop( + &self, + app: &AppHandle, + layout: Layout, + pg_password: String, + install_id: String, + cancel: CancellationToken, + ) { + const TICK: Duration = Duration::from_secs(30); + const MIRROR_EVERY: Duration = Duration::from_secs(120); + let (site_url, mut was_online) = { + let inner = self.inner.lock().await; + (inner.settings.site_url.clone(), inner.online) + }; + let mut mirrored = Instant::now(); + loop { + tokio::select! { + _ = cancel.cancelled() => break, + _ = tokio::time::sleep(TICK) => {} + } + let probe = site::probe(&self.http, &site_url).await; + { + let mut inner = self.inner.lock().await; + inner.online = probe.online; + if probe.engine_url.is_some() { + inner.engine_url = probe.engine_url.clone(); + } + } + if !probe.online { + if was_online { + self.log(app, "warn", "The DHP is out of reach: ODIAN is working from this PC. Your changes stay here until it is back."); + } + was_online = false; + self.read_live_state().await; + self.emit(app).await; + continue; + } + let Ok(token) = self.site_access_token(app).await else { + self.read_live_state().await; + continue; + }; + if !was_online { + // Hold the API on this PC's data until this PC's offline work has reached the + // DHP: nothing may be edited there on top of edits that have not arrived yet. + self.write_upstream_token(&layout, None); + self.log(app, "info", "Back on the DHP: sending the changes made offline…"); + self.sync_run(app, &layout, &pg_password, &install_id, false).await; + mirrored = Instant::now(); + } + self.write_upstream_token(&layout, Some(&token)); + if !was_online { + self.log(app, "info", "ODIAN is back on the DHP's data."); + was_online = true; + } + if mirrored.elapsed() >= MIRROR_EVERY { + self.sync_run(app, &layout, &pg_password, &install_id, true).await; + mirrored = Instant::now(); + } + self.read_live_state().await; + self.emit(app).await; + } + } + + /// Sync on demand while ODIAN is stopped. + pub async fn sync_now(&self, app: &AppHandle) -> Result<(), String> { + let (layout, password, install_id) = { + let mut inner = self.inner.lock().await; + if !matches!(inner.phase, Phase::Idle | Phase::Error) { + return Err("ODIAN is busy: sync once it has stopped.".to_string()); + } + let version = inner.state.installed_version.clone().ok_or("ODIAN is not installed.")?; + if inner.state.pg_password.is_empty() || inner.state.db_snapshot.is_none() { + return Err("The local database was never set up. Reinstall from the DHP or a drive.".to_string()); + } + inner.phase = Phase::Syncing; + inner.error = None; + (Layout::new(&inner.settings.install_path(), &version), inner.state.pg_password.clone(), inner.state.install_id.clone()) + }; + self.emit(app).await; + self.sync_run(app, &layout, &password, &install_id, true).await; + self.finish(app, None).await; + Ok(()) + } + + /// Run `work` on this PC's database while ODIAN is stopped, starting Postgres just for it. + async fn with_local_db( + &self, + app: &AppHandle, + work: impl FnOnce(&Pg) -> Result + Send + 'static, + ) -> Result { + let (layout, password) = { + let mut inner = self.inner.lock().await; + if !matches!(inner.phase, Phase::Idle | Phase::Error) { + return Err("ODIAN is busy: review sync once it has stopped.".to_string()); + } + let version = inner.state.installed_version.clone().ok_or("ODIAN is not installed.")?; + if inner.state.pg_password.is_empty() { + return Err("The local database was never set up.".to_string()); + } + inner.phase = Phase::Syncing; + (Layout::new(&inner.settings.install_path(), &version), inner.state.pg_password.clone()) + }; + self.emit(app).await; + let pg = Pg::new(&layout, &password); + let result = tauri::async_runtime::spawn_blocking(move || { + let started = !pg.running(); + if started { + pg.start()?; + } + let result = work(&pg); + if started { + let _ = pg.stop(); + } + result + }) + .await + .map_err(|e| e.to_string()) + .and_then(|result| result); + { + let mut inner = self.inner.lock().await; + if inner.phase == Phase::Syncing { + inner.phase = Phase::Idle; + } + } + self.emit(app).await; + result + } + + /// What the operator has to decide, oldest first. + pub async fn sync_conflicts(&self, app: &AppHandle) -> Result, String> { + self + .with_local_db(app, |pg| match sync::joined_db_id(pg)? { + None => Ok(Vec::new()), + Some(_) => sync::conflicts(pg), + }) + .await + } + + /// Record one decision; returns what is left to decide. + pub async fn resolve_sync_conflict( + &self, + app: &AppHandle, + tbl: String, + key: serde_json::Value, + choice: String, + value: Option, + ) -> Result, String> { + let (left, status) = self + .with_local_db(app, move |pg| { + sync::resolve(pg, &tbl, &key, &choice, value.as_ref())?; + Ok((sync::conflicts(pg)?, sync::status(pg)?)) + }) + .await?; + self.inner.lock().await.sync = Some(status); + self.emit(app).await; + Ok(left) + } + /// The launcher is exiting: stop the stack cleanly rather than leave Postgres to crash recovery. pub async fn shutdown(&self) { if let Ok(slot) = self.install_cancel.lock() { diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 3a83160..922cdfd 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -10,6 +10,7 @@ mod settings; mod site; mod stack; mod state; +mod sync; mod tokens; use tauri::Manager; @@ -53,6 +54,9 @@ pub fn run() { commands::get_settings, commands::set_settings, commands::pick_folder, + commands::sync_now, + commands::get_sync_conflicts, + commands::resolve_sync_conflict, ]) .build(tauri::generate_context!()) .expect("error while building the launcher") diff --git a/src-tauri/src/localdb.rs b/src-tauri/src/localdb.rs index 8b92712..3292c2a 100644 --- a/src-tauri/src/localdb.rs +++ b/src-tauri/src/localdb.rs @@ -107,6 +107,11 @@ impl Pg { if self.running() { return Ok(()); } + // Another install's Postgres on the shared port would leave only "exit code 1" in the log. + if crate::stack::port_taken(PG_PORT) { + return Err("Another ODIAN install is already running on this PC: it holds the database port. Stop it in the launcher window that started it, then try again." + .to_string()); + } let mut cmd = self.tool("pg_ctl"); cmd .arg("start") @@ -151,6 +156,34 @@ impl Pg { Self::run(cmd, "psql") } + /// Run a script on the `odian` database through psql's stdin and return what it printed + /// (unaligned, tuples only). For sync, whose JSON payloads are too large for a command line. + pub fn script(&self, sql: &str) -> Result { + use std::io::Write; + let port = PG_PORT.to_string(); + let mut cmd = self.tool("psql"); + cmd + .env("PGCLIENTENCODING", "UTF8") + .args(["-X", "-q", "-A", "-t", "-v", "ON_ERROR_STOP=1", "-h", "127.0.0.1", "-p", port.as_str()]) + .args(["-U", PG_USER, "-d", PG_DATABASE, "-f", "-"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + let mut child = cmd.spawn().map_err(|e| format!("cannot run psql: {e}"))?; + let mut stdin = child.stdin.take().ok_or("cannot write to psql")?; + // Feed from a thread: psql filling its stdout pipe while we still write would deadlock. + let body = sql.as_bytes().to_vec(); + let writer = std::thread::spawn(move || stdin.write_all(&body)); + let out = child.wait_with_output().map_err(|e| format!("psql failed: {e}"))?; + let _ = writer.join(); + if out.status.success() { + return Ok(String::from_utf8_lossy(&out.stdout).into_owned()); + } + let stderr = String::from_utf8_lossy(&out.stderr); + let tail: String = stderr.chars().rev().take(1500).collect::>().into_iter().rev().collect(); + Err(format!("psql failed ({}): {}", out.status, tail.trim())) + } + /// Replace the `odian` database with a snapshot. Only for a PC whose restore never /// completed: anything an operator did in the old database is dropped. pub fn restore( @@ -225,24 +258,26 @@ impl Pg { .args(["-h", "127.0.0.1", "-p", port.as_str(), "-U", PG_USER, "-d", PG_DATABASE]) .args(["--no-owner", "--no-privileges", "--exit-on-error"]); if pre_data { - cmd.arg("-L").arg(self.pre_data_list(&dump)?); + // A fresh database already has schema public; --exit-on-error would stop on the duplicate. + let drop = |line: &str| line.contains(" SCHEMA - public ") || line.contains(" COMMENT - SCHEMA public "); + cmd.arg("-L").arg(self.toc_without(&dump, "pre-data.list", &drop)?); + } else if file.section == "post-data" { + // A DHP running sync has marking triggers on its tables; they call sync.mark(), which + // this database gets only when the launcher loads the version's sync schema. + let drop = |line: &str| line.contains(" TRIGGER ") && line.contains(" sync_mark_"); + cmd.arg("-L").arg(self.toc_without(&dump, "post-data.list", &drop)?); } cmd.arg(&dump); Self::run(cmd, &format!("pg_restore {}", file.path)).map(|_| ()) } - /// The pre-data table of contents without `CREATE SCHEMA public`: a fresh database - /// already has that schema, and --exit-on-error would stop on the duplicate. - fn pre_data_list(&self, dump: &Path) -> Result { + /// The dump's table of contents without the entries `drop` names, for `pg_restore -L`. + fn toc_without(&self, dump: &Path, list_name: &str, drop: &dyn Fn(&str) -> bool) -> Result { let mut cmd = self.tool("pg_restore"); cmd.arg("-l").arg(dump); let listing = Self::run(cmd, "pg_restore -l")?; - let kept: String = listing - .lines() - .filter(|line| !(line.contains(" SCHEMA - public ") || line.contains(" COMMENT - SCHEMA public "))) - .map(|line| format!("{line}\n")) - .collect(); - let path = self.logs.join("pre-data.list"); + let kept: String = listing.lines().filter(|line| !drop(line)).map(|line| format!("{line}\n")).collect(); + let path = self.logs.join(list_name); std::fs::write(&path, kept).map_err(|e| format!("cannot write {}: {e}", path.display()))?; Ok(path) } diff --git a/src-tauri/src/site.rs b/src-tauri/src/site.rs index 569719e..581f788 100644 --- a/src-tauri/src/site.rs +++ b/src-tauri/src/site.rs @@ -11,10 +11,13 @@ pub struct Probe { pub online: bool, /// The site's CURRENT version. pub version: Option, + /// The DHP's Engine API, as this PC can reach it: where live mode forwards while + /// online. Absent when the DHP offers none, and this PC then serves its own data. + pub engine_url: Option, } pub async fn probe(http: &reqwest::Client, site_url: &str) -> Probe { - let offline = Probe { online: false, version: None }; + let offline = Probe { online: false, version: None, engine_url: None }; if site_url.is_empty() { return offline; } @@ -25,10 +28,8 @@ pub async fn probe(http: &reqwest::Client, site_url: &str) -> Probe { return offline; } let body: serde_json::Value = response.json().await.unwrap_or_default(); - Probe { - online: true, - version: body.get("version").and_then(|v| v.as_str()).map(str::to_string), - } + let text = |key: &str| body.get(key).and_then(|v| v.as_str()).filter(|s| !s.is_empty()).map(str::to_string); + Probe { online: true, version: text("version"), engine_url: text("engine_api_url") } } pub enum TokenCheck { diff --git a/src-tauri/src/stack.rs b/src-tauri/src/stack.rs index a3a5b94..3a15ca0 100644 --- a/src-tauri/src/stack.rs +++ b/src-tauri/src/stack.rs @@ -115,6 +115,10 @@ impl Layout { pub fn pg_data(&self) -> PathBuf { self.data_dir().join("pg") } + /// The operator-data sync schema this version ships (`sync\*.sql`); absent before M3. + pub fn sync_sql_dir(&self) -> PathBuf { + self.version_dir().join("sync") + } fn python(&self) -> PathBuf { self.app_file(PYTHON_EXE) } @@ -140,6 +144,15 @@ pub struct StackConfig { pub install_id: String, /// (style, dataset) of every tile set the installed version carries. pub tile_sets: Vec<(String, String)>, + /// The DHP's Engine API (from its health answer). With one, the local API serves + /// the DHP's data while it is reachable; without, this PC always serves its own. + pub upstream_url: Option, +} + +/// Where the launcher keeps the DHP token the local API forwards with. A file, not an +/// env var: the launcher rewrites it whenever it renews, without restarting the stack. +pub fn upstream_token_file(layout: &Layout) -> PathBuf { + layout.state_dir().join("api").join("upstream.token") } /// A process started from a clean environment, in `cwd`, with `search_path` first on PATH. @@ -223,7 +236,26 @@ impl RunningStack { } } +/// Is something already listening on 127.0.0.1:`port`? +pub fn port_taken(port: u16) -> bool { + std::net::TcpListener::bind(("127.0.0.1", port)).is_err() +} + +/// Every install on this PC uses the same three loopback ports, so only one runs at a time. +/// Said plainly here, rather than as "Postgres did not start" once the ports collide. +fn check_ports_free() -> Result<(), String> { + for (port, what) in [(PG_PORT, "the database"), (API_PORT, "the Engine API"), (TILES_PORT, "the tile server")] { + if port_taken(port) { + return Err(format!( + "Another ODIAN install is already running on this PC ({what} holds 127.0.0.1:{port}). Stop it in the launcher window that started it, then launch again." + )); + } + } + Ok(()) +} + pub async fn start(http: &reqwest::Client, config: &StackConfig, emit: Emit<'_>) -> Result { + check_ports_free()?; prepare(config)?; let mut running = RunningStack { job: job::Job::new()?, @@ -413,7 +445,7 @@ fn datasets(config: &StackConfig) -> serde_json::Value { fn api_env(config: &StackConfig) -> Vec<(&'static str, String)> { let api = config.layout.state_dir().join("api"); let logs = api.join("logs"); - vec![ + let mut env = vec![ ("ODIAN_API_BIND", "127.0.0.1".to_string()), ("ODIAN_API_PORT", API_PORT.to_string()), ("ODIAN_PG_INI", text(api.join("pg.ini"))), @@ -428,7 +460,15 @@ fn api_env(config: &StackConfig) -> Vec<(&'static str, String)> { ("ODIAN_API_LOGDIR", text(logs.clone())), ("ODIAN_API_LOG", text(logs.join("odian_api.log"))), ("ODIAN_API_THREAD_STATE", text(logs.join("thread_state.txt"))), - ] + ]; + // Live mode: with a DHP to forward to, the local API serves the DHP's data while it + // answers, and this PC's own the moment it stops. The token file may not exist yet + // (signed out, or offline) — the API simply stays local until it does. + if let Some(url) = &config.upstream_url { + env.push(("ODIAN_UPSTREAM_URL", url.clone())); + env.push(("ODIAN_UPSTREAM_TOKEN_FILE", text(upstream_token_file(&config.layout)))); + } + env } fn tiles_env(config: &StackConfig) -> Vec<(&'static str, String)> { @@ -504,6 +544,14 @@ pub fn installed_tile_sets(layout: &Layout) -> Result, Str #[cfg(test)] mod tests { + #[test] + fn a_bound_port_reads_as_taken() { + let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).expect("bind"); + let port = listener.local_addr().expect("addr").port(); + assert!(super::port_taken(port), "a port someone is listening on must read as taken"); + drop(listener); + } + use super::*; fn config(tile_sets: &[(&str, &str)]) -> StackConfig { @@ -514,6 +562,7 @@ mod tests { pg_password: "pw".to_string(), install_id: "install".to_string(), tile_sets: tile_sets.iter().map(|(s, d)| (s.to_string(), d.to_string())).collect(), + upstream_url: None, } } @@ -521,6 +570,19 @@ mod tests { env.iter().find(|(k, _)| *k == key).map(|(_, v)| v.as_str()) } + #[test] + fn the_api_is_told_about_the_dhp_only_when_there_is_one() { + let alone = config(&[]); + assert_eq!(value(&api_env(&alone), "ODIAN_UPSTREAM_URL"), None, "no DHP: this PC serves its own data"); + assert_eq!(value(&api_env(&alone), "ODIAN_UPSTREAM_TOKEN_FILE"), None); + + let mut live = config(&[]); + live.upstream_url = Some("http://dhp.example:30684".to_string()); + let env = api_env(&live); + assert_eq!(value(&env, "ODIAN_UPSTREAM_URL"), Some("http://dhp.example:30684")); + assert!(value(&env, "ODIAN_UPSTREAM_TOKEN_FILE").is_some_and(|path| path.ends_with("upstream.token"))); + } + #[test] fn layout_resolves_manifest_paths_inside_the_version() { let layout = Layout::new(Path::new(r"D:\ODIAN"), "0.52.0"); diff --git a/src-tauri/src/state.rs b/src-tauri/src/state.rs index e71a74e..9321c1d 100644 --- a/src-tauri/src/state.rs +++ b/src-tauri/src/state.rs @@ -30,6 +30,18 @@ pub struct LauncherState { /// database holds no operator work yet and may be (re)created from a snapshot. #[serde(default)] pub db_snapshot: Option, + /// Where that snapshot starts syncing (the manifest's `db.sync`). None for a snapshot + /// from before sync: its first sync then starts from the DHP's state at that moment. + #[serde(default)] + pub sync_origin: Option, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct SyncOrigin { + /// The DHP database the snapshot came from. + pub db_id: String, + /// The first transaction the snapshot may be missing. + pub horizon: String, } fn state_dir(install_dir: &Path) -> PathBuf { diff --git a/src-tauri/src/sync.rs b/src-tauri/src/sync.rs new file mode 100644 index 0000000..6938e3f --- /dev/null +++ b/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\\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, + pub base: Option, + pub mine: Option, + pub theirs: Option, + pub reason: Option, +} + +#[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 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 { + 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( + pg: &Pg, + work: impl FnOnce(&Pg) -> Result + Send + 'static, +) -> Result { + 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, 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 { + 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, 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 { + 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::().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 { + 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 = 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}"); + } +} diff --git a/src/App.svelte b/src/App.svelte index b69ce64..f3187a9 100644 --- a/src/App.svelte +++ b/src/App.svelte @@ -6,6 +6,7 @@ import KeyArt from "./components/KeyArt.svelte"; import NewsPanel from "./components/NewsPanel.svelte"; import SettingsDrawer from "./components/SettingsDrawer.svelte"; + import SyncPanel from "./components/SyncPanel.svelte"; import StatusBar from "./components/StatusBar.svelte"; import TitleBar from "./components/TitleBar.svelte"; import { launcher } from "./lib/launcher.svelte"; @@ -31,6 +32,10 @@
{/if} + {#if launcher.syncOpen} +
+ {/if} + @@ -83,6 +88,14 @@ width: min(640px, calc(100% - 64px)); } + .sync-slot { + position: absolute; + z-index: 31; + right: 32px; + bottom: calc(var(--statusbar-h) + 10px); + width: min(560px, calc(100% - 64px)); + } + @media (max-width: 1140px) { .hub { grid-template-columns: minmax(300px, 360px) minmax(0, 1fr) minmax(340px, 380px); @@ -92,6 +105,9 @@ .activity-slot { left: 24px; } + .sync-slot { + right: 24px; + } } @media (max-height: 700px) { diff --git a/src/components/ActionPanel.svelte b/src/components/ActionPanel.svelte index f352986..283fb11 100644 --- a/src/components/ActionPanel.svelte +++ b/src/components/ActionPanel.svelte @@ -60,6 +60,8 @@ return { label: s?.installed ? "Updating" : "Installing", tone: "accent", pulse: true, meta: stage }; case "restoring": return { label: "Restoring database", tone: "accent", pulse: true, meta: stage }; + case "syncing": + return { label: "Syncing with the DHP", tone: "accent", pulse: true, meta: version(s?.installed) }; case "update": return { label: "Update required", tone: "warn", pulse: false, meta: `${s?.installed} → ${s?.available}` }; case "launch": @@ -165,6 +167,21 @@ Offline: ODIAN {s?.installed} runs on local data and map tiles. {/if}

+ {#if launcher.syncAttention > 0} +

+ + + {launcher.syncAttention} + {launcher.syncAttention === 1 ? "change needs" : "changes need"} your decision. + + +

+ {:else if s?.online || s?.sync} +

+ Operator data syncs with {site} before every launch. + +

+ {/if} {:else if action === "installing"}
@@ -211,6 +228,14 @@ {#if p?.stage === "restore" && p.message}

{p.message}

{/if}

This can take a few minutes. Keep the launcher open until it finishes.

+ {:else if action === "syncing"} +
+
+ Exchanging operator data +
+ +
+

Sending this PC's changes to {site} and taking in the other operators' work.

{:else if action === "stack"} + {/if} {s.install_dir} @@ -122,6 +137,28 @@ font-weight: 500; } + .sync-item { + border-top: 0; + border-bottom: 0; + border-left: 0; + background: transparent; + color: inherit; + font: inherit; + transition: + background-color var(--dur-fast) var(--ease), + color var(--dur-fast) var(--ease); + } + .sync-item:hover { + background: var(--surface-2); + color: var(--ink); + } + .sync-item:focus-visible { + outline-offset: -2px; + } + .item strong.review { + color: var(--warn); + } + .dir { flex-shrink: 1; } diff --git a/src/components/SyncPanel.svelte b/src/components/SyncPanel.svelte new file mode 100644 index 0000000..90d86bb --- /dev/null +++ b/src/components/SyncPanel.svelte @@ -0,0 +1,316 @@ + + + + +
+
+

Sync

+ {#if launcher.conflicts?.length}{launcher.conflicts.length}{/if} + + + +
+ + {#if s?.sync} +

+ {s.sync.pending} waiting to send + {#if s.sync.waiting > 0}{s.sync.waiting} DHP changes not applied yet{/if} +

+ {/if} + + {#if launcher.locked} +

Decisions can be made once ODIAN has stopped.

+ {:else if launcher.conflicts === null} +

{launcher.busy === "conflicts" ? "Reading this PC's database…" : "Nothing loaded yet."}

+ {:else if launcher.conflicts.length === 0} +

Nothing needs your decision. Operator data syncs before every launch and after ODIAN closes.

+ {:else} +
    + {#each launcher.conflicts as c (c.tbl + JSON.stringify(c.key))} +
  1. +
    + {kind(c)} + {title(c)} + {c.kind === "rejected" ? "Refused" : "Conflict"} +
    +

    {explain(c)}

    + {#if rows(c).length} +
    + + + + + + {#each rows(c) as row (row.field)} + + + + + + {/each} + +
    FieldThis PCDHP
    {row.field}{row.mine}{row.theirs}
    +
    + {/if} +
    + {#each choices(c) as [choice, label] (choice)} + + {/each} +
    +
    2. + {/each} +
+ {/if} +
+ + diff --git a/src/lib/backend.ts b/src/lib/backend.ts index 7d68ff0..4b227f8 100644 --- a/src/lib/backend.ts +++ b/src/lib/backend.ts @@ -13,6 +13,7 @@ import type { SignInResult, StackEvent, Status, + SyncConflict, Unlisten, } from "./types"; @@ -52,6 +53,10 @@ function createTauriBackend(): Backend { getSettings: () => invoke("get_settings"), setSettings: (settings) => invoke("set_settings", { settings }), pickFolder: () => invoke("pick_folder"), + syncNow: () => invoke("sync_now"), + getSyncConflicts: () => invoke("get_sync_conflicts"), + resolveSyncConflict: (tbl, key, choice, value) => + invoke("resolve_sync_conflict", { tbl, key, choice, value: value === undefined ? null : value }), onStatus: (handler) => subscribe("status", handler), onInstallProgress: (handler) => subscribe("install-progress", handler), diff --git a/src/lib/launcher.svelte.ts b/src/lib/launcher.svelte.ts index a69d486..217b029 100644 --- a/src/lib/launcher.svelte.ts +++ b/src/lib/launcher.svelte.ts @@ -4,6 +4,7 @@ import { backend } from "./backend"; import { errorText, hostOf } from "./format"; import { SIGN_IN_CANCELLED } from "./types"; import type { + ConflictChoice, InstallProgress, LogEvent, LogLevel, @@ -12,6 +13,7 @@ import type { StackComponent, StackEvent, Status, + SyncConflict, Unlisten, } from "./types"; @@ -24,6 +26,7 @@ export type ActionKind = | "signed_out_offline" | "installing" | "restoring" + | "syncing" | "update" | "launch" | "stack" @@ -45,6 +48,8 @@ export function deriveAction(s: Status | null): ActionKind { return "installing"; case "restoring": return "restoring"; + case "syncing": + return "syncing"; case "launching": case "running": return "stack"; @@ -75,6 +80,7 @@ const LOCKED_PHASES: ReadonlySet = new Set([ "signing_in", "installing", "restoring", + "syncing", "launching", "running", ]); @@ -104,6 +110,10 @@ class Launcher { busy = $state(null); refreshing = $state(false); stopping = $state(false); + /** The sync review panel. */ + syncOpen = $state(false); + /** Decisions waiting in this PC's database; null until the panel loads them. */ + conflicts = $state(null); notes = $state({ key: null, loading: false, @@ -117,6 +127,8 @@ class Launcher { /** Settings can't change while something is installing or running. */ locked = $derived(this.status ? LOCKED_PHASES.has(this.status.phase) : false); siteName = $derived(this.status ? (this.status.site.name ?? hostOf(this.status.site.url)) : ""); + /** Conflicts plus refused edits from the last sync: what needs the operator. */ + syncAttention = $derived(this.status?.sync ? this.status.sync.conflicts + this.status.sync.rejected : 0); notesVersion = $derived( this.status ? (this.status.online ? (this.status.available ?? this.status.installed) : this.status.installed) : null, ); @@ -339,6 +351,32 @@ class Launcher { }); } + syncNow() { + return this.run("sync", async () => { + await backend.syncNow(); + // The panel lists what the sync just found (or settled). + if (this.syncOpen) this.conflicts = await backend.getSyncConflicts(); + }); + } + + /** Opens the review panel and loads what waits for a decision. */ + async openSync() { + this.syncOpen = true; + await this.run("conflicts", async () => { + this.conflicts = await backend.getSyncConflicts(); + }); + } + + closeSync() { + this.syncOpen = false; + } + + resolveConflict(conflict: SyncConflict, choice: ConflictChoice, value?: unknown) { + return this.run("resolve", async () => { + this.conflicts = await backend.resolveSyncConflict(conflict.tbl, conflict.key, choice, value); + }); + } + async pickFolder(): Promise { try { return await backend.pickFolder(); diff --git a/src/lib/mock.ts b/src/lib/mock.ts index d11252f..730ef25 100644 --- a/src/lib/mock.ts +++ b/src/lib/mock.ts @@ -2,7 +2,7 @@ // `npm run dev`. Same shapes as docs/CONTRACT.md, driven by timers. // // ?scenario=first-run | update-required | ready | offline-installed | -// offline-not-installed | installing | running | error +// offline-not-installed | installing | running | sync-conflicts | error // &fail=sign_in | install | restore | launch | notes make that flow fail // &speed=4 run timers 4x faster // ¬es=no-roadmap notes without a Roadmap @@ -14,6 +14,7 @@ import { SIGN_IN_CANCELLED, type Backend, type InstallProgress, + type ConflictChoice, type InstallStage, type LogEvent, type LogLevel, @@ -24,6 +25,8 @@ import { type StackComponent, type StackEvent, type Status, + type SyncConflict, + type SyncStatus, type Unlisten, } from "./types"; @@ -131,6 +134,42 @@ const STAGE_DONE: Record = { finalize: "Install finalized", }; +const POI = "poi_0192f3a1c4e87b5d9a2e6f10b3c4d5e6"; +const TARGET = "tar_0192f3a2d91c7e4fa6b80c1d2e3f4a5b"; +const AREA = "are_0192f3a3e0b1c2d3e4f5a6b7c8d9e0f1"; +const MOCK_CONFLICTS: SyncConflict[] = [ + { + tbl: "db_ground_poi", + key: { poi_id: POI }, + kind: "conflict", + fields: ["name", "color"], + base: { poi_id: POI, name: "Relay hill", color: "red", lat: 48.61, lon: 2.35 }, + mine: { poi_id: POI, name: "Relay hill (north slope)", color: "orange", lat: 48.61, lon: 2.35 }, + theirs: { poi_id: POI, name: "Relay hill, confirmed", color: "yellow", lat: 48.61, lon: 2.35 }, + reason: null, + }, + { + tbl: "db_ground_target", + key: { target_id: TARGET }, + kind: "conflict", + fields: ["*exists"], + base: { target_id: TARGET, name: "Bridge 14", status_id: null }, + mine: null, + theirs: { target_id: TARGET, name: "Bridge 14", status_id: "st_damaged" }, + reason: null, + }, + { + tbl: "db_ground_area", + key: { area_id: AREA }, + kind: "rejected", + fields: [], + base: null, + mine: { area_id: AREA, name: "NFA Delta (wider)" }, + theirs: { area_id: AREA, name: "NFA Delta" }, + reason: "only the owner can change or delete this", + }, +]; + type Payloads = { status: Status; "install-progress": InstallProgress; @@ -173,6 +212,8 @@ function initialStatus(scenario: MockScenario): Status { phase: "idle", error: null, install_dir: DEFAULT_DIR, + sync: null, + live: null, }; switch (scenario) { case "first-run": @@ -193,7 +234,15 @@ function initialStatus(scenario: MockScenario): Status { Object.assign(s, { operator: OPERATOR, signed_in: true, phase: "installing" }); break; case "running": - Object.assign(s, { installed: CURRENT, operator: OPERATOR, signed_in: true, phase: "running" }); + Object.assign(s, { installed: CURRENT, operator: OPERATOR, signed_in: true, phase: "running", live: true }); + break; + case "sync-conflicts": + Object.assign(s, { + installed: CURRENT, + operator: OPERATOR, + signed_in: true, + sync: { pending: 1, conflicts: 2, rejected: 1, waiting: 0 }, + }); break; case "error": Object.assign(s, { @@ -225,6 +274,8 @@ class MockBackend implements Backend { private job = 0; private booted = false; private signInCancelled = false; + /** What waits for a decision in the mock "local database". */ + private conflicts: SyncConflict[]; /** Latest event per component, as `get_stack` reports it. */ private readonly stackStates = new Map(); private readonly listeners: Listeners = { @@ -241,6 +292,7 @@ class MockBackend implements Backend { private readonly notesVariant: NotesVariant, ) { this.status = initialStatus(scenario); + this.conflicts = scenario === "sync-conflicts" ? structuredClone(MOCK_CONFLICTS) : []; this.network = this.status.online; this.settings = { site_url: SITE_URL, mcp_url: MCP_URL, install_dir: DEFAULT_DIR }; @@ -475,6 +527,58 @@ class MockBackend implements Backend { return this.snapshot(); } + private syncCounts(pending: number): SyncStatus { + return { + pending, + conflicts: this.conflicts.filter((c) => c.kind === "conflict").length, + rejected: this.conflicts.filter((c) => c.kind === "rejected").length, + waiting: 0, + }; + } + + async syncNow(): Promise { + if (!this.status.online) throw "Your DHP is unreachable: changes stay on this PC until the next connected sync."; + if (this.status.phase !== "idle" && this.status.phase !== "error") throw "ODIAN is busy: sync once it has stopped."; + const sent = this.status.sync?.pending ?? 0; + this.commit({ phase: "syncing", error: null }); + await this.wait(1400); + const counts = this.syncCounts(0); + this.commit({ phase: "idle", sync: counts }); + this.log( + "info", + `Synced with the DHP: ${sent} change(s) sent, 4 received. 0 waiting to send, ${counts.conflicts} conflict(s), ${counts.rejected} refused.`, + ); + } + + async getSyncConflicts(): Promise { + await this.wait(300); + return structuredClone(this.conflicts); + } + + async resolveSyncConflict( + tbl: string, + key: Record, + choice: ConflictChoice, + ): Promise { + if (this.status.phase !== "idle" && this.status.phase !== "error") throw "ODIAN is busy: review sync once it has stopped."; + const index = this.conflicts.findIndex((c) => c.tbl === tbl && JSON.stringify(c.key) === JSON.stringify(key)); + if (index < 0) throw `Nothing to resolve for ${tbl}.`; + const conflict = this.conflicts[index]; + if (choice === "dismiss" && conflict.kind !== "rejected") throw `"dismiss" is not a valid decision for a conflict`; + await this.wait(250); + this.conflicts.splice(index, 1); + const pending = (this.status.sync?.pending ?? 0) + (choice === "mine" || choice === "value" ? 1 : 0); + this.commit({ sync: this.syncCounts(pending) }); + const outcome = + choice === "theirs" + ? "took the DHP's version" + : choice === "dismiss" + ? "kept the refused change on this PC only" + : "kept this PC's version; the next sync sends it"; + this.log("info", `${tbl}: ${outcome}`); + return structuredClone(this.conflicts); + } + async pickFolder(): Promise { await new Promise((resolve) => setTimeout(resolve, 450)); return "E:\\ODIAN"; diff --git a/src/lib/types.ts b/src/lib/types.ts index abebdc5..315463a 100644 --- a/src/lib/types.ts +++ b/src/lib/types.ts @@ -7,6 +7,7 @@ export type Phase = | "signing_in" // browser sign-in in progress | "installing" // downloading + verifying a version | "restoring" // loading the database snapshot into local Postgres + | "syncing" // exchanging operator data with the DHP (ODIAN is not running) | "launching" // starting Postgres → API → tile server → Odian.exe | "running" // Odian.exe is open | "error"; @@ -37,6 +38,44 @@ export interface Status { error: string | null; /** e.g. "D:\\ODIAN". */ install_dir: string; + /** Operator-data sync counts from the last sync this session; null before one. */ + sync: SyncStatus | null; + /** + * While ODIAN runs: true when its data comes from the DHP (live), false when from + * this PC. null when ODIAN is not running, or the DHP offers no Engine API. + */ + live: boolean | null; +} + +export interface SyncStatus { + /** Local edits the DHP has not taken yet. */ + pending: number; + /** Records both sides changed: the operator decides. */ + conflicts: number; + /** Edits the DHP refused (another operator's record, ...). */ + rejected: number; + /** DHP changes this PC could not apply yet. */ + waiting: number; +} + +/** + * theirs: take the DHP's version. mine: keep this PC's (the next sync sends it). + * value: keep a field-by-field pick. dismiss: keep a refused edit on this PC only. + */ +export type ConflictChoice = "theirs" | "mine" | "value" | "dismiss"; + +export interface SyncConflict { + /** Database table, e.g. "db_ground_poi". */ + tbl: string; + key: Record; + kind: "conflict" | "rejected"; + /** Clashing fields; "*exists": one side deleted it; "*value": changed differently as a whole. */ + fields: string[]; + base: unknown; + mine: unknown; + theirs: unknown; + /** Why the DHP refused it (rejected only). */ + reason: string | null; } export interface Settings { @@ -115,6 +154,7 @@ export const MOCK_SCENARIOS = [ "offline-not-installed", "installing", "running", + "sync-conflicts", "error", ] as const; @@ -152,6 +192,17 @@ export interface Backend { getSettings(): Promise; setSettings(settings: Settings): Promise; pickFolder(): Promise; + /** Push and pull operator data now; only while ODIAN is stopped. */ + syncNow(): Promise; + /** What the operator has to decide, oldest first. */ + getSyncConflicts(): Promise; + /** Records one decision; resolves with what is left to decide. */ + resolveSyncConflict( + tbl: string, + key: Record, + choice: ConflictChoice, + value?: unknown, + ): Promise; onStatus(handler: (status: Status) => void): Promise; onInstallProgress(handler: (progress: InstallProgress) => void): Promise;