//! The live endpoints over real HTTP: rooms, operations, the event stream, presence
//! and the flush of a room's text into the vault.
//!
//! Time is injected, never waited for: the flush is called directly and the presence
//! is pruned with an explicit clock. The event stream is read by a thread that takes
//! the Server-Sent Events line by line.

mod common;

use std::io::BufRead;
use std::sync::mpsc::Receiver;
use std::time::Duration;

use common::Server;
use notescraft_collab::{Client, HISTORY_CAP, TextOp};
use notescraft_sync_server::{AppState, Store, Timings, flush, now_ms};
use serde_json::{Value, json};

/// The live windows the tests use: a short keepalive, no background sweep.
fn windows() -> Timings {
    Timings { keepalive_ms: 200, idle_ms: 5_000, presence_ttl_ms: 30_000, flush_debounce_ms: 60_000, sweep_ms: 60_000, max_rooms: 8, max_peers: 64 }
}

/// A server over a fresh folder with the given live windows.
fn server(name: &str, windows: Timings) -> 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)
}

/// `POST .../join`: the snapshot a peer adopts.
fn join(server: &Server, vault: &str, token: &str, path: &str, client: &str, name: &str) -> (u16, Value) {
    server.json("POST", &format!("/v1/vaults/{vault}/live/{path}/join"), Some(token), &[], Some(&json!({"client": client, "name": name})))
}

/// `POST .../ops`: one operation of a peer.
fn ops(server: &Server, vault: &str, token: &str, path: &str, client: &str, base_rev: u64, op: &TextOp) -> (u16, Value) {
    server.json("POST", &format!("/v1/vaults/{vault}/live/{path}/ops"), Some(token), &[], Some(&json!({"client": client, "base_rev": base_rev, "op": op})))
}

/// The event stream of a peer, read line by line in a thread.
struct Stream {
    lines: Receiver<String>,
}

impl Stream {
    /// Opens the stream of a peer that already joined the room.
    fn open(server: &Server, vault: &str, token: &str, path: &str, client: &str) -> Stream {
        let url = format!("{}/v1/vaults/{vault}/live/{path}/events?client={client}", server.base);
        let token = token.to_owned();
        let agent = server.agent.clone();
        let (tx, lines) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let Ok(response) = agent.get(&url).header("Authorization", format!("Bearer {token}")).call() else { return };
            let reader = std::io::BufReader::new(response.into_body().into_reader());
            for line in reader.lines() {
                let Ok(line) = line else { break };
                if tx.send(line).is_err() {
                    break;
                }
            }
        });
        Stream { lines }
    }

    /// The next raw line of the stream.
    fn line(&self) -> Option<String> {
        self.lines.recv_timeout(Duration::from_secs(10)).ok()
    }

    /// The next event `(name, data)`; comments are skipped.
    fn next(&self) -> Option<(String, String)> {
        let mut name = String::new();
        let mut data = String::new();
        loop {
            let line = self.line()?;
            if line.is_empty() {
                if !name.is_empty() {
                    return Some((name, data));
                }
                continue;
            }
            if let Some(rest) = line.strip_prefix("event: ") {
                rest.clone_into(&mut name);
            } else if let Some(rest) = line.strip_prefix("data: ") {
                rest.clone_into(&mut data);
            }
        }
    }

    /// The next event with the given name.
    fn next_named(&self, wanted: &str) -> Option<(String, String)> {
        loop {
            let (name, data) = self.next()?;
            if name == wanted {
                return Some((name, data));
            }
        }
    }
}

/// The operation an event carries.
fn event_op(data: &str) -> TextOp {
    let value: Value = serde_json::from_str(data).expect("event data");
    serde_json::from_value(value["op"].clone()).expect("operation")
}

/// The body the file endpoint serves.
fn file_body(server: &Server, vault: &str, token: &str, path: &str) -> (u16, String) {
    let (status, body) = server.send("GET", &format!("/v1/vaults/{vault}/files/{path}"), Some(token), &[], None);
    (status, String::from_utf8_lossy(&body).into_owned())
}

/// A path with the spaces and brackets a conflict copy carries, in URL form.
fn url_path(path: &str) -> String {
    path.replace(' ', "%20").replace('(', "%28").replace(')', "%29")
}

