haematite-cli 0.21.8

#![cfg(unix)]
#![forbid(unsafe_code)]

use std::error::Error;
use std::io::Write;
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::process::{Command, Output, Stdio};

use haematite_cli::serve::receipt_fragments::minimum_cap;
use serde_json::{Value, json};

#[path = "support/served_child.rs"]
mod served_child;
use served_child::Running;

#[path = "support/stand_in.rs"]
mod support;
use support::{StandIn, answer, stand_in};

type TestResult = Result<(), Box<dyn Error>>;

/// A created service on `dir`, returned once it says it is ready.
fn start(dir: &Path) -> Result<(Running, PathBuf), Box<dyn Error>> {
    std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700))?;
    let config = dir.join("serve.json");
    std::fs::write(
        &config,
        serde_json::to_vec(&json!({
            "data_dir":dir.join("database"), "socket":dir.join("serve.sock"),
            "name":dir.file_name().and_then(|name| name.to_str()).ok_or("fixture name is not text")?,
            "branch_cache_bytes_per_shard":1_048_576, "max_frame":1_048_576,
            "max_clients":4, "max_page":20, "timeout_ms":2000,
            "create":{"shards":2,"cache_bytes_total":2_097_152,"executor_threads":2}
        }))?,
    )?;
    let running = Running::start(&config, true)?;
    Ok((running, config))
}

/// `haematite call` with `args`, and `input` on its standard input.
fn call(args: &[&str], input: &str) -> Result<Output, Box<dyn Error>> {
    haematite("call", args, input)
}

/// `haematite <command>` with `args`, and `input` on its standard input.
fn haematite(command: &str, args: &[&str], input: &str) -> Result<Output, Box<dyn Error>> {
    let mut child = Command::new(env!("CARGO_BIN_EXE_haematite"))
        .arg(command)
        .args(args)
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;
    child
        .stdin
        .take()
        .ok_or("missing child stdin")?
        .write_all(input.as_bytes())?;
    Ok(child.wait_with_output()?)
}

fn text(bytes: &[u8]) -> Result<&str, Box<dyn Error>> {
    Ok(std::str::from_utf8(bytes)?)
}

fn answered(output: &Output) -> Result<Value, Box<dyn Error>> {
    assert_eq!(output.status.code(), Some(0), "{}", text(&output.stderr)?);
    Ok(serde_json::from_slice(&output.stdout)?)
}

