semora-cli 1.1.0

mod common;

use std::io::{BufRead, BufReader};
use std::process::Stdio;
use std::sync::mpsc;
use std::time::Duration;

use common::*;

/// Spawns `semora watch` and returns its stdout lines through a channel.
fn spawn_watch(cli: &Cli, args: &[&str]) -> (std::process::Child, mpsc::Receiver<String>) {
    spawn_watch_then(cli, args, || {})
}

/// Like `spawn_watch`, and runs `on_first_line` on the reader thread the instant the first line
/// of output arrives, before anything polls the channel. That is the earliest an edit can
/// follow the first push.
fn spawn_watch_then(
    cli: &Cli,
    args: &[&str],
    on_first_line: impl FnOnce() + Send + 'static,
) -> (std::process::Child, mpsc::Receiver<String>) {
    let mut cmd = cli.command(args);
    cmd.stdout(Stdio::piped()).stderr(Stdio::inherit());
    let mut child = cmd.spawn().unwrap();
    let stdout = child.stdout.take().unwrap();
    let (tx, rx) = mpsc::channel();
    std::thread::spawn(move || {
        let mut hook = Some(on_first_line);
        for line in BufReader::new(stdout).lines().map_while(Result::ok) {
            if let Some(run) = hook.take() {
                run();
            }
            if tx.send(line).is_err() {
                break;
            }
        }
    });
    (child, rx)
}

/// Kills the watch process when dropped, so a failing assertion can't leave it running
/// (it holds the test's stderr, and the run would hang).
struct Running(std::process::Child);

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

async fn next_push(rx: &mpsc::Receiver<String>, secs: u64) -> Option<serde_json::Value> {
    let deadline = std::time::Instant::now() + Duration::from_secs(secs);
    loop {
        let left = deadline.saturating_duration_since(std::time::Instant::now());
        if left.is_zero() {
            return None;
        }
        // Poll without blocking the runtime that serves the test server.
        match rx.try_recv() {
            Ok(line) => {
                if let Ok(v) = serde_json::from_str::<serde_json::Value>(&line)
                    && v.get("rev").is_some()
                {
                    return Some(v);
                }
            }
            Err(mpsc::TryRecvError::Empty) => tokio::time::sleep(Duration::from_millis(50)).await,
            Err(mpsc::TryRecvError::Disconnected) => return None,
        }
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn watch_repushes_a_folder_once_per_burst() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    cli.write("site/index.md", "# One");
    cli.write("site/other.md", "# Other");
    let (child, rx) = spawn_watch(&cli, &["watch", "site", "--json"]);
    let _child = Running(child);

    let first = next_push(&rx, 15).await.expect("initial push");
    assert_eq!(first["rev"], 1);
    // The push itself must not trigger another push.
    assert!(
        next_push(&rx, 2).await.is_none(),
        "no push without a change"
    );

    // A burst of edits becomes one push.
    for n in 0..4 {
        cli.write("site/index.md", format!("# Edit {n}"));
        tokio::time::sleep(Duration::from_millis(80)).await;
    }
    cli.write("site/other.md", "# Other changed");
    let second = next_push(&rx, 15).await.expect("push after edits");
    assert_eq!(second["rev"], 2);
    assert_eq!(second["id"], first["id"]);
    assert!(
        next_push(&rx, 2).await.is_none(),
        "the burst was coalesced into one push"
    );

    // Hidden files are ignored.
    cli.write("site/.notes", "scratch");
    assert!(next_push(&rx, 2).await.is_none());
}

#[tokio::test(flavor = "multi_thread")]
async fn watch_a_single_file() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    cli.write("doc.md", "# Start");
    cli.write("unrelated.md", "# Unrelated");
    let (child, rx) = spawn_watch(&cli, &["watch", "doc.md", "--json"]);
    let _child = Running(child);
    let first = next_push(&rx, 15).await.expect("initial push");
    cli.write("unrelated.md", "# Changed but not watched");
    assert!(
        next_push(&rx, 2).await.is_none(),
        "other files in the folder do not count"
    );
    cli.write("doc.md", "# Changed");
    let second = next_push(&rx, 15).await.expect("push after edit");
    assert_eq!(second["id"], first["id"]);
    assert_eq!(second["rev"], 2);
}