#[test]
fn joining_answers_the_snapshot_and_the_revision() {
    let server = server("live-join", windows());
    let (vault, token, _) = server.owned_vault("live-join");
    assert_eq!(server.put(&vault, &token, "note.md", "hello", 0).0, 201);

    let (status, reply) = join(&server, &vault, &token, "note.md", "client-a", "Ana");
    assert_eq!(status, 200);
    assert_eq!(reply["text"], json!("hello"), "the room starts with the file's text");
    assert_eq!(reply["rev"], json!(0), "a new room is at revision zero");
    let peers = reply["peers"].as_array().expect("peers");
    assert_eq!(peers.len(), 1);
    assert_eq!(peers[0]["client"], json!("client-a"));
    assert_eq!(peers[0]["name"], json!("Ana"));

    // A second peer sees the same room.
    let (status, reply) = join(&server, &vault, &token, "note.md", "client-b", "Bo");
    assert_eq!(status, 200);
    assert_eq!(reply["rev"], json!(0));
    assert_eq!(reply["peers"].as_array().map(Vec::len), Some(2));
}

#[test]
fn two_clients_converge_on_the_room_text() {
    let server = server("live-converge", windows());
    let (vault, token, _) = server.owned_vault("live-converge");
    assert_eq!(server.put(&vault, &token, "note.md", "hi", 0).0, 201);
    assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
    assert_eq!(join(&server, &vault, &token, "note.md", "b", "Bo").0, 200);

    let stream_a = Stream::open(&server, &vault, &token, "note.md", "a");
    let stream_b = Stream::open(&server, &vault, &token, "note.md", "b");

    // Both edit the same base revision.
    let mut client_a = Client::new(0);
    let mut client_b = Client::new(0);
    let edit_a = TextOp::new().retain(2).insert("A");
    let edit_b = TextOp::new().retain(2).insert("B");
    let submit_a = client_a.local_edit(edit_a.clone()).expect("edit").expect("submit");
    let submit_b = client_b.local_edit(edit_b.clone()).expect("edit").expect("submit");
    let mut text_a = edit_a.apply("hi").expect("apply");
    let mut text_b = edit_b.apply("hi").expect("apply");

    // The first operation is accepted as it is.
    let (status, reply) = ops(&server, &vault, &token, "note.md", "a", submit_a.base_rev, &submit_a.op);
    assert_eq!(status, 200);
    assert_eq!(reply["rev"], json!(1));
    // The second one is transformed against it.
    let (status, reply) = ops(&server, &vault, &token, "note.md", "b", submit_b.base_rev, &submit_b.op);
    assert_eq!(status, 200);
    assert_eq!(reply["rev"], json!(2));

    // A hears its own acknowledgement first, then the other operation.
    let (name, data) = stream_a.next_named("ack").expect("ack");
    assert_eq!(name, "ack");
    assert_eq!(data, json!({"rev": 1}).to_string());
    client_a.server_ack().expect("ack");
    let (_name, data) = stream_a.next_named("op").expect("op");
    let applied = client_a.remote_op(event_op(&data)).expect("remote");
    text_a = applied.apply(&text_a).expect("apply");

    // B hears the other operation first, then its own acknowledgement.
    let (_name, data) = stream_b.next_named("op").expect("op");
    let applied = client_b.remote_op(event_op(&data)).expect("remote");
    text_b = applied.apply(&text_b).expect("apply");
    client_b.server_ack().expect("ack");

    // Both hold the text the room holds.
    let (status, joined) = join(&server, &vault, &token, "note.md", "c", "Cy");
    assert_eq!(status, 200);
    let room = joined["text"].as_str().expect("text");
    assert_eq!(text_a, room, "the first client converged");
    assert_eq!(text_b, room, "the second client converged");
}

#[test]
fn a_viewer_may_join_but_not_edit() {
    let server = server("live-viewer", windows());
    let (vault, owner, _) = server.owned_vault("live-viewer");
    let (_, viewer) = server.paired("live-viewer");
    let (status, code) = server.invite(&vault, &owner, "viewer");
    assert_eq!(status, 200);
    assert_eq!(server.accept(&viewer, &code).0, 200);

    // The viewer joins the room and sees it.
    let (status, reply) = join(&server, &vault, &viewer, "note.md", "v", "Vic");
    assert_eq!(status, 200);
    assert_eq!(reply["text"], json!(""));

    // Its operations are refused.
    let op = TextOp::new().insert("x");
    let (status, _reply) = ops(&server, &vault, &viewer, "note.md", "v", 0, &op);
    assert_eq!(status, 403);
}