fn head(socket: &str) -> Result<Value, Box<dyn Error>> {
    let branches = answered(&call(
        &["--socket", socket, "branches.list", r#"{"limit":5}"#],
        "",
    )?)?;
    let rows = branches["entries"].as_array().ok_or("missing branches")?;
    Ok(rows
        .iter()
        .find(|row| row["branch"] == "main")
        .ok_or("main missing")?["version"]
        .clone())
}

fn socket_of(dir: &Path) -> Result<String, Box<dyn Error>> {
    Ok(dir
        .join("serve.sock")
        .to_str()
        .ok_or("socket path is not text")?
        .to_owned())
}

#[test]
fn hello_answers_the_protocol() -> TestResult {
    let dir = tempfile::tempdir()?;
    let (_service, _) = start(dir.path())?;
    let hello = answered(&call(&["--socket", &socket_of(dir.path())?, "hello"], "")?)?;
    assert_eq!(hello["protocol"], "haem/1");
    assert_eq!(hello["max_frame"], 1_048_576);
    Ok(())
}

#[test]
fn params_given_as_an_argument_and_on_standard_input_print_the_same_bytes() -> TestResult {
    let dir = tempfile::tempdir()?;
    let (_service, _) = start(dir.path())?;
    let socket = socket_of(dir.path())?;
    let params = r#"{"limit":5}"#;
    let argument = call(&["--socket", &socket, "branches.list", params], "")?;
    let input = call(&["--socket", &socket, "branches.list", "-"], params)?;
    assert_eq!(answered(&argument)?["entries"][0]["branch"], "main");
    assert_eq!(argument.status.code(), input.status.code());
    assert_eq!(argument.stdout, input.stdout);
    Ok(())
}

#[test]
fn numbers_come_back_digit_for_digit() -> TestResult {
    let dir = tempfile::tempdir()?;
    let (_service, _) = start(dir.path())?;
    let socket = socket_of(dir.path())?;
    let declare = json!({"request_id":"declare", "branch":"main", "expected":head(&socket)?,
        "change":{"kind":"collection.put", "value":{"collection":"records", "schema":null}}});
    let declare = declare.to_string();
    answered(&call(
        &["--socket", &socket, "mutation.apply", &declare],
        "",
    )?)?;
    // Written into the params as text: json! would read 1.00 as a float.
    let document = r#"{"big":9007199254740993,"price":1.00}"#;
    let put = json!({"request_id":"put", "branch":"main", "expected":head(&socket)?,
    "change":{"kind":"document.put", "value":{
        "collection":"records", "document_id":"one", "document":"DOCUMENT",
    }}});
    let put = put.to_string().replace(r#""DOCUMENT""#, document);
    answered(&call(&["--socket", &socket, "mutation.apply", "-"], &put)?)?;
    let read = answered(&call(
        &[
            "--socket",
            &socket,
            "documents.get",
            r#"{"selection":{"branch":"main"},"collection":"records","document_id":"one"}"#,
        ],
        "",
    )?)?;
    assert_eq!(read["document"]["raw_json"], document);
    Ok(())
}

#[test]
fn an_unknown_method_is_refused_by_its_code_with_nothing_on_standard_output() -> TestResult {
    let dir = tempfile::tempdir()?;
    let (_service, _) = start(dir.path())?;
    let output = call(&["--socket", &socket_of(dir.path())?, "modules.list"], "")?;
    assert_eq!(output.status.code(), Some(2));
    assert!(output.stdout.is_empty());
    assert!(
        text(&output.stderr)?.starts_with("haematite: refused unknown_method: "),
        "{}",
        text(&output.stderr)?
    );
    Ok(())
}

#[test]
fn no_socket_at_the_path_is_could_not_ask_naming_the_path() -> TestResult {
    let dir = tempfile::tempdir()?;
    let socket = socket_of(dir.path())?;
    let output = call(&["--socket", &socket, "hello"], "")?;
    let stderr = text(&output.stderr)?;
    assert_eq!(output.status.code(), Some(1));
    assert!(output.stdout.is_empty());
    assert!(stderr.starts_with("haematite: could not ask: "), "{stderr}");
    assert!(stderr.contains(&socket), "{stderr}");
    Ok(())
}

#[test]
fn both_or_neither_of_socket_and_config_is_a_usage_error() -> TestResult {
    for args in [
        &["hello"][..],
        &["--socket", "/a.sock", "--config", "/b.json", "hello"][..],
        &["--config", "/b.json", "--socket", "/a.sock", "hello"][..],
    ] {
        let output = call(args, "")?;
        let stderr = text(&output.stderr)?;
        assert_eq!(output.status.code(), Some(1));
        assert!(output.stdout.is_empty());
        assert!(
            stderr.starts_with("haematite: give exactly one of --socket and --config\n"),
            "{stderr}"
        );
        assert!(stderr.contains("usage: haematite call"), "{stderr}");
    }
    Ok(())
}

#[test]
fn config_finds_the_socket() -> TestResult {
    let dir = tempfile::tempdir()?;
    let (_service, config_path) = start(dir.path())?;
    let config = config_path.to_str().ok_or("config path is not text")?;
    let hello = answered(&call(&["--config", config, "hello"], "")?)?;
    assert_eq!(hello["protocol"], "haem/1");
    Ok(())
}

/// The one request a stand-in of one connection read, or none when the
/// client closed instead.
fn request(service: StandIn) -> Result<Option<Vec<u8>>, Box<dyn Error>> {
    let frames = service.frames()?;
    let count = frames.len();
    let [frame] = <[_; 1]>::try_from(frames)
        .map_err(|_| format!("the stand-in read {count} requests, not one"))?;
    Ok(frame)
}

#[test]
fn a_result_is_printed_as_the_bytes_the_service_sent() -> TestResult {
    let dir = tempfile::tempdir()?;
    let result = r#"{ "big": 9007199254740993, "price": 1.00, "none": null }"#;
    let service = stand_in(dir.path(), 1_048_576, vec![vec![Some(answer(result))]])?;
    let output = call(
        &["--socket", &socket_of(dir.path())?, "store.stats", "{}"],
        "",
    )?;
    assert_eq!(output.status.code(), Some(0), "{}", text(&output.stderr)?);
    assert_eq!(text(&output.stdout)?, format!("{result}\n"));
    let sent = request(service)?;
    assert_eq!(
        text(&sent.ok_or("the request was never sent")?)?,
        r#"{"id":"haem-call","operation":{"method":"store.stats","params":{}}}"#
    );
    Ok(())
}

#[test]
fn a_service_that_closes_before_answering_leaves_the_outcome_not_known() -> TestResult {
    let dir = tempfile::tempdir()?;
    let service = stand_in(dir.path(), 1_048_576, vec![vec![None]])?;
    let params = r#"{"limit":5}"#;
    let output = call(
        &["--socket", &socket_of(dir.path())?, "branches.list", params],
        "",
    )?;
    let stderr = text(&output.stderr)?;
    assert_eq!(output.status.code(), Some(3));
    assert!(output.stdout.is_empty());
    assert!(
        stderr.starts_with("haematite: the outcome is not known: "),
        "{stderr}"
    );
    let read = request(service)?;
    assert!(read.is_some(), "the request was never sent");
    Ok(())
}

#[test]
fn a_request_larger_than_the_service_reads_is_refused_before_sending() -> TestResult {
    let dir = tempfile::tempdir()?;
    // The smallest frame a service may offer: below the typed receipt floor
    // hello is refused (BUSYSHARDNODEADLINE-001 R12, acceptance 17).
    let floor = minimum_cap()?;
    let service = stand_in(dir.path(), u32::try_from(floor)?, vec![vec![None]])?;
    let params = format!(r#"{{"after":"{}","limit":5}}"#, "x".repeat(floor));
    let output = call(
        &[
            "--socket",
            &socket_of(dir.path())?,
            "branches.list",
            &params,
        ],
        "",
    )?;
    let stderr = text(&output.stderr)?;
    assert_eq!(output.status.code(), Some(1));
    assert!(
        stderr.starts_with("haematite: could not ask: request_too_large: "),
        "{stderr}"
    );
    let read = request(service)?;
    assert_eq!(read, None, "the request was sent");
    Ok(())
}

#[test]
fn hello_is_asked_once() -> TestResult {
    let dir = tempfile::tempdir()?;
    let service = stand_in(dir.path(), 1_048_576, vec![vec![None]])?;
    let hello = answered(&call(&["--socket", &socket_of(dir.path())?, "hello"], "")?)?;
    assert_eq!(hello["protocol"], "haem/1");
    let read = request(service)?;
    assert_eq!(read, None, "a second hello was sent");
    Ok(())
}

// The lines of docs/design/product/command-line.md, run as the page gives
// them: each value below is on the page byte for byte, the socket's path and
// the fork's version apart. A value changed here is changed there.

const PAGE_COLLECTION: &str = r#"{"collection":"records","schema":null}"#;
const PAGE_ADA: &str =
    r#"{"collection":"records","document_id":"ada","document":{"name":"Ada","ward":"4B"}}"#;
const PAGE_BO: &str =
    r#"{"collection":"records","document_id":"bo","document":{"name":"Bo","ward":"2A"}}"#;
const PAGE_GET: &str =
    r#"{"selection":{"branch":"main"},"collection":"records","document_id":"ada"}"#;
const PAGE_COLLECTIONS: &str = r#"{"selection":{"branch":"main"},"limit":20}"#;
const PAGE_ADA_VECTOR: &str =
    r#"{"collection":"records","document_id":"ada","field":"embedding","vector":[1.0,0.0,0.5]}"#;
const PAGE_BO_VECTOR: &str =
    r#"{"collection":"records","document_id":"bo","field":"embedding","vector":[0.0,1.0,0.5]}"#;
const PAGE_VECTOR_GET: &str = concat!(
    r#"{"selection":{"branch":"main"},"collection":"records","#,
    r#""document_id":"ada","field":"embedding"}"#
);
const PAGE_SEARCH: &str = concat!(
    r#"{"selection":{"branch":"main"},"collection":"records","field":"embedding","#,
    r#""query":[0.9,0.1,0.5],"k":1,"metric":"cosine"}"#
);
const PAGE_QUERY: &str = "records | '.name'";
const PAGE_DISCARD: &str = r#"{"name":"work"}"#;

/// `haematite write` on `branch`, the value given on the command line as the page
/// gives it.
fn write(socket: &str, branch: &str, kind: &str, value: &str) -> Result<Output, Box<dyn Error>> {
    haematite(
        "write",
        &["--socket", socket, "--branch", branch, kind, value],
        "",
    )
}

/// `haematite call` with params given on the command line as the page gives them.
fn ask(socket: &str, method: &str, params: &str) -> Result<Value, Box<dyn Error>> {
    answered(&call(&["--socket", socket, method, params], "")?)
}

#[test]
fn the_pages_record_lines_put_list_and_read_a_record() -> TestResult {
    let dir = tempfile::tempdir()?;
    let (_service, _) = start(dir.path())?;
    let socket = socket_of(dir.path())?;
    answered(&write(&socket, "main", "collection.put", PAGE_COLLECTION)?)?;
    answered(&write(&socket, "main", "document.put", PAGE_ADA)?)?;
    let read = ask(&socket, "documents.get", PAGE_GET)?;
    assert_eq!(
        read["document"]["raw_json"],
        r#"{"name":"Ada","ward":"4B"}"#
    );
    let listed = ask(&socket, "collections.list", PAGE_COLLECTIONS)?;
    assert_eq!(listed["entries"][0]["name"], "records");
    // With no query settings in serve.json, as here, a query is refused by name.
    let args = ["--socket", socket.as_str(), "--branch", "main", PAGE_QUERY];
    let query = haematite("query", &args, "")?;
    let stderr = text(&query.stderr)?;
    assert_eq!(query.status.code(), Some(2), "{stderr}");
    assert!(
        stderr.starts_with("haematite: refused query_engine_unavailable: "),
        "{stderr}"
    );
    Ok(())
}

#[test]
fn the_pages_vector_lines_put_read_and_search_vectors() -> TestResult {
    let dir = tempfile::tempdir()?;
    let (_service, _) = start(dir.path())?;
    let socket = socket_of(dir.path())?;
    answered(&write(&socket, "main", "collection.put", PAGE_COLLECTION)?)?;
    answered(&write(&socket, "main", "document.put", PAGE_ADA)?)?;
    answered(&write(&socket, "main", "document.put", PAGE_BO)?)?;
    answered(&write(&socket, "main", "vector.put", PAGE_ADA_VECTOR)?)?;
    answered(&write(&socket, "main", "vector.put", PAGE_BO_VECTOR)?)?;
    let vector = ask(&socket, "vectors.get", PAGE_VECTOR_GET)?;
    assert_eq!(vector["vector"], json!([1.0, 0.0, 0.5]));
    let found = ask(&socket, "vectors.search", PAGE_SEARCH)?;
    let hits = found["hits"].as_array().ok_or("missing hits")?;
    assert_eq!(hits.len(), 1, "{found}");
    assert_eq!(hits[0]["document_id"], "ada");
    Ok(())
}

#[test]
fn the_pages_branch_lines_fork_and_discard() -> TestResult {
    let dir = tempfile::tempdir()?;
    let (_service, _) = start(dir.path())?;
    let socket = socket_of(dir.path())?;
    let branches = ask(&socket, "branches.list", r#"{"limit":20}"#)?;
    let version = &branches["entries"][0]["version"];
    let generation = version["generation"].as_str().ok_or("no generation")?;
    let sequence = version["sequence"].as_str().ok_or("no sequence")?;
    let fork = format!(
        r#"{{"name":"work","from":{{"generation":"{generation}","sequence":"{sequence}"}}}}"#
    );
    answered(&write(&socket, "main", "branch.fork", &fork)?)?;
    let listed = ask(&socket, "branches.list", r#"{"limit":20}"#)?;
    assert_eq!(listed["entries"][1]["branch"], "work");
    answered(&write(&socket, "work", "branch.discard", PAGE_DISCARD)?)?;
    let listed = ask(&socket, "branches.list", r#"{"limit":20}"#)?;
    assert_eq!(listed["entries"].as_array().map(Vec::len), Some(1));
    Ok(())
}

/// A reply refusing with `code`.
fn refuse(code: &str) -> String {
    format!(r#"{{"id":"haem-call","error":{{"code":"{code}","message":"stand-in","path":null}}}}"#)
}

#[test]
fn a_change_that_may_have_been_applied_is_not_known_not_refused() -> TestResult {
    let dir = tempfile::tempdir()?;
    let script = vec![vec![Some(refuse("durability_unknown"))]];
    let service = stand_in(dir.path(), 1_048_576, script)?;
    let args = ["--socket", &socket_of(dir.path())?, "mutation.apply", "{}"];
    let output = call(&args, "")?;
    let stderr = text(&output.stderr)?;
    assert_eq!(output.status.code(), Some(3), "{stderr}");
    assert!(output.stdout.is_empty());
    let not_known = "haematite: the outcome is not known: the service answered durability_unknown";
    assert!(stderr.starts_with(not_known), "{stderr}");
    assert!(request(service)?.is_some());
    Ok(())
}

#[test]
fn a_change_refused_before_anything_changed_is_still_refused() -> TestResult {
    let dir = tempfile::tempdir()?;
    let script = vec![vec![Some(refuse("version_conflict"))]];
    let service = stand_in(dir.path(), 1_048_576, script)?;
    let args = ["--socket", &socket_of(dir.path())?, "mutation.apply", "{}"];
    let output = call(&args, "")?;
    let stderr = text(&output.stderr)?;
    assert_eq!(output.status.code(), Some(2), "{stderr}");
    assert!(
        stderr.starts_with("haematite: refused version_conflict: "),
        "{stderr}"
    );
    assert!(request(service)?.is_some());
    Ok(())
}

#[test]
fn a_code_this_client_does_not_know_on_a_read_is_still_refused() -> TestResult {
    let dir = tempfile::tempdir()?;
    let script = vec![vec![Some(refuse("brand_new_code"))]];
    let service = stand_in(dir.path(), 1_048_576, script)?;
    let args = [
        "--socket",
        &socket_of(dir.path())?,
        "branches.list",
        r#"{"limit":5}"#,
    ];
    let output = call(&args, "")?;
    let stderr = text(&output.stderr)?;
    assert_eq!(output.status.code(), Some(2), "{stderr}");
    assert!(
        stderr.starts_with("haematite: refused brand_new_code: "),
        "{stderr}"
    );
    assert!(request(service)?.is_some());
    Ok(())
}

#[test]
fn a_code_this_client_does_not_know_on_a_change_is_not_known() -> TestResult {
    let dir = tempfile::tempdir()?;
    let script = vec![vec![Some(refuse("brand_new_code"))]];
    let service = stand_in(dir.path(), 1_048_576, script)?;
    let args = ["--socket", &socket_of(dir.path())?, "mutation.apply", "{}"];
    let output = call(&args, "")?;
    let stderr = text(&output.stderr)?;
    assert_eq!(output.status.code(), Some(3), "{stderr}");
    assert!(
        stderr.contains("this client does not know that code"),
        "{stderr}"
    );
    assert!(request(service)?.is_some());
    Ok(())
}

#[test]
fn independent_stores_with_the_same_basename_start_together() -> TestResult {
    let first = tempfile::tempdir()?;
    let second = tempfile::tempdir()?;
    let (_first_service, first_config) = start(first.path())?;
    let (_second_service, second_config) = start(second.path())?;
    let first_value: Value = serde_json::from_slice(&std::fs::read(first_config)?)?;
    let second_value: Value = serde_json::from_slice(&std::fs::read(second_config)?)?;
    assert_ne!(first_value["name"], second_value["name"]);
    for value in [&first_value, &second_value] {
        let data_dir = Path::new(value["data_dir"].as_str().ok_or("data path is not text")?);
        assert_eq!(data_dir.file_name(), Some(std::ffi::OsStr::new("database")));
    }
    for directory in [first.path(), second.path()] {
        let reply = answered(&call(&["--socket", &socket_of(directory)?, "hello"], "")?)?;
        assert_eq!(reply["protocol"], "haem/1");
    }
    Ok(())
}

#[test]
fn duplicate_name_startup_keeps_the_refusal_and_child_status() -> TestResult {
    let first = tempfile::tempdir()?;
    let second = tempfile::tempdir()?;
    let (_first_service, first_config) = start(first.path())?;
    std::fs::set_permissions(second.path(), std::fs::Permissions::from_mode(0o700))?;
    let mut value: Value = serde_json::from_slice(&std::fs::read(first_config)?)?;
    value["data_dir"] = json!(second.path().join("database"));
    value["socket"] = json!(second.path().join("serve.sock"));
    let second_config = second.path().join("serve.json");
    std::fs::write(&second_config, serde_json::to_vec(&value)?)?;
    let error = Running::start(&second_config, true)
        .err()
        .ok_or("duplicate name became ready")?;
    let failure = error
        .downcast_ref::<served_child::StartupFailure>()
        .ok_or("startup error omitted child evidence")?;
    let status = failure.status.ok_or("child status missing")?;
    assert!(!status.success());
    let stderr = std::str::from_utf8(&failure.stderr)?;
    assert!(stderr.contains("bind service socket:"), "{stderr}");
    assert!(!second.path().join("database").try_exists()?);
    let message = error.to_string();
    assert!(message.contains(stderr), "{message}");
    assert!(message.contains(&status.to_string()), "{message}");
    assert!(message.contains("service exited before ready"), "{message}");
    let reply = answered(&call(
        &["--socket", &socket_of(first.path())?, "hello"],
        "",
    )?)?;
    assert_eq!(reply["protocol"], "haem/1");
    Ok(())
}