haematite-cli 0.21.0

//! `head.follow` on a served connection: a stream of frames under the
//! request's id, decided through Lys like every other verb.

use std::error::Error;
use std::num::{NonZeroU32, NonZeroUsize};
use std::os::unix::net::UnixStream;
use std::sync::Arc;
use std::time::Duration;

use base64::Engine;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use ed25519_dalek::{Signer, SigningKey};
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::{Value, json};

use super::{Limits, connection};
use crate::serve::pass::{AUDIENCE, Gate, Unreachable, now};
use crate::serve::protocol::{BranchPage, Operation, Version};
use crate::serve::{frame, service::Service, storage::Storage};

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

const PAGE: NonZeroU32 = NonZeroU32::MIN.saturating_add(9);
const MAX_FRAME: NonZeroU32 = NonZeroU32::MIN.saturating_add(65_535);
/// Bounds each read of a frame that is due, so a broken follow fails the
/// test rather than hanging it; no frame due ever waits on it.
const WAIT: Duration = Duration::from_secs(30);
const ISSUER: &str = "https://lys.example";
const KID: &str = "haematite-follow-test";

fn signer() -> SigningKey {
    SigningKey::from_bytes(&[11; 32])
}

fn keys() -> String {
    let x = URL_SAFE_NO_PAD.encode(signer().verifying_key().as_bytes());
    json!({"keys": [{"kty": "OKP", "crv": "Ed25519", "kid": KID, "x": x, "alg": "EdDSA", "use": "sig"}]})
        .to_string()
}

/// A pass for `holder` with read on the branches named, under database
/// `wards`, good for ten minutes from the real clock the service reads.
fn pass(holder: &str, branches: &[&str]) -> String {
    let issued = now();
    let rights: Vec<Value> = branches
        .iter()
        .map(|branch| {
            json!({"resource": {"kind": "haematite.branch", "id": format!("wards.{branch}")},
                "actions": ["read"], "mode": "outright", "grant": "follow-grant"})
        })
        .collect();
    let header =
        URL_SAFE_NO_PAD.encode(json!({"alg": "EdDSA", "typ": "JWT", "kid": KID}).to_string());
    let claims = URL_SAFE_NO_PAD.encode(
        json!({
            "iss": ISSUER, "sub": holder, "aud": AUDIENCE, "iat": issued - 10, "exp": issued + 600,
            "holder": {"id": holder, "kind": "person", "responsible": null},
            "rights": rights,
        })
        .to_string(),
    );
    let signature = URL_SAFE_NO_PAD.encode(
        signer()
            .sign(format!("{header}.{claims}").as_bytes())
            .to_bytes(),
    );
    format!("{header}.{claims}.{signature}")
}

fn service(directory: &std::path::Path, lys: bool) -> Result<Service, Box<dyn Error>> {
    let storage = 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),
    )?;
    let service = Service::new(storage, PAGE)?;
    Ok(if lys {
        service.with_lys(Gate::new(&keys(), ISSUER, "wards", Arc::new(Unreachable))?)
    } else {
        service
    })
}