#[test]
fn a_base_revision_older_than_the_history_is_refused() {
    let server = server("live-stale", windows());
    let (vault, token, _) = server.owned_vault("live-stale");
    assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);

    // The room keeps a bounded history; past it a client has to rejoin.
    let mut len = 0usize;
    for rev in 0..=HISTORY_CAP as u64 {
        let op = TextOp::new().retain(len).insert("x");
        let (status, _reply) = ops(&server, &vault, &token, "note.md", "a", rev, &op);
        assert_eq!(status, 200, "revision {rev}");
        len += 1;
    }

    // The first revision is out of the kept history now.
    let op = TextOp::new().retain(len).insert("x");
    let (status, reply) = ops(&server, &vault, &token, "note.md", "a", 0, &op);
    assert_eq!(status, 409);
    assert_eq!(reply["error"], json!("stale"), "the client must rejoin");
}

#[test]
fn a_malformed_operation_is_a_bad_request() {
    let server = server("live-malformed", windows());
    let (vault, token, _) = server.owned_vault("live-malformed");
    assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);

    // Not an operation at all.
    let (status, _reply) =
        server.json("POST", &format!("/v1/vaults/{vault}/live/note.md/ops"), Some(&token), &[], Some(&json!({"client": "a", "base_rev": 0, "op": "nonsense"})));
    assert_eq!(status, 400);

    // Not JSON at all.
    let (status, _body) = server.send("POST", &format!("/v1/vaults/{vault}/live/note.md/ops"), Some(&token), &[], Some(b"{not json"));
    assert_eq!(status, 400);
}

#[test]
fn a_body_over_the_cap_is_refused() {
    let server = server("live-oversize", windows());
    let (vault, token, _) = server.owned_vault("live-oversize");
    assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);

    let big = "x".repeat(1024 * 1024 + 64);
    let body = json!({"client": "a", "base_rev": 0, "op": TextOp::new().insert(&big)}).to_string();
    let (status, _reply) = server.send("POST", &format!("/v1/vaults/{vault}/live/note.md/ops"), Some(&token), &[], Some(body.as_bytes()));
    assert_eq!(status, 413);
}

#[test]
fn a_file_that_is_too_large_or_not_text_cannot_be_joined() {
    let server = server("live-files", windows());
    let (vault, token, _) = server.owned_vault("live-files");

    // A file over the document cap.
    let big = "x".repeat(2 * 1024 * 1024 + 1);
    let (status, _body) = server.send("PUT", &format!("/v1/vaults/{vault}/files/big.md"), Some(&token), &[("X-Base-Seq", "0")], Some(big.as_bytes()));
    assert_eq!(status, 201);
    let (status, _reply) = join(&server, &vault, &token, "big.md", "a", "Ana");
    assert_eq!(status, 413);

    // A file that is not UTF-8.
    let (status, _body) = server.send("PUT", &format!("/v1/vaults/{vault}/files/raw.md"), Some(&token), &[("X-Base-Seq", "0")], Some(&[0xff, 0xfe, 0x00]));
    assert_eq!(status, 201);
    let (status, _reply) = join(&server, &vault, &token, "raw.md", "a", "Ana");
    assert_eq!(status, 415);
}

#[test]
fn presence_lists_the_peers_and_prunes_the_silent_ones() {
    let server = server("live-presence", windows());
    let (vault, token, _) = server.owned_vault("live-presence");
    assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
    assert_eq!(join(&server, &vault, &token, "note.md", "b", "Bo").0, 200);

    // A heartbeat with a caret.
    let (status, reply) = server.json(
        "POST",
        &format!("/v1/vaults/{vault}/live/note.md/presence"),
        Some(&token),
        &[],
        Some(&json!({"client": "a", "selection": {"anchor": 1, "head": 1}})),
    );
    assert_eq!(status, 200);
    let peers = reply["peers"].as_array().expect("peers");
    assert_eq!(peers.len(), 2);
    let first = peers.iter().find(|peer| peer["client"] == json!("a")).expect("peer a");
    assert_eq!(first["selection"], json!({"anchor": 1, "head": 1}));

    // Time passes: the silent peers are dropped.
    let pruned = server.state.prune_live_presence(&vault, "note.md", now_ms() + 60_000);
    assert_eq!(pruned.len(), 2, "both peers fell silent: {pruned:?}");

    // A pruned peer is gone.
    let (status, _reply) = server.json("POST", &format!("/v1/vaults/{vault}/live/note.md/presence"), Some(&token), &[], Some(&json!({"client": "a"})));
    assert_eq!(status, 404);
}

