semora-cli 0.8.0

mod common;

use std::process::Stdio;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use common::*;
use semora_proto::{Level, Scope};
use semora_server::auth::{apikey, users};
use semora_server::projects::access::{self, ProjectAction};
use serde_json::json;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

async fn push_doc(cli: &Cli) -> String {
    cli.write("plan.md", "# Plan\n\n## Overview\n\nSome text.");
    let out = cli.run(&["push", "plan.md", "--json"]).await;
    assert_eq!(out.code, 0, "{}", out.stderr);
    out.json()["id"].as_str().unwrap().to_string()
}

fn comment(text: &str) -> serde_json::Value {
    json!({"type": "comment", "text": text,
           "anchor": {"path": "plan.md", "quote": "Some text", "prefix": "", "suffix": ".", "heading": "Overview"}})
}

/// Spawns the CLI in the background and returns a handle that yields its output.
fn spawn(cli: &Cli, args: &[&str]) -> tokio::task::JoinHandle<Out> {
    let mut cmd = cli.command(args);
    cmd.stdout(Stdio::piped()).stderr(Stdio::piped());
    tokio::task::spawn_blocking(move || {
        let o = cmd.output().unwrap();
        Out {
            code: o.status.code().unwrap_or(-1),
            stdout: String::from_utf8_lossy(&o.stdout).into_owned(),
            stderr: String::from_utf8_lossy(&o.stderr).into_owned(),
        }
    })
}

#[tokio::test(flavor = "multi_thread")]
async fn wait_collects_a_burst_prints_it_and_acks() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    let id = push_doc(&cli).await;
    let mut browser = server.browser(&id).await;

    let waiting = spawn(
        &cli,
        &[
            "wait",
            &id,
            "--timeout",
            WAIT_TIMEOUT,
            "--collect",
            BURST_WINDOW,
            "--json",
        ],
    );
    browser.wait_for_agent().await;
    browser.send(comment("Reword this sentence.")).await;
    browser
        .send(json!({"type": "message", "text": "Also add a summary."}))
        .await;
    let out = waiting.await.unwrap();
    assert_eq!(out.code, 0, "{}", out.stderr);
    let v = out.json();
    assert_eq!(v["doc_id"], id.as_str());
    let events = v["events"].as_array().unwrap();
    assert_eq!(events.len(), 2, "{v}");
    assert_eq!(events[0]["kind"], "comment");
    assert_eq!(events[0]["payload"]["anchor"]["heading"], "Overview");
    assert_eq!(events[1]["payload"]["text"], "Also add a summary.");

    until_pending(
        &server,
        &id,
        &server.key,
        "the burst was never acked",
        |n| n == 0,
    )
    .await;
}

#[tokio::test(flavor = "multi_thread")]
async fn wait_text_output_is_readable_for_agents() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    let id = push_doc(&cli).await;
    let mut browser = server.browser(&id).await;
    // The comment is stored, so `wait` gets it whether it has joined the channel by the time it is
    // sent or not. The test does not wait for it to join: a CLI that is slow to start is not a
    // failure here, and that is not what the test is about.
    let waiting = spawn(
        &cli,
        &["wait", &id, "--timeout", WAIT_TIMEOUT, "--collect", "0.5"],
    );
    browser.send(comment("Reword this sentence.")).await;
    let out = waiting.await.unwrap();
    assert_eq!(out.code, 0, "{}", out.stderr);
    for needle in [
        "comment",
        "plan.md",
        "Overview",
        "Some text",
        "Reword this sentence.",
        &format!("semora say {id}"),
    ] {
        assert!(
            out.stdout.contains(needle),
            "missing {needle:?} in:\n{}",
            out.stdout
        );
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn feedback_sent_while_no_agent_waits_is_delivered_later() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    let id = push_doc(&cli).await;
    let mut browser = server.browser(&id).await;
    browser
        .next_of("hello", HUNG_MS)
        .await
        .expect("the server says hello");
    browser
        .send(json!({"type": "page_event", "name": "approve", "data": {"option": "B"}}))
        .await;
    // The event is stored before `wait` starts, so that it is delivered later, not live.
    until_pending(
        &server,
        &id,
        &server.key,
        "the event was never stored",
        |n| n > 0,
    )
    .await;
    let out = cli
        .run(&[
            "wait",
            &id,
            "--timeout",
            WAIT_TIMEOUT,
            "--collect",
            "0.2",
            "--json",
        ])
        .await;
    assert_eq!(out.code, 0, "{}", out.stderr);
    assert_eq!(out.json()["events"][0]["payload"]["data"]["option"], "B");
}

