liminal-server 0.20.1

//! Durable stores seeded once per test binary and handed to each test as a
//! private copy, through LIM-008's per-binary fixtures
//! ([`crate::test_fixture::fixture_copy`]), and the forwarding wrapper the
//! count pins lend a store through.

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

use liminal::durability::bridge::block_on;
use liminal::durability::{DurabilityError, DurableStore, ScanCursor, ScanPage, StoredEntry};
use tempfile::TempDir;

use super::tests::open_disk_store_for_tests;

/// A private copy of the seed `name`, built by `seed` the first time any test
/// of this binary asks for it. The copy's directory lives as long as the
/// returned [`TempDir`].
pub(super) fn seeded_copy(
    name: &str,
    seed: impl FnOnce(&Arc<dyn DurableStore>) -> Result<(), Box<dyn Error>>,
) -> Result<(TempDir, Arc<dyn DurableStore>), Box<dyn Error>> {
    let copy = crate::test_fixture::fixture_copy(name, |data| {
        let store = open_disk_store_for_tests(data)?;
        seed(&store)?;
        block_on(store.flush())??;
        drop(store);
        Ok(())
    })?;
    let store = open_disk_store_for_tests(copy.path())?;
    Ok((copy, store))
}

/// Lends a shared store to [`CountingStore`], forwarding every call
/// unchanged, `append_all` included, so the counts are the caller's own.
#[derive(Debug)]
pub(super) struct Shared(pub(super) Arc<dyn DurableStore>);

#[async_trait::async_trait]
impl DurableStore for Shared {
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), liminal::durability::DurabilityError> {
        self.0.apply_routed_batch(route_key, items).await
    }

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

    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.0.append(stream_key, payload, expected_seq).await
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.0.append_all(stream_key, payloads, expected_seq).await
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        self.0.read_from(stream_key, offset, limit).await
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        self.0.read_at(stream_key, sequence).await
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        self.0.cas(key, old_value, new_value).await
    }

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

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

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        self.0.scan_page(prefix, cursor, limit).await
    }

    async fn flush(&self) -> Result<(), DurabilityError> {
        self.0.flush().await
    }
}