haematite-cli 0.21.8

use std::error::Error;

use super::Admission;
use crate::serve::contracts::{MutationRequest, Receipt, ResolvedMutation};
use crate::serve::protocol::{Change, Mutation, Selection, Version};
use crate::serve::storage::{Storage, branch_name, receipt_key};
use haematite::branch::{CommitDurability, CommitRequest, current_timestamp, open};
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::{Value, json};

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

fn current(storage: &Storage) -> Result<Version, Box<dyn Error>> {
    Ok(storage
        .capture(&Selection {
            branch: "main".to_owned(),
            version: None,
        })?
        .version)
}

fn apply(storage: &mut Storage, id: &str, change: &Value) -> TestResult {
    let change: Change = serde_json::from_slice(&serde_json::to_vec(change)?)?;
    storage.mutate(&Mutation {
        request_id: id.to_owned(),
        branch: "main".to_owned(),
        expected: current(storage)?,
        change,
    })?;
    Ok(())
}

fn put(id: &str, value: u64) -> Value {
    json!({"kind":"document.put", "value":{"collection":"charts", "document_id":id, "document":{"value":value}}})
}

fn delete(id: &str) -> Value {
    json!({"kind":"document.delete", "value":{"collection":"charts", "document_id":id}})
}

fn setup() -> Result<(tempfile::TempDir, Storage), Box<dyn Error>> {
    let directory = tempfile::tempdir()?;
    let mut storage = Storage::create(
        DatabaseConfig {
            data_dir: directory.path().join("database"),
            shard_count: 2,
            distributed: None,
            executor_threads: Some(2),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::KeepAll,
        },
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    apply(
        &mut storage,
        "fixture-collection",
        &json!({"kind":"collection.put", "value":{"collection":"charts", "schema":null}}),
    )?;
    Ok((directory, storage))
}

fn resolve(
    id: &str,
    caller: &str,
    contract: &Value,
    change: &Value,
) -> Result<ResolvedMutation, Box<dyn Error>> {
    Ok(MutationRequest::decode(&serde_json::to_vec(&json!({
        "request_id":id, "branch":"main", "caller":{"issuer":"local", "subject":caller},
        "contract":contract, "change":change,
    }))?)?
    .resolve(caller)?)
}

fn dependency(
    id: &str,
    origin: Version,
    reads: &Value,
    change: &Value,
) -> Result<ResolvedMutation, Box<dyn Error>> {
    resolve(
        id,
        "doctor-a",
        &json!({"kind":"dependencies", "origin":origin, "reads":reads}),
        change,
    )
}

fn admitted(storage: &mut Storage, request: &ResolvedMutation) -> TestResult {
    let decision = storage.decide_resolved(request)?;
    assert!(matches!(decision, Admission::Admitted(_)));
    Ok(())
}

fn refused(storage: &mut Storage, request: &ResolvedMutation, code: &str) -> TestResult {
    let error = storage
        .decide_resolved(request)
        .err()
        .ok_or("request was admitted")?;
    assert_eq!(error.code, code);
    Ok(())
}

fn store_receipt(
    storage: &mut Storage,
    request: &ResolvedMutation,
) -> Result<Receipt, Box<dyn Error>> {
    let name = branch_name("main");
    let record = storage.refs.get(&name).ok_or("missing main")?.clone();
    storage.retain_current("main", &record)?;
    let installed = Version {
        generation: record.created,
        sequence: record.seq + 1,
    };
    let receipt = Receipt::for_request(request, installed)?;
    let key = receipt_key("main", installed.generation, request.request_id());
    let branch = open(&name, &storage.refs, &storage.registry)?;
    branch.put(
        storage.writer.shard_for(&key),
        key,
        serde_json::to_vec(&receipt)?,
    )?;
    storage.writer.commit(
        &branch,
        &storage.registry,
        CommitRequest {
            durability: CommitDurability::Durable {
                refs: &mut storage.refs,
            },
            extra_parents: &[],
            timestamp: current_timestamp(),
        },
    )?;
    storage.publish_record(&name)?;
    Ok(receipt)
}

#[test]
fn different_callers_creating_different_documents_from_one_origin_are_admitted() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let origin = current(&storage)?;
    let first = resolve(
        "first",
        "doctor-a",
        &json!({"kind":"dependencies", "origin":origin, "reads":[]}),
        &put("a", 1),
    )?;
    let second = resolve(
        "second",
        "doctor-b",
        &json!({"kind":"dependencies", "origin":origin, "reads":[]}),
        &put("b", 2),
    )?;
    admitted(&mut storage, &first)?;
    apply(&mut storage, "fixture-first", &put("a", 1))?;
    admitted(&mut storage, &second)
}

#[test]
fn changed_declared_document_read_is_a_named_conflict() -> TestResult {
    let (_directory, mut storage) = setup()?;
    apply(&mut storage, "fixture-a", &put("a", 1))?;
    let request = dependency(
        "write",
        current(&storage)?,
        &json!([{"kind":"document", "collection":"charts", "document_id":"a"}]),
        &put("b", 1),
    )?;
    apply(&mut storage, "fixture-change", &put("a", 2))?;
    refused(&mut storage, &request, "conflict")
}

#[test]
fn undeclared_overwrite_of_a_changed_document_is_refused() -> TestResult {
    let (_directory, mut storage) = setup()?;
    apply(&mut storage, "fixture-a", &put("a", 1))?;
    let request = dependency("write", current(&storage)?, &json!([]), &put("a", 3))?;
    apply(&mut storage, "fixture-change", &put("a", 2))?;
    refused(&mut storage, &request, "conflict")
}

#[test]
fn creation_checks_implicit_absence() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let request = dependency("write", current(&storage)?, &json!([]), &put("a", 3))?;
    apply(&mut storage, "fixture-create", &put("a", 2))?;
    refused(&mut storage, &request, "conflict")
}

#[test]
fn declared_absence_detects_a_new_document() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let request = dependency(
        "write",
        current(&storage)?,
        &json!([{"kind":"document", "collection":"charts", "document_id":"a"}]),
        &put("b", 3),
    )?;
    apply(&mut storage, "fixture-create", &put("a", 2))?;
    refused(&mut storage, &request, "conflict")
}

#[test]
fn interval_detects_a_created_phantom() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let request = dependency(
        "write",
        current(&storage)?,
        &json!([{"kind":"source_interval", "collection":"charts", "lower":"b", "upper":"y"}]),
        &put("z", 1),
    )?;
    apply(&mut storage, "fixture-phantom", &put("m", 2))?;
    refused(&mut storage, &request, "conflict")
}

#[test]
fn interval_detects_deleted_documents() -> TestResult {
    let (_directory, mut storage) = setup()?;
    apply(&mut storage, "fixture-a", &put("m", 1))?;
    let request = dependency(
        "write",
        current(&storage)?,
        &json!([{"kind":"source_interval", "collection":"charts", "lower":"b", "upper":"y"}]),
        &put("z", 1),
    )?;
    apply(&mut storage, "fixture-delete", &delete("m"))?;
    refused(&mut storage, &request, "conflict")
}

#[test]
fn interval_detects_same_body_rewrite_through_its_revision() -> TestResult {
    let (_directory, mut storage) = setup()?;
    apply(&mut storage, "fixture-a", &put("m", 1))?;
    let request = dependency(
        "write",
        current(&storage)?,
        &json!([{"kind":"source_interval", "collection":"charts", "lower":"b", "upper":"y"}]),
        &put("z", 1),
    )?;
    apply(&mut storage, "fixture-rewrite", &put("m", 1))?;
    refused(&mut storage, &request, "conflict")
}

#[test]
fn interval_bounds_exclude_the_upper_id() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let request = dependency(
        "write",
        current(&storage)?,
        &json!([{"kind":"source_interval", "collection":"charts", "lower":"b", "upper":"y"}]),
        &put("z", 1),
    )?;
    apply(&mut storage, "fixture-outside", &put("y", 1))?;
    admitted(&mut storage, &request)
}

#[test]
fn schema_drift_is_an_implicit_dependency_for_put() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let request = dependency("write", current(&storage)?, &json!([]), &put("a", 1))?;
    apply(
        &mut storage,
        "fixture-schema",
        &json!({"kind":"collection.put", "value":{"collection":"charts", "schema":{"type":"object"}}}),
    )?;
    refused(&mut storage, &request, "conflict")
}

#[test]
fn collection_definition_read_is_checked() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let request = dependency(
        "write",
        current(&storage)?,
        &json!([{"kind":"collection", "collection":"charts"}]),
        &put("a", 1),
    )?;
    apply(
        &mut storage,
        "fixture-schema",
        &json!({"kind":"collection.put", "value":{"collection":"charts", "schema":false}}),
    )?;
    refused(&mut storage, &request, "conflict")
}

#[test]
fn exact_version_still_refuses_any_intervening_write() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let request = resolve(
        "write",
        "doctor-a",
        &json!({"kind":"exact_version", "expected":current(&storage)?}),
        &put("a", 1),
    )?;
    apply(&mut storage, "fixture-other", &put("b", 1))?;
    refused(&mut storage, &request, "version_conflict")
}

#[test]
fn replay_after_head_moves_preserves_original_receipt_bytes() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let request = dependency("write", current(&storage)?, &json!([]), &put("a", 1))?;
    let original = store_receipt(&mut storage, &request)?;
    apply(&mut storage, "fixture-other", &put("a", 2))?;
    let Admission::Replayed(replay) = storage.decide_resolved(&request)? else {
        return Err("did not replay".into());
    };
    assert_eq!(serde_json::to_vec(&original)?, serde_json::to_vec(&replay)?);
    assert_eq!(replay.request_origin(), request.origin());
    assert!(current(&storage)?.sequence > replay.version().sequence);
    Ok(())
}

#[test]
fn a_reused_id_is_refused_before_changed_dependencies() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let origin = current(&storage)?;
    let request = dependency("write", origin, &json!([]), &put("a", 1))?;
    store_receipt(&mut storage, &request)?;
    let changed = dependency("write", origin, &json!([]), &put("a", 2))?;
    refused(&mut storage, &changed, "request_id_reused")
}

#[test]
fn a_delete_checks_incoming_incident_edges() -> TestResult {
    let (_directory, mut storage) = setup()?;
    apply(&mut storage, "fixture-a", &put("a", 1))?;
    apply(&mut storage, "fixture-b", &put("b", 1))?;
    let request = dependency("delete", current(&storage)?, &json!([]), &delete("a"))?;
    apply(
        &mut storage,
        "fixture-edge",
        &json!({"kind":"edge.put", "value":{
            "relation":"link", "from":{"collection":"charts", "document_id":"b"},
            "to":{"collection":"charts", "document_id":"a"}, "properties":{}
        }}),
    )?;
    refused(&mut storage, &request, "conflict")
}

#[test]
fn a_delete_checks_existing_vector_values() -> TestResult {
    let (_directory, mut storage) = setup()?;
    apply(&mut storage, "fixture-a", &put("a", 1))?;
    apply(
        &mut storage,
        "fixture-vector",
        &json!({"kind":"vector.put", "value":{"collection":"charts", "document_id":"a", "field":"embedding", "vector":[1.0]}}),
    )?;
    let request = dependency("delete", current(&storage)?, &json!([]), &delete("a"))?;
    apply(
        &mut storage,
        "fixture-change",
        &json!({"kind":"vector.put", "value":{"collection":"charts", "document_id":"a", "field":"embedding", "vector":[2.0]}}),
    )?;
    refused(&mut storage, &request, "conflict")
}

#[test]
fn dependencies_refuse_unversioned_edge_and_vector_mutations() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let origin = current(&storage)?;
    for (index, change) in [
        json!({"kind":"edge.delete", "value":{"relation":"link", "from":{"collection":"charts", "document_id":"a"}, "to":{"collection":"charts", "document_id":"b"}}}),
        json!({"kind":"vector.delete", "value":{"collection":"charts", "document_id":"a", "field":"embedding"}}),
    ].into_iter().enumerate() {
        let request = dependency(&format!("write-{index}"), origin, &json!([]), &change)?;
        refused(&mut storage, &request, "dependency_kind_not_admitted")?;
    }
    Ok(())
}

#[test]
fn empty_interval_does_not_conflict_with_an_outside_write() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let request = dependency(
        "write",
        current(&storage)?,
        &json!([{"kind":"source_interval", "collection":"charts", "lower":"m", "upper":"m"}]),
        &put("z", 1),
    )?;
    apply(&mut storage, "fixture-outside", &put("m", 2))?;
    admitted(&mut storage, &request)
}

#[test]
fn delete_refuses_an_undeclared_document_update() -> TestResult {
    let (_directory, mut storage) = setup()?;
    apply(&mut storage, "fixture-a", &put("a", 1))?;
    let request = dependency("delete", current(&storage)?, &json!([]), &delete("a"))?;
    apply(&mut storage, "fixture-change", &put("a", 2))?;
    refused(&mut storage, &request, "conflict")
}

#[test]
fn a_delete_checks_new_vector_field_definitions() -> TestResult {
    let (_directory, mut storage) = setup()?;
    apply(&mut storage, "fixture-a", &put("a", 1))?;
    apply(&mut storage, "fixture-b", &put("b", 1))?;
    let request = dependency("delete", current(&storage)?, &json!([]), &delete("a"))?;
    // The new field's first vector belongs to another document, so only the
    // field definition row can refuse this delete.
    apply(
        &mut storage,
        "fixture-vector",
        &json!({"kind":"vector.put", "value":{"collection":"charts", "document_id":"b", "field":"new-field", "vector":[1.0]}}),
    )?;
    refused(&mut storage, &request, "conflict")
}

#[test]
fn a_delete_admits_an_unrelated_vector_update_in_an_existing_field() -> TestResult {
    let (_directory, mut storage) = setup()?;
    apply(&mut storage, "fixture-a", &put("a", 1))?;
    apply(&mut storage, "fixture-b", &put("b", 1))?;
    apply(
        &mut storage,
        "fixture-vector",
        &json!({"kind":"vector.put", "value":{"collection":"charts", "document_id":"b", "field":"embedding", "vector":[1.0]}}),
    )?;
    let request = dependency("delete", current(&storage)?, &json!([]), &delete("a"))?;
    apply(
        &mut storage,
        "fixture-change",
        &json!({"kind":"vector.put", "value":{"collection":"charts", "document_id":"b", "field":"embedding", "vector":[2.0]}}),
    )?;
    admitted(&mut storage, &request)
}

#[test]
fn exact_version_admits_unversioned_edge_and_vector_mutations_at_head() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let origin = current(&storage)?;
    for (index, change) in [
        json!({"kind":"edge.delete", "value":{"relation":"link", "from":{"collection":"charts", "document_id":"a"}, "to":{"collection":"charts", "document_id":"b"}}}),
        json!({"kind":"vector.delete", "value":{"collection":"charts", "document_id":"a", "field":"embedding"}}),
    ].into_iter().enumerate() {
        let request = resolve(&format!("write-{index}"), "doctor-a", &json!({"kind":"exact_version", "expected":origin}), &change)?;
        admitted(&mut storage, &request)?;
    }
    Ok(())
}

#[test]
fn future_and_wrong_generation_origins_are_refused() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let origin = current(&storage)?;
    let future = Version {
        generation: origin.generation,
        sequence: origin.sequence + 1,
    };
    let wrong = Version {
        generation: origin.generation + 1,
        sequence: origin.sequence,
    };
    refused(
        &mut storage,
        &dependency("future", future, &json!([]), &put("a", 1))?,
        "version_missing",
    )?;
    refused(
        &mut storage,
        &dependency("wrong", wrong, &json!([]), &put("a", 1))?,
        "generation_mismatch",
    )
}

#[test]
fn an_old_receipt_shape_is_refused_instead_of_assigning_an_author() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let origin = current(&storage)?;
    apply(&mut storage, "old-write", &put("a", 1))?;
    let request = dependency("old-write", origin, &json!([]), &put("a", 1))?;
    refused(&mut storage, &request, "stored_receipt_format_unsupported")
}

#[test]
fn interval_conflict_carries_the_changed_document_revision_not_the_head() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let request = dependency(
        "write",
        current(&storage)?,
        &json!([{"kind":"source_interval", "collection":"charts", "lower":"b", "upper":"y"}]),
        &put("z", 1),
    )?;
    apply(&mut storage, "fixture-phantom", &put("m", 2))?;
    let changed = current(&storage)?;
    apply(&mut storage, "fixture-unrelated", &put("a", 2))?;
    let error = storage
        .decide_resolved(&request)
        .err()
        .ok_or("request admitted")?;
    assert_eq!(error.code, "conflict");
    assert!(error.message.contains(&format!(
        "current_revision={}:{}",
        changed.generation, changed.sequence
    )));
    assert_eq!(error.path.as_deref(), Some("/documents/charts/m"));
    Ok(())
}

#[test]
fn a_delete_checks_outgoing_incident_edges() -> TestResult {
    let (_directory, mut storage) = setup()?;
    apply(&mut storage, "fixture-a", &put("a", 1))?;
    apply(&mut storage, "fixture-b", &put("b", 1))?;
    let request = dependency("delete", current(&storage)?, &json!([]), &delete("a"))?;
    apply(
        &mut storage,
        "fixture-edge",
        &json!({"kind":"edge.put", "value":{
            "relation":"link", "from":{"collection":"charts", "document_id":"a"},
            "to":{"collection":"charts", "document_id":"b"}, "properties":{}
        }}),
    )?;
    refused(&mut storage, &request, "conflict")
}

fn import(ids: &[&str]) -> Value {
    let documents: Vec<Value> = ids
        .iter()
        .map(|id| json!({"document_id":id, "document":{"value":1}}))
        .collect();
    json!({"kind":"documents.put_many", "value":{"collection":"charts", "documents":documents}})
}

#[test]
fn an_import_checks_every_row_it_would_overwrite() -> TestResult {
    let (_directory, mut storage) = setup()?;
    apply(&mut storage, "fixture-b", &put("b", 1))?;
    let origin = current(&storage)?;
    let overwriting = dependency("import", origin, &json!([]), &import(&["a", "b"]))?;
    let elsewhere = dependency("other-import", origin, &json!([]), &import(&["a", "c"]))?;
    apply(&mut storage, "fixture-change", &put("b", 2))?;
    refused(&mut storage, &overwriting, "conflict")?;
    admitted(&mut storage, &elsewhere)
}

#[test]
fn an_import_checks_its_collection_definition_once_for_all_rows() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let request = dependency(
        "import",
        current(&storage)?,
        &json!([]),
        &import(&["a", "b"]),
    )?;
    apply(
        &mut storage,
        "fixture-schema",
        &json!({"kind":"collection.put", "value":{"collection":"charts", "schema":{"type":"object"}}}),
    )?;
    refused(&mut storage, &request, "conflict")
}

#[test]
fn source_interval_includes_its_lower_bound() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let request = dependency(
        "write",
        current(&storage)?,
        &json!([{"kind":"source_interval", "collection":"charts", "lower":"b", "upper":"y"}]),
        &put("z", 1),
    )?;
    apply(&mut storage, "fixture-phantom", &put("b", 2))?;
    refused(&mut storage, &request, "conflict")
}

fn append_prune(suffix: &str) -> Value {
    json!({"kind":"document.append_prune", "value":{
        "collection":"charts", "scope":["ward"], "suffix":suffix,
        "document":{"value":2}, "keep":1,
    }})
}

#[test]
fn append_prune_dependencies_refuse_without_writing() -> TestResult {
    let (_directory, mut storage) = setup()?;
    apply(&mut storage, "seed-prune", &put("1/ward/000000", 1))?;
    let before = storage.capture(&Selection {
        branch: "main".to_owned(),
        version: None,
    })?;
    let request = dependency(
        "dependency-prune",
        before.version,
        &json!([]),
        &append_prune("000001"),
    )?;
    refused(&mut storage, &request, "dependency_kind_not_admitted")?;
    let after = storage.capture(&Selection {
        branch: "main".to_owned(),
        version: None,
    })?;
    assert_eq!(after.version, before.version);
    assert!(
        before
            .shards
            .iter()
            .map(|shard| (shard.shard_id, shard.head))
            .eq(after
                .shards
                .iter()
                .map(|shard| (shard.shard_id, shard.head)))
    );
    assert!(
        crate::serve::storage::reads::get_bytes(
            &storage.reader,
            &after,
            &receipt_key("main", after.version.generation, "dependency-prune"),
        )?
        .is_none()
    );
    for key in [
        crate::serve::storage::document_key("charts", "1/ward/000001"),
        crate::serve::storage::revision_key("charts", "1/ward/000001"),
    ] {
        assert!(crate::serve::storage::reads::get_bytes(&storage.reader, &after, &key)?.is_none());
    }
    for key in [
        crate::serve::storage::document_key("charts", "1/ward/000000"),
        crate::serve::storage::revision_key("charts", "1/ward/000000"),
    ] {
        assert_eq!(
            crate::serve::storage::reads::get_bytes(&storage.reader, &before, &key)?,
            crate::serve::storage::reads::get_bytes(&storage.reader, &after, &key)?
        );
    }
    Ok(())
}

#[test]
fn append_prune_exact_version_still_appends_and_prunes_in_one_version() -> TestResult {
    let (_directory, mut storage) = setup()?;
    apply(&mut storage, "seed-prune", &put("1/ward/000000", 1))?;
    let before = current(&storage)?;
    let change = append_prune("000001");
    let request = resolve(
        "exact-prune",
        "doctor-a",
        &json!({"kind":"exact_version", "expected":before}),
        &change,
    )?;
    admitted(&mut storage, &request)?;
    assert_eq!(current(&storage)?, before);
    let receipt = storage.mutate(&Mutation {
        request_id: "exact-prune".to_owned(),
        branch: "main".to_owned(),
        expected: before,
        change: serde_json::from_slice(&serde_json::to_vec(&change)?)?,
    })?;
    let after = storage.capture(&Selection {
        branch: "main".to_owned(),
        version: None,
    })?;
    assert_eq!(after.version.generation, before.generation);
    assert_eq!(after.version.sequence, before.sequence + 1);
    assert_eq!(receipt.version, after.version);
    assert_eq!(
        serde_json::to_value(receipt)?["retention"],
        json!({"document_id":"1/ward/000001", "removed":1, "more":false})
    );
    for key in [
        crate::serve::storage::document_key("charts", "1/ward/000000"),
        crate::serve::storage::revision_key("charts", "1/ward/000000"),
    ] {
        assert!(crate::serve::storage::reads::get_bytes(&storage.reader, &after, &key)?.is_none());
    }
    for key in [
        crate::serve::storage::document_key("charts", "1/ward/000001"),
        crate::serve::storage::revision_key("charts", "1/ward/000001"),
    ] {
        assert!(crate::serve::storage::reads::get_bytes(&storage.reader, &after, &key)?.is_some());
    }
    Ok(())
}

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