#[tokio::test(flavor = "multi_thread")]
async fn wait_exit_codes() {
    let server = Server::start().await;
    let mut cli = Cli::new(&server);
    let id = push_doc(&cli).await;
    assert_eq!(
        cli.run(&["wait", &id, "--timeout", "1"]).await.code,
        2,
        "timeout"
    );
    assert_eq!(
        cli.run(&["wait", "AAAAAAAAAA", "--timeout", "1"])
            .await
            .code,
        4,
        "not found"
    );
    let ws_url = format!("{}/ws/d/{id}", server.url.replacen("http", "ws", 1));
    assert_eq!(
        cli.run(&["wait", &ws_url, "--timeout", "1"]).await.code,
        2,
        "websocket URLs work too"
    );
    cli.key = Some("sm_AAAAAAAA_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".into());
    assert_eq!(
        cli.run(&["wait", &id, "--timeout", "1"]).await.code,
        3,
        "auth"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn wait_downloads_screenshots() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    let id = push_doc(&cli).await;
    let mut browser = server.browser(&id).await;
    let png = b"\x89PNG\r\n\x1a\nscreenshot-bytes".to_vec();
    let resp = reqwest::Client::new()
        .post(format!("{}/api/v1/docs/{id}/attachments", server.url))
        .header("cookie", &browser.cookie)
        .header("origin", &server.url)
        .header("x-csrf-token", &browser.csrf)
        .header("content-type", "image/png")
        .body(png.clone())
        .send()
        .await
        .unwrap();
    assert!(resp.status().is_success());
    let attachment = resp.json::<serde_json::Value>().await.unwrap()["attachment_id"]
        .as_str()
        .unwrap()
        .to_string();

    let save = cli.path("shots");
    let save_s = save.to_string_lossy().into_owned();
    let waiting = spawn(
        &cli,
        &[
            "wait",
            &id,
            "--timeout",
            WAIT_TIMEOUT,
            "--collect",
            "0.3",
            "--json",
            "--save-dir",
            &save_s,
        ],
    );
    browser.wait_for_agent().await;
    browser
        .send(json!({"type": "drawing", "path": "plan.md", "note": "this box", "attachment_id": attachment,
                     "viewport": {"width": 800, "height": 600}, "scroll": {"x": 0, "y": 0}}))
        .await;
    let out = waiting.await.unwrap();
    assert_eq!(out.code, 0, "{}", out.stderr);
    let ev = &out.json()["events"][0];
    let path = ev["attachment_path"].as_str().expect("screenshot saved");
    assert_eq!(std::fs::read(path).unwrap(), png);
    assert!(path.starts_with(&save_s));
}

#[tokio::test(flavor = "multi_thread")]
async fn follow_streams_json_lines() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    let id = push_doc(&cli).await;
    let mut browser = server.browser(&id).await;
    let mut cmd = cli.command(&["wait", &id, "--follow"]);
    cmd.stdout(Stdio::piped());
    let mut child = cmd.spawn().unwrap();
    let stdout = child.stdout.take().unwrap();
    let (tx, rx) = std::sync::mpsc::channel();
    std::thread::spawn(move || {
        use std::io::BufRead;
        for line in std::io::BufReader::new(stdout)
            .lines()
            .map_while(Result::ok)
        {
            let _ = tx.send(line);
        }
    });
    browser.wait_for_agent().await;
    browser.send(comment("one")).await;
    browser.send(comment("two")).await;
    let mut got = Vec::new();
    let deadline = std::time::Instant::now() + HUNG;
    while got.len() < 2 && std::time::Instant::now() < deadline {
        match rx.try_recv() {
            Ok(line) => got.push(serde_json::from_str::<serde_json::Value>(&line).unwrap()),
            Err(_) => tokio::time::sleep(Duration::from_millis(50)).await,
        }
    }
    child.kill().unwrap();
    let _ = child.wait();
    assert_eq!(got.len(), 2);
    assert_eq!(got[0]["payload"]["text"], "one");
    assert_eq!(got[1]["payload"]["text"], "two");
}

#[tokio::test(flavor = "multi_thread")]
async fn say_shows_a_message_in_the_page() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    let id = push_doc(&cli).await;
    let mut browser = server.browser(&id).await;
    browser
        .next_of("hello", HUNG_MS)
        .await
        .expect("the server says hello");
    let out = cli
        .run(&["say", &id, "On it — fixing the intro now."])
        .await;
    assert_eq!(out.code, 0, "{}", out.stderr);
    let s = browser
        .next_of("say", HUNG_MS)
        .await
        .expect("page gets the message");
    assert_eq!(s["text"], "On it — fixing the intro now.");
}

#[tokio::test(flavor = "multi_thread")]
async fn push_with_wait_publishes_then_waits() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    cli.write("draft.md", "# Draft");
    let waiting = spawn(
        &cli,
        &[
            "push",
            "draft.md",
            "--wait",
            "--json",
            "--timeout",
            WAIT_TIMEOUT,
            "--collect",
            "0.3",
        ],
    );
    let deadline = std::time::Instant::now() + HUNG;
    while cli.cache().is_empty() && std::time::Instant::now() < deadline {
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    let id = cli.cache()[0][2].clone();
    let mut browser = server.browser(&id).await;
    browser.wait_for_agent().await;
    browser
        .send(json!({"type": "message", "text": "ship it"}))
        .await;
    let out = waiting.await.unwrap();
    assert_eq!(out.code, 0, "{}", out.stderr);
    assert_eq!(out.json()["events"][0]["payload"]["text"], "ship it");
}

