haematite-cli 0.21.1

use std::num::NonZeroU32;

use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use lys_pass::drafts::{ApprovedDraft, Execution, RunError, execute_once, request_digest};
use lys_pass::{Mode, Target};
use serde_json::{Value, json};

use super::Drafts;
use crate::serve::live::LysLive;
use crate::serve::protocol::{BranchPage, Operation, Version};
use crate::serve::service::Service;
use crate::serve::storage::Storage;

type TestResult = Result<(), Box<dyn std::error::Error>>;
const LIMIT: NonZeroU32 = NonZeroU32::MIN.saturating_add(9);

fn with_service(test: impl FnOnce(&Service) -> TestResult) -> TestResult {
    let directory = tempfile::tempdir()?;
    let storage = Storage::create(
        DatabaseConfig {
            data_dir: directory.path().join("database"),
            shard_count: 1,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::KeepAll,
        },
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    let service = Service::new(storage, LIMIT)?;
    test(&service)
}

fn branches(service: &Service) -> Result<Vec<Value>, Box<dyn std::error::Error>> {
    let answer = service.dispatch(Operation::Branches(BranchPage {
        after: None,
        limit: LIMIT,
    }))?;
    Ok(answer["entries"]
        .as_array()
        .ok_or("branches has no entries")?
        .clone())
}

fn head(service: &Service) -> Result<Version, Box<dyn std::error::Error>> {
    let main = branches(service)?
        .into_iter()
        .find(|row| row["branch"] == "main")
        .ok_or("main branch missing")?;
    Ok(serde_json::from_value(main["version"].clone())?)
}

/// A fork of main as a client frames it, at main's head.
fn fork_words(service: &Service) -> Result<String, Box<dyn std::error::Error>> {
    let head = head(service)?;
    Ok(
        json!({"id": "7", "operation": {"method": "mutation.apply", "params": {
            "request_id": "f", "branch": "main", "expected": head,
            "change": {"kind": "branch.fork", "value": {"name": "trial", "from": head}},
        }}})
        .to_string(),
    )
}

fn approved(words: &str, target: Target) -> ApprovedDraft {
    ApprovedDraft {
        id: "draft-1".to_owned(),
        app: "haematite".to_owned(),
        grant: "branch-by-draft".to_owned(),
        target,
        request_digest: request_digest(words),
        approved_by: vec!["person-tom".to_owned()],
        mode: Mode::ByDraft,
        words: words.to_owned(),
    }
}

fn forks(service: &Service) -> Result<usize, Box<dyn std::error::Error>> {
    Ok(branches(service)?
        .iter()
        .filter(|row| row["branch"] == "trial")
        .count())
}

/// D4: an approved draft is applied once. A second run, as after Lys missed
/// the first acknowledgment, answers the same receipt and applies nothing.
#[test]
fn an_approved_change_is_applied_exactly_once() -> TestResult {
    with_service(|service| {
        let words = fork_words(service)?;
        let draft = approved(
            &words,
            Target::new("haematite.database", "wards", "branch")?,
        );
        let mut drafts = Drafts::new(service, "wards");
        let first = execute_once(&draft, &mut drafts)?;
        let Execution::Executed { request_digest, .. } = &first else {
            return Err(format!("the draft was not executed: {first:?}").into());
        };
        assert_eq!(request_digest, &draft.request_digest);
        assert_eq!(forks(service)?, 1);
        assert_eq!(execute_once(&draft, &mut drafts)?, first);
        assert_eq!(forks(service)?, 1, "a second run forked again");
        Ok(())
    })
}

/// A draft approved for one resource never applies words acting on another:
/// it is refused on execution by name, before anything is written, so Lys
/// records the refusal rather than offering the draft again.
#[test]
fn a_draft_whose_words_act_on_another_resource_is_refused() -> TestResult {
    with_service(|service| {
        let words = fork_words(service)?;
        let draft = approved(
            &words,
            Target::new("haematite.database", "clinic", "branch")?,
        );
        let mut drafts = Drafts::new(service, "wards");
        let refused = execute_once(&draft, &mut drafts)?;
        let Execution::RefusedOnExecution { refusal, .. } = &refused else {
            return Err(format!("a draft for another database was applied: {refused:?}").into());
        };
        assert_eq!(refusal, "draft_target_differs");
        assert_eq!(forks(service)?, 0);
        Ok(())
    })
}

/// Only a change is applied from a draft.
#[test]
fn a_draft_of_a_read_is_refused() -> TestResult {
    with_service(|service| {
        let words = r#"{"id":"8","operation":{"method":"branch.merge_prepare","params":{"branch":"main","limit":10}}}"#;
        let draft = approved(
            words,
            Target::new("haematite.branch", "wards.main", "merge")?,
        );
        let mut drafts = Drafts::new(service, "wards");
        let refused = execute_once(&draft, &mut drafts)?;
        let Execution::RefusedOnExecution {
            refusal, reason, ..
        } = &refused
        else {
            return Err(format!("a read was run from a draft: {refused:?}").into());
        };
        assert_eq!(refusal, "draft_not_a_change");
        assert!(
            !reason.is_empty(),
            "a refusal on execution carries no reason"
        );
        Ok(())
    })
}

/// D10: with Lys unreachable the runner pulls nothing and applies nothing.
#[test]
fn a_run_against_an_unreachable_lys_applies_nothing() -> TestResult {
    with_service(|service| {
        let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
        let port = listener.local_addr()?.port();
        drop(listener);
        let live = LysLive::new(&format!("http://127.0.0.1:{port}/"))?;
        let mut drafts = Drafts::new(service, "wards");
        let Err(RunError::Pull(error)) = live
            .run_approved("connector", &mut drafts, None)
            .map_err(|error| *error)
        else {
            return Err("an unreachable Lys answered a page".into());
        };
        assert_eq!(error.name(), "lys_could_not_be_asked");
        assert_eq!(forks(service)?, 0);
        Ok(())
    })
}

const fn fixture_long_wait_interval() -> std::time::Duration {
    std::time::Duration::from_secs(1)
}

/// A live check whose one approved page is `draft`, applied through the
/// runner exactly as lys-pass's pull applies each draft it is given.
struct ApprovedPage(ApprovedDraft);

impl crate::serve::pass::LiveCheck for ApprovedPage {
    fn ask(
        &self,
        _token: &str,
        _holder: &str,
        _target: &Target,
    ) -> Result<lys_pass::deliberate::CheckAnswer, lys_pass::Error> {
        Err(lys_pass::Error::CannotAsk("this double answers no check"))
    }

    fn draft(
        &self,
        _token: &str,
        _request: &lys_pass::drafts::DraftRequest,
        _judgment: &lys_pass::deliberate::CheckAnswer,
    ) -> Result<String, lys_pass::Error> {
        Err(lys_pass::Error::CannotAsk("this double records no draft"))
    }

    fn run_drafts(
        &self,
        _connector: &str,
        drafts: &mut Drafts<'_>,
        _after: Option<&str>,
    ) -> Result<lys_pass::drafts::RunReport, Box<RunError<crate::serve::storage::Failure>>> {
        execute_once(&self.0, drafts).map_err(|source| {
            Box::new(RunError::Execution {
                draft: self.0.id.clone(),
                source,
            })
        })?;
        Ok(lys_pass::drafts::RunReport {
            acknowledged: 1,
            next: None,
        })
    }
}

fn gate(
    live: impl crate::serve::pass::LiveCheck + Send + Sync + 'static,
) -> Result<crate::serve::pass::Gate, lys_pass::Error> {
    crate::serve::pass::Gate::new(
        lys_pass::conformance::KEYS,
        lys_pass::conformance::ISSUER,
        "wards",
        std::sync::Arc::new(live),
    )
}

/// `drafts.run` applies the page Lys approved with the connection's pass as
/// the connector's, and answers what was acknowledged and where to go next.
#[test]
fn drafts_run_applies_the_approved_page_and_answers_its_report() -> TestResult {
    with_service(|service| {
        let words = fork_words(service)?;
        let draft = approved(
            &words,
            Target::new("haematite.database", "wards", "branch")?,
        );
        let answer = gate(ApprovedPage(draft))?.run_drafts(service, Some("connector"), None)?;
        assert_eq!(answer, json!({"acknowledged": 1, "next": null}));
        assert_eq!(forks(service)?, 1);
        Ok(())
    })
}

/// Without a pass the runner runs nothing; with Lys unreachable it is
/// refused by name and applies nothing (D10).
#[test]
fn drafts_run_needs_a_pass_and_a_reachable_lys() -> TestResult {
    with_service(|service| {
        let words = fork_words(service)?;
        let draft = approved(
            &words,
            Target::new("haematite.database", "wards", "branch")?,
        );
        let missing = gate(ApprovedPage(draft))?
            .run_drafts(service, None, None)
            .err()
            .ok_or("drafts ran without a pass")?;
        assert_eq!(missing.code, "pass_missing");
        let unreachable = gate(crate::serve::pass::Unreachable)?
            .run_drafts(service, Some("connector"), None)
            .err()
            .ok_or("drafts ran with no Lys to ask")?;
        assert_eq!(unreachable.code, "lys_could_not_be_asked");
        assert_eq!(forks(service)?, 0);
        Ok(())
    })
}

/// A service not served through Lys runs no drafts.
#[test]
fn drafts_run_without_lys_is_an_invalid_operation() -> TestResult {
    with_service(|service| {
        let refused = service
            .run_drafts(
                Some("connector"),
                &crate::serve::protocol::DraftsRun { after: None },
            )
            .err()
            .ok_or("a service without Lys ran drafts")?;
        assert_eq!(refused.code, "invalid_operation");
        Ok(())
    })
}