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

//! The engine's live session against the real sync server: two peers type in the
//! same room, their texts meet and the room reaches the vault through the flush.
//!
//! The time is never waited for: the windows of the room are long, the flush is
//! called by hand and the clients are polled until they are quiet.

mod common;

use std::time::{Duration, Instant};

use common::Server;
use notescraft_engine::{LiveConfig, LiveSession};
use notescraft_sync_server::{AppState, Store, Timings, flush, now_ms};
use serde_json::json;

/// How many rounds both peers type in.
const ROUNDS: usize = 5;
/// How long a test waits for the clients to settle.
const DEADLINE: Duration = Duration::from_secs(10);
/// How many polls without an event mean that the clients are quiet.
const QUIET_POLLS: usize = 40;
/// The note both peers type in.
const NOTE: &str = "notes/note.md";

/// The windows of the room: a short keepalive and long debounces, so the only
/// write of the test is the flush at the end.
fn windows() -> Timings {
    Timings { keepalive_ms: 200, idle_ms: 30_000, presence_ttl_ms: 60_000, flush_debounce_ms: 60_000, sweep_ms: 60_000, max_rooms: 8, max_peers: 64 }
}

/// A server over a fresh folder with the windows of the test.
fn server(name: &str) -> Server {
    let dir = common::temp_dir(name);
    let state = AppState::new(Store::open(&dir.join("sync.db")).expect("db"), dir.join("blobs"), common::ADMIN);
    state.set_live_timings(windows());
    common::start_with_state(state)
}

/// The connection details of one peer of the room.
fn config(server: &Server, vault: &str, token: &str, client: &str, name: &str) -> LiveConfig {
    LiveConfig { server: server.base.clone(), vault_id: vault.to_owned(), token: token.to_owned(), client_id: client.to_owned(), name: name.to_owned() }
}

/// A generator of positions that repeats itself, so a failing run repeats too.
struct Lcg(u64);

impl Lcg {
    /// A generator with the given seed.
    fn new(seed: u64) -> Self {
        Self(seed)
    }

    /// The next number of the sequence.
    fn next(&mut self) -> u64 {
        self.0 = self.0.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1_442_695_040_888_963_407);
        self.0 >> 33
    }

    /// A position inside a text of `len` characters.
    fn position(&mut self, len: usize) -> usize {
        let span = u64::try_from(len).unwrap_or(u64::MAX).max(1);
        usize::try_from(self.next() % span).unwrap_or(0)
    }
}

/// The places a marker may go: the start of a line. A marker put there keeps
/// its shape, whatever the other peers type later.
fn line_starts(text: &str) -> Vec<usize> {
    let mut starts = vec![0];
    for (index, character) in text.chars().enumerate() {
        if character == '\n' {
            starts.push(index + 1);
        }
    }
    starts
}

/// The text with `marker` put in at the given position.
fn insert_at(text: &str, at: usize, marker: &str) -> String {
    let mut out = String::with_capacity(text.len() + marker.len());
    for (index, character) in text.chars().enumerate() {
        if index == at {
            out.push_str(marker);
        }
        out.push(character);
    }
    if at >= text.chars().count() {
        out.push_str(marker);
    }
    out
}

/// Polls both clients until they are quiet and hold the same text.
fn settle(a: &mut LiveSession, b: &mut LiveSession) -> bool {
    let deadline = Instant::now() + DEADLINE;
    let mut quiet = 0;
    while Instant::now() < deadline {
        let events = a.poll().len() + b.poll().len();
        if events == 0 {
            quiet += 1;
        } else {
            quiet = 0;
        }
        if quiet >= QUIET_POLLS && a.text() == b.text() && a.rev() == b.rev() {
            return true;
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    false
}

/// Polls the client until it sees the peer with that id.
fn sees_peer(session: &mut LiveSession, client: &str) -> bool {
    let deadline = Instant::now() + DEADLINE;
    while Instant::now() < deadline {
        session.poll();
        if session.peers().iter().any(|peer| peer.client == client) {
            return true;
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    false
}

/// The text and the revision of the room, read by joining as a third peer.
fn room_snapshot(server: &Server, vault: &str, token: &str, path: &str) -> (String, u64) {
    let (status, reply) =
        server.json("POST", &format!("/v1/vaults/{vault}/live/{path}/join"), Some(token), &[], Some(&json!({"client": "observer", "name": "Observer"})));
    assert_eq!(status, 200, "{reply}");
    let text = reply["text"].as_str().expect("the text of the room").to_owned();
    let rev = reply["rev"].as_u64().expect("the revision of the room");
    (text, rev)
}

#[test]
fn two_peers_type_in_the_same_room() {
    let server = server("live-e2e");
    let (vault, token, _user) = server.owned_vault("live-e2e");
    let (status, _) = server.put(&vault, &token, NOTE, "one\ntwo\nthree\n", 0);
    assert_eq!(status, 201, "the note is in the vault");

    let mut lcg = Lcg::new(0x5EED_1234);
    let (mut a, text_a) = LiveSession::join(config(&server, &vault, &token, "client-a", "Ana"), NOTE).expect("join a");
    assert_eq!(text_a, "one\ntwo\nthree\n", "the room hands its text to the first client");
    let (mut b, text_b) = LiveSession::join(config(&server, &vault, &token, "client-b", "Ben"), NOTE).expect("join b");
    assert_eq!(text_b, "one\ntwo\nthree\n", "the room hands its text to the second client");

    for round in 0..ROUNDS {
        let marker_a = format!("A{round} ");
        let marker_b = format!("B{round} ");
        let starts_a = line_starts(a.text());
        let starts_b = line_starts(b.text());
        let at_a = starts_a.get(lcg.position(starts_a.len())).copied().unwrap_or(0);
        let at_b = starts_b.get(lcg.position(starts_b.len())).copied().unwrap_or(0);
        let edited_a = insert_at(a.text(), at_a, &marker_a);
        let edited_b = insert_at(b.text(), at_b, &marker_b);
        a.local_edit(&edited_a).expect("the edit of the first client");
        b.local_edit(&edited_b).expect("the edit of the second client");
        assert!(settle(&mut a, &mut b), "the clients did not settle in round {round}");
    }

    assert!(sees_peer(&mut a, "client-b"), "the first client sees the second one");
    assert!(sees_peer(&mut b, "client-a"), "the second client sees the first one");

    let (room_text, room_rev) = room_snapshot(&server, &vault, &token, NOTE);
    assert_eq!(a.text(), room_text, "the first client holds the text of the room");
    assert_eq!(b.text(), room_text, "the second client holds the text of the room");
    assert_eq!(a.rev(), room_rev, "the first client is at the revision of the room");
    assert_eq!(b.rev(), room_rev, "the second client is at the revision of the room");
    for round in 0..ROUNDS {
        let marker_a = format!("A{round} ");
        let marker_b = format!("B{round} ");
        assert!(room_text.contains(&marker_a), "the marker {marker_a} was lost");
        assert!(room_text.contains(&marker_b), "the marker {marker_b} was lost");
    }

    a.leave();
    b.leave();

    let report = flush(&server.state, &vault, NOTE, now_ms()).expect("the room is written");
    assert!(report.flushed, "the room was written to the vault");
    assert!(!report.conflict, "the room did not conflict with the vault");

    let (status, body) = server.send("GET", &format!("/v1/vaults/{vault}/files/{NOTE}"), Some(&token), &[], None);
    assert_eq!(status, 200);
    assert_eq!(String::from_utf8_lossy(&body), room_text, "the file holds the text of the room");
}