haematite-cli 0.21.8

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

use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::value::RawValue;

use super::super::super::Storage;
use super::*;
use crate::serve::protocol::{Change, CollectionWrite, DocumentWrite, Fork, Mutation};

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

fn fixture() -> Result<(Storage, tempfile::TempDir), Box<dyn Error>> {
    let directory = tempfile::tempdir()?;
    let mut 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 expected = storage
        .capture(&Selection {
            branch: "main".to_owned(),
            version: None,
        })?
        .version;
    let collection = storage.mutate(&Mutation {
        request_id: "compare-collection".to_owned(),
        branch: "main".to_owned(),
        expected,
        change: Change::Collection(CollectionWrite {
            collection: "records".to_owned(),
            schema: None,
        }),
    })?;
    let fork = storage.mutate(&Mutation {
        request_id: "compare-fork".to_owned(),
        branch: "main".to_owned(),
        expected: collection.version,
        change: Change::Fork(Fork {
            name: "feature".to_owned(),
            from: collection.version,
        }),
    })?;
    storage.mutate(&Mutation {
        request_id: "compare-document".to_owned(),
        branch: "feature".to_owned(),
        expected: fork.version,
        change: Change::Put(DocumentWrite {
            collection: "records".to_owned(),
            document_id: "one".to_owned(),
            document: RawValue::from_string(r#"{"n":1}"#.to_owned())?,
        }),
    })?;
    storage.mutate(&Mutation {
        request_id: "compare-target-document".to_owned(),
        branch: "main".to_owned(),
        expected: collection.version,
        change: Change::Put(DocumentWrite {
            collection: "records".to_owned(),
            document_id: "one".to_owned(),
            document: RawValue::from_string(r#"{"n":2}"#.to_owned())?,
        }),
    })?;
    Ok((storage, directory))
}

#[test]
fn a_node_budget_below_one_unequal_root_pair_refuses_without_progress() -> TestResult {
    let (storage, directory) = fixture()?;
    let before = storage.capture(&Selection {
        branch: "feature".to_owned(),
        version: None,
    })?;
    let request = ReviewPage {
        branch: "feature".to_owned(),
        after: None,
        limit: NonZeroU32::MIN,
    };
    let result = page_with_node_budget(
        &storage.reader()?,
        &storage.catalogue(),
        &request,
        NonZeroUsize::MIN,
        2,
    );
    assert_eq!(
        result.err().ok_or("node budget accepted")?.code,
        "node_budget_too_small"
    );
    let after = storage.capture(&Selection {
        branch: "feature".to_owned(),
        version: None,
    })?;
    assert_eq!(before.version, after.version);
    assert_eq!(before.shards, after.shards);
    assert!(directory.path().join("database").is_dir());
    drop(storage);
    drop(directory);
    Ok(())
}

#[test]
fn both_comparisons_stay_inside_their_shared_node_and_change_budgets() -> TestResult {
    let (storage, directory) = fixture()?;
    let reader = storage.reader()?;
    let catalogue = storage.catalogue();
    let mut request = ReviewPage {
        branch: "feature".to_owned(),
        after: None,
        limit: NonZeroU32::MIN,
    };
    let mut target_seen = false;
    for _ in 0..32 {
        let result = page_with_node_budget(&reader, &catalogue, &request, NonZeroUsize::MIN, 32)?;
        assert!(result.nodes_loaded <= 32);
        assert!(result.source.len() <= 1);
        assert!(result.target.len() <= 1);
        target_seen |= !result.target.is_empty();
        let Some(next) = result.next_after else {
            assert!(target_seen);
            drop(reader);
            drop(catalogue);
            drop(storage);
            drop(directory);
            return Ok(());
        };
        assert_ne!(request.after.as_ref(), Some(&next));
        request.after = Some(next);
    }
    Err("comparison fixture did not exhaust".into())
}

fn request(after: Option<String>, limit: NonZeroU32) -> ReviewPage {
    ReviewPage {
        branch: "feature".to_owned(),
        after,
        limit,
    }
}

fn limits(limit: usize) -> Result<(NonZeroU32, NonZeroUsize), Box<dyn Error>> {
    Ok((
        NonZeroU32::new(u32::try_from(limit)?).ok_or("no limit")?,
        NonZeroUsize::new(limit).ok_or("no limit")?,
    ))
}

/// The keys of `changes`, in order.
fn keys(changes: &[ObservedChange]) -> Vec<Vec<u8>> {
    changes.iter().map(|change| change.key().to_vec()).collect()
}

/// The one shard's changes scanned on each side of the comparison, read
/// whole: the most either side holds, and the whole page.
fn compared_whole(storage: &Storage) -> Result<(usize, Page), Box<dyn Error>> {
    let (wide, wide_size) = limits(64)?;
    let whole = page(
        &storage.reader()?,
        &storage.catalogue(),
        &request(None, wide),
        wide_size,
    )?;
    if whole.next_after.is_some() {
        return Err("the fixture did not fit one page of 64".into());
    }
    Ok((whole.source_scanned.max(whole.target_scanned), whole))
}

/// JT2. Fails while a comparison whose fuller side holds exactly its limit
/// says there is more: on one shard, a page whose limit is the most either
/// side scans is the whole comparison and is done.
#[test]
fn a_comparison_holding_exactly_its_limit_is_done_on_its_first_page() -> TestResult {
    let (storage, _directory) = fixture()?;
    let (most, whole) = compared_whole(&storage)?;
    let (limit, size) = limits(most)?;
    let first = page(
        &storage.reader()?,
        &storage.catalogue(),
        &request(None, limit),
        size,
    )?;
    assert_eq!(first.next_after, None);
    assert_eq!(keys(&first.source), keys(&whole.source));
    assert_eq!(keys(&first.target), keys(&whole.target));
    Ok(())
}

/// JT2. Guards the change past the limit: a limit one below the most either
/// side scans gives a first page that is not done and a second that is, and
/// the two pages hold every change of the whole comparison once, in order.
#[test]
fn a_comparison_one_past_its_limit_takes_the_one_left_on_its_second_page() -> TestResult {
    let (storage, _directory) = fixture()?;
    let (most, whole) = compared_whole(&storage)?;
    let (limit, size) = limits(most.checked_sub(1).ok_or("nothing scanned")?)?;
    let reader = storage.reader()?;
    let catalogue = storage.catalogue();
    let first = page(&reader, &catalogue, &request(None, limit), size)?;
    let after = first
        .next_after
        .clone()
        .ok_or("the first page claimed the end")?;
    let second = page(&reader, &catalogue, &request(Some(after), limit), size)?;
    assert_eq!(second.next_after, None);
    let source = [keys(&first.source), keys(&second.source)].concat();
    let target = [keys(&first.target), keys(&second.target)].concat();
    assert_eq!((source, target), (keys(&whole.source), keys(&whole.target)));
    Ok(())
}

/// A review page's lines as JSON, in order.
fn lines(page: &super::super::Page) -> Result<Vec<serde_json::Value>, serde_json::Error> {
    page.diff.iter().map(serde_json::to_value).collect()
}

/// The review's one-shard changes scanned from the fork to the branch's head,
/// read whole, and its whole page as JSON lines.
fn reviewed_whole(storage: &Storage) -> Result<(usize, Vec<serde_json::Value>), Box<dyn Error>> {
    let (wide, wide_size) = limits(64)?;
    let reader = storage.reader()?;
    let whole = super::super::page(
        &reader,
        &storage.catalogue(),
        &request(None, wide),
        wide_size,
    )?;
    if whole.next_after.is_some() {
        return Err("the fixture did not fit one page of 64".into());
    }
    let parent = storage.capture(&Selection {
        branch: whole.from.branch.clone(),
        version: Some(whole.from.version),
    })?;
    let (branch, fork) = (
        whole.selection.shards.first().ok_or("no branch shard")?,
        parent.shards.first().ok_or("no fork shard")?,
    );
    let budget = DiffBudget {
        max_changes: wide_size,
        max_nodes: NonZeroUsize::new(NODE_BUDGET).ok_or("no node budget")?,
    };
    let (_, scanned) = owner_changes_page(
        &reader,
        branch.shard_id,
        fork.head,
        branch.head,
        None,
        budget,
    )?;
    Ok((scanned, lines(&whole)?))
}

/// JT2. Fails while a review whose one shard holds exactly its limit says
/// there is more: a page whose limit is every change scanned is the whole
/// review and is done.
#[test]
fn a_review_holding_exactly_its_limit_is_done_on_its_first_page() -> TestResult {
    let (storage, _directory) = fixture()?;
    let (scanned, whole) = reviewed_whole(&storage)?;
    let (limit, size) = limits(scanned)?;
    let reader = storage.reader()?;
    let first = super::super::page(&reader, &storage.catalogue(), &request(None, limit), size)?;
    assert_eq!(first.next_after, None);
    assert_eq!(lines(&first)?, whole);
    Ok(())
}

/// JT2. Guards the change past the limit: a limit one below every change
/// scanned gives a first page that is not done and a second that is, and the
/// two pages hold the whole review's lines once, in order.
#[test]
fn a_review_one_past_its_limit_takes_the_one_left_on_its_second_page() -> TestResult {
    let (storage, _directory) = fixture()?;
    let (scanned, whole) = reviewed_whole(&storage)?;
    let (limit, size) = limits(scanned.checked_sub(1).ok_or("nothing scanned")?)?;
    let reader = storage.reader()?;
    let catalogue = storage.catalogue();
    let first = super::super::page(&reader, &catalogue, &request(None, limit), size)?;
    let after = first
        .next_after
        .clone()
        .ok_or("the first page claimed the end")?;
    let second = super::super::page(&reader, &catalogue, &request(Some(after), limit), size)?;
    assert_eq!(second.next_after, None);
    assert_eq!([lines(&first)?, lines(&second)?].concat(), whole);
    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
}