#[test]
fn flushing_a_room_writes_its_text_to_the_vault() {
    let server = server("live-flush", windows());
    let (vault, token, _) = server.owned_vault("live-flush");
    assert_eq!(server.put(&vault, &token, "note.md", "hi", 0).0, 201);
    assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
    let edit = TextOp::new().retain(2).insert("!");
    let (status, reply) = ops(&server, &vault, &token, "note.md", "a", 0, &edit);
    assert_eq!(status, 200);
    assert_eq!(reply["rev"], json!(1));

    // The flush is a plain call: no waiting.
    let report = flush(&server.state, &vault, "note.md", now_ms()).expect("flush");
    assert!(report.flushed);
    assert!(!report.conflict);
    assert_eq!(report.path, "note.md");

    // The file the normal endpoint serves is the room's text.
    let (status, body) = file_body(&server, &vault, &token, "note.md");
    assert_eq!(status, 200);
    assert_eq!(body, "hi!");
}

#[test]
fn flushing_after_an_external_write_makes_a_conflict_copy() {
    let server = server("live-conflict", windows());
    let (vault, token, _) = server.owned_vault("live-conflict");
    assert_eq!(server.put(&vault, &token, "note.md", "one", 0).0, 201);
    assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
    let edit = TextOp::new().retain(3).insert("!");
    assert_eq!(ops(&server, &vault, &token, "note.md", "a", 0, &edit).0, 200);

    // Someone else writes the file while the room is open.
    assert_eq!(server.put(&vault, &token, "note.md", "two", 1).0, 201);

    let report = flush(&server.state, &vault, "note.md", now_ms()).expect("flush");
    assert!(report.flushed);
    assert!(report.conflict, "the ledger held a newer version");
    assert_ne!(report.path, "note.md");

    // The file itself is untouched.
    let (status, body) = file_body(&server, &vault, &token, "note.md");
    assert_eq!(status, 200);
    assert_eq!(body, "two", "the external write stands");

    // The room's text went to the conflict copy.
    let (status, body) = file_body(&server, &vault, &token, &url_path(&report.path));
    assert_eq!(status, 200);
    assert_eq!(body, "one!");
}

#[test]
fn the_last_peer_leaving_writes_the_room_text() {
    let server = server("live-leave", windows());
    let (vault, token, _) = server.owned_vault("live-leave");
    assert_eq!(server.put(&vault, &token, "note.md", "hi", 0).0, 201);
    assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);
    let edit = TextOp::new().retain(2).insert("!");
    assert_eq!(ops(&server, &vault, &token, "note.md", "a", 0, &edit).0, 200);

    let (status, _reply) = server.json("POST", &format!("/v1/vaults/{vault}/live/note.md/leave"), Some(&token), &[], Some(&json!({"client": "a"})));
    assert_eq!(status, 204);

    let (status, body) = file_body(&server, &vault, &token, "note.md");
    assert_eq!(status, 200);
    assert_eq!(body, "hi!", "the room's text reached the vault");
}

#[test]
fn the_room_and_peer_caps_are_enforced() {
    let mut windows = windows();
    windows.max_rooms = 2;
    windows.max_peers = 2;
    let server = server("live-caps", windows);
    let (vault, token, _) = server.owned_vault("live-caps");

    // Two rooms fit.
    assert_eq!(join(&server, &vault, &token, "one.md", "a", "Ana").0, 200);
    assert_eq!(join(&server, &vault, &token, "two.md", "b", "Bo").0, 200);
    // The third room has no room left.
    let (status, _reply) = join(&server, &vault, &token, "three.md", "c", "Cy");
    assert_eq!(status, 503);

    // The peer cap holds per room.
    assert_eq!(join(&server, &vault, &token, "one.md", "d", "Di").0, 200);
    let (status, _reply) = join(&server, &vault, &token, "one.md", "e", "Ed");
    assert_eq!(status, 409);
}

#[test]
fn the_event_stream_keeps_alive() {
    let server = server("live-keepalive", windows());
    let (vault, token, _) = server.owned_vault("live-keepalive");
    assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);

    let stream = Stream::open(&server, &vault, &token, "note.md", "a");
    let line = stream.line().expect("a line");
    assert!(line.contains("keepalive"), "the stream comments itself alive: {line}");
}

#[test]
fn the_audit_log_names_the_live_events_without_the_text() {
    let server = server("live-audit", windows());
    let (vault, token, _) = server.owned_vault("live-audit");
    assert_eq!(server.put(&vault, &token, "note.md", "secret text", 0).0, 201);
    assert_eq!(join(&server, &vault, &token, "note.md", "a", "Ana").0, 200);

    let rows = server.audit();
    let joined = rows.iter().find(|row| row["kind"] == json!("live_joined")).expect("a join row");
    assert_eq!(joined["detail"], json!("note.md"));
    assert!(!joined.to_string().contains("secret text"), "the note text is not logged");
}