haematite-cli 0.21.0

//! ACCESS-004: every collection and branch is placed in Lys under its
//! database, after the write that makes it commits and once at start, with a
//! stable operation id; a refused placement leaves the write standing.

use std::io::{Read, Write};
use std::net::TcpListener;
use std::num::NonZeroU32;
use std::path::{Path, PathBuf};
use std::sync::mpsc;
use std::time::Duration;

use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::{Value, json};

use super::{Child, Kind, Lys, Outcome, Placer, operation_id};
use crate::serve::cli::LysConfig;
use crate::serve::protocol::{BranchPage, CollectionPage, Operation, Selection, Version};
use crate::serve::service::Service;
use crate::serve::storage::Storage;

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

const CREDENTIAL: &str = "lys-app.haematite.never-shown";
const DATABASE: &str = "wards";
const LIMIT: NonZeroU32 = NonZeroU32::MIN.saturating_add(9);
/// The bound on waiting for a placement that is due; one never comes late
/// unless placing is broken, so this only stops a broken test hanging.
const WAIT: Duration = Duration::from_secs(30);

/// One request the double of Lys was sent.
#[derive(Debug)]
struct Seen {
    path: String,
    authorization: String,
    body: Value,
}

/// How the double answers a placement: its status and JSON body.
type Answer = fn(&Value) -> (u16, Value);

/// Lys's answer to a placement it keeps: the operation it was kept under.
fn kept(body: &Value) -> (u16, Value) {
    (200, json!({"placed": {}, "operation": body["operation"]}))
}

fn already(_: &Value) -> (u16, Value) {
    (
        400,
        json!({
            "refusal": "placement_invalid",
            "reason": "placement_invalid: the resource is already placed in a parent",
            "fields": [],
        }),
    )
}

fn unavailable(_: &Value) -> (u16, Value) {
    (
        503,
        json!({"refusal": "apps_unavailable", "reason": "apps_unavailable: down", "fields": []}),
    )
}

/// A double of Lys answering every request by `answer`; each request goes to
/// the returned channel once it has been answered.
fn double(answer: Answer) -> Result<(String, mpsc::Receiver<Seen>), std::io::Error> {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let origin = format!("http://{}/", listener.local_addr()?);
    let (sent, seen) = mpsc::channel();
    std::thread::spawn(move || {
        for stream in listener.incoming() {
            let Ok(mut stream) = stream else { return };
            let Some((head, body)) = request(&mut stream) else {
                continue;
            };
            let body: Value = serde_json::from_str(&body).unwrap_or(Value::Null);
            let (status, answer) = answer(&body);
            let answer = answer.to_string();
            let response = format!(
                "HTTP/1.1 {status} Answered\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{answer}",
                answer.len()
            );
            if stream.write_all(response.as_bytes()).is_err() {
                return;
            }
            let path = head.split(' ').nth(1).unwrap_or_default().to_owned();
            let authorization = header(&head, "authorization").unwrap_or_default();
            if sent
                .send(Seen {
                    path,
                    authorization,
                    body,
                })
                .is_err()
            {
                return;
            }
        }
    });
    Ok((origin, seen))
}

/// One HTTP request's head and body, read whole.
fn request(stream: &mut std::net::TcpStream) -> Option<(String, String)> {
    let mut bytes = Vec::new();
    loop {
        let mut buffer = [0; 4096];
        let count = stream.read(&mut buffer).ok()?;
        if count == 0 {
            return None;
        }
        bytes.extend_from_slice(buffer.get(..count).unwrap_or_default());
        let text = String::from_utf8_lossy(&bytes).into_owned();
        let Some((head, body)) = text.split_once("\r\n\r\n") else {
            continue;
        };
        let length = header(head, "content-length")
            .and_then(|value| value.parse::<usize>().ok())
            .unwrap_or(0);
        if body.len() >= length {
            return Some((head.to_owned(), body.to_owned()));
        }
    }
}

