notescraft/crates/engine/tests/live_session.rs

//! The live session against a stub server: the events of the room are pushed
//! by the test and the requests of the client are recorded.

mod live_stub;

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

use live_stub::Stub;
use notescraft_engine::{LiveConfig, LiveEvent, LiveSession};
use serde_json::{Value, json};

/// How long a test waits for the client to catch up.
const DEADLINE: Duration = Duration::from_secs(10);

/// The connection details of a client of the stub.
fn config(stub: &Stub, client: &str) -> LiveConfig {
    LiveConfig { server: stub.base.clone(), vault_id: "vault".into(), token: "secret-token".into(), client_id: client.into(), name: "Ana".into() }
}

/// Polls the session until the events satisfy `wanted`, or the deadline ends.
fn wait_for(session: &mut LiveSession, wanted: impl Fn(&[LiveEvent]) -> bool) -> Vec<LiveEvent> {
    let deadline = Instant::now() + DEADLINE;
    let mut all = Vec::new();
    while Instant::now() < deadline {
        let events = session.poll();
        let done = wanted(&events);
        all.extend(events);
        if done {
            return all;
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    all
}

/// Waits until the stub opened that many event streams.
fn wait_for_streams(stub: &Stub, count: usize) {
    let deadline = Instant::now() + DEADLINE;
    while stub.stream_count() < count && Instant::now() < deadline {
        std::thread::sleep(Duration::from_millis(10));
    }
}

/// The first event of the list that matches.
fn find(events: &[LiveEvent], wanted: impl Fn(&LiveEvent) -> bool) -> Option<&LiveEvent> {
    events.iter().find(|event| wanted(event))
}

#[test]
fn joining_takes_the_text_of_the_room() {
    let stub = Stub::start("hello");
    let (session, text) = LiveSession::join(config(&stub, "a"), "note.md").expect("join");
    assert_eq!(text, "hello", "the room hands its text to the client");
    assert_eq!(session.text(), "hello");
    assert_eq!(session.rev(), 0);
    let joins = stub.joins();
    assert_eq!(joins.len(), 1, "the client joined once");
    let body: Value = serde_json::from_str(&joins[0].body).expect("join body");
    assert_eq!(body["client"], json!("a"));
    session.leave();
}

#[test]
fn an_operation_of_a_peer_becomes_a_remote_event() {
    let stub = Stub::start("hello");
    let (mut session, _) = LiveSession::join(config(&stub, "a"), "note.md").expect("join");
    wait_for_streams(&stub, 1);
    stub.push_event("event: op\ndata: {\"rev\":1,\"client\":\"b\",\"op\":[5,\"!\"]}\n\n");
    let events = wait_for(&mut session, |events| find(events, |event| matches!(event, LiveEvent::Remote { .. })).is_some());
    let remote = find(&events, |event| matches!(event, LiveEvent::Remote { .. })).expect("a remote event");
    let LiveEvent::Remote { op, text } = remote else {
        panic!("not a remote event");
    };
    assert_eq!(text, "hello!", "the operation of the peer reached the mirror");
    assert_eq!(op.target_len(), 6);
    assert_eq!(session.text(), "hello!");
    session.leave();
}

#[test]
fn a_local_edit_is_sent_to_the_room() {
    let stub = Stub::start("hello");
    let (mut session, _) = LiveSession::join(config(&stub, "a"), "note.md").expect("join");
    session.local_edit("hello!").expect("edit");
    let deadline = Instant::now() + DEADLINE;
    while stub.ops().is_empty() && Instant::now() < deadline {
        std::thread::sleep(Duration::from_millis(10));
    }
    let ops = stub.ops();
    assert_eq!(ops.len(), 1, "the edit went out once");
    let body: Value = serde_json::from_str(&ops[0].body).expect("op body");
    assert_eq!(body["client"], json!("a"));
    assert_eq!(body["base_rev"], json!(0), "the operation is based on the revision of the room");
    assert_eq!(body["op"], json!([5, "!"]), "the operation is in the wire format");
    session.leave();
}

#[test]
fn a_second_edit_waits_for_the_acknowledgement() {
    let stub = Stub::start("hello");
    let (mut session, _) = LiveSession::join(config(&stub, "a"), "note.md").expect("join");
    wait_for_streams(&stub, 1);
    session.local_edit("hello!").expect("first edit");
    session.local_edit("hello!!").expect("second edit");
    let deadline = Instant::now() + DEADLINE;
    while stub.ops().is_empty() && Instant::now() < deadline {
        std::thread::sleep(Duration::from_millis(10));
    }
    std::thread::sleep(Duration::from_millis(100));
    assert_eq!(stub.ops().len(), 1, "the second edit waits for the confirmation");

    stub.push_event("event: ack\ndata: {\"rev\":1}\n\n");
    let events = wait_for(&mut session, |events| find(events, |event| matches!(event, LiveEvent::Acked)).is_some());
    assert!(find(&events, |event| matches!(event, LiveEvent::Acked)).is_some(), "the client heard the confirmation");

    let deadline = Instant::now() + DEADLINE;
    while stub.ops().len() < 2 && Instant::now() < deadline {
        std::thread::sleep(Duration::from_millis(10));
    }
    let ops = stub.ops();
    assert_eq!(ops.len(), 2, "the buffered edit went out after the confirmation");
    let body: Value = serde_json::from_str(&ops[1].body).expect("op body");
    assert_eq!(body["base_rev"], json!(1), "the buffered operation is based on the new revision");
    assert_eq!(body["op"], json!([6, "!"]));
    session.leave();
}

#[test]
fn the_stream_survives_comments_and_split_data() {
    let stub = Stub::start("hello");
    let (mut session, _) = LiveSession::join(config(&stub, "a"), "note.md").expect("join");
    wait_for_streams(&stub, 1);
    // A comment, then an operation whose payload is split over two data lines.
    stub.push_event(": keepalive\n\n");
    stub.push_event("event: op\ndata: {\"rev\":1,\ndata: \"client\":\"b\",\"op\":[5,\"!\"]}\n\n");
    let events = wait_for(&mut session, |events| find(events, |event| matches!(event, LiveEvent::Remote { .. })).is_some());
    assert!(find(&events, |event| matches!(event, LiveEvent::Remote { .. })).is_some(), "the split payload was read");
    assert_eq!(session.text(), "hello!");
    session.leave();
}

#[test]
fn a_dropped_stream_is_restored() {
    let stub = Stub::start("hello");
    let (mut session, _) = LiveSession::join(config(&stub, "a"), "note.md").expect("join");
    wait_for_streams(&stub, 1);
    assert_eq!(stub.stream_count(), 1, "the client reads the stream of the room");

    stub.drop_streams();
    wait_for_streams(&stub, 2);
    assert!(stub.stream_count() >= 2, "the client opened the stream again");
    assert!(stub.joins().len() >= 2, "the client joined the room again");

    // The restored stream still carries the events of the room.
    stub.push_event("event: op\ndata: {\"rev\":1,\"client\":\"b\",\"op\":[5,\"!\"]}\n\n");
    let events = wait_for(&mut session, |events| find(events, |event| matches!(event, LiveEvent::Remote { .. })).is_some());
    assert!(find(&events, |event| matches!(event, LiveEvent::Remote { .. })).is_some(), "the restored stream works");
    session.leave();
}

#[test]
fn a_stale_operation_joins_the_room_again() {
    let stub = Stub::start("hello");
    let (mut session, _) = LiveSession::join(config(&stub, "a"), "note.md").expect("join");
    stub.set_ops_status(409);
    session.local_edit("hello!").expect("edit");
    let events = wait_for(&mut session, |events| find(events, |event| matches!(event, LiveEvent::Resynced { .. })).is_some());
    let resynced = find(&events, |event| matches!(event, LiveEvent::Resynced { .. })).expect("a resync");
    let LiveEvent::Resynced { text, conflict_copy } = resynced else {
        panic!("not a resync");
    };
    assert_eq!(text, "hello!", "our edit is in the merged text");
    assert_eq!(*conflict_copy, None, "the two sides did not overlap");
    assert!(stub.joins().len() >= 2, "the client joined the room again");
    session.leave();
}

#[test]
fn a_conflicting_room_keeps_our_text() {
    let stub = Stub::start("hello");
    let (mut session, _) = LiveSession::join(config(&stub, "a"), "note.md").expect("join");
    stub.set_room_text("goodbye");
    stub.set_ops_status(409);
    session.local_edit("hello!").expect("edit");
    let events = wait_for(&mut session, |events| find(events, |event| matches!(event, LiveEvent::Resynced { .. })).is_some());
    let resynced = find(&events, |event| matches!(event, LiveEvent::Resynced { .. })).expect("a resync");
    let LiveEvent::Resynced { text, conflict_copy } = resynced else {
        panic!("not a resync");
    };
    assert_eq!(text, "goodbye", "the text of the room wins");
    assert_eq!(*conflict_copy, Some("hello!".to_owned()), "our text is kept as a conflict copy");
    session.leave();
}