liminal-server 0.24.9

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

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

/// The conversation every body here is written for.
const CONVERSATION: u64 = 7;

async fn store_layout(
    store: &MemoryStore,
    key: CheckpointBodyKey,
    body: Vec<u8>,
    skip_part: Option<usize>,
) -> Result<usize, Box<dyn std::error::Error>> {
    let layout = lay_out(CONVERSATION, body);
    let parts = layout.parts.len();
    let mut items = vec![RoutedBatchItem {
        key: key.routed_key(CONVERSATION),
        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(&checkpoint_conversation_route(CONVERSATION), 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, key, body.clone(), None).await?, 0);
    assert_eq!(read_body(&store, CONVERSATION, 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, key, body.clone(), None).await?, 3);
    let read = read_body(&store, CONVERSATION, 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, key, body, Some(1)).await?;
    assert_eq!(read_body(&store, CONVERSATION, key).await?, None);
    Ok(())
}