liminal-server 0.23.1

use liminal::durability::{DurableStore, RoutedBatchItem};

use super::super::restore_checkpoint_locator::CheckpointBodyKey;
use super::super::tests_memory_store::MemoryStore;
use super::{BODY_PART_BYTES, lay_out, read_body};

async fn store_layout(
    store: &MemoryStore,
    route: &str,
    key: CheckpointBodyKey,
    body: Vec<u8>,
    skip_part: Option<usize>,
) -> Result<usize, Box<dyn std::error::Error>> {
    let layout = lay_out(body);
    let parts = layout.parts.len();
    let mut items = vec![RoutedBatchItem {
        key: key.routed_key(),
        expected: None,
        value: layout.head,
        ttl: None,
    }];
    for (index, (part_key, value)) in layout.parts.into_iter().enumerate() {
        if skip_part == Some(index) {
            continue;
        }
        items.push(RoutedBatchItem {
            key: part_key,
            expected: None,
            value,
            ttl: None,
        });
    }
    store.apply_routed_batch(route, items).await?;
    Ok(parts)
}

#[tokio::test]
async fn a_small_body_is_stored_whole_and_read_back() -> Result<(), Box<dyn std::error::Error>> {
    let store = MemoryStore::default();
    let body = vec![5_u8; 1_000];
    let key = CheckpointBodyKey::for_body(&body);
    assert_eq!(store_layout(&store, "r", key, body.clone(), None).await?, 0);
    assert_eq!(read_body(&store, "r", key).await?, Some(body));
    Ok(())
}

#[tokio::test]
async fn a_large_body_is_cut_into_parts_and_reassembled() -> Result<(), Box<dyn std::error::Error>>
{
    let store = MemoryStore::default();
    let body: Vec<u8> = (0..BODY_PART_BYTES * 2 + 17)
        .map(|index| u8::try_from(index % 251).unwrap_or(0))
        .collect();
    let key = CheckpointBodyKey::for_body(&body);
    assert_eq!(store_layout(&store, "r", key, body.clone(), None).await?, 3);
    let read = read_body(&store, "r", key).await?;
    assert_eq!(read.as_ref().map(Vec::len), Some(body.len()));
    assert_eq!(read, Some(body));
    Ok(())
}

#[tokio::test]
async fn a_missing_part_reads_as_no_body() -> Result<(), Box<dyn std::error::Error>> {
    let store = MemoryStore::default();
    let body = vec![9_u8; BODY_PART_BYTES + 1];
    let key = CheckpointBodyKey::for_body(&body);
    store_layout(&store, "r", key, body, Some(1)).await?;
    assert_eq!(read_body(&store, "r", key).await?, None);
    Ok(())
}