notescraft/crates/engine/tests/sync_commands.rs

//! The sync commands and the worker, driven end to end.
//!
//! The server is a fake over [`Ledger`] (same helper as `tests/sync_driver.rs`), so no
//! network is needed; the one network test pairs against a local `TcpListener` stub that
//! speaks `POST /v1/devices`. The waits are shortened, so no test sleeps long.

use std::collections::HashMap;
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::{Duration, Instant};

use notescraft_engine::{
    Binding, Changes, PutReply, Registry, Remote, Session, SyncError, SyncResult, SyncWorker, Timings, load_binding, load_token, register_core, save_binding,
    save_token,
};
use notescraft_sync::{Ledger, Outcome, Write, conflict_path, hash_hex};
use notescraft_vault::Vault;
use serde_json::{Value, json};

/// A clock that never moves: the stamp of every conflict copy is the same day.
const FIXED_MS: u64 = 1_759_545_600_000;

fn fixed_clock() -> u64 {
    FIXED_MS
}

/// The waits a test wants: everything happens within a second or two.
fn short_timings() -> Timings {
    Timings {
        debounce: Duration::from_millis(50),
        poll: Duration::from_millis(300),
        backoff: Duration::from_millis(100),
        backoff_max: Duration::from_millis(200),
    }
}

/// Waits up to `secs` seconds for `check` to hold, polling in short steps.
fn wait_for(secs: u64, check: impl Fn() -> bool) -> bool {
    let deadline = Instant::now() + Duration::from_secs(secs);
    while Instant::now() < deadline {
        if check() {
            return true;
        }
        std::thread::sleep(Duration::from_millis(20));
    }
    check()
}

