liminal-rs 0.16.2

use std::error::Error;

use haematite::tree::Hash;

use super::bridge::block_on;
use super::tests_memory_store::MemoryStore;
use super::{CountingStore, DurableStore, RoutedBatchItem};

fn row(key: &[u8], expected: Option<Hash>, value: &[u8]) -> RoutedBatchItem {
    RoutedBatchItem {
        key: key.to_vec(),
        expected,
        value: value.to_vec(),
        ttl: None,
    }
}

#[test]
fn a_late_precondition_failure_publishes_neither_operations_nor_checkpoint()
-> Result<(), Box<dyn Error>> {
    let store = CountingStore::new(MemoryStore::default());
    block_on(store.apply_routed_batch("conversation", vec![row(b"checkpoint", None, b"old")]))??;
    let before = store.snapshot();
    let refused = block_on(store.apply_routed_batch(
        "conversation",
        vec![
            row(b"operation", None, b"covered"),
            row(b"checkpoint", Some(Hash::of(b"wrong")), b"new"),
        ],
    ))?;
    assert!(refused.is_err());
    assert_eq!(
        block_on(store.get_routed("conversation", b"operation"))??,
        None
    );
    assert_eq!(
        block_on(store.get_routed("conversation", b"checkpoint"))??,
        Some(b"old".to_vec())
    );
    let counts = store.snapshot().since(&before);
    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);
    Ok(())
}

#[test]
fn a_routed_batch_has_one_route_and_keeps_content_hash_preconditions() -> Result<(), Box<dyn Error>>
{
    let store = CountingStore::new(MemoryStore::default());
    block_on(store.apply_routed_batch("left", vec![row(b"key", None, b"old")]))??;
    assert_eq!(block_on(store.get_routed("right", b"key"))??, None);
    block_on(store.apply_routed_batch("left", vec![row(b"key", Some(Hash::of(b"old")), b"new")]))??;
    assert_eq!(
        block_on(store.get_routed("left", b"key"))??,
        Some(b"new".to_vec())
    );
    assert!(
        block_on(store.apply_routed_batch("left", vec![row(b"key", None, b"duplicate")]))?.is_err()
    );
    assert_eq!(
        block_on(store.get_routed("left", b"key"))??,
        Some(b"new".to_vec())
    );
    let counts = store.snapshot();
    assert_eq!(counts.apply_routed_batch_calls, 3);
    assert_eq!(counts.apply_routed_batch_items, 2);
    assert_eq!(counts.get_routed_calls, 3);
    assert_eq!(counts.get_routed_rows, 2);
    assert_eq!(counts.flushes, 0);
    Ok(())
}

#[derive(Debug)]
struct BeforeCommitFault<'a>(&'a dyn DurableStore);

#[async_trait::async_trait]
impl DurableStore for BeforeCommitFault<'_> {
    async fn apply_routed_batch(
        &self,
        _: &str,
        _: Vec<RoutedBatchItem>,
    ) -> Result<(), super::DurabilityError> {
        Err(super::DurabilityError::ConfigError(
            "atomic batch refused before commit".into(),
        ))
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, super::DurabilityError> {
        self.0.get_routed(route_key, key).await
    }

    async fn append(&self, _: &str, _: Vec<u8>, _: u64) -> Result<u64, super::DurabilityError> {
        Err(super::DurabilityError::BatchUnsupported {
            store: std::any::type_name::<Self>(),
        })
    }

    async fn append_all(
        &self,
        _: &str,
        _: Vec<Vec<u8>>,
        _: u64,
    ) -> Result<u64, super::DurabilityError> {
        Err(super::DurabilityError::BatchUnsupported {
            store: std::any::type_name::<Self>(),
        })
    }

    async fn read_from(
        &self,
        _: &str,
        _: u64,
        _: usize,
    ) -> Result<Vec<super::StoredEntry>, super::DurabilityError> {
        Err(super::DurabilityError::BatchUnsupported {
            store: std::any::type_name::<Self>(),
        })
    }

    async fn cas(&self, _: &str, _: u64, _: u64) -> Result<(), super::DurabilityError> {
        Err(super::DurabilityError::BatchUnsupported {
            store: std::any::type_name::<Self>(),
        })
    }

    async fn read_stream_next_seq(&self, key: &str) -> Result<Option<u64>, super::DurabilityError> {
        self.0.read_stream_next_seq(key).await
    }

    async fn read_value(&self, _: &str) -> Result<Option<u64>, super::DurabilityError> {
        Err(super::DurabilityError::BatchUnsupported {
            store: std::any::type_name::<Self>(),
        })
    }

    async fn scan_page(
        &self,
        _: &str,
        _: Option<super::ScanCursor>,
        _: usize,
    ) -> Result<super::ScanPage, super::DurabilityError> {
        Err(super::DurabilityError::BatchUnsupported {
            store: std::any::type_name::<Self>(),
        })
    }

    async fn flush(&self) -> Result<(), super::DurabilityError> {
        Err(super::DurabilityError::BatchUnsupported {
            store: std::any::type_name::<Self>(),
        })
    }
}

#[test]
fn a_store_fault_before_commit_leaves_no_operation_or_checkpoint() -> Result<(), Box<dyn Error>> {
    let memory = MemoryStore::default();
    let store = CountingStore::new(BeforeCommitFault(&memory));
    let refused = block_on(store.apply_routed_batch(
        "conversation",
        vec![
            row(b"operation", None, b"covered"),
            row(b"checkpoint", None, b"new"),
        ],
    ))?;
    assert!(
        matches!(refused, Err(super::DurabilityError::ConfigError(message))
        if message == "atomic batch refused before commit")
    );
    assert_eq!(
        block_on(store.get_routed("conversation", b"operation"))??,
        None
    );
    assert_eq!(
        block_on(store.get_routed("conversation", b"checkpoint"))??,
        None
    );
    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);
    Ok(())
}