#[tokio::test(flavor = "multi_thread")]
async fn watch_takes_the_new_stored_key_after_a_login_revoked_its_own() {
    use semora_proto::Scope;
    use semora_server::auth::apikey;

    let server = Server::start().await;
    let mut cli = Cli::new(&server);
    // The key comes from the credentials file, as it does after `semora login`.
    cli.key = None;
    let store = semora_cli::credentials::Store::new(cli.config.path(), false);
    store.set(&server.url, &server.key, "cli-test").unwrap();
    cli.write("doc.md", "# One");
    let (child, rx) = spawn_watch(&cli, &["watch", "doc.md", "--json"]);
    let _child = Running(child);
    let first = next_push(&rx, 15).await.expect("initial push");

    // What a second `semora login` does: store the new key, then revoke the old one.
    let (_, fresh) = apikey::create(
        &server.state,
        server.user_id,
        "again",
        &Scope::all(),
        None,
        "web",
    )
    .await
    .unwrap();
    store.set(&server.url, &fresh, "again").unwrap();
    let (old, _) = apikey::authenticate(&server.state.db, &server.key)
        .await
        .unwrap()
        .expect("the first key");
    assert!(
        apikey::revoke(&server.state.db, None, old.id)
            .await
            .unwrap()
    );

    cli.write("doc.md", "# Two");
    let second = next_push(&rx, 15)
        .await
        .expect("the push after the login uses the key that is stored now");
    assert_eq!(second["id"], first["id"]);
    assert_eq!(second["rev"], 2);
}

#[tokio::test(flavor = "multi_thread")]
async fn watch_in_a_project_pushes_by_key() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    cli.write("docs/plan.md", "# One");
    assert_eq!(
        cli.run(&["project", "init", ".", "--name", "semora"])
            .await
            .code,
        0
    );
    let (child, rx) = spawn_watch(&cli, &["watch", "docs/plan.md", "--json"]);
    let _child = Running(child);
    let first = next_push(&rx, 15).await.expect("initial push");
    assert_eq!(first["project"], "semora");
    assert_eq!(first["rev"], 1);
    // Keyed: even if the document is deleted, the next push makes a new one at the same key.
    let gone = cli.run(&["rm", first["id"].as_str().unwrap()]).await;
    assert_eq!(gone.code, 0, "{}", gone.stderr);
    cli.write("docs/plan.md", "# Two");
    let second = next_push(&rx, 15).await.expect("push after the edit");
    assert_ne!(
        second["id"], first["id"],
        "pushed by key, so a deleted document is made again"
    );
    assert_eq!(second["rev"], 1);
    assert_eq!(second["path"], "docs/plan.md");
    cli.write("docs/plan.md", "# Three");
    let third = next_push(&rx, 15).await.expect("another push");
    assert_eq!(third["id"], second["id"]);
    assert_eq!(third["rev"], 2);
}

#[tokio::test(flavor = "multi_thread")]
async fn watch_in_a_project_pushes_an_edit_made_the_instant_the_first_push_prints() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    cli.write("docs/plan.md", "# One");
    assert_eq!(
        cli.run(&["project", "init", ".", "--name", "semora"])
            .await
            .code,
        0
    );
    // The edit is made on the reader thread the instant the first line prints, with no
    // polling in between. The watcher must already be listening.
    let path = cli.path("docs/plan.md");
    let (child, rx) = spawn_watch_then(&cli, &["watch", "docs/plan.md", "--json"], move || {
        std::fs::write(path, "# Two").unwrap()
    });
    let _child = Running(child);
    let first = next_push(&rx, 15).await.expect("initial push");
    assert_eq!(first["project"], "semora");
    assert_eq!(first["rev"], 1);
    let second = next_push(&rx, 15).await.expect("push after the edit");
    assert_eq!(second["id"], first["id"]);
    assert_eq!(second["rev"], 2);
    assert_eq!(second["unchanged"], false);
}

#[tokio::test(flavor = "multi_thread")]
async fn watch_with_id_in_a_project_keeps_pushing_to_that_document() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    cli.write("docs/plan.md", "# One");
    assert_eq!(
        cli.run(&["project", "init", ".", "--name", "semora"])
            .await
            .code,
        0
    );
    // A document outside the key, then watch it explicitly.
    let standalone = cli
        .run(&["push", "docs/plan.md", "--new", "--json"])
        .await
        .json();
    let id = standalone["id"].as_str().unwrap().to_string();
    // The edit is made the instant the first line prints: the earliest any edit can follow
    // the first push, with no polling in between. It must still be pushed.
    let path = cli.path("docs/plan.md");
    let (child, rx) = spawn_watch_then(
        &cli,
        &["watch", "docs/plan.md", "--id", &id, "--json"],
        move || std::fs::write(path, "# Two").unwrap(),
    );
    let child = Running(child);
    let first = next_push(&rx, 15).await.expect("initial push");
    assert_eq!(first["id"], id.as_str());
    assert_eq!(
        first["rev"], 1,
        "the same bytes as the push --new: the current revision, no new one"
    );
    assert_eq!(first["unchanged"], true);
    let second = next_push(&rx, 15).await.expect("push after the edit");
    assert_eq!(
        second["id"],
        id.as_str(),
        "still X, not the document at the key"
    );
    assert_eq!(second["rev"], 2);
    assert_eq!(second["unchanged"], false);
    drop(child);
    let projects = cli.run(&["project", "ls", "--json"]).await.json();
    assert_eq!(
        projects[0]["documents"], 0,
        "no document was made at the key"
    );
}