/// A vault in a fresh folder.
fn temp_vault(name: &str) -> (Vault, std::path::PathBuf) {
    let root = std::env::temp_dir().join(format!("nc-sync-cmds-{name}-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&root);
    std::fs::create_dir_all(&root).expect("temp folder");
    let vault = Vault::open(&root).expect("vault");
    (vault, root)
}

/// What can fail on the fake server, on top of a healthy one.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Failure {
    /// The server is unreachable.
    Never,
    /// The token is refused with this status.
    Status(u16),
}

/// The state the fake server keeps.
struct Inner {
    ledger: Ledger,
    blobs: HashMap<String, Vec<u8>>,
    changes_calls: usize,
    failure: Option<Failure>,
}

/// The server the device talks to.
struct Server {
    inner: Mutex<Inner>,
}

impl Server {
    fn new(failure: Option<Failure>) -> Arc<Server> {
        Arc::new(Server { inner: Mutex::new(Inner { ledger: Ledger::new(), blobs: HashMap::new(), changes_calls: 0, failure }) })
    }

    fn lock(&self) -> MutexGuard<'_, Inner> {
        self.inner.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    fn healthy(&self) {
        self.lock().failure = None;
    }

    fn changes_calls(&self) -> usize {
        self.lock().changes_calls
    }
}

/// The client of the fake server.
struct FakeRemote {
    server: Arc<Server>,
}

impl Remote for FakeRemote {
    fn changes(&self, since: u64, limit: usize, _wait: u64) -> SyncResult<Changes> {
        let mut inner = self.server.lock();
        inner.changes_calls += 1;
        match inner.failure {
            Some(Failure::Never) => Err(SyncError::Http("connection refused (the server is unreachable)".to_owned())),
            Some(Failure::Status(status)) => Err(SyncError::Status { status, body: "the token is not good".to_owned() }),
            None => Ok(inner.ledger.changes_since(since, limit)),
        }
    }

    fn get(&self, path: &str) -> SyncResult<Vec<u8>> {
        self.server.lock().blobs.get(path).cloned().ok_or_else(|| SyncError::Status { status: 404, body: path.to_owned() })
    }

    fn put(&self, path: &str, bytes: &[u8], base_seq: u64, mtime_ms: u64) -> SyncResult<PutReply> {
        let mut inner = self.server.lock();
        let outcome = inner
            .ledger
            .apply(Write { path: path.to_owned(), hash: Some(hash_hex(bytes)), size: bytes.len() as u64, mtime_ms, base_seq })
            .map_err(|e| SyncError::Status { status: 400, body: e.to_string() })?;
        match outcome {
            Outcome::Applied(record) => {
                inner.blobs.insert(path.to_owned(), bytes.to_vec());
                Ok(PutReply::Applied(record))
            }
            Outcome::Unchanged(record) => Ok(PutReply::Unchanged(record)),
            Outcome::Conflict(record) => Ok(PutReply::Conflict(record)),
        }
    }

    fn delete(&self, path: &str, base_seq: u64) -> SyncResult<PutReply> {
        let mut inner = self.server.lock();
        let outcome = inner
            .ledger
            .apply(Write { path: path.to_owned(), hash: None, size: 0, mtime_ms: 0, base_seq })
            .map_err(|e| SyncError::Status { status: 400, body: e.to_string() })?;
        Ok(match outcome {
            Outcome::Applied(record) => PutReply::Applied(record),
            Outcome::Unchanged(record) => PutReply::Unchanged(record),
            Outcome::Conflict(record) => PutReply::Conflict(record),
        })
    }
}

/// Whether the status file on disk shows `cycle` with `phase` and a scan of `pending`.
/// The file is written after every cycle, so a test that saw its cycle number knows the
/// in-memory status of that cycle is final.
fn file_is(root: &std::path::Path, cycle: u64, phase: &str, pending: usize) -> bool {
    let Ok(bytes) = std::fs::read(root.join(".notescraft/sync/status.json")) else { return false };
    let Ok(file) = serde_json::from_slice::<Value>(&bytes) else { return false };
    file.get("cycle") == Some(&json!(cycle)) && file.get("phase") == Some(&json!(phase)) && file.get("pending") == Some(&json!(pending))
}

/// A worker against the fake server, with waits short enough for the test.
fn worker_with(vault: &Vault, name: &str, failure: Option<Failure>) -> (SyncWorker, Arc<Server>) {
    let server = Server::new(failure);
    let worker = SyncWorker::start_with(vault, name, Box::new(FakeRemote { server: Arc::clone(&server) }), fixed_clock);
    worker.set_timings(short_timings());
    (worker, server)
}

/// A registry with the core commands and a session on a fresh vault folder.
fn fixture(name: &str) -> (Session, Registry<Session>, std::path::PathBuf) {
    let root = std::env::temp_dir().join(format!("nc-sync-cmds-{name}-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&root);
    std::fs::create_dir_all(&root).expect("temp folder");
    let mut reg = Registry::new();
    register_core(&mut reg).expect("register the core commands");
    let session = Session::open(&root).expect("open the session");
    (session, reg, root)
}

#[test]
fn worker_syncs_after_request_and_reports_counts() {
    let (vault, root) = temp_vault("worker-sync");
    vault.write("a.md", b"# one\n").expect("write a");
    let (worker, server) = worker_with(&vault, "testdevice", None);

    // The start cycle arrives by itself.
    assert!(wait_for(5, || worker.status().last_report.is_some()), "the start cycle ran");
    assert_eq!(worker.status().phase, notescraft_engine::SyncPhase::Idle);

    // An edit plus a request: the debounce passes and one run uploads the edit.
    vault.write("b.md", b"# two\n").expect("write b");
    let calls = server.changes_calls();
    worker.request_sync();
    assert!(wait_for(5, || server.changes_calls() > calls && file_is(&root, 2, "idle", 0)), "the request ran one cycle to its end");

    let status = worker.status();
    let report = status.last_report.expect("a report");
    assert_eq!(report, notescraft_engine::ReportCounts { uploaded: 1, downloaded: 0, deleted: 0, merged: 0, conflicts: 0 });
    assert_eq!(status.pending, 0, "everything the scan saw is in the bases now");
    assert_eq!(status.last_seq, 1, "the worker saw the record the changes log showed before its own upload");
    assert_eq!(status.last_sync_ms, Some(FIXED_MS));
    assert_eq!(status.server, None, "start_with does not know a server name");
    assert!(status.last_error.is_none());
    assert!(status.bound);
}

#[test]
fn network_error_goes_offline_then_recovers() {
    let (vault, root) = temp_vault("worker-off");
    vault.write("a.md", b"# one\n").expect("write");
    let (worker, server) = worker_with(&vault, "testdevice", Some(Failure::Never));

    assert!(wait_for(5, || file_is(&root, 1, "offline", 1)), "a network error is offline");
    let offline = worker.status();
    assert!(offline.last_error.as_deref().is_some_and(|error| !error.is_empty()), "{:?} names the failure", offline.last_error);
    assert!(offline.last_sync_ms.is_none(), "nothing has synced yet");

    server.healthy();
    assert!(wait_for(5, || file_is(&root, 2, "idle", 0)), "a healthy server recovers");
    assert_eq!(worker.status().pending, 0);
    assert!(worker.status().last_error.is_none());
}

#[test]
fn status_401_needs_sign_in_and_stops_retrying_until_requested() {
    let (vault, root) = temp_vault("worker-401");
    vault.write("a.md", b"# one\n").expect("write");
    let (worker, server) = worker_with(&vault, "testdevice", Some(Failure::Status(401)));

    assert!(wait_for(5, || server.changes_calls() == 1 && file_is(&root, 1, "sign_in_again", 1)), "a refused token asks to sign in");
    assert_eq!(server.changes_calls(), 1);

    // No retry happens on its own, not even after the backoff or the idle poll.
    std::thread::sleep(Duration::from_millis(400));
    assert_eq!(server.changes_calls(), 1, "the worker waits for a request");
    assert_eq!(worker.status().last_error.as_deref(), Some("the token was refused (HTTP 401): sign in again"));

    worker.request_sync();
    assert!(wait_for(5, || server.changes_calls() == 2), "a request retries once");
    assert!(wait_for(5, || file_is(&root, 2, "sign_in_again", 1)), "the retried cycle wrote its state down");
    assert_eq!(worker.status().phase, notescraft_engine::SyncPhase::SignInAgain, "still refused, still signed out");
}

#[test]
fn status_json_is_written_and_readable_by_another_process() {
    let (vault, root) = temp_vault("worker-file");
    vault.write("a.md", b"# one\n").expect("write");
    let (_worker, _server) = worker_with(&vault, "testdevice", None);
    assert!(wait_for(5, || file_is(&root, 1, "idle", 0)), "a cycle ran");

    // Read the way another process would: the plain file, no worker in hand.
    let bytes = std::fs::read(root.join(".notescraft/sync/status.json")).expect("the status file exists");
    let text = String::from_utf8(bytes).expect("text");
    assert!(!text.contains("token"), "no token in the status file: {text}");
    let file: Value = serde_json::from_str(&text).expect("json");
    assert_eq!(file.get("cycle"), Some(&json!(1)), "the start cycle wrote once");
    assert_eq!(file.get("phase"), Some(&json!("idle")));
    assert_eq!(file.get("pending"), Some(&json!(0)));
    assert_eq!(file.get("conflicts"), Some(&json!([])));
    assert_eq!(file.get("last_sync_ms"), Some(&json!(FIXED_MS)));
}

/// A saved binding stays readable and the token stays outside it and mode `0o600`.
#[cfg(unix)]
#[test]
fn token_file_has_mode_600_and_binding_has_no_token() {
    use std::os::unix::fs::PermissionsExt;
    let (vault, root) = temp_vault("binding-token");
    save_binding(&vault, &Binding { server: "https://sync.example".into(), vault_id: "v-42".into(), device_id: "d-1".into(), user_id: None })
        .expect("save binding");
    save_token(&vault, "pair-token-secret").expect("save token");

    let binding = load_binding(&vault).expect("load binding").expect("a binding");
    assert_eq!(binding.vault_id, "v-42");
    let text = std::fs::read_to_string(root.join(".notescraft/sync/binding.json")).expect("binding file");
    assert!(!text.contains("pair-token-secret"), "the token stays out of the binding: {text}");

    let credentials = root.join(".notescraft/sync/credentials.json");
    let mode = std::fs::metadata(credentials).expect("credentials").permissions().mode() & 0o777;
    assert_eq!(mode, 0o600, "the token file is private to the owner");
    assert_eq!(load_token(&vault).expect("load token").as_deref(), Some("pair-token-secret"));
}

/// A one-shot `POST /v1/devices` server: it answers once and closes.
struct Stub {
    port: u16,
    handle: Option<std::thread::JoinHandle<()>>,
}

impl Stub {
    fn start() -> Stub {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind the stub");
        let port = listener.local_addr().expect("address").port();
        let handle = std::thread::Builder::new().name("stub-server".to_owned()).spawn(move || answer_once(&listener)).expect("spawn");
        Stub { port, handle: Some(handle) }
    }

    fn url(&self) -> String {
        format!("http://127.0.0.1:{}", self.port)
    }
}

impl Drop for Stub {
    fn drop(&mut self) {
        if let Some(handle) = self.handle.take() {
            let _ = handle.join();
        }
    }
}

use std::io::{Read, Write as IoWrite};

/// Answers the one pairing request this stub exists for.
fn answer_once(listener: &std::net::TcpListener) {
    let Ok((mut stream, _)) = listener.accept() else { return };
    let mut buffer = [0u8; 1024];
    let read = stream.read(&mut buffer).unwrap_or(0);
    let request = String::from_utf8_lossy(&buffer[..read.min(buffer.len())]);
    if !request.contains("POST /v1/devices") {
        let _ = stream.write_all(b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
        return;
    }
    let body = json!({ "device_id": "d-1", "vault_id": "v-42", "token": "pair-token-secret" }).to_string();
    let answer = format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len());
    let _ = std::io::Write::write_all(&mut stream, answer.as_bytes());
    let _ = stream.flush();
    let _ = stream.shutdown(std::net::Shutdown::Both);
}

#[cfg(unix)]
#[test]
fn pair_stores_binding_against_a_stub_server() {
    use std::os::unix::fs::PermissionsExt;
    let (mut session, reg, root) = fixture("pair");
    let stub = Stub::start();

    let answer = reg.run("sync.pair", &mut session, &json!({ "server": stub.url(), "code": "the-code", "name": "cli laptop" })).expect("pair");
    assert_eq!(answer.get("vault_id"), Some(&json!("v-42")));
    assert_eq!(answer.get("device_id"), Some(&json!("d-1")));
    assert!(!answer.to_string().contains("pair-token-secret"), "the token never leaves the vault: {answer}");

    // The binding sits in the vault, the token in its own private file.
    let binding = std::fs::read_to_string(root.join(".notescraft/sync/binding.json")).expect("binding file");
    assert!(!binding.contains("pair-token-secret"), "no token in the binding: {binding}");
    let credentials = root.join(".notescraft/sync/credentials.json");
    let mode = std::fs::metadata(credentials).expect("credentials").permissions().mode() & 0o777;
    assert_eq!(mode, 0o600);

    // The worker runs and knows the server it was paired against.
    let worker = session.sync_worker().expect("the pair starts the worker");
    assert!(wait_for(5, || worker.status().server.as_deref() == Some(stub.url().as_str())), "the worker carries the binding");
    assert_eq!(worker.status().vault_id, Some("v-42".to_owned()));
}

/// Mine, theirs and both settle a pair through the `sync.*` commands.
#[test]
fn conflict_resolution_mine_theirs_both() {
    for (name, keep) in [("conflict-mine", "mine"), ("conflict-theirs", "theirs"), ("conflict-both", "both")] {
        let (mut session, reg, _root) = fixture(name);
        let copy = conflict_path("note.md", "cliclient", "2026-01-01");
        {
            let vault = session.vault();
            vault.write("note.md", b"their version").expect("the server version");
            vault.write(&copy, b"local version").expect("the local version");
        }

        let listed = reg.run("sync.conflicts", &mut session, &Value::Null).expect("list conflicts");
        assert_eq!(listed.get("conflicts"), Some(&json!([{ "original": "note.md", "copy": copy.clone() }])));

        reg.run("sync.resolve", &mut session, &json!({ "copy": copy, "keep": keep })).expect("resolve");

        // `mine` puts the local version at the original; the others keep the server one.
        let kept = if keep == "mine" { "local version" } else { "their version" };
        assert_eq!(session.vault().read("note.md").expect("original"), kept.as_bytes());
        if keep == "both" {
            assert_eq!(session.vault().read("note (2).md").expect("the local version survives"), b"local version");
        }
        let listed = reg.run("sync.conflicts", &mut session, &Value::Null).expect("list conflicts after");
        assert_eq!(listed.get("conflicts"), Some(&json!([])), "{keep} leaves no pair behind");
    }
}

#[test]
fn unbound_vault_status_and_start_error() {
    let (mut session, reg, _root) = fixture("unbound");

    let status = reg.run("sync.status", &mut session, &Value::Null).expect("status");
    assert_eq!(status, json!({ "bound": false }));

    let now = reg.run("sync.now", &mut session, &Value::Null).expect("now");
    assert_eq!(now, json!({ "requested": false }));

    let error = session.start_sync("cli laptop").expect_err("no worker without a binding");
    assert!(matches!(error, SyncError::State(_)), "{error:?}");
}

#[test]
fn stop_joins_quickly() {
    let (vault, _root) = temp_vault("worker-stop");
    vault.write("a.md", b"# one\n").expect("write");
    let (worker, _server) = worker_with(&vault, "testdevice", None);
    assert!(wait_for(5, || worker.status().last_report.is_some()), "the start cycle ran");

    let began = Instant::now();
    worker.stop();
    let wait = began.elapsed();
    assert!(wait < Duration::from_secs(2), "a stop does not wait long: {wait:?}");
}

// (tests continue)