liminal-server 0.24.9

use std::error::Error;

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

use super::{RoutedStreamError, RoutedStreamRefusal, read_head, read_page};
use crate::server::participant::production::routed_stream_codec::{
    encode_head, encode_row, head_key, row_key,
};
use crate::server::participant::production::tests_memory_store::MemoryStore;

fn fixture(
    rows: &[StoredEntry],
    next_sequence: u64,
) -> Result<CountingStore<MemoryStore>, Box<dyn Error>> {
    let store = CountingStore::new(MemoryStore::default());
    let mut items = vec![RoutedBatchItem {
        key: head_key("source")?,
        expected: None,
        value: encode_head(next_sequence).to_vec(),
        ttl: None,
    }];
    for row in rows {
        items.push(RoutedBatchItem {
            key: row_key("source", row.sequence)?,
            expected: None,
            value: encode_row(row)?,
            ttl: None,
        });
    }
    block_on(store.apply_routed_batch("route", items))??;
    Ok(store)
}

#[test]
fn a_head_read_preserves_absence_without_granting_recovery_authority() -> Result<(), Box<dyn Error>>
{
    let store = CountingStore::new(MemoryStore::default());
    assert!(block_on(read_head(&store, "route", "source"))??.is_none());
    assert_eq!(store.snapshot().get_routed_calls, 1);
    assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
    Ok(())
}

#[test]
fn page_reads_are_bounded_by_the_validated_head() -> Result<(), Box<dyn Error>> {
    let store = fixture(
        &[
            StoredEntry {
                sequence: 1,
                timestamp: 55,
                payload: vec![0, 255],
            },
            StoredEntry {
                sequence: 2,
                timestamp: 56,
                payload: vec![],
            },
        ],
        3,
    )?;
    let head = block_on(read_head(&store, "route", "source"))??.ok_or("head missing")?;
    let before = store.snapshot();
    let page = block_on(read_page(&store, "route", "source", &head, 1, 1))??;
    assert_eq!(
        page,
        vec![StoredEntry {
            sequence: 1,
            timestamp: 55,
            payload: vec![0, 255]
        }]
    );
    assert_eq!(store.snapshot().since(&before).get_routed_calls, 1);
    let tail = block_on(read_page(&store, "route", "source", &head, 2, 100))??;
    assert_eq!(
        tail,
        vec![StoredEntry {
            sequence: 2,
            timestamp: 56,
            payload: vec![]
        }]
    );
    let eof = store.snapshot();
    assert!(block_on(read_page(&store, "route", "source", &head, 3, 1))??.is_empty());
    assert_eq!(store.snapshot().since(&eof).get_routed_calls, 0);
    assert_eq!(store.snapshot().since(&before).append_calls, 0);
    assert_eq!(store.snapshot().since(&before).flushes, 0);
    Ok(())
}

#[test]
fn a_missing_row_below_the_head_is_refused_instead_of_eof() -> Result<(), Box<dyn Error>> {
    let store = fixture(&[], 2)?;
    let head = block_on(read_head(&store, "route", "source"))??.ok_or("head missing")?;
    assert!(matches!(
        block_on(read_page(&store, "route", "source", &head, 1, 1))?,
        Err(RoutedStreamError::Refusal(
            RoutedStreamRefusal::MissingRow { sequence: 1, .. }
        ))
    ));
    Ok(())
}

#[test]
fn malformed_heads_and_row_sequence_drift_keep_the_typed_refusal() -> Result<(), Box<dyn Error>> {
    let store = fixture(
        &[StoredEntry {
            sequence: 1,
            timestamp: 0,
            payload: vec![1],
        }],
        2,
    )?;
    let head = block_on(read_head(&store, "route", "source"))??.ok_or("head missing")?;
    let wrong = encode_row(&StoredEntry {
        sequence: 0,
        timestamp: 0,
        payload: vec![2],
    })?;
    let original = encode_row(&StoredEntry {
        sequence: 1,
        timestamp: 0,
        payload: vec![1],
    })?;
    block_on(store.apply_routed_batch(
        "route",
        vec![RoutedBatchItem {
            key: row_key("source", 1)?,
            expected: Some(haematite::tree::Hash::of(&original)),
            value: wrong,
            ttl: None,
        }],
    ))??;
    assert!(matches!(
        block_on(read_page(&store, "route", "source", &head, 1, 1))?,
        Err(RoutedStreamError::Refusal(RoutedStreamRefusal::Codec(
            super::super::routed_stream_codec::CodecError::SequenceMismatch {
                expected: 1,
                actual: 0
            }
        )))
    ));
    block_on(store.apply_routed_batch(
        "route",
        vec![RoutedBatchItem {
            key: head_key("source")?,
            expected: Some(head.expected),
            value: vec![0],
            ttl: None,
        }],
    ))??;
    assert!(matches!(
        block_on(read_head(&store, "route", "source"))?,
        Err(RoutedStreamError::Refusal(RoutedStreamRefusal::Codec(_)))
    ));
    Ok(())
}

#[test]
fn invalid_page_bounds_do_not_read_rows() -> Result<(), Box<dyn Error>> {
    let store = fixture(&[], 0)?;
    let head = block_on(read_head(&store, "route", "source"))??.ok_or("head missing")?;
    let before = store.snapshot();
    assert!(matches!(
        block_on(read_page(&store, "route", "source", &head, 0, 0))?,
        Err(RoutedStreamError::Refusal(
            RoutedStreamRefusal::InvalidLimit
        ))
    ));
    assert!(matches!(
        block_on(read_page(&store, "route", "source", &head, 1, 1))?,
        Err(RoutedStreamError::Refusal(
            RoutedStreamRefusal::OffsetBeyondHead { offset: 1, head: 0 }
        ))
    ));
    assert_eq!(store.snapshot().since(&before).get_routed_calls, 0);
    Ok(())
}

#[test]
fn a_head_from_another_stream_or_route_is_refused_without_row_reads() -> Result<(), Box<dyn Error>>
{
    let store = fixture(&[], 0)?;
    let head = block_on(read_head(&store, "route", "source"))??.ok_or("head missing")?;
    let before = store.snapshot();
    assert!(matches!(
        block_on(read_page(&store, "other", "source", &head, 0, 1))?,
        Err(RoutedStreamError::Refusal(
            RoutedStreamRefusal::HeadIdentityMismatch
        ))
    ));
    assert!(matches!(
        block_on(read_page(&store, "route", "other", &head, 0, 1))?,
        Err(RoutedStreamError::Refusal(
            RoutedStreamRefusal::HeadIdentityMismatch
        ))
    ));
    assert_eq!(store.snapshot().since(&before).get_routed_calls, 0);
    Ok(())
}