fn header(head: &str, wanted: &str) -> Option<String> {
    head.lines().find_map(|line| {
        let (name, value) = line.split_once(':')?;
        name.eq_ignore_ascii_case(wanted)
            .then(|| value.trim().to_owned())
    })
}

fn credential(directory: &Path) -> Result<PathBuf, std::io::Error> {
    let path = directory.join("credential");
    std::fs::write(&path, format!("{CREDENTIAL}\n"))?;
    Ok(path)
}

fn child(kind: Kind, name: &str) -> Child {
    Child {
        kind,
        name: name.to_owned(),
    }
}

fn place(lys: &Lys, child: &Child) -> Result<Result<Outcome, String>, std::io::Error> {
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()?;
    Ok(runtime.block_on(lys.place(child)))
}

/// The placement Lys must be sent for `child` of the database `wards`.
fn placement(kind: Kind, name: &str) -> Value {
    json!({
        "operation": operation_id(DATABASE, &child(kind, name)),
        "child": {"kind": kind.lys(), "id": format!("{DATABASE}.{name}")},
        "parent": {"kind": "haematite.database", "id": DATABASE},
        "restricted": false,
    })
}

#[test]
fn the_operation_id_is_lys_shaped_and_one_per_database_kind_and_name() {
    let patients = child(Kind::Collection, "patients");
    let id = operation_id(DATABASE, &patients);
    let digits = id.strip_prefix("op-").unwrap_or_default();
    assert_eq!(digits.len(), 32, "{id} is not op- and 32 digits");
    assert!(
        digits
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)),
        "{id} is not lowercase hex"
    );
    assert_eq!(
        id,
        operation_id(DATABASE, &patients),
        "a retry is a new act"
    );
    for other in [
        operation_id("theatres", &patients),
        operation_id(DATABASE, &child(Kind::Branch, "patients")),
        operation_id(DATABASE, &child(Kind::Collection, "patient")),
    ] {
        assert_ne!(id, other, "two placements share an operation");
    }
}

#[test]
fn a_placement_goes_to_the_route_with_the_app_credential_under_the_database() -> TestResult {
    let directory = tempfile::tempdir()?;
    let (issuer, seen) = double(kept)?;
    let lys = Lys::new(&issuer, DATABASE, &credential(directory.path())?)?;
    let patients = child(Kind::Collection, "patients");
    assert_eq!(place(&lys, &patients)?, Ok(Outcome::Placed));
    let request = seen.recv_timeout(WAIT)?;
    assert_eq!(request.path, "/api/apps/haematite/placements");
    assert_eq!(request.authorization, format!("Bearer {CREDENTIAL}"));
    assert_eq!(request.body, placement(Kind::Collection, "patients"));
    assert_eq!(
        place(&lys, &child(Kind::Branch, "trial"))?,
        Ok(Outcome::Placed)
    );
    assert_eq!(
        seen.recv_timeout(WAIT)?.body,
        placement(Kind::Branch, "trial")
    );
    Ok(())
}

#[test]
fn a_child_placed_by_another_act_is_already_placed_not_a_failure() -> TestResult {
    let directory = tempfile::tempdir()?;
    let (issuer, _seen) = double(already)?;
    let lys = Lys::new(&issuer, DATABASE, &credential(directory.path())?)?;
    assert_eq!(
        place(&lys, &child(Kind::Collection, "patients"))?,
        Ok(Outcome::AlreadyPlaced)
    );
    Ok(())
}

