notescraft/apps/notescraft-sync-server/tests/chunk_client_e2e.rs

//! The real engine sync client against the real server: a big file travels in
//! chunks, a cut connection resumes with the chunks that never arrived.

use std::path::{Path, PathBuf};
use std::sync::Mutex;

use notescraft_engine::{Changes, HttpRemote, Limits, Progress, PutReply, Remote, SyncError, SyncOptions, SyncReport, SyncResult, SyncState, sync_once_with};
use notescraft_sync_server::{AppState, Store, router};
use notescraft_vault::Vault;

const ADMIN: &str = "test-admin-token-0123456789abcdef";
/// The size of one chunk the client sends.
const CHUNK: usize = 32 * 1024;
/// The size above which the client sends a file in chunks.
const THRESHOLD: u64 = 16 * 1024;
/// The biggest chunk the server takes.
const SERVER_CHUNK_MAX: usize = 64 * 1024;
/// The biggest file the server assembles.
const SERVER_FILE_CAP: u64 = 1024 * 1024;
/// The size of the file that travels.
const FILE_BYTES: usize = 300 * 1024;

/// A temp folder that is empty when the test starts.
fn temp(name: &str) -> PathBuf {
    let dir = std::env::temp_dir().join(format!("nc-chunk-e2e-{name}-{}", std::process::id()));
    std::fs::remove_dir_all(&dir).ok();
    std::fs::create_dir_all(&dir).expect("dir");
    dir
}

/// Starts the server on an ephemeral port and returns its base url.
fn start(data: &Path) -> String {
    std::fs::create_dir_all(data.join("blobs")).expect("blobs");
    let state = AppState::new(Store::open(&data.join("sync.db")).expect("db"), data.join("blobs"), ADMIN);
    // The chunk limits are set before the router is built: the route takes its
    // body cap from the state.
    state.set_chunk_limits(SERVER_CHUNK_MAX, SERVER_FILE_CAP);
    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
    listener.set_nonblocking(true).expect("nonblocking");
    let port = listener.local_addr().expect("addr").port();
    std::thread::spawn(move || {
        let runtime = tokio::runtime::Builder::new_multi_thread().enable_all().build().expect("runtime");
        runtime.block_on(async {
            let listener = tokio::net::TcpListener::from_std(listener).expect("listener");
            axum::serve(listener, router(state)).await.ok();
        });
    });
    format!("http://127.0.0.1:{port}")
}

fn admin(base: &str, path: &str) -> serde_json::Value {
    let agent: ureq::Agent = ureq::Agent::config_builder().http_status_as_error(false).build().into();
    let mut response = agent.post(format!("{base}{path}")).header("Authorization", format!("Bearer {ADMIN}")).send_empty().expect("admin call");
    response.body_mut().read_json().expect("json")
}

/// Issues a pairing code for the vault, bound to `user` when given.
fn issue(base: &str, vault_id: &str, user: Option<&str>) -> String {
    let path = user.map_or_else(|| format!("/v1/vaults/{vault_id}/pairing"), |user| format!("/v1/vaults/{vault_id}/pairing?user={user}"));
    admin(base, &path)["code"].as_str().expect("code").to_owned()
}

/// Redeems a code with raw HTTP to also learn the account the device joined.
fn pair_raw(base: &str, code: &str, name: &str) -> (String, String) {
    let agent: ureq::Agent = ureq::Agent::config_builder().http_status_as_error(false).build().into();
    let mut response = agent.post(format!("{base}/v1/devices")).send_json(serde_json::json!({"code": code, "name": name})).expect("pair");
    let device: serde_json::Value = response.body_mut().read_json().expect("json");
    (device["token"].as_str().expect("token").to_owned(), device["user_id"].as_str().expect("user id").to_owned())
}

/// A remote that can cut the connection in the middle of a chunked upload.
struct CutRemote {
    inner: HttpRemote,
    /// How many chunks the client sent.
    sent: Mutex<usize>,
    /// From this send on, the connection is cut.
    cut_from: Mutex<Option<usize>>,
}

impl CutRemote {
    fn new(inner: HttpRemote) -> Self {
        Self { inner, sent: Mutex::new(0), cut_from: Mutex::new(None) }
    }

    /// Cuts the connection from the given chunk send on.
    fn cut_from(&self, call: usize) {
        *self.cut_from.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(call);
    }

    /// Lets the connection through again.
    fn mend(&self) {
        *self.cut_from.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = None;
    }

    /// How many chunks were sent since the last count.
    fn take_sent(&self) -> usize {
        let mut sent = self.sent.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        std::mem::take(&mut *sent)
    }
}

impl Remote for CutRemote {
    fn changes(&self, since: u64, limit: usize, wait: u64) -> SyncResult<Changes> {
        self.inner.changes(since, limit, wait)
    }

    fn get(&self, path: &str) -> SyncResult<Vec<u8>> {
        self.inner.get(path)
    }

    fn put(&self, path: &str, bytes: &[u8], base_seq: u64, mtime_ms: u64) -> SyncResult<PutReply> {
        self.inner.put(path, bytes, base_seq, mtime_ms)
    }

    fn delete(&self, path: &str, base_seq: u64) -> SyncResult<PutReply> {
        self.inner.delete(path, base_seq)
    }

