liminal-server 0.24.1

use std::cell::Cell;
use std::error::Error;
use std::sync::Arc;

use liminal::durability::bridge::block_on;
use liminal::durability::{CountingStore, DurableStore, RoutedBatchItem};

use super::OperationBatch;
use crate::server::participant::production::tests_memory_store::MemoryStore;
use crate::server::participant::production::tests_seed::Shared;

#[test]
fn covered_operations_and_checkpoint_publish_in_one_store_call() -> Result<(), Box<dyn Error>> {
    let memory = Arc::new(MemoryStore::default());
    let store = CountingStore::new(Shared(Arc::clone(&memory) as Arc<dyn DurableStore>));
    let mut batch = OperationBatch::new("conversation");
    for operation in 0_u64..257 {
        batch.stage(RoutedBatchItem {
            key: format!("operation/{operation}").into_bytes(),
            expected: None,
            value: operation.to_be_bytes().to_vec(),
            ttl: None,
        })?;
    }
    assert!(OperationBatch::checkpoint_due(0, 257, false)?);
    batch.stage(RoutedBatchItem {
        key: b"checkpoint".to_vec(),
        expected: None,
        value: b"covered:257".to_vec(),
        ttl: None,
    })?;
    let submissions = Cell::new(0_usize);
    block_on(batch.commit(&store, || {
        assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
        submissions.set(submissions.get() + 1);
    }))??;
    assert_eq!(submissions.get(), 1);
    let counts = store.snapshot();
    assert_eq!(counts.apply_routed_batch_calls, 1);
    assert_eq!(counts.apply_routed_batch_items, 258);
    assert_eq!(counts.append_calls, 0);
    assert_eq!(counts.append_all_calls, 0);
    assert_eq!(counts.flushes, 0);
    for operation in 0_u64..257 {
        assert_eq!(
            block_on(
                store.get_routed("conversation", format!("operation/{operation}").as_bytes())
            )??,
            Some(operation.to_be_bytes().to_vec())
        );
    }
    assert_eq!(
        block_on(store.get_routed("conversation", b"checkpoint"))??,
        Some(b"covered:257".to_vec())
    );
    Ok(())
}

#[test]
fn sealing_and_crossings_require_a_checkpoint_without_clock_waits() -> Result<(), Box<dyn Error>> {
    assert!(!OperationBatch::checkpoint_due(0, 255, false)?);
    assert!(OperationBatch::checkpoint_due(255, 256, false)?);
    assert!(!OperationBatch::checkpoint_due(256, 257, false)?);
    assert!(OperationBatch::checkpoint_due(257, 257, true)?);
    assert!(OperationBatch::checkpoint_due(257, 256, false).is_err());
    Ok(())
}

#[test]
fn repeated_keys_are_refused_before_commit() -> Result<(), Box<dyn Error>> {
    let mut batch = OperationBatch::new("conversation");
    batch.stage(RoutedBatchItem {
        key: b"operation".to_vec(),
        expected: None,
        value: vec![1],
        ttl: None,
    })?;
    assert!(
        batch
            .stage(RoutedBatchItem {
                key: b"operation".to_vec(),
                expected: None,
                value: vec![2],
                ttl: None
            })
            .is_err()
    );
    Ok(())
}

#[test]
fn an_empty_logical_operation_is_refused_without_submitting() -> Result<(), Box<dyn Error>> {
    let store = CountingStore::new(MemoryStore::default());
    let submissions = Cell::new(0_usize);
    let refused = block_on(OperationBatch::new("conversation").commit(&store, || {
        assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
        submissions.set(submissions.get() + 1);
    }))?;
    assert!(matches!(
        refused,
        Err(liminal::durability::DurabilityError::EmptyOperationBatch)
    ));
    assert_eq!(submissions.get(), 0);
    assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
    Ok(())
}

#[test]
fn an_unpolled_commit_does_not_mark_or_submit() -> Result<(), Box<dyn Error>> {
    let store = CountingStore::new(MemoryStore::default());
    let mut batch = OperationBatch::new("conversation");
    batch.stage(RoutedBatchItem {
        key: b"operation".to_vec(),
        expected: None,
        value: vec![1],
        ttl: None,
    })?;
    let submissions = Cell::new(0_usize);
    let pending = batch.commit(&store, || {
        assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
        submissions.set(submissions.get() + 1);
    });
    assert_eq!(submissions.get(), 0);
    assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
    drop(pending);
    assert_eq!(submissions.get(), 0);
    assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
    Ok(())
}

#[test]
fn a_refused_batch_marks_once_and_publishes_no_partial_checkpoint() -> Result<(), Box<dyn Error>> {
    let memory = Arc::new(MemoryStore::default());
    block_on(memory.apply_routed_batch(
        "conversation",
        vec![RoutedBatchItem {
            key: b"operation".to_vec(),
            expected: None,
            value: vec![1],
            ttl: None,
        }],
    ))??;
    let store = CountingStore::new(Shared(Arc::clone(&memory) as Arc<dyn DurableStore>));
    let mut batch = OperationBatch::new("conversation");
    batch.stage(RoutedBatchItem {
        key: b"checkpoint".to_vec(),
        expected: None,
        value: b"covered:1".to_vec(),
        ttl: None,
    })?;
    batch.stage(RoutedBatchItem {
        key: b"operation".to_vec(),
        expected: None,
        value: vec![2],
        ttl: None,
    })?;
    let submissions = Cell::new(0_usize);
    let refused = block_on(batch.commit(&store, || {
        assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
        submissions.set(submissions.get() + 1);
    }))?;
    assert!(matches!(
        refused,
        Err(liminal::durability::DurabilityError::ConfigError(message))
            if message.contains("content-hash mismatch")
    ));
    assert_eq!(submissions.get(), 1);
    let counts = store.snapshot();
    assert_eq!(counts.apply_routed_batch_calls, 1);
    assert_eq!(counts.apply_routed_batch_items, 0);
    assert_eq!(counts.append_calls, 0);
    assert_eq!(counts.append_all_calls, 0);
    assert_eq!(counts.flushes, 0);
    assert_eq!(
        block_on(store.get_routed("conversation", b"operation"))??,
        Some(vec![1])
    );
    assert_eq!(
        block_on(store.get_routed("conversation", b"checkpoint"))??,
        None
    );
    Ok(())
}