#[test]
fn a_refused_or_unreachable_placement_is_named() -> TestResult {
    let directory = tempfile::tempdir()?;
    let path = credential(directory.path())?;
    let (issuer, _seen) = double(unavailable)?;
    let refused = place(
        &Lys::new(&issuer, DATABASE, &path)?,
        &child(Kind::Collection, "patients"),
    )?
    .err()
    .ok_or("a refused placement was taken as placed")?;
    assert!(refused.starts_with("apps_unavailable"), "named {refused}");
    let closed = TcpListener::bind("127.0.0.1:0")?;
    let origin = format!("http://{}/", closed.local_addr()?);
    drop(closed);
    let unreachable = place(
        &Lys::new(&origin, DATABASE, &path)?,
        &child(Kind::Collection, "patients"),
    )?
    .err()
    .ok_or("an unreachable Lys was taken as placed")?;
    assert!(
        unreachable.starts_with("lys_could_not_be_asked"),
        "named {unreachable}"
    );
    Ok(())
}

#[test]
fn a_credential_file_that_is_relative_or_not_haematites_is_refused() -> TestResult {
    let directory = tempfile::tempdir()?;
    let relative = Lys::new("http://127.0.0.1:1/", DATABASE, Path::new("credential"))
        .err()
        .ok_or("a relative credential file was taken")?;
    assert!(relative.contains("absolute"), "refused as {relative}");
    let wrong = directory.path().join("wrong");
    std::fs::write(&wrong, "lys-app.cambium.x\n")?;
    let refused = Lys::new("http://127.0.0.1:1/", DATABASE, &wrong)
        .err()
        .ok_or("another app's credential was taken")?;
    assert!(
        !refused.contains("lys-app.cambium.x"),
        "the refusal shows the credential: {refused}"
    );
    Ok(())
}

#[test]
fn the_lys_section_takes_an_optional_placement_credential_file() -> TestResult {
    let config: LysConfig = serde_json::from_str(
        r#"{"issuer": "https://lys.example", "database": "wards",
            "placement_credential_file": "/etc/haematite/lys-app.credential"}"#,
    )?;
    assert_eq!(
        config.placement_credential_file.as_deref(),
        Some(Path::new("/etc/haematite/lys-app.credential"))
    );
    let config: LysConfig =
        serde_json::from_str(r#"{"issuer": "https://lys.example", "database": "wards"}"#)?;
    assert!(config.placement_credential_file.is_none());
    Ok(())
}

fn storage(directory: &Path) -> Result<Storage, Box<dyn std::error::Error>> {
    Ok(Storage::create(
        DatabaseConfig {
            data_dir: directory.join("database"),
            shard_count: 1,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::KeepAll,
        },
        NodeCacheBudget::Unlimited,
        Duration::from_secs(1),
    )?)
}

fn head(service: &Service, branch: &str) -> Result<Version, Box<dyn std::error::Error>> {
    let answer = service.dispatch(Operation::Branches(BranchPage {
        after: None,
        limit: LIMIT,
    }))?;
    let row = answer["entries"]
        .as_array()
        .ok_or("branches has no entries")?
        .iter()
        .find(|row| row["branch"] == branch)
        .ok_or_else(|| format!("branch {branch} missing"))?;
    Ok(serde_json::from_value(row["version"].clone())?)
}

/// Apply `change` to `branch` at its head, as request `request_id`.
fn mutate(service: &Service, branch: &str, request_id: &str, change: &Value) -> TestResult {
    let expected = head(service, branch)?;
    let operation: Operation = serde_json::from_value(json!({
        "method": "mutation.apply",
        "params": {"request_id": request_id, "branch": branch, "expected": expected, "change": change},
    }))?;
    service.dispatch(operation)?;
    Ok(())
}

fn put_collection(service: &Service, name: &str) -> TestResult {
    mutate(
        service,
        "main",
        &format!("put-{name}"),
        &json!({"kind": "collection.put", "value": {"collection": name, "schema": null}}),
    )
}

fn fork(service: &Service, name: &str) -> TestResult {
    let from = head(service, "main")?;
    mutate(
        service,
        "main",
        &format!("fork-{name}"),
        &json!({"kind": "branch.fork", "value": {"name": name, "from": from}}),
    )
}