    fn missing_chunks(&self, hashes: &[String]) -> SyncResult<Vec<String>> {
        self.inner.missing_chunks(hashes)
    }

    fn put_chunk(&self, hash: &str, bytes: &[u8]) -> SyncResult<()> {
        let mut sent = self.sent.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        *sent = sent.saturating_add(1);
        let cut = *self.cut_from.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        if cut.is_some_and(|from| *sent >= from) {
            return Err(SyncError::Http("the connection was cut".to_owned()));
        }
        drop(sent);
        self.inner.put_chunk(hash, bytes)
    }

    fn put_chunked(&self, path: &str, chunk_hashes: &[String], content_hash: &str, base_seq: u64, mtime_ms: u64) -> SyncResult<PutReply> {
        self.inner.put_chunked(path, chunk_hashes, content_hash, base_seq, mtime_ms)
    }
}

/// A device with its own vault, its own state and a remote that can cut.
struct Device {
    vault: Vault,
    remote: CutRemote,
    state: SyncState,
    name: &'static str,
}

impl Device {
    fn new(base: &str, vault_id: &str, token: &str, name: &'static str) -> Self {
        let root = temp(name);
        let vault = Vault::open(&root).expect("vault");
        Self { vault, remote: CutRemote::new(HttpRemote::new(base, vault_id, token)), state: SyncState::default(), name }
    }

    fn write(&self, path: &str, bytes: &[u8]) {
        self.vault.write(path, bytes).expect("write");
    }

    fn read(&self, path: &str) -> Vec<u8> {
        self.vault.read(path).expect("read")
    }

    /// Syncs once and returns the report.
    fn sync(&mut self) -> SyncReport {
        let options = SyncOptions { limits: Limits { threshold: THRESHOLD, chunk: CHUNK }, progress: Box::new(|_: &Progress| {}) };
        sync_once_with(&self.vault, &self.remote, &mut self.state, self.name, "2026-10-20", options).expect("sync")
    }
}

/// Bytes that look random but come from a small generator, so the test is stable.
fn random_bytes(len: usize, seed: u64) -> Vec<u8> {
    let mut state = seed;
    (0..len)
        .map(|_| {
            state = state.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1_442_695_040_888_963_407);
            u8::try_from((state >> 33) & 0xff).expect("byte")
        })
        .collect()
}

/// How many chunks the server holds for the vault.
fn stored_chunks(blobs: &Path, vault_id: &str) -> usize {
    let dir = blobs.join(vault_id).join("chunks");
    let Ok(entries) = std::fs::read_dir(dir) else { return 0 };
    entries.filter_map(std::result::Result::ok).filter(|entry| entry.path().is_file()).count()
}

#[test]
fn a_big_file_travels_in_chunks_and_a_cut_resumes_with_the_rest() {
    let data = temp("server-data");
    let base = start(&data);
    let blobs = data.join("blobs");
    let vault_id = format!("v-chunk-{}", std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.subsec_nanos()));
    let (phone_token, user_id) = pair_raw(&base, &issue(&base, &vault_id, None), "phone");
    let laptop = pair_raw(&base, &issue(&base, &vault_id, Some(&user_id)), "laptop");
    let mut phone = Device::new(&base, &vault_id, &phone_token, "phone");
    let mut laptop = Device::new(&base, &vault_id, &laptop.0, "laptop");

    // A file that is too big for one request travels in chunks.
    let bytes = random_bytes(FILE_BYTES, 42);
    phone.write("big.bin", &bytes);
    let report = phone.sync();
    assert_eq!(report.uploaded, vec!["big.bin".to_owned()]);
    let chunks = FILE_BYTES.div_ceil(CHUNK);
    assert_eq!(stored_chunks(&blobs, &vault_id), chunks, "the server holds every chunk");

    // The other device gets the very same bytes.
    let report = laptop.sync();
    assert_eq!(report.downloaded, vec!["big.bin".to_owned()]);
    assert_eq!(laptop.read("big.bin"), bytes);

    // The connection dies on the fourth chunk of the next file.
    let other = random_bytes(FILE_BYTES, 7);
    phone.write("other.bin", &other);
    phone.remote.take_sent();
    phone.remote.cut_from(4);
    let error = sync_once_with(
        &phone.vault,
        &phone.remote,
        &mut phone.state,
        "phone",
        "2026-10-20",
        SyncOptions { limits: Limits { threshold: THRESHOLD, chunk: CHUNK }, progress: Box::new(|_: &Progress| {}) },
    )
    .expect_err("cut");
    assert!(matches!(error, SyncError::Http(_)), "{error:?}");
    phone.remote.take_sent();
    assert_eq!(stored_chunks(&blobs, &vault_id), chunks + 3, "the three chunks that arrived are kept");

    // The next pass sends the chunks that never arrived and nothing else.
    phone.remote.mend();
    phone.sync();
    assert_eq!(phone.remote.take_sent(), chunks - 3, "only the missing chunks travel again");
    assert_eq!(stored_chunks(&blobs, &vault_id), chunks * 2, "the server holds every chunk of both files");

    // The other device converges on the second file too.
    laptop.sync();
    assert_eq!(laptop.read("other.bin"), other);
    assert_eq!(laptop.read("big.bin"), bytes);
}