fn head(service: &Service, branch: &str) -> Result<Version, Box<dyn Error>> {
    let answer = service.dispatch(Operation::Branches(BranchPage {
        after: None,
        limit: PAGE,
    }))?;
    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` in this process, beside the connection.
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 fork_trial(service: &Service) -> TestResult {
    let from = head(service, "main")?;
    mutate(
        service,
        "main",
        "fork-trial",
        &json!({"kind": "branch.fork", "value": {"name": "trial", "from": from}}),
    )
}

fn discard_trial(service: &Service) -> TestResult {
    mutate(
        service,
        "trial",
        "discard-trial",
        &json!({"kind": "branch.discard", "value": {"name": "trial"}}),
    )
}

fn send(client: &mut UnixStream, request: &Value) -> TestResult {
    frame::write(client, &serde_json::to_vec(request)?, MAX_FRAME)?;
    Ok(())
}

fn receive(client: &mut UnixStream) -> Result<Value, Box<dyn Error>> {
    let bytes = frame::read(client, MAX_FRAME)?.ok_or("the connection closed")?;
    Ok(serde_json::from_slice(&bytes)?)
}

fn ask(client: &mut UnixStream, request: &Value) -> Result<Value, Box<dyn Error>> {
    send(client, request)?;
    receive(client)
}

fn hello(client: &mut UnixStream) -> TestResult {
    let answer = ask(
        client,
        &json!({"id": "hello", "operation": {"method": "hello", "params": {
            "protocol": crate::serve::protocol::PROTOCOL,
            "capabilities": [crate::serve::protocol::RECEIPT_CAPABILITY],
            "max_frame": MAX_FRAME,
        }}}),
    )?;
    assert!(answer.get("error").is_none(), "hello refused: {answer}");
    Ok(())
}

fn follow(id: &str, branch: &str) -> Value {
    json!({"id": id, "operation": {"method": "head.follow", "params": {"branch": branch}}})
}

fn refused(answer: &Value, id: &str, code: &str) {
    assert_eq!(answer["id"], id, "{answer}");
    assert!(answer.get("result").is_none(), "{answer}");
    assert_eq!(answer["error"]["code"], code, "{answer}");
}

fn headed(answer: &Value, id: &str, branch: &str, version: Version) -> TestResult {
    assert_eq!(answer["id"], id, "{answer}");
    assert_eq!(
        answer["result"],
        json!({"branch": branch, "version": serde_json::to_value(version)?}),
        "{answer}"
    );
    Ok(())
}

/// Run one served connection on its own thread while `test` talks to it,
/// then close the client. A follow still waiting is then ended by
/// discarding trial, so a failed step never leaves the test hanging.
fn with_connection(
    service: &Service,
    test: impl FnOnce(&mut UnixStream) -> TestResult,
) -> TestResult {
    let limits = Limits {
        max_frame: MAX_FRAME,
        max_clients: NonZeroUsize::MIN,
        frame_timeout: Duration::from_secs(60),
    };
    let (mut client, server) = UnixStream::pair()?;
    client.set_read_timeout(Some(WAIT))?;
    std::thread::scope(|scope| {
        let worker = scope.spawn(move || connection(server, service, limits));
        let outcome = test(&mut client);
        drop(client);
        let released = if published(service, "trial")? {
            discard_trial(service)
        } else {
            Ok(())
        };
        let ended = worker.join().map_err(|_| "the connection panicked")?;
        outcome?;
        released?;
        ended?;
        Ok(())
    })
}

fn published(service: &Service, branch: &str) -> Result<bool, Box<dyn Error>> {
    let answer = service.dispatch(Operation::Branches(BranchPage {
        after: None,
        limit: PAGE,
    }))?;
    Ok(answer["entries"]
        .as_array()
        .ok_or("branches has no entries")?
        .iter()
        .any(|row| row["branch"] == branch))
}

/// Fails while a follow answers one frame and stops, sends a head under
/// another id, or leaves the connection unusable once it ends: the head at
/// once, one frame per change, the ending refusal, then the next request is
/// answered on the same connection.
#[test]
fn a_follow_streams_heads_under_its_id_then_ends_and_the_connection_answers() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path(), false)?;
    fork_trial(&service)?;
    with_connection(&service, |client| {
        hello(client)?;
        send(client, &follow("f", "trial"))?;
        headed(&receive(client)?, "f", "trial", head(&service, "trial")?)?;
        mutate(
            &service,
            "trial",
            "one",
            &json!({"kind": "collection.put", "value": {"collection": "one", "schema": null}}),
        )?;
        headed(&receive(client)?, "f", "trial", head(&service, "trial")?)?;
        discard_trial(&service)?;
        refused(&receive(client)?, "f", "branch_unknown");
        let listed = ask(
            client,
            &json!({"id": "after", "operation": {"method": "branches.list", "params": {"limit": 1}}}),
        )?;
        assert_eq!(listed["id"], "after");
        assert!(listed.get("result").is_some(), "{listed}");
        Ok(())
    })
}

/// Fails while a follow runs undecided, or a refused one is sent a head: with
/// no pass it is refused `pass_missing`, with a pass lacking read on the
/// branch it is refused `permission_refused`, each as its one frame; a pass
/// with read on the branch is sent its head.
#[test]
fn a_follow_is_decided_through_lys_and_a_refused_one_gets_no_head() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path(), true)?;
    fork_trial(&service)?;
    with_connection(&service, |client| {
        hello(client)?;
        let bare = ask(client, &follow("bare", "trial"))?;
        refused(&bare, "bare", "pass_missing");
        let present = |token: String| json!({"id": "pass", "operation": {"method": "session.pass", "params": {"pass": token}}});
        let other = ask(client, &present(pass("clinician", &["main"])))?;
        assert!(other.get("result").is_some(), "{other}");
        refused(
            &ask(client, &follow("denied", "trial"))?,
            "denied",
            "permission_refused",
        );
        let reader = ask(client, &present(pass("clinician", &["trial"])))?;
        assert!(reader.get("result").is_some(), "{reader}");
        send(client, &follow("allowed", "trial"))?;
        headed(
            &receive(client)?,
            "allowed",
            "trial",
            head(&service, "trial")?,
        )?;
        discard_trial(&service)?;
        refused(&receive(client)?, "allowed", "branch_unknown");
        Ok(())
    })
}