/// The collection names `main` holds at its head.
fn collections(service: &Service) -> Result<Vec<String>, Box<dyn std::error::Error>> {
    let answer = service.dispatch(Operation::Collections(CollectionPage {
        selection: Selection {
            branch: "main".to_owned(),
            version: None,
        },
        after: None,
        limit: LIMIT,
    }))?;
    Ok(answer["entries"]
        .as_array()
        .ok_or("collections has no entries")?
        .iter()
        .filter_map(|row| row["name"].as_str().map(str::to_owned))
        .collect())
}

/// A service over a new store whose placements go to a double answering by
/// `answer`, and what the double was sent.
fn placed_service(
    directory: &Path,
    answer: Answer,
) -> Result<(Service, mpsc::Receiver<Seen>), Box<dyn std::error::Error>> {
    let (issuer, seen) = double(answer)?;
    let lys = Lys::new(&issuer, DATABASE, &credential(directory)?)?;
    let service = Service::new(storage(directory)?, LIMIT)?.with_placer(Placer::start(lys)?)?;
    Ok((service, seen))
}

#[test]
fn a_new_collection_and_branch_are_placed_after_their_writes_commit() -> TestResult {
    let directory = tempfile::tempdir()?;
    let (service, seen) = placed_service(directory.path(), kept)?;
    // At start, the one branch a new store holds.
    assert_eq!(
        seen.recv_timeout(WAIT)?.body,
        placement(Kind::Branch, "main")
    );
    put_collection(&service, "patients")?;
    assert_eq!(
        seen.recv_timeout(WAIT)?.body,
        placement(Kind::Collection, "patients")
    );
    fork(&service, "trial")?;
    assert_eq!(
        seen.recv_timeout(WAIT)?.body,
        placement(Kind::Branch, "trial")
    );
    // A write that makes neither places nothing.
    mutate(
        &service,
        "main",
        "put-one",
        &json!({"kind": "document.put",
            "value": {"collection": "patients", "document_id": "one", "document": {}}}),
    )?;
    assert!(
        seen.recv_timeout(Duration::from_millis(200)).is_err(),
        "a document write was placed"
    );
    Ok(())
}

#[test]
fn a_refused_placement_leaves_the_committed_write_standing() -> TestResult {
    let directory = tempfile::tempdir()?;
    let (service, seen) = placed_service(directory.path(), unavailable)?;
    assert_eq!(
        seen.recv_timeout(WAIT)?.body,
        placement(Kind::Branch, "main")
    );
    put_collection(&service, "patients")?;
    assert_eq!(
        seen.recv_timeout(WAIT)?.body,
        placement(Kind::Collection, "patients")
    );
    assert_eq!(collections(&service)?, vec!["patients".to_owned()]);
    Ok(())
}

/// A store made before placing, or whose placements failed, is healed at
/// the next start: every branch, then every collection on any of them.
#[test]
fn at_start_every_branch_and_collection_the_store_holds_is_placed() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = Service::new(storage(directory.path())?, LIMIT)?;
    put_collection(&service, "patients")?;
    fork(&service, "trial")?;
    mutate(
        &service,
        "trial",
        "put-theatres",
        &json!({"kind": "collection.put", "value": {"collection": "theatres", "schema": null}}),
    )?;
    let (issuer, seen) = double(kept)?;
    let lys = Lys::new(&issuer, DATABASE, &credential(directory.path())?)?;
    let service = service.with_placer(Placer::start(lys)?)?;
    let mut sent = Vec::new();
    for _ in 0..4 {
        sent.push(seen.recv_timeout(WAIT)?.body);
    }
    assert_eq!(
        sent,
        vec![
            placement(Kind::Branch, "main"),
            placement(Kind::Branch, "trial"),
            placement(Kind::Collection, "patients"),
            placement(Kind::Collection, "theatres"),
        ]
    );
    assert!(
        seen.recv_timeout(Duration::from_millis(200)).is_err(),
        "a resource was placed twice"
    );
    assert_eq!(collections(&service)?, vec!["patients".to_owned()]);
    Ok(())
}