lix 0.12.2

use crate::storage::conformance::{
    ConformanceReport, ConformanceResult, SingleSpaceStorageRead, StorageFactory, StorageFixture,
    fixtures::{full_put, key, put_batch, space},
};
use crate::storage::{
    GetOptions, ProjectedValue, ReadOptions, SpaceId, Storage, StorageSpace, StorageWrite,
    WriteOptions,
};

/// Single space used by these fixtures; cross-space isolation is pinned
/// by the baseline cross-space tests.
const TEST_SPACE: StorageSpace = StorageSpace::mutable(SpaceId(7), "storage.persistence.test");

pub(crate) async fn register<F>(report: &mut ConformanceReport, factory: &F)
where
    F: StorageFactory,
{
    report
        .run(
            "persistence::committed_data_survives_reopen",
            committed_data_survives_reopen(factory),
        )
        .await;
    report
        .run(
            "persistence::rolled_back_data_does_not_survive_reopen",
            rolled_back_data_does_not_survive_reopen(factory),
        )
        .await;
    report
        .run(
            "persistence::overwrite_and_delete_final_state_survives_reopen",
            overwrite_and_delete_final_state_survives_reopen(factory),
        )
        .await;
    report
        .run(
            "persistence::await_durable_commit_is_accepted_and_survives_reopen",
            await_durable_commit_is_accepted_and_survives_reopen(factory),
        )
        .await;
}

async fn committed_data_survives_reopen<F>(factory: &F) -> ConformanceResult
where
    F: StorageFactory,
{
    let fixture = factory.create_fixture();
    let test_space = space(71);
    let alpha = key("alpha");
    let beta = key("beta");

    {
        let storage = fixture.open().await;
        let mut write = storage
            .begin_write(WriteOptions::default())
            .await
            .map_err(|error| format!("begin_write failed: {error}"))?;
        write
            .put_many(
                TEST_SPACE,
                put_batch([
                    full_put(alpha.clone(), "persisted-alpha"),
                    full_put(beta.clone(), "persisted-beta"),
                ]),
            )
            .await
            .map_err(|error| format!("put_many failed: {error}"))?;
        write
            .commit()
            .await
            .map_err(|error| format!("commit failed: {error}"))?;
    }

    let reopened = fixture.open().await;
    assert_full_values(
        &reopened,
        test_space,
        &[
            (alpha, Some("persisted-alpha")),
            (beta, Some("persisted-beta")),
        ],
    )
    .await
}

/// An adapter must accept `WriteOptions::await_durable` and apply the write
/// set unchanged under it.
///
/// # Read the name literally
///
/// This proves **acceptance and round-trip, not durability.** A durable
/// acknowledgement is invisible from inside the process — on RocksDB it is an
/// `fdatasync` on the WAL, and no in-process observation distinguishes a synced
/// commit from an unsynced one. An adapter that ignores the flag entirely
/// passes this case, and for four rounds one did.
///
/// It is here for the failure it *can* catch cheaply: an adapter that rejects
/// the flag, or that takes a different and buggy code path under it. Verifying
/// that the sync actually happens requires an external syscall census, and
/// verifying that the store survives losing power requires block-level fault
/// injection; see `WriteOptions::await_durable`. **Do not extend this case to
/// claim more than its name says** — a test that looks like it checks
/// durability while checking acceptance is worse than no test.
async fn await_durable_commit_is_accepted_and_survives_reopen<F>(factory: &F) -> ConformanceResult
where
    F: StorageFactory,
{
    let fixture = factory.create_fixture();
    let test_space = space(71);
    let durable = key("durable");

    {
        let storage = fixture.open().await;
        let mut write = storage
            .begin_write(WriteOptions {
                await_durable: true,
                ..WriteOptions::default()
            })
            .await
            .map_err(|error| format!("begin_write with await_durable failed: {error}"))?;
        write
            .put_many(
                TEST_SPACE,
                put_batch([full_put(durable.clone(), "durable-value")]),
            )
            .await
            .map_err(|error| format!("put_many under await_durable failed: {error}"))?;
        write
            .commit()
            .await
            .map_err(|error| format!("await_durable commit failed: {error}"))?;
    }

    let reopened = fixture.open().await;
    assert_full_values(&reopened, test_space, &[(durable, Some("durable-value"))]).await
}

async fn rolled_back_data_does_not_survive_reopen<F>(factory: &F) -> ConformanceResult
where
    F: StorageFactory,
{
    let fixture = factory.create_fixture();
    let test_space = space(72);
    let rolled_back = key("rolled-back");

    {
        let storage = fixture.open().await;
        let mut write = storage
            .begin_write(WriteOptions::default())
            .await
            .map_err(|error| format!("begin_write failed: {error}"))?;
        write
            .put_many(
                TEST_SPACE,
                put_batch([full_put(rolled_back.clone(), "should-not-persist")]),
            )
            .await
            .map_err(|error| format!("put_many failed: {error}"))?;
        write
            .rollback()
            .await
            .map_err(|error| format!("rollback failed: {error}"))?;
    }

    let reopened = fixture.open().await;
    assert_full_values(&reopened, test_space, &[(rolled_back, None)]).await
}

async fn overwrite_and_delete_final_state_survives_reopen<F>(factory: &F) -> ConformanceResult
where
    F: StorageFactory,
{
    let fixture = factory.create_fixture();
    let test_space = space(73);
    let overwritten = key("overwritten");
    let deleted = key("deleted");

    {
        let storage = fixture.open().await;
        let mut write = storage
            .begin_write(WriteOptions::default())
            .await
            .map_err(|error| format!("begin_write failed: {error}"))?;
        write
            .put_many(
                TEST_SPACE,
                put_batch([
                    full_put(overwritten.clone(), "old"),
                    full_put(deleted.clone(), "delete-me"),
                ]),
            )
            .await
            .map_err(|error| format!("initial put_many failed: {error}"))?;
        write
            .commit()
            .await
            .map_err(|error| format!("initial commit failed: {error}"))?;
    }

    {
        let storage = fixture.open().await;
        let mut write = storage
            .begin_write(WriteOptions::default())
            .await
            .map_err(|error| format!("begin_write failed: {error}"))?;
        write
            .put_many(
                TEST_SPACE,
                put_batch([full_put(overwritten.clone(), "new")]),
            )
            .await
            .map_err(|error| format!("overwrite put_many failed: {error}"))?;
        write
            .delete_many(TEST_SPACE, std::slice::from_ref(&deleted))
            .await
            .map_err(|error| format!("delete_many failed: {error}"))?;
        write
            .commit()
            .await
            .map_err(|error| format!("final commit failed: {error}"))?;
    }

    let reopened = fixture.open().await;
    assert_full_values(
        &reopened,
        test_space,
        &[(overwritten, Some("new")), (deleted, None)],
    )
    .await
}

async fn assert_full_values<StorageImpl>(
    storage: &StorageImpl,
    _test_space: StorageSpace,
    expected: &[(crate::storage::Key, Option<&str>)],
) -> ConformanceResult
where
    StorageImpl: Storage,
{
    let keys = expected
        .iter()
        .map(|(key, _)| key.clone())
        .collect::<Vec<_>>();
    let read = storage
        .begin_read(ReadOptions::default())
        .await
        .map_err(|error| format!("begin_read failed: {error}"))?;
    let result = read
        .get_many_in_space(TEST_SPACE, &keys, GetOptions::default())
        .await
        .map_err(|error| format!("get_many failed: {error}"))?;

    for (index, (key, expected_value)) in expected.iter().enumerate() {
        let actual = match result.values.get(index).and_then(|value| value.as_ref()) {
            Some(ProjectedValue::FullValue(bytes)) => Some(
                std::str::from_utf8(bytes.as_ref())
                    .map_err(|error| format!("slot {index} contained non-utf8 bytes: {error}"))?,
            ),
            Some(other) => {
                return Err(format!("slot {index} returned unexpected value: {other:?}"));
            }
            None => None,
        };
        if actual != *expected_value {
            return Err(format!(
                "key {key:?} value mismatch: expected {expected_value:?}, got {actual:?}"
            ));
        }
    }

    Ok(())
}