#[tokio::test(flavor = "multi_thread")]
async fn wait_prints_who_sent_each_event() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    let id = push_doc(&cli).await;
    let mut browser = server.browser(&id).await;
    let waiting = spawn(
        &cli,
        &[
            "wait",
            &id,
            "--timeout",
            WAIT_TIMEOUT,
            "--collect",
            BURST_WINDOW,
        ],
    );
    browser.wait_for_agent().await;
    browser.send(comment("Reword this sentence.")).await;
    browser
        .send(json!({"type": "message", "text": "Also add a summary."}))
        .await;
    let out = waiting.await.unwrap();
    assert_eq!(out.code, 0, "{}", out.stderr);
    assert_eq!(
        out.stdout.matches(" from Admin]").count(),
        2,
        "{}",
        out.stdout
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn wait_json_carries_from() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    let id = push_doc(&cli).await;
    let mut browser = server.browser(&id).await;
    let waiting = spawn(
        &cli,
        &[
            "wait",
            &id,
            "--timeout",
            WAIT_TIMEOUT,
            "--collect",
            "0.5",
            "--json",
        ],
    );
    browser.wait_for_agent().await;
    browser.send(comment("Reword this sentence.")).await;
    let out = waiting.await.unwrap();
    let v = out.json();
    assert_eq!(v["events"][0]["from"]["name"], "Admin");
    assert_eq!(v["events"][0]["from"]["id"], server.user_id);
}

#[tokio::test(flavor = "multi_thread")]
async fn wait_as_a_reviewer_exits_3_with_the_level_message() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    // A project and a document in it, made with the owner's CLI.
    assert_eq!(
        cli.run(&["project", "init", "--name", "proj"]).await.code,
        0
    );
    cli.write("docs/plan.md", "# Plan");
    let out = cli.run(&["push", "docs/plan.md", "--json"]).await;
    assert_eq!(out.code, 0, "{}", out.stderr);
    let id = out.json()["id"].as_str().unwrap().to_string();
    // A second account, added as a Reviewer.
    let bo = users::create(
        &server.state.db,
        "bo@example.com",
        "Bo",
        "correct horse battery",
        false,
    )
    .await
    .unwrap();
    let (_, bo_key) = apikey::create(&server.state, bo.id, "bo", &Scope::all(), None, "web")
        .await
        .unwrap();
    let project = access::project_for(
        &server.state.db,
        server.user_id,
        "proj",
        ProjectAction::View,
    )
    .await
    .unwrap();
    semora_server::collab::add_collaborator(
        &server.state.db,
        project.id,
        bo.id,
        Level::Reviewer,
        Some(server.user_id),
    )
    .await
    .unwrap();
    let mut bo_cli = Cli::new(&server);
    bo_cli.key = Some(bo_key);
    let out = bo_cli.run(&["wait", &id, "--timeout", "5"]).await;
    assert_eq!(out.code, 3, "{}", out.stderr);
    assert!(
        out.stderr.contains("needs the Editor level"),
        "{}",
        out.stderr
    );
}

// ---------------------------------------------------------------------------
// An outage: a proxy between the CLI and the server that the test can cut
// ---------------------------------------------------------------------------

#[derive(Clone, Copy)]
enum Mode {
    /// Relay every connection to the server.
    Open,
    /// Relay the server's frames, and drop what the client sends: the connection stays open but
    /// its acks never arrive.
    Deaf,
    /// Close each new connection at once, as a server that is gone does.
    Refuse,
    /// Accept each new connection and never answer it, as a host that fell off the network does.
    Hang,
}

/// The `--timeout` of a `wait` that must get something: twice `HUNG`, so a `wait` that gives up on
/// its own timer is told apart from one that was still waiting when the test stopped looking.
const WAIT_TIMEOUT: &str = "120";

/// The burst window of a test that sends two events and wants both in one burst. The second event
/// follows the first by as long as the server takes to store and relay it, which on a machine
/// starved of CPU was more than half a second. The test waits out the whole window, so it cannot
/// be `HUNG`; it is long enough for that gap and short enough to cost little.
const BURST_WINDOW: &str = "10";

/// `HUNG` in milliseconds, for `next_of`.
const HUNG_MS: u64 = HUNG.as_millis() as u64;

const HELLO: &[u8] = br#""type":"hello""#;
const EVENT: &[u8] = br#""type":"event""#;

fn contains(haystack: &[u8], needle: &[u8]) -> bool {
    haystack.windows(needle.len()).any(|w| w == needle)
}

/// Polls `ready` until it holds, and panics with `what` if it has not within `HUNG`.
async fn until(what: &str, mut ready: impl FnMut() -> bool) {
    let deadline = std::time::Instant::now() + HUNG;
    while !ready() {
        assert!(std::time::Instant::now() < deadline, "{what}");
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
}

/// Polls the server until the number of events pending for `token`'s key satisfies `ready`, and
/// panics with `what` if it does not within `HUNG`. The server records an ack when it gets to it,
/// which can be after the CLI that sent it has exited.
async fn until_pending(
    server: &Server,
    id: &str,
    token: &str,
    what: &str,
    ready: impl Fn(usize) -> bool,
) {
    let deadline = std::time::Instant::now() + HUNG;
    while !ready(pending_for(server, id, token).await.len()) {
        assert!(std::time::Instant::now() < deadline, "{what}");
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
}

/// What the proxy does at the next event frame it sends a client. An outage that begins there, in
/// the proxy, the moment the frame has been written, falls inside the burst window whatever the
/// machine is doing. Cutting the line from the test instead can only guess when the CLI has read
/// the event and how much of its window is left, and a test thread that waits for CPU for a second
/// has guessed wrong.
enum Cut {
    Idle,
    /// The outage with this `Mode` begins at the next event frame.
    Armed(Mode),
    /// A connection was cut at its event frame.
    Done,
}

/// A TCP proxy in front of the server. `outage` drops every open connection and makes the server
/// unreachable; `restore` brings it back. The real server and its database stay untouched behind
/// it, so what the server recorded can be checked afterwards.
struct Proxy {
    url: String,
    mode: Arc<Mutex<Mode>>,
    from_server: Arc<Mutex<Vec<u8>>>,
    relays: Arc<Mutex<Vec<tokio::task::JoinHandle<()>>>>,
    held: Arc<Mutex<Vec<TcpStream>>>,
    cut: Arc<Mutex<Cut>>,
    restore_on: Arc<Mutex<Option<Arc<AtomicBool>>>>,
    accept: tokio::task::JoinHandle<()>,
}

async fn relay(
    mut client: TcpStream,
    target: String,
    from_server: Arc<Mutex<Vec<u8>>>,
    mode: Arc<Mutex<Mode>>,
    cut: Arc<Mutex<Cut>>,
) {
    let Ok(mut server) = TcpStream::connect(&target).await else {
        return;
    };
    let (mut client_in, mut client_out) = client.split();
    let (mut server_in, mut server_out) = server.split();
    // Set when the connection is cut: nothing the client says reaches the server after that.
    let severed = AtomicBool::new(false);
    let up = async {
        let mut buf = [0u8; 8192];
        loop {
            match client_in.read(&mut buf).await {
                Ok(0) | Err(_) => return,
                Ok(n) => {
                    let deaf = matches!(*mode.lock().unwrap(), Mode::Deaf);
                    let severed = severed.load(Ordering::SeqCst);
                    if !(deaf || severed) && server_out.write_all(&buf[..n]).await.is_err() {
                        return;
                    }
                }
            }
        }
    };
    let down = async {
        let mut buf = [0u8; 8192];
        let mut sent = Vec::new();
        loop {
            match server_in.read(&mut buf).await {
                Ok(0) | Err(_) => return,
                Ok(n) => {
                    from_server.lock().unwrap().extend_from_slice(&buf[..n]);
                    if client_out.write_all(&buf[..n]).await.is_err() {
                        return;
                    }
                    sent.extend_from_slice(&buf[..n]);
                    let outage = match *cut.lock().unwrap() {
                        Cut::Armed(outage) if contains(&sent, EVENT) => Some(outage),
                        _ => None,
                    };
                    if let Some(outage) = outage {
                        *mode.lock().unwrap() = outage;
                        severed.store(true, Ordering::SeqCst);
                        *cut.lock().unwrap() = Cut::Done;
                        // The event is on its way and the stream ends after it, as when a server
                        // goes away. The client's side stays open until the client hangs up, so
                        // that closing here cannot reset the connection over the event.
                        let _ = client_out.shutdown().await;
                        std::future::pending::<()>().await;
                    }
                }
            }
        }
    };
    tokio::select! { _ = up => {}, _ = down => {} }
}

impl Proxy {
    async fn start(server_url: &str) -> Proxy {
        let target = server_url.trim_start_matches("http://").to_string();
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let url = format!("http://{}", listener.local_addr().unwrap());
        let mode = Arc::new(Mutex::new(Mode::Open));
        let from_server = Arc::new(Mutex::new(Vec::new()));
        let relays = Arc::new(Mutex::new(Vec::new()));
        let held = Arc::new(Mutex::new(Vec::new()));
        let cut = Arc::new(Mutex::new(Cut::Idle));
        let restore_on = Arc::new(Mutex::new(None::<Arc<AtomicBool>>));
        let accept = tokio::spawn({
            let (mode, from_server) = (mode.clone(), from_server.clone());
            let (relays, held, cut) = (relays.clone(), held.clone(), cut.clone());
            let restore_on = restore_on.clone();
            async move {
                while let Ok((client, _)) = listener.accept().await {
                    // The server returns for the first connection that comes after the signal
                    // `restore_once` was given.
                    let signal = restore_on.lock().unwrap().clone();
                    if signal.is_some_and(|s| s.load(Ordering::SeqCst)) {
                        *restore_on.lock().unwrap() = None;
                        *mode.lock().unwrap() = Mode::Open;
                        held.lock().unwrap().clear();
                    }
                    let now = *mode.lock().unwrap();
                    match now {
                        Mode::Open | Mode::Deaf => {
                            let task = tokio::spawn(relay(
                                client,
                                target.clone(),
                                from_server.clone(),
                                mode.clone(),
                                cut.clone(),
                            ));
                            relays.lock().unwrap().push(task);
                        }
                        Mode::Refuse => drop(client),
                        Mode::Hang => held.lock().unwrap().push(client),
                    }
                }
            }
        });
        Proxy {
            url,
            mode,
            from_server,
            relays,
            held,
            cut,
            restore_on,
            accept,
        }
    }

    /// The server goes away: open connections drop, new ones fail the way `mode` says.
    fn outage(&self, mode: Mode) {
        *self.mode.lock().unwrap() = mode;
        self.held.lock().unwrap().clear();
        for task in self.relays.lock().unwrap().drain(..) {
            task.abort();
        }
    }

    fn restore(&self) {
        *self.mode.lock().unwrap() = Mode::Open;
        self.held.lock().unwrap().clear();
    }

    /// The server returns for the first connection that comes after `signal` is set. A `wait`
    /// that has printed its burst tries to ack it for `ACK_GRACE`, 5 s, and only a server that is
    /// back by then gets the ack. Setting the signal when the CLI prints, and letting the proxy
    /// act on it at the CLI's own next attempt, keeps the return inside that time however long
    /// the test thread waits for CPU; a `restore` called from the test has no such guarantee.
    fn restore_once(&self, signal: Arc<AtomicBool>) {
        *self.restore_on.lock().unwrap() = Some(signal);
    }

    /// Open connections stay up, but nothing the client sends reaches the server any more, so
    /// every ack is lost. `outage` or `restore` ends it.
    fn go_deaf(&self) {
        *self.mode.lock().unwrap() = Mode::Deaf;
    }

    /// Waits until the server has said hello to a client: the client has joined the channel, so an
    /// event sent now reaches it.
    async fn wait_for_hello(&self) {
        until("the server never said hello to the client", || {
            contains(&self.from_server.lock().unwrap(), HELLO)
        })
        .await;
    }

    /// The next event frame the server sends a client through the proxy is the last thing that
    /// client hears: the connection ends right after it, and the outage `mode` begins.
    fn cut_after_event(&self, mode: Mode) {
        *self.cut.lock().unwrap() = Cut::Armed(mode);
    }

    /// Waits until `cut_after_event` has cut a connection.
    async fn wait_for_cut(&self) {
        until("the event never reached the client", || {
            matches!(*self.cut.lock().unwrap(), Cut::Done)
        })
        .await;
    }
}

impl Drop for Proxy {
    fn drop(&mut self) {
        self.accept.abort();
        for task in self.relays.lock().unwrap().drain(..) {
            task.abort();
        }
    }
}

/// The CLI running in the background, its stdout read line by line as it arrives.
struct Running {
    child: std::process::Child,
    lines: std::sync::mpsc::Receiver<String>,
    stderr: Option<std::thread::JoinHandle<String>>,
    stdout: String,
    /// Set when the CLI has written its first line of stdout.
    printed: Arc<AtomicBool>,
    _cli: Cli,
}

impl Running {
    fn start(cli: Cli, args: &[&str]) -> Running {
        let mut cmd = cli.command(args);
        cmd.stdout(Stdio::piped()).stderr(Stdio::piped());
        let mut child = cmd.spawn().unwrap();
        let stdout = child.stdout.take().unwrap();
        let mut stderr = child.stderr.take().unwrap();
        let (tx, lines) = std::sync::mpsc::channel();
        let printed = Arc::new(AtomicBool::new(false));
        std::thread::spawn({
            let printed = printed.clone();
            move || {
                use std::io::BufRead;
                for line in std::io::BufReader::new(stdout)
                    .lines()
                    .map_while(Result::ok)
                {
                    printed.store(true, Ordering::SeqCst);
                    let _ = tx.send(line);
                }
            }
        });
        let stderr = std::thread::spawn(move || {
            use std::io::Read;
            let mut text = String::new();
            let _ = stderr.read_to_string(&mut text);
            text
        });
        Running {
            child,
            lines,
            stderr: Some(stderr),
            stdout: String::new(),
            printed,
            _cli: cli,
        }
    }

    /// Becomes true when the CLI prints its first line.
    fn printed(&self) -> Arc<AtomicBool> {
        self.printed.clone()
    }

    /// The next line of stdout, or `None` when none comes within `within`.
    async fn next_line(&mut self, within: Duration) -> Option<String> {
        let deadline = std::time::Instant::now() + within;
        loop {
            match self.lines.try_recv() {
                Ok(line) => {
                    self.stdout.push_str(&line);
                    self.stdout.push('\n');
                    return Some(line);
                }
                Err(std::sync::mpsc::TryRecvError::Empty)
                    if std::time::Instant::now() < deadline =>
                {
                    tokio::time::sleep(Duration::from_millis(25)).await;
                }
                Err(_) => return None,
            }
        }
    }

    /// Stops a CLI that would run on, and returns what it wrote.
    async fn stop(mut self) -> Out {
        let _ = self.child.kill();
        self.finish().await
    }

    /// Waits for the CLI to exit. Its stdout includes the lines `next_line` already took.
    async fn finish(mut self) -> Out {
        tokio::task::spawn_blocking(move || {
            let status = self.child.wait().unwrap();
            for line in self.lines.iter() {
                self.stdout.push_str(&line);
                self.stdout.push('\n');
            }
            Out {
                code: status.code().unwrap_or(-1),
                stdout: std::mem::take(&mut self.stdout),
                stderr: self.stderr.take().unwrap().join().unwrap(),
            }
        })
        .await
        .unwrap()
    }
}

impl Drop for Running {
    fn drop(&mut self) {
        let _ = self.child.kill();
    }
}

/// A `wait` run by `cli` with `opts`, reaching the server through `proxy`, that has the first
/// event of a burst in hand. The event is the last thing the server sends it: the outage `outage`
/// begins at the proxy right after that frame, so it falls inside the burst window however slowly
/// the test or the CLI runs.
async fn holding_an_event(
    server: &Server,
    mut cli: Cli,
    proxy: &Proxy,
    id: &str,
    opts: &[&str],
    outage: Mode,
) -> Running {
    cli.url = proxy.url.clone();
    let mut browser = server.browser(id).await;
    let args: Vec<&str> = ["wait", id]
        .into_iter()
        .chain(opts.iter().copied())
        .collect();
    let run = Running::start(cli, &args);
    proxy.wait_for_hello().await;
    proxy.cut_after_event(outage);
    browser.send(comment("Reword this sentence.")).await;
    proxy.wait_for_cut().await;
    run
}

/// A `wait` with a 1 s burst window and a timeout of 120 s, twice `HUNG`, that has the first event
/// of a burst in hand: a burst printed at the timeout cannot pass for one printed at the window.
async fn wait_holding_an_event(server: &Server, proxy: &Proxy, id: &str, outage: Mode) -> Running {
    let opts = ["--timeout", WAIT_TIMEOUT, "--collect", "1"];
    holding_an_event(server, Cli::new(server), proxy, id, &opts, outage).await
}

#[tokio::test(flavor = "multi_thread")]
async fn a_burst_that_closes_during_an_outage_is_printed_then_and_acked_when_the_server_returns() {
    let server = Server::start().await;
    let direct = Cli::new(&server);
    let id = push_doc(&direct).await;
    let proxy = Proxy::start(&server.url).await;
    let mut run = wait_holding_an_event(&server, &proxy, &id, Mode::Refuse).await;

    // The server returns while wait is still trying to ack: at its first attempt after it printed.
    proxy.restore_once(run.printed());

    // The 1 s window closes while the server is out of reach. The events are printed then, not
    // when the 120 s timeout ends.
    let head = run
        .next_line(HUNG)
        .await
        .expect("the burst is printed while the server is out");
    assert!(head.starts_with("1 event on"), "{head}");

    let out = run.finish().await;
    assert_eq!(out.code, 0, "{}", out.stderr);
    assert!(
        out.stdout.contains("Reword this sentence."),
        "{}",
        out.stdout
    );
    assert!(!out.stderr.contains("not acked"), "{}", out.stderr);

    // Acked: the event is no longer pending for the key.
    until_pending(
        &server,
        &id,
        &server.key,
        "the burst was never acked",
        |n| n == 0,
    )
    .await;
}

/// The burst closes during an outage and the server stays out of reach: wait prints at the
/// window, tries to ack for a while, then exits 0 with one warning. The events were not acked,
/// so the next wait gets them again.
async fn printed_in_an_outage_that_does_not_end(mode: Mode) {
    let server = Server::start().await;
    let direct = Cli::new(&server);
    let id = push_doc(&direct).await;
    let proxy = Proxy::start(&server.url).await;
    let mut run = wait_holding_an_event(&server, &proxy, &id, mode).await;

    let head = run
        .next_line(HUNG)
        .await
        .expect("the burst is printed when its window closes, not at the 120 s timeout");
    assert!(head.starts_with("1 event on"), "{head}");

    let out = run.finish().await;
    assert_eq!(out.code, 0, "{}", out.stderr);
    assert_eq!(
        out.stderr.matches("printed but not acked").count(),
        1,
        "{}",
        out.stderr
    );
    // `wait` knows that it could not reach the server, not that no ack was recorded: an ack it
    // wrote to a socket that then failed may have landed. So the warning says that the next `wait`
    // delivers the event again unless the ack reached the server.
    assert!(
        out.stderr.contains(&format!(
            "the next `semora wait {id}` delivers it again unless the ack reached the server"
        )),
        "{}",
        out.stderr
    );

    // No ack reached the server, so the event is still pending for the key.
    assert_eq!(
        pending_for(&server, &id, &server.key).await.len(),
        1,
        "an ack reached the server while it was out of reach"
    );
    let out = direct
        .run(&["wait", &id, "--timeout", WAIT_TIMEOUT, "--collect", "0.2"])
        .await;
    assert_eq!(out.code, 0, "{}", out.stderr);
    assert!(
        out.stdout.contains("Reword this sentence."),
        "the event was delivered again: {}",
        out.stdout
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn a_burst_printed_during_an_outage_warns_when_no_ack_gets_through() {
    printed_in_an_outage_that_does_not_end(Mode::Refuse).await;
}

#[tokio::test(flavor = "multi_thread")]
async fn a_server_that_never_answers_does_not_hold_the_burst_window_open() {
    printed_in_an_outage_that_does_not_end(Mode::Hang).await;
}

// ---------------------------------------------------------------------------
// `--follow` after a lost ack, and a refusal after a reconnect
// ---------------------------------------------------------------------------

/// The ids of the events the server still holds pending for the key `token`.
async fn pending_for(server: &Server, id: &str, token: &str) -> Vec<i64> {
    let (key, _) = apikey::authenticate(&server.state.db, token)
        .await
        .unwrap()
        .expect("the key is valid");
    semora_server::events::pending(&server.state, id, key.id, &[])
        .await
        .unwrap()
        .into_iter()
        .map(|e| e.id)
        .collect()
}

/// `--follow` prints an event once, but its ack can be lost. The server sends the event again
/// when the socket returns; follow does not print it again, and acks it again, so it is not
/// pending for the next `wait`.
#[tokio::test(flavor = "multi_thread")]
async fn follow_acks_an_event_again_when_the_server_sends_it_again() {
    let server = Server::start().await;
    let direct = Cli::new(&server);
    let id = push_doc(&direct).await;
    let proxy = Proxy::start(&server.url).await;
    let mut cli = Cli::new(&server);
    cli.url = proxy.url.clone();
    let mut browser = server.browser(&id).await;
    let mut run = Running::start(cli, &["wait", &id, "--follow"]);
    browser.wait_for_agent().await;

    // The event is printed, and its ack goes nowhere.
    proxy.go_deaf();
    browser.send(comment("Reword this sentence.")).await;
    let line = run.next_line(HUNG).await.expect("follow prints the event");
    assert!(line.contains("Reword this sentence."), "{line}");
    // Follow acks as it prints. Nothing tells the test when the ack has been sent; it cannot
    // arrive, so the check below holds whether or not it has.
    tokio::time::sleep(Duration::from_millis(300)).await;
    assert_eq!(
        pending_for(&server, &id, &server.key).await.len(),
        1,
        "the ack was lost, so the event is still pending"
    );

    // The socket drops and returns, and the server sends the event again.
    proxy.outage(Mode::Refuse);
    proxy.restore();
    until_pending(
        &server,
        &id,
        &server.key,
        "follow never acked the event it was sent again",
        |n| n == 0,
    )
    .await;

    let out = run.stop().await;
    assert_eq!(
        out.stdout.matches("Reword this sentence.").count(),
        1,
        "the event is printed once: {}",
        out.stdout
    );
}

/// What the server refuses a returning `wait` with.
#[derive(Clone, Copy)]
enum Refusal {
    KeyRevoked,
    DocumentDeleted,
}

impl Refusal {
    fn code(self) -> i32 {
        match self {
            Refusal::KeyRevoked => 3,
            Refusal::DocumentDeleted => 4,
        }
    }

    /// A phrase the CLI's message for it contains.
    fn words(self) -> &'static str {
        match self {
            Refusal::KeyRevoked => "API key was rejected",
            Refusal::DocumentDeleted => "not found",
        }
    }
}

/// What a `wait` has in hand when its connection is cut.
#[derive(Clone, Copy)]
enum Held {
    /// Nothing has arrived yet.
    Nothing,
    /// A burst, collected and not printed: the window is long and has most of its time left.
    Burst,
    /// `--follow`, which has printed and acked an event.
    Followed,
}

/// A `wait` whose connection is cut. While it is out of reach the key is revoked or the document
/// deleted, then the server returns and refuses the reconnect.
async fn refused_after_an_outage(refusal: Refusal, held: Held) {
    let server = Server::start().await;
    let direct = Cli::new(&server);
    let id = push_doc(&direct).await;
    let proxy = Proxy::start(&server.url).await;
    // A key of its own, so that revoking it leaves `direct` working.
    let (key, token) = apikey::create(
        &server.state,
        server.user_id,
        "waiter",
        &Scope::all(),
        None,
        "web",
    )
    .await
    .unwrap();
    let mut cli = Cli::new(&server);
    cli.key = Some(token.clone());
    cli.url = proxy.url.clone();
    let mut run = match held {
        Held::Burst => {
            // A window of 120 s, twice `HUNG`, and a timeout beyond it: a burst printed when the
            // window ends cannot pass for one printed at the refusal.
            let opts = ["--timeout", "240", "--collect", "120"];
            holding_an_event(&server, cli, &proxy, &id, &opts, Mode::Refuse).await
        }
        // The server's hello has reached the CLI through the proxy before the outage begins: the
        // CLI has joined, and its first connection is not the one the outage cuts. Without it
        // the browser can see the agent while the proxy has not yet passed the CLI its upgrade
        // response, and an outage then ends the CLI's first connect with exit 1, not 3 or 4.
        Held::Nothing => {
            let run = Running::start(cli, &["wait", &id, "--timeout", WAIT_TIMEOUT]);
            proxy.wait_for_hello().await;
            run
        }
        Held::Followed => {
            let mut browser = server.browser(&id).await;
            let mut run =
                Running::start(cli, &["wait", &id, "--follow", "--timeout", WAIT_TIMEOUT]);
            proxy.wait_for_hello().await;
            browser.send(comment("Reword this sentence.")).await;
            let line = run.next_line(HUNG).await.expect("follow prints the event");
            assert!(line.contains("Reword this sentence."), "{line}");
            // Printed and acked: the server no longer holds the event for the key.
            until_pending(&server, &id, &token, "follow never acked the event", |n| {
                n == 0
            })
            .await;
            run
        }
    };

    proxy.outage(Mode::Refuse);
    match refusal {
        Refusal::KeyRevoked => {
            assert!(
                apikey::revoke(&server.state.db, None, key.id)
                    .await
                    .unwrap()
            );
        }
        Refusal::DocumentDeleted => assert_eq!(direct.run(&["rm", &id]).await.code, 0),
    }
    proxy.restore();

    match held {
        Held::Burst => {
            // The refusal comes with the 120 s window nearly whole ahead: the burst is printed
            // then, not when the window ends.
            let head = run
                .next_line(HUNG)
                .await
                .expect("the burst is printed when the server refuses the reconnect");
            assert!(head.starts_with("1 event on"), "{head}");
            let out = run.finish().await;
            assert_eq!(out.code, 0, "{}", out.stderr);
            assert!(
                out.stdout.contains("Reword this sentence."),
                "{}",
                out.stdout
            );
            assert_eq!(
                out.stderr.matches("printed but not acked").count(),
                1,
                "{}",
                out.stderr
            );
            assert!(out.stderr.contains(refusal.words()), "{}", out.stderr);
        }
        Held::Nothing | Held::Followed => {
            let out = run.finish().await;
            assert_eq!(out.code, refusal.code(), "{}", out.stderr);
            assert!(out.stderr.contains(refusal.words()), "{}", out.stderr);
            assert!(!out.stderr.contains("not acked"), "{}", out.stderr);
            let printed = out.stdout.matches("Reword this sentence.").count();
            assert_eq!(printed, usize::from(matches!(held, Held::Followed)));
        }
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn a_burst_is_printed_when_the_reconnect_finds_the_key_revoked() {
    refused_after_an_outage(Refusal::KeyRevoked, Held::Burst).await;
}

#[tokio::test(flavor = "multi_thread")]
async fn a_burst_is_printed_when_the_reconnect_finds_the_document_deleted() {
    refused_after_an_outage(Refusal::DocumentDeleted, Held::Burst).await;
}

#[tokio::test(flavor = "multi_thread")]
async fn with_no_burst_a_revoked_key_on_reconnect_exits_3() {
    refused_after_an_outage(Refusal::KeyRevoked, Held::Nothing).await;
}

#[tokio::test(flavor = "multi_thread")]
async fn with_no_burst_a_deleted_document_on_reconnect_exits_4() {
    refused_after_an_outage(Refusal::DocumentDeleted, Held::Nothing).await;
}

#[tokio::test(flavor = "multi_thread")]
async fn follow_exits_3_when_the_reconnect_finds_the_key_revoked() {
    refused_after_an_outage(Refusal::KeyRevoked, Held::Followed).await;
}

#[tokio::test(flavor = "multi_thread")]
async fn follow_exits_4_when_the_reconnect_finds_the_document_deleted() {
    refused_after_an_outage(Refusal::DocumentDeleted, Held::Followed).await;
}

/// The window closes during the outage, so the burst is printed and the ack is tried over a new
/// socket; the key was revoked meanwhile. The warning must not promise that the next `wait`
/// delivers the events again: a revoked key (or a deleted document) may never get them.
#[tokio::test(flavor = "multi_thread")]
async fn a_refused_ack_after_an_outage_does_not_promise_the_events_again() {
    let server = Server::start().await;
    let direct = Cli::new(&server);
    let id = push_doc(&direct).await;
    let proxy = Proxy::start(&server.url).await;
    let (key, token) = apikey::create(
        &server.state,
        server.user_id,
        "waiter",
        &Scope::all(),
        None,
        "web",
    )
    .await
    .unwrap();
    let mut cli = Cli::new(&server);
    cli.key = Some(token);
    let opts = ["--timeout", WAIT_TIMEOUT, "--collect", "1"];
    let mut run = holding_an_event(&server, cli, &proxy, &id, &opts, Mode::Refuse).await;

    assert!(
        apikey::revoke(&server.state.db, None, key.id)
            .await
            .unwrap()
    );
    // The server returns, and refuses the key, at the first attempt after the burst is printed.
    // It comes after the revoke: an ack tried before it would get through.
    proxy.restore_once(run.printed());
    let head = run
        .next_line(HUNG)
        .await
        .expect("the burst is printed when its window closes");
    assert!(head.starts_with("1 event on"), "{head}");

    let out = run.finish().await;
    assert_eq!(out.code, 0, "{}", out.stderr);
    assert_eq!(
        out.stderr.matches("printed but not acked").count(),
        1,
        "{}",
        out.stderr
    );
    assert!(
        out.stderr.contains(Refusal::KeyRevoked.words()),
        "{}",
        out.stderr
    );
    assert!(!out.stderr.contains("delivers"), "{}", out.stderr);
    assert!(!out.stderr.contains("unless"), "{}", out.stderr);
}