haematite-cli 0.21.3

//! `changes.list` asked through the service's own dispatch, as a client's
//! request is read: the method, its members and its refusals by name.

#![forbid(unsafe_code)]

use std::error::Error;
use std::num::{NonZeroU32, NonZeroUsize};

use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use haematite_cli::serve::protocol::{
    Change, CollectionWrite, DocumentWrite, Mutation, Operation, Selection, Version,
};
use haematite_cli::serve::service::Service;
use haematite_cli::serve::storage::Storage;
use serde_json::{Value, json, value::RawValue};

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

const WIRE_PAGE: NonZeroU32 = NonZeroU32::MIN.saturating_add(9);
const BYTES: NonZeroUsize = NonZeroUsize::MIN.saturating_add((1 << 20) - 1);

/// A served store with one collection, and the versions before and after one
/// write of `patients/p1`.
fn served() -> Result<(tempfile::TempDir, Service, Version, Version), Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let mut storage = Storage::create(
        DatabaseConfig {
            data_dir: dir.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(),
    )?;
    let head = Selection {
        branch: "main".to_owned(),
        version: None,
    };
    let mut version = storage.capture(&head)?.version;
    for (id, change) in [
        (
            "declare",
            Change::Collection(CollectionWrite {
                collection: "patients".to_owned(),
                schema: None,
            }),
        ),
        (
            "write",
            Change::Put(DocumentWrite {
                collection: "patients".to_owned(),
                document_id: "p1".to_owned(),
                document: RawValue::from_string(r#"{"name":"Ada"}"#.to_owned())?,
            }),
        ),
    ] {
        let before = version;
        let mutation = Mutation {
            request_id: id.to_owned(),
            branch: "main".to_owned(),
            expected: version,
            change,
        };
        version = storage.mutate(&mutation)?.version;
        if id == "write" {
            let service = Service::with_query(storage, WIRE_PAGE, None)?;
            return Ok((dir, service, before, version));
        }
    }
    Err("the write was not made".into())
}

/// `changes.list` with `params`, read as the transport reads a request.
fn changes(params: Value) -> Result<Operation, serde_json::Error> {
    serde_json::from_value(Value::Object(serde_json::Map::from_iter([
        ("method".to_owned(), json!("changes.list")),
        ("params".to_owned(), params),
    ])))
}

fn moment(version: Version) -> Result<Value, serde_json::Error> {
    Ok(json!({"branch": "main", "version": serde_json::to_value(version)?}))
}

#[test]
fn changes_list_answers_one_writes_change_through_the_service() -> TestResult {
    let (_dir, service, before, after) = served()?;
    let params = json!({
        "from": moment(before)?,
        "to": moment(after)?,
        "page": {"rows": 10, "bytes": 65_536},
    });
    let answer = service.dispatch_bounded(changes(params)?, BYTES)?;
    let row = json!({
        "kind": "added",
        "origin": {"collection": "patients", "id": "p1"},
        "before": null,
        "after": serde_json::to_value(after)?,
    });
    assert_eq!(answer["rows"], json!([row]));
    assert!(answer["next"].is_null(), "{answer}");
    assert_eq!(answer["to"], moment(after)?);
    Ok(())
}

#[test]
fn an_unknown_member_is_refused_by_name() -> TestResult {
    let (_dir, _service, before, after) = served()?;
    let params = json!({
        "from": moment(before)?,
        "to": moment(after)?,
        "page": {"rows": 10, "bytes": 65_536},
        "since": "yesterday",
    });
    let error = changes(params)
        .err()
        .ok_or("a request with an unknown member was read")?;
    assert!(
        error.to_string().contains("unknown field `since`"),
        "{error}"
    );
    Ok(())
}

#[test]
fn a_moment_with_no_version_is_refused_by_name() -> TestResult {
    let (_dir, service, _before, after) = served()?;
    let params = json!({
        "from": {"branch": "main"},
        "to": moment(after)?,
        "page": {"rows": 10, "bytes": 65_536},
    });
    let refused = service
        .dispatch_bounded(changes(params)?, BYTES)
        .err()
        .ok_or("a moment with no version was answered")?;
    assert_eq!(refused.code, "changes_version_required");
    Ok(())
}

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