//! [`ManagedDb`] implementation for QMDB [`current`](commonware_storage::qmdb::current) databases.
//!
//! The QMDB batch API passes `&db` to `get()` and `merkleize()` for
//! read-through to committed state. This module provides wrapper types
//! that capture `Arc<TracedAsyncRwLock<Db>>` alongside the raw batch so the
//! [`Unmerkleized`](super::Unmerkleized) and [`Merkleized`](super::Merkleized)
//! traits can be implemented without a DB parameter.

use crate::stateful::db::{
    ManagedDb, Merkleized as MerkleizedTrait, StateSyncDb, SyncEngineConfig,
    Unmerkleized as UnmerkleizedTrait,
};
use commonware_codec::{Codec, Read as CodecRead};
use commonware_cryptography::Hasher;
use commonware_parallel::Strategy;
use commonware_storage::{
    index::{
        ordered::Index as OrderedIdx, unordered::Index as UnorderedIdx, Ordered as OrderedIndex,
        Unordered as UnorderedIndex,
    },
    journal::contiguous::{
        fixed::Journal as FixedJournal, variable::Journal as VariableJournal, Contiguous, Mutable,
    },
    merkle::{Graftable, Location},
    qmdb::{
        any::{
            initial_root,
            operation::{Operation, Update},
            ordered, unordered,
            value::{self, FixedEncoding, ValueEncoding, VariableEncoding},
        },
        current::{
            batch::{MerkleizedBatch, Staged, UnmerkleizedBatch},
            db::Db,
            FixedConfig, VariableConfig,
        },
        operation::Key,
        sync::{self, resolver::Resolver, Target as CurrentSyncTarget},
        Error,
    },
    translator::Translator,
    Context,
};
use commonware_utils::{channel::mpsc, non_empty_range, sync::TracedAsyncRwLock, Array};
use std::{
    ops::{Deref, Range},
    sync::Arc,
};

type CurrentDbHandle<F, E, C, I, H, U, const N: usize, S> =
    Arc<TracedAsyncRwLock<Db<F, E, C, I, H, U, N, S>>>;

/// Wraps a QMDB [`UnmerkleizedBatch`] with a reference to the parent
/// database, implementing the [`Unmerkleized`](super::Unmerkleized) trait.
pub struct CurrentUnmerkleized<F, E, C, I, H, U, const N: usize, S>
where
    F: Graftable,
    E: Context,
    U: Update,
    C: Contiguous<Item = Operation<F, U>>,
    I: UnorderedIndex<Value = Location<F>>,
    H: Hasher,
    S: Strategy,
    Operation<F, U>: Codec,
{
    batch: UnmerkleizedBatch<F, H, U, N, S>,
    db: CurrentDbHandle<F, E, C, I, H, U, N, S>,
    metadata: Option<U::Value>,
}

/// Staged batch returned by [`CurrentUnmerkleized::stage`], wrapping a QMDB [`Staged`] with a
/// reference to the parent database.
///
/// Like any speculative batch, this handle is a branch-scoped view of the shared database: it
/// stays valid only while every batch finalized on the database is an ancestor of this batch
/// (see [`MerkleizedBatch`]'s branch-validity contract).
pub struct CurrentStaged<F, E, C, I, H, U, const N: usize, S>
where
    F: Graftable,
    E: Context,
    U: Update,
    C: Contiguous<Item = Operation<F, U>>,
    I: UnorderedIndex<Value = Location<F>>,
    H: Hasher,
    S: Strategy,
    Operation<F, U>: Codec,
{
    staged: Staged<F, H, U, N, S>,
    db: CurrentDbHandle<F, E, C, I, H, U, N, S>,
    metadata: Option<U::Value>,
}

/// Key-value operations shared by both `current` update kinds.
impl<F, E, C, I, H, U, const N: usize, S> CurrentUnmerkleized<F, E, C, I, H, U, N, S>
where
    F: Graftable,
    E: Context,
    U: Update,
    C: Contiguous<Item = Operation<F, U>>,
    I: UnorderedIndex<Value = Location<F>> + 'static,
    H: Hasher,
    S: Strategy,
    Operation<F, U>: Codec,
{
    /// Set commit metadata included in the next
    /// [`merkleize`](UnmerkleizedTrait::merkleize) call.
    pub fn with_metadata(mut self, metadata: U::Value) -> Self {
        self.metadata = Some(metadata);
        self
    }

    /// Read a value by key, falling back to committed state.
    pub async fn get(&self, key: &U::Key) -> Result<Option<U::Value>, Error<F>> {
        let db = self.db.read().await;
        self.batch.get(key, &*db).await
    }

    /// Read multiple values by key, falling back to committed state.
    ///
    /// Returns results in the same order as the input keys.
    pub async fn get_many(&self, keys: &[&U::Key]) -> Result<Vec<Option<U::Value>>, Error<F>> {
        let db = self.db.read().await;
        self.batch.get_many(keys, &*db).await
    }

    /// Read multiple values and return a staged batch for the same keys.
    ///
    /// Returns results in the same order as the input keys.
    pub async fn stage(
        self,
        keys: &[&U::Key],
    ) -> Result<(Vec<Option<U::Value>>, CurrentStaged<F, E, C, I, H, U, N, S>), Error<F>> {
        let Self {
            batch,
            db,
            metadata,
        } = self;
        let (values, staged) = {
            let guard = db.read().await;
            batch.stage(keys, &*guard).await?
        };
        Ok((
            values,
            CurrentStaged {
                staged,
                db,
                metadata,
            },
        ))
    }

    /// Record a mutation. `Some(value)` for upsert, `None` for delete.
    pub fn write(mut self, key: U::Key, value: Option<U::Value>) -> Self {
        self.batch = self.batch.write(key, value);
        self
    }
}

/// Wraps a QMDB [`MerkleizedBatch`] with a reference to the parent
/// database, implementing the [`Merkleized`](super::Merkleized) trait.
pub struct CurrentMerkleized<F, E, C, I, H, U, const N: usize, S>
where
    F: Graftable,
    E: Context,
    U: Update,
    C: Contiguous<Item = Operation<F, U>>,
    I: UnorderedIndex<Value = Location<F>>,
    H: Hasher,
    S: Strategy,
    Operation<F, U>: Codec,
{
    inner: Arc<MerkleizedBatch<F, H::Digest, U, N, S>>,
    db: CurrentDbHandle<F, E, C, I, H, U, N, S>,
}

impl<F, E, C, I, H, U, const N: usize, S> Deref for CurrentUnmerkleized<F, E, C, I, H, U, N, S>
where
    F: Graftable,
    E: Context,
    U: Update,
    C: Contiguous<Item = Operation<F, U>>,
    I: UnorderedIndex<Value = Location<F>>,
    H: Hasher,
    S: Strategy,
    Operation<F, U>: Codec,
{
    type Target = UnmerkleizedBatch<F, H, U, N, S>;

    fn deref(&self) -> &Self::Target {
        &self.batch
    }
}

impl<F, E, C, I, H, U, const N: usize, S> Deref for CurrentMerkleized<F, E, C, I, H, U, N, S>
where
    F: Graftable,
    E: Context,
    U: Update,
    C: Contiguous<Item = Operation<F, U>>,
    I: UnorderedIndex<Value = Location<F>>,
    H: Hasher,
    S: Strategy,
    Operation<F, U>: Codec,
{
    type Target = MerkleizedBatch<F, H::Digest, U, N, S>;

    fn deref(&self) -> &Self::Target {
        &self.inner
    }
}

/// Read-expansion operations for the `current` staged batch.
impl<F, E, C, I, H, U, const N: usize, S> CurrentStaged<F, E, C, I, H, U, N, S>
where
    F: Graftable,
    E: Context,
    U: Update,
    C: Contiguous<Item = Operation<F, U>>,
    I: UnorderedIndex<Value = Location<F>> + 'static,
    H: Hasher,
    S: Strategy,
    Operation<F, U>: Codec,
{
    /// Set commit metadata included in the [`merkleize`](Self::merkleize) call, replacing any
    /// metadata set before staging.
    pub fn with_metadata(mut self, metadata: U::Value) -> Self {
        self.metadata = Some(metadata);
        self
    }

    /// Expand this staged batch with more reads.
    ///
    /// Existing read indices remain stable. Newly read keys are appended to the staged read set and
    /// assigned the returned range. Expansion does not deduplicate against previously staged keys
    /// and does not observe values computed for earlier staged slots but not yet passed to
    /// `merkleize`.
    pub async fn expand(
        self,
        keys: &[&U::Key],
    ) -> Result<(Range<usize>, Vec<Option<U::Value>>, Self), Error<F>> {
        let Self {
            staged,
            db,
            metadata,
        } = self;
        let (range, values, staged) = {
            let guard = db.read().await;
            staged.expand(keys, &*guard).await?
        };
        Ok((
            range,
            values,
            Self {
                staged,
                db,
                metadata,
            },
        ))
    }
}

/// Staged merkleize for the `current` unordered update kind.
impl<F, E, C, I, H, K, V, const N: usize, S>
    CurrentStaged<F, E, C, I, H, unordered::Update<K, V>, N, S>
where
    F: Graftable,
    E: Context,
    K: Key,
    V: ValueEncoding + 'static,
    C: Mutable<Item = Operation<F, unordered::Update<K, V>>>,
    I: UnorderedIndex<Value = Location<F>> + 'static,
    H: Hasher,
    S: Strategy,
    Operation<F, unordered::Update<K, V>>: Codec,
{
    /// Record updates for staged reads and upserts for unread keys, then merkleize.
    ///
    /// Consumes the staged handle and write vectors. Call [`expand`](CurrentStaged::expand)
    /// before this method if more keys must be read into the staged index space.
    ///
    /// A `Some` value is an upsert. `None` is a delete. Update indices refer to the staged read
    /// set: the initial `stage` input followed by any [`expand`](CurrentStaged::expand) ranges.
    /// Metadata set via [`with_metadata`](CurrentStaged::with_metadata) (or before staging) is
    /// committed with the returned batch.
    ///
    /// # Panics
    ///
    /// Panics if any update's `read_index` is out of the staged read range.
    pub async fn merkleize(
        self,
        updates: Vec<(usize, Option<V::Value>)>,
        upserts: Vec<(K, Option<V::Value>)>,
    ) -> Result<CurrentMerkleized<F, E, C, I, H, unordered::Update<K, V>, N, S>, Error<F>> {
        let Self {
            staged,
            db,
            metadata,
        } = self;
        let inner = {
            let guard = db.read().await;
            staged
                .merkleize(updates, upserts, metadata, &*guard)
                .await?
        };
        Ok(CurrentMerkleized { inner, db })
    }
}

/// Staged merkleize for the `current` ordered update kind.
impl<F, E, C, I, H, K, V, const N: usize, S>
    CurrentStaged<F, E, C, I, H, ordered::Update<K, V>, N, S>
where
    F: Graftable,
    E: Context,
    K: Key,
    V: ValueEncoding + 'static,
    C: Mutable<Item = Operation<F, ordered::Update<K, V>>>,
    I: OrderedIndex<Value = Location<F>> + 'static,
    H: Hasher,
    S: Strategy,
    Operation<F, ordered::Update<K, V>>: Codec,
{
    /// Record updates for staged reads and upserts for unread keys, then merkleize.
    ///
    /// Consumes the staged handle and write vectors. Call [`expand`](CurrentStaged::expand)
    /// before this method if more keys must be read into the staged index space.
    ///
    /// A `Some` value is an upsert. `None` is a delete. Update indices refer to the staged read
    /// set: the initial `stage` input followed by any [`expand`](CurrentStaged::expand) ranges.
    /// Metadata set via [`with_metadata`](CurrentStaged::with_metadata) (or before staging) is
    /// committed with the returned batch.
    ///
    /// # Panics
    ///
    /// Panics if any update's `read_index` is out of the staged read range.
    pub async fn merkleize(
        self,
        updates: Vec<(usize, Option<V::Value>)>,
        upserts: Vec<(K, Option<V::Value>)>,
    ) -> Result<CurrentMerkleized<F, E, C, I, H, ordered::Update<K, V>, N, S>, Error<F>> {
        let Self {
            staged,
            db,
            metadata,
        } = self;
        let inner = {
            let guard = db.read().await;
            staged
                .merkleize(updates, upserts, metadata, &*guard)
                .await?
        };
        Ok(CurrentMerkleized { inner, db })
    }
}

/// Read-through operations for the `current` merkleized batch.
impl<F, E, C, I, H, U, const N: usize, S> CurrentMerkleized<F, E, C, I, H, U, N, S>
where
    F: Graftable,
    E: Context,
    U: Update,
    C: Contiguous<Item = Operation<F, U>>,
    I: UnorderedIndex<Value = Location<F>> + 'static,
    H: Hasher,
    S: Strategy,
    Operation<F, U>: Codec,
{
    /// Read a value by key, falling back to committed state.
    pub async fn get(&self, key: &U::Key) -> Result<Option<U::Value>, Error<F>> {
        let db = self.db.read().await;
        self.inner.get(key, &*db).await
    }

    /// Read multiple values by key, falling back to committed state.
    ///
    /// Returns results in the same order as the input keys.
    pub async fn get_many(&self, keys: &[&U::Key]) -> Result<Vec<Option<U::Value>>, Error<F>> {
        let db = self.db.read().await;
        self.inner.get_many(keys, &*db).await
    }
}

/// Implement [`Unmerkleized`](UnmerkleizedTrait) for the `current` unordered update kind.
impl<F, E, C, I, H, K, V, const N: usize, S> UnmerkleizedTrait
    for CurrentUnmerkleized<F, E, C, I, H, unordered::Update<K, V>, N, S>
where
    F: Graftable,
    E: Context,
    K: Key,
    V: ValueEncoding + 'static,
    C: Mutable<Item = Operation<F, unordered::Update<K, V>>>,
    I: UnorderedIndex<Value = Location<F>> + 'static,
    H: Hasher,
    S: Strategy,
    Operation<F, unordered::Update<K, V>>: Codec,
{
    type Merkleized = CurrentMerkleized<F, E, C, I, H, unordered::Update<K, V>, N, S>;
    type Error = Error<F>;

    async fn merkleize(self) -> Result<Self::Merkleized, Error<F>> {
        let db = self.db.read().await;
        let merkleized = self.batch.merkleize(&*db, self.metadata).await?;
        Ok(CurrentMerkleized {
            inner: merkleized,
            db: self.db.clone(),
        })
    }
}

/// Implement [`Unmerkleized`](UnmerkleizedTrait) for the `current` ordered update kind.
impl<F, E, C, I, H, K, V, const N: usize, S> UnmerkleizedTrait
    for CurrentUnmerkleized<F, E, C, I, H, ordered::Update<K, V>, N, S>
where
    F: Graftable,
    E: Context,
    K: Key,
    V: ValueEncoding + 'static,
    C: Mutable<Item = Operation<F, ordered::Update<K, V>>>,
    I: OrderedIndex<Value = Location<F>> + 'static,
    H: Hasher,
    S: Strategy,
    Operation<F, ordered::Update<K, V>>: Codec,
{
    type Merkleized = CurrentMerkleized<F, E, C, I, H, ordered::Update<K, V>, N, S>;
    type Error = Error<F>;

    async fn merkleize(self) -> Result<Self::Merkleized, Error<F>> {
        let db = self.db.read().await;
        let merkleized = self.batch.merkleize(&*db, self.metadata).await?;
        Ok(CurrentMerkleized {
            inner: merkleized,
            db: self.db.clone(),
        })
    }
}

/// Implement [`Merkleized`](MerkleizedTrait) for all supported `current` update kinds.
impl<F, E, C, I, H, U, const N: usize, S> MerkleizedTrait
    for CurrentMerkleized<F, E, C, I, H, U, N, S>
where
    F: Graftable,
    E: Context,
    U: Update,
    C: Mutable<Item = Operation<F, U>>,
    I: UnorderedIndex<Value = Location<F>> + 'static,
    H: Hasher,
    S: Strategy,
    Operation<F, U>: Codec,
    CurrentUnmerkleized<F, E, C, I, H, U, N, S>: UnmerkleizedTrait,
{
    type Digest = H::Digest;
    type Unmerkleized = CurrentUnmerkleized<F, E, C, I, H, U, N, S>;

    fn root(&self) -> H::Digest {
        self.inner.root()
    }

    fn new_batch(&self) -> Self::Unmerkleized {
        CurrentUnmerkleized {
            batch: self.inner.new_batch::<H>(),
            db: self.db.clone(),
            metadata: None,
        }
    }
}

/// Implement [`ManagedDb`] for unordered current QMDB databases with fixed-size values.
impl<F, E, K, V, H, T, const N: usize, S> ManagedDb<E>
    for Db<
        F,
        E,
        FixedJournal<E, Operation<F, unordered::Update<K, FixedEncoding<V>>>>,
        UnorderedIdx<T, Location<F>>,
        H,
        unordered::Update<K, FixedEncoding<V>>,
        N,
        S,
    >
where
    F: Graftable,
    E: Context,
    K: Array,
    V: value::FixedValue + 'static,
    H: Hasher + 'static,
    T: Translator,
    S: Strategy,
{
    type Unmerkleized = CurrentUnmerkleized<
        F,
        E,
        FixedJournal<E, Operation<F, unordered::Update<K, FixedEncoding<V>>>>,
        UnorderedIdx<T, Location<F>>,
        H,
        unordered::Update<K, FixedEncoding<V>>,
        N,
        S,
    >;
    type Merkleized = CurrentMerkleized<
        F,
        E,
        FixedJournal<E, Operation<F, unordered::Update<K, FixedEncoding<V>>>>,
        UnorderedIdx<T, Location<F>>,
        H,
        unordered::Update<K, FixedEncoding<V>>,
        N,
        S,
    >;
    type Error = Error<F>;
    type Config = FixedConfig<T, S>;
    type SyncTarget = CurrentSyncTarget<F, H::Digest>;

    async fn init(context: E, config: Self::Config) -> Result<Self, Error<F>> {
        <Self>::init(context, config).await
    }

    fn initial_sync_target() -> Self::SyncTarget {
        CurrentSyncTarget::new(
            initial_root::<F, unordered::Update<K, FixedEncoding<V>>, H>(),
            non_empty_range!(Location::new(0), Location::new(1)),
        )
    }

    async fn new_batch(db: &Arc<TracedAsyncRwLock<Self>>) -> Self::Unmerkleized {
        let inner = db.read().await;
        CurrentUnmerkleized {
            batch: inner.new_batch(),
            db: db.clone(),
            metadata: None,
        }
    }

    fn matches_sync_target(batch: &Self::Merkleized, target: &Self::SyncTarget) -> bool {
        batch.ops_root() == target.root
            && *target.range.start() == batch.sync_boundary()
            && *target.range.end() == Location::<F>::new(batch.bounds().total_size)
    }

    async fn finalize(&mut self, batch: Self::Merkleized) -> Result<(), Error<F>> {
        self.apply_batch(batch.inner).await?;
        self.sync().await
    }

    async fn prune(&mut self, target: &Self::SyncTarget) -> Result<(), Error<F>> {
        self.prune((*target.range.start()).into()).await
    }

    fn sync_target(&self) -> Self::SyncTarget {
        let bounds = self.bounds();
        CurrentSyncTarget::new(
            self.ops_root(),
            non_empty_range!(self.sync_boundary(), bounds.end),
        )
    }

    async fn rewind_to_target(&mut self, target: Self::SyncTarget) -> Result<(), Error<F>> {
        self.rewind(target.range.end()).await?;
        self.sync().await?;

        let rewound_target = self.sync_target();
        assert_eq!(
            rewound_target, target,
            "rewound database target mismatch after rewind",
        );
        Ok(())
    }
}

/// Implement [`ManagedDb`] for ordered current QMDB databases with fixed-size values.
impl<F, E, K, V, H, T, const N: usize, S> ManagedDb<E>
    for Db<
        F,
        E,
        FixedJournal<E, Operation<F, ordered::Update<K, FixedEncoding<V>>>>,
        OrderedIdx<T, Location<F>>,
        H,
        ordered::Update<K, FixedEncoding<V>>,
        N,
        S,
    >
where
    F: Graftable,
    E: Context,
    K: Array,
    V: value::FixedValue + 'static,
    H: Hasher + 'static,
    T: Translator,
    S: Strategy,
{
    type Unmerkleized = CurrentUnmerkleized<
        F,
        E,
        FixedJournal<E, Operation<F, ordered::Update<K, FixedEncoding<V>>>>,
        OrderedIdx<T, Location<F>>,
        H,
        ordered::Update<K, FixedEncoding<V>>,
        N,
        S,
    >;
    type Merkleized = CurrentMerkleized<
        F,
        E,
        FixedJournal<E, Operation<F, ordered::Update<K, FixedEncoding<V>>>>,
        OrderedIdx<T, Location<F>>,
        H,
        ordered::Update<K, FixedEncoding<V>>,
        N,
        S,
    >;
    type Error = Error<F>;
    type Config = FixedConfig<T, S>;
    type SyncTarget = CurrentSyncTarget<F, H::Digest>;

    async fn init(context: E, config: Self::Config) -> Result<Self, Error<F>> {
        <Self>::init(context, config).await
    }

    fn initial_sync_target() -> Self::SyncTarget {
        CurrentSyncTarget::new(
            initial_root::<F, ordered::Update<K, FixedEncoding<V>>, H>(),
            non_empty_range!(Location::new(0), Location::new(1)),
        )
    }

    async fn new_batch(db: &Arc<TracedAsyncRwLock<Self>>) -> Self::Unmerkleized {
        let inner = db.read().await;
        CurrentUnmerkleized {
            batch: inner.new_batch(),
            db: db.clone(),
            metadata: None,
        }
    }

    fn matches_sync_target(batch: &Self::Merkleized, target: &Self::SyncTarget) -> bool {
        batch.ops_root() == target.root
            && *target.range.start() == batch.sync_boundary()
            && *target.range.end() == Location::<F>::new(batch.bounds().total_size)
    }

    async fn finalize(&mut self, batch: Self::Merkleized) -> Result<(), Error<F>> {
        self.apply_batch(batch.inner).await?;
        self.sync().await
    }

    async fn prune(&mut self, target: &Self::SyncTarget) -> Result<(), Error<F>> {
        self.prune((*target.range.start()).into()).await
    }

    fn sync_target(&self) -> Self::SyncTarget {
        let bounds = self.bounds();
        CurrentSyncTarget::new(
            self.ops_root(),
            non_empty_range!(self.sync_boundary(), bounds.end),
        )
    }

    async fn rewind_to_target(&mut self, target: Self::SyncTarget) -> Result<(), Error<F>> {
        self.rewind(target.range.end()).await?;
        self.sync().await?;

        let rewound_target = self.sync_target();
        assert_eq!(
            rewound_target, target,
            "rewound database target mismatch after rewind",
        );
        Ok(())
    }
}

/// Workaround for <https://github.com/rust-lang/rust/issues/115188>.
///
/// Inside a `ManagedDb` trait impl, `<Self>::init(...)` in a non-async `fn`
/// resolves to the *trait* method (infinite recursion), while in an
/// `async fn` it resolves correctly to the inherent method but the compiler
/// cannot verify the RPITIT future is `Send`. By placing the call in this
/// module -- which does not import `ManagedDb` -- the compiler
/// unambiguously picks the inherent `Db::init`.
mod open {
    use commonware_codec::{Codec, Read};
    use commonware_cryptography::Hasher;
    use commonware_parallel::Strategy;
    use commonware_storage::{
        merkle::Graftable,
        qmdb::{
            any::{
                operation::Operation,
                ordered, unordered,
                value::{VariableEncoding, VariableValue},
            },
            current::{
                ordered::variable::Db as OrderedVariableDb, unordered::variable::Db, VariableConfig,
            },
            Error,
        },
        Context,
    };
    use commonware_utils::Array;

    type VConfig<T, F, K, V, S> = VariableConfig<
        T,
        <Operation<F, unordered::Update<K, VariableEncoding<V>>> as Read>::Cfg,
        S,
    >;
    type OrderedVConfig<T, F, K, V, S> =
        VariableConfig<T, <Operation<F, ordered::Update<K, VariableEncoding<V>>> as Read>::Cfg, S>;

    pub(super) async fn variable<F, E, K, V, H, T, const N: usize, S>(
        context: E,
        config: VConfig<T, F, K, V, S>,
    ) -> Result<Db<F, E, K, V, H, T, N, S>, Error<F>>
    where
        F: Graftable,
        E: Context,
        K: Array,
        V: VariableValue + 'static,
        H: Hasher,
        T: commonware_storage::translator::Translator,
        S: Strategy,
        Operation<F, unordered::Update<K, VariableEncoding<V>>>: Codec,
    {
        Db::init(context, config).await
    }

    pub(super) async fn ordered_variable<F, E, K, V, H, T, const N: usize, S>(
        context: E,
        config: OrderedVConfig<T, F, K, V, S>,
    ) -> Result<OrderedVariableDb<F, E, K, V, H, T, N, S>, Error<F>>
    where
        F: Graftable,
        E: Context,
        K: commonware_storage::qmdb::operation::Key,
        V: VariableValue + 'static,
        H: Hasher,
        T: commonware_storage::translator::Translator,
        S: Strategy,
        Operation<F, ordered::Update<K, VariableEncoding<V>>>: Codec,
    {
        OrderedVariableDb::init(context, config).await
    }
}

/// Implement [`ManagedDb`] for unordered current QMDB databases with variable-size values.
impl<F, E, K, V, H, T, const N: usize, S> ManagedDb<E>
    for Db<
        F,
        E,
        VariableJournal<E, Operation<F, unordered::Update<K, VariableEncoding<V>>>>,
        UnorderedIdx<T, Location<F>>,
        H,
        unordered::Update<K, VariableEncoding<V>>,
        N,
        S,
    >
where
    F: Graftable,
    E: Context,
    K: Key + Array,
    V: value::VariableValue + 'static,
    H: Hasher,
    T: Translator,
    S: Strategy,
    Operation<F, unordered::Update<K, VariableEncoding<V>>>: Codec,
{
    type Unmerkleized = CurrentUnmerkleized<
        F,
        E,
        VariableJournal<E, Operation<F, unordered::Update<K, VariableEncoding<V>>>>,
        UnorderedIdx<T, Location<F>>,
        H,
        unordered::Update<K, VariableEncoding<V>>,
        N,
        S,
    >;
    type Merkleized = CurrentMerkleized<
        F,
        E,
        VariableJournal<E, Operation<F, unordered::Update<K, VariableEncoding<V>>>>,
        UnorderedIdx<T, Location<F>>,
        H,
        unordered::Update<K, VariableEncoding<V>>,
        N,
        S,
    >;
    type Error = Error<F>;
    type Config = VariableConfig<
        T,
        <Operation<F, unordered::Update<K, VariableEncoding<V>>> as CodecRead>::Cfg,
        S,
    >;
    type SyncTarget = CurrentSyncTarget<F, H::Digest>;

    async fn init(context: E, config: Self::Config) -> Result<Self, Error<F>> {
        open::variable(context, config).await
    }

    fn initial_sync_target() -> Self::SyncTarget {
        CurrentSyncTarget::new(
            initial_root::<F, unordered::Update<K, VariableEncoding<V>>, H>(),
            non_empty_range!(Location::new(0), Location::new(1)),
        )
    }

    async fn new_batch(db: &Arc<TracedAsyncRwLock<Self>>) -> Self::Unmerkleized {
        let inner = db.read().await;
        CurrentUnmerkleized {
            batch: inner.new_batch(),
            db: db.clone(),
            metadata: None,
        }
    }

    fn matches_sync_target(batch: &Self::Merkleized, target: &Self::SyncTarget) -> bool {
        batch.ops_root() == target.root
            && *target.range.start() == batch.sync_boundary()
            && *target.range.end() == Location::<F>::new(batch.bounds().total_size)
    }

    async fn finalize(&mut self, batch: Self::Merkleized) -> Result<(), Error<F>> {
        self.apply_batch(batch.inner).await?;
        self.sync().await
    }

    async fn prune(&mut self, target: &Self::SyncTarget) -> Result<(), Error<F>> {
        self.prune((*target.range.start()).into()).await
    }

    fn sync_target(&self) -> Self::SyncTarget {
        let bounds = self.bounds();
        CurrentSyncTarget::new(
            self.ops_root(),
            non_empty_range!(self.sync_boundary(), bounds.end),
        )
    }

    async fn rewind_to_target(&mut self, target: Self::SyncTarget) -> Result<(), Error<F>> {
        self.rewind(target.range.end()).await?;
        self.sync().await?;

        let rewound_target = self.sync_target();
        assert_eq!(
            rewound_target, target,
            "rewound database target mismatch after rewind",
        );
        Ok(())
    }
}

/// Implement [`ManagedDb`] for ordered current QMDB databases with variable-size values.
impl<F, E, K, V, H, T, const N: usize, S> ManagedDb<E>
    for Db<
        F,
        E,
        VariableJournal<E, Operation<F, ordered::Update<K, VariableEncoding<V>>>>,
        OrderedIdx<T, Location<F>>,
        H,
        ordered::Update<K, VariableEncoding<V>>,
        N,
        S,
    >
where
    F: Graftable,
    E: Context,
    K: Key,
    V: value::VariableValue + 'static,
    H: Hasher,
    T: Translator,
    S: Strategy,
    Operation<F, ordered::Update<K, VariableEncoding<V>>>: Codec,
{
    type Unmerkleized = CurrentUnmerkleized<
        F,
        E,
        VariableJournal<E, Operation<F, ordered::Update<K, VariableEncoding<V>>>>,
        OrderedIdx<T, Location<F>>,
        H,
        ordered::Update<K, VariableEncoding<V>>,
        N,
        S,
    >;
    type Merkleized = CurrentMerkleized<
        F,
        E,
        VariableJournal<E, Operation<F, ordered::Update<K, VariableEncoding<V>>>>,
        OrderedIdx<T, Location<F>>,
        H,
        ordered::Update<K, VariableEncoding<V>>,
        N,
        S,
    >;
    type Error = Error<F>;
    type Config = VariableConfig<
        T,
        <Operation<F, ordered::Update<K, VariableEncoding<V>>> as CodecRead>::Cfg,
        S,
    >;
    type SyncTarget = CurrentSyncTarget<F, H::Digest>;

    async fn init(context: E, config: Self::Config) -> Result<Self, Error<F>> {
        open::ordered_variable(context, config).await
    }

    fn initial_sync_target() -> Self::SyncTarget {
        CurrentSyncTarget::new(
            initial_root::<F, ordered::Update<K, VariableEncoding<V>>, H>(),
            non_empty_range!(Location::new(0), Location::new(1)),
        )
    }

    async fn new_batch(db: &Arc<TracedAsyncRwLock<Self>>) -> Self::Unmerkleized {
        let inner = db.read().await;
        CurrentUnmerkleized {
            batch: inner.new_batch(),
            db: db.clone(),
            metadata: None,
        }
    }

    fn matches_sync_target(batch: &Self::Merkleized, target: &Self::SyncTarget) -> bool {
        batch.ops_root() == target.root
            && *target.range.start() == batch.sync_boundary()
            && *target.range.end() == Location::<F>::new(batch.bounds().total_size)
    }

    async fn finalize(&mut self, batch: Self::Merkleized) -> Result<(), Error<F>> {
        self.apply_batch(batch.inner).await?;
        self.sync().await
    }

    async fn prune(&mut self, target: &Self::SyncTarget) -> Result<(), Error<F>> {
        self.prune((*target.range.start()).into()).await
    }

    fn sync_target(&self) -> Self::SyncTarget {
        let bounds = self.bounds();
        CurrentSyncTarget::new(
            self.ops_root(),
            non_empty_range!(self.sync_boundary(), bounds.end),
        )
    }

    async fn rewind_to_target(&mut self, target: Self::SyncTarget) -> Result<(), Error<F>> {
        self.rewind(target.range.end()).await?;
        self.sync().await?;

        let rewound_target = self.sync_target();
        assert_eq!(
            rewound_target, target,
            "rewound database target mismatch after rewind",
        );
        Ok(())
    }
}

/// Implement [`StateSyncDb`] for unordered current QMDB databases with fixed-size values.
impl<F, E, K, V, H, T, R, const N: usize, S> StateSyncDb<E, R>
    for Db<
        F,
        E,
        FixedJournal<E, Operation<F, unordered::Update<K, FixedEncoding<V>>>>,
        UnorderedIdx<T, Location<F>>,
        H,
        unordered::Update<K, FixedEncoding<V>>,
        N,
        S,
    >
where
    F: Graftable,
    E: Context,
    K: Array,
    V: value::FixedValue + 'static,
    H: Hasher,
    T: Translator,
    S: Strategy,
    R: Resolver<
        Family = F,
        Op = Operation<F, unordered::Update<K, FixedEncoding<V>>>,
        Digest = H::Digest,
    >,
{
    type SyncError = sync::Error<F, R::Error, H::Digest>;

    async fn sync_db(
        context: E,
        config: Self::Config,
        resolver: R,
        target: Self::SyncTarget,
        tip_updates: mpsc::Receiver<Self::SyncTarget>,
        finish: Option<mpsc::Receiver<()>>,
        reached_target: Option<mpsc::Sender<Self::SyncTarget>>,
        sync_config: SyncEngineConfig,
    ) -> Result<Self, Self::SyncError> {
        sync::sync(sync::engine::Config {
            context,
            resolver,
            target,
            max_outstanding_requests: sync_config.max_outstanding_requests,
            fetch_batch_size: sync_config.fetch_batch_size,
            apply_batch_size: sync_config.apply_batch_size,
            db_config: config,
            update_rx: Some(tip_updates),
            finish_rx: finish,
            reached_target_tx: reached_target,
            max_retained_roots: sync_config.max_retained_roots,
        })
        .await
    }
}

/// Implement [`StateSyncDb`] for ordered current QMDB databases with fixed-size values.
impl<F, E, K, V, H, T, R, const N: usize, S> StateSyncDb<E, R>
    for Db<
        F,
        E,
        FixedJournal<E, Operation<F, ordered::Update<K, FixedEncoding<V>>>>,
        OrderedIdx<T, Location<F>>,
        H,
        ordered::Update<K, FixedEncoding<V>>,
        N,
        S,
    >
where
    F: Graftable,
    E: Context,
    K: Array,
    V: value::FixedValue + 'static,
    H: Hasher,
    T: Translator,
    S: Strategy,
    R: Resolver<
        Family = F,
        Op = Operation<F, ordered::Update<K, FixedEncoding<V>>>,
        Digest = H::Digest,
    >,
{
    type SyncError = sync::Error<F, R::Error, H::Digest>;

    async fn sync_db(
        context: E,
        config: Self::Config,
        resolver: R,
        target: Self::SyncTarget,
        tip_updates: mpsc::Receiver<Self::SyncTarget>,
        finish: Option<mpsc::Receiver<()>>,
        reached_target: Option<mpsc::Sender<Self::SyncTarget>>,
        sync_config: SyncEngineConfig,
    ) -> Result<Self, Self::SyncError> {
        sync::sync(sync::engine::Config {
            context,
            resolver,
            target,
            max_outstanding_requests: sync_config.max_outstanding_requests,
            fetch_batch_size: sync_config.fetch_batch_size,
            apply_batch_size: sync_config.apply_batch_size,
            db_config: config,
            update_rx: Some(tip_updates),
            finish_rx: finish,
            reached_target_tx: reached_target,
            max_retained_roots: sync_config.max_retained_roots,
        })
        .await
    }
}

/// Implement [`StateSyncDb`] for unordered current QMDB databases with variable-size values.
impl<F, E, K, V, H, T, R, const N: usize, S> StateSyncDb<E, R>
    for Db<
        F,
        E,
        VariableJournal<E, Operation<F, unordered::Update<K, VariableEncoding<V>>>>,
        UnorderedIdx<T, Location<F>>,
        H,
        unordered::Update<K, VariableEncoding<V>>,
        N,
        S,
    >
where
    F: Graftable,
    E: Context,
    K: Key + Array,
    V: value::VariableValue + 'static,
    H: Hasher,
    T: Translator,
    S: Strategy,
    Operation<F, unordered::Update<K, VariableEncoding<V>>>: Codec,
    R: Resolver<
        Family = F,
        Op = Operation<F, unordered::Update<K, VariableEncoding<V>>>,
        Digest = H::Digest,
    >,
{
    type SyncError = sync::Error<F, R::Error, H::Digest>;

    async fn sync_db(
        context: E,
        config: Self::Config,
        resolver: R,
        target: Self::SyncTarget,
        tip_updates: mpsc::Receiver<Self::SyncTarget>,
        finish: Option<mpsc::Receiver<()>>,
        reached_target: Option<mpsc::Sender<Self::SyncTarget>>,
        sync_config: SyncEngineConfig,
    ) -> Result<Self, Self::SyncError> {
        sync::sync(sync::engine::Config {
            context,
            resolver,
            target,
            max_outstanding_requests: sync_config.max_outstanding_requests,
            fetch_batch_size: sync_config.fetch_batch_size,
            apply_batch_size: sync_config.apply_batch_size,
            db_config: config,
            update_rx: Some(tip_updates),
            finish_rx: finish,
            reached_target_tx: reached_target,
            max_retained_roots: sync_config.max_retained_roots,
        })
        .await
    }
}

/// Implement [`StateSyncDb`] for ordered current QMDB databases with variable-size values.
impl<F, E, K, V, H, T, R, const N: usize, S> StateSyncDb<E, R>
    for Db<
        F,
        E,
        VariableJournal<E, Operation<F, ordered::Update<K, VariableEncoding<V>>>>,
        OrderedIdx<T, Location<F>>,
        H,
        ordered::Update<K, VariableEncoding<V>>,
        N,
        S,
    >
where
    F: Graftable,
    E: Context,
    K: Key,
    V: value::VariableValue + 'static,
    H: Hasher,
    T: Translator,
    S: Strategy,
    Operation<F, ordered::Update<K, VariableEncoding<V>>>: Codec,
    R: Resolver<
        Family = F,
        Op = Operation<F, ordered::Update<K, VariableEncoding<V>>>,
        Digest = H::Digest,
    >,
{
    type SyncError = sync::Error<F, R::Error, H::Digest>;

    async fn sync_db(
        context: E,
        config: Self::Config,
        resolver: R,
        target: Self::SyncTarget,
        tip_updates: mpsc::Receiver<Self::SyncTarget>,
        finish: Option<mpsc::Receiver<()>>,
        reached_target: Option<mpsc::Sender<Self::SyncTarget>>,
        sync_config: SyncEngineConfig,
    ) -> Result<Self, Self::SyncError> {
        sync::sync(sync::engine::Config {
            context,
            resolver,
            target,
            max_outstanding_requests: sync_config.max_outstanding_requests,
            fetch_batch_size: sync_config.fetch_batch_size,
            apply_batch_size: sync_config.apply_batch_size,
            db_config: config,
            update_rx: Some(tip_updates),
            finish_rx: finish,
            reached_target_tx: reached_target,
            max_retained_roots: sync_config.max_retained_roots,
        })
        .await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use commonware_cryptography::{sha256::Digest, Sha256};
    use commonware_parallel::Sequential;
    use commonware_runtime::{
        buffer::paged::CacheRef, deterministic, BufferPooler, Runner as _, Supervisor as _,
    };
    use commonware_storage::{
        journal::contiguous::{
            fixed::Config as FixedJournalConfig, variable::Config as VariableJournalConfig,
        },
        merkle::{full::Config as MerkleConfig, mmr},
        qmdb::current::{
            ordered::{fixed as ordered_fixed, variable as ordered_variable},
            unordered::fixed,
        },
        translator::TwoCap,
    };
    use commonware_utils::{non_empty_range, NZUsize, NZU16, NZU64};
    use std::num::{NonZeroU16, NonZeroUsize};

    type FixedDb = fixed::Db<
        mmr::Family,
        deterministic::Context,
        Digest,
        Digest,
        Sha256,
        TwoCap,
        64,
        Sequential,
    >;
    type OrderedFixedDb = ordered_fixed::Db<
        mmr::Family,
        deterministic::Context,
        Digest,
        Digest,
        Sha256,
        TwoCap,
        64,
        Sequential,
    >;
    type OrderedVariableDb = ordered_variable::Db<
        mmr::Family,
        deterministic::Context,
        Digest,
        Digest,
        Sha256,
        TwoCap,
        64,
        Sequential,
    >;

    const PAGE_SIZE: NonZeroU16 = NZU16!(101);
    const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(11);

    fn fixed_config(suffix: &str, pooler: &impl BufferPooler) -> FixedConfig<TwoCap, Sequential> {
        let page_cache = CacheRef::from_pooler(pooler, PAGE_SIZE, PAGE_CACHE_SIZE);
        FixedConfig {
            merkle_config: MerkleConfig {
                journal_partition: format!("stateful-current-journal-{suffix}"),
                metadata_partition: format!("stateful-current-metadata-{suffix}"),
                items_per_blob: NZU64!(11),
                write_buffer: NZUsize!(1024),
                strategy: Sequential,
                page_cache: page_cache.clone(),
            },
            journal_config: FixedJournalConfig {
                partition: format!("stateful-current-log-{suffix}"),
                items_per_blob: NZU64!(7),
                page_cache,
                write_buffer: NZUsize!(1024),
            },
            grafted_metadata_partition: format!("stateful-current-grafted-{suffix}"),
            translator: TwoCap,
            init_cache_size: Some(NZUsize!(1024)),
        }
    }

    fn variable_config(
        suffix: &str,
        pooler: &impl BufferPooler,
    ) -> VariableConfig<TwoCap, ((), ()), Sequential> {
        let page_cache = CacheRef::from_pooler(pooler, PAGE_SIZE, PAGE_CACHE_SIZE);
        VariableConfig {
            merkle_config: MerkleConfig {
                journal_partition: format!("stateful-current-journal-{suffix}"),
                metadata_partition: format!("stateful-current-metadata-{suffix}"),
                items_per_blob: NZU64!(11),
                write_buffer: NZUsize!(1024),
                strategy: Sequential,
                page_cache: page_cache.clone(),
            },
            journal_config: VariableJournalConfig {
                partition: format!("stateful-current-log-{suffix}"),
                items_per_section: NZU64!(7),
                compression: None,
                codec_config: ((), ()),
                page_cache,
                write_buffer: NZUsize!(1024),
            },
            grafted_metadata_partition: format!("stateful-current-grafted-{suffix}"),
            translator: TwoCap,
            init_cache_size: Some(NZUsize!(1024)),
        }
    }

    fn assert_managed_db<T: ManagedDb<deterministic::Context>>() {}

    fn assert_state_sync_db<T, R>()
    where
        T: StateSyncDb<deterministic::Context, R>,
    {
    }

    fn assert_database_set<T: crate::stateful::db::DatabaseSet<deterministic::Context>>() {}

    #[test]
    fn ordered_current_db_trait_impls_compile() {
        assert_managed_db::<OrderedFixedDb>();
        assert_managed_db::<OrderedVariableDb>();
        assert_state_sync_db::<OrderedFixedDb, Arc<OrderedFixedDb>>();
        assert_state_sync_db::<OrderedVariableDb, Arc<OrderedVariableDb>>();
        assert_database_set::<Arc<TracedAsyncRwLock<OrderedFixedDb>>>();
        assert_database_set::<Arc<TracedAsyncRwLock<OrderedVariableDb>>>();
    }

    #[test]
    fn ordered_fixed_managed_db_finalizes_batch_and_proves_exclusion() {
        deterministic::Runner::default().start(|context| async move {
            let config = fixed_config("ordered-fixed-managed-db", &context);
            let db = <OrderedFixedDb as ManagedDb<_>>::init(context.child("db"), config)
                .await
                .unwrap();
            let db = Arc::new(TracedAsyncRwLock::new("test", db));
            let key = Sha256::hash(b"key");
            let value = Sha256::hash(b"value");
            let metadata = Sha256::hash(b"metadata");
            let missing = Sha256::hash(b"missing");

            let batch = <OrderedFixedDb as ManagedDb<_>>::new_batch(&db)
                .await
                .write(key, Some(value))
                .with_metadata(metadata);
            let merkleized = crate::stateful::db::Unmerkleized::merkleize(batch)
                .await
                .unwrap();
            let expected_root = merkleized.root();

            {
                let mut guard = db.write().await;
                <OrderedFixedDb as ManagedDb<_>>::finalize(&mut *guard, merkleized)
                    .await
                    .unwrap();
            }

            let guard = db.read().await;
            assert_eq!(guard.root(), expected_root);
            assert_eq!(guard.get(&key).await.unwrap(), Some(value));

            let proof = guard.exclusion_proof(&missing).await.unwrap();
            assert!(OrderedFixedDb::verify_exclusion_proof(
                &missing,
                &proof,
                &guard.root(),
            ));
        });
    }

    /// The glue staged wrapper (`CurrentUnmerkleized::stage` -> `CurrentStaged::expand` ->
    /// `CurrentStaged::merkleize`) must return the same values and root as an explicit `get_many` +
    /// `write` + `merkleize`, including a staged delete, an upsert, and metadata flow (both set
    /// on the staged handle via `with_metadata` and carried from before staging). This guards
    /// metadata flow and db-handle pairing through the wrapper.
    #[test]
    fn ordered_fixed_staged_merkleize_matches_explicit_writes() {
        deterministic::Runner::default().start(|context| async move {
            let config = fixed_config("ordered-fixed-glue-staged", &context);
            let db = <OrderedFixedDb as ManagedDb<_>>::init(context.child("db"), config)
                .await
                .unwrap();
            let db = Arc::new(TracedAsyncRwLock::new("test", db));

            let key = |i: u64| Sha256::hash(&i.to_be_bytes());
            let val = |i: u64| Sha256::hash(&(i + 10_000).to_be_bytes());
            let metadata = Sha256::hash(b"metadata");

            // Seed keys 0..50 and finalize.
            let mut seed = <OrderedFixedDb as ManagedDb<_>>::new_batch(&db).await;
            for i in 0..50u64 {
                seed = seed.write(key(i), Some(val(i)));
            }
            let merkleized = crate::stateful::db::Unmerkleized::merkleize(seed)
                .await
                .unwrap();
            {
                let mut guard = db.write().await;
                <OrderedFixedDb as ManagedDb<_>>::finalize(&mut *guard, merkleized)
                    .await
                    .unwrap();
            }

            // Read set: key(1) updated, key(2) deleted, key(999) missing -> created.
            let read_keys = [key(1), key(2), key(999)];
            let keys: Vec<&Digest> = read_keys.iter().collect();
            let indexed_updates = vec![(0, Some(val(1_000))), (1, None), (2, Some(val(1_001)))];
            let upserts = vec![(key(3), Some(val(1_002)))];

            // Explicit path.
            let mut explicit = <OrderedFixedDb as ManagedDb<_>>::new_batch(&db).await;
            let explicit_values = explicit.get_many(&keys).await.unwrap();
            for (slot, value) in &indexed_updates {
                explicit = explicit.write(read_keys[*slot], *value);
            }
            for (k, v) in &upserts {
                explicit = explicit.write(*k, *v);
            }
            let explicit_root =
                crate::stateful::db::Unmerkleized::merkleize(explicit.with_metadata(metadata))
                    .await
                    .unwrap()
                    .root();

            // Staged path, with metadata set on the staged handle.
            let staged_batch = <OrderedFixedDb as ManagedDb<_>>::new_batch(&db).await;
            let split = 2;
            let (mut staged_values, staged) = staged_batch.stage(&keys[..split]).await.unwrap();
            let (range, suffix_values, staged) = staged.expand(&keys[split..]).await.unwrap();
            assert_eq!(range, split..keys.len());
            staged_values.extend(suffix_values);
            let staged_root = staged
                .with_metadata(metadata)
                .merkleize(indexed_updates.clone(), upserts.clone())
                .await
                .unwrap()
                .root();

            assert_eq!(explicit_values, staged_values);
            assert_eq!(explicit_root, staged_root);

            // Metadata set before staging must be carried through to staged merkleize.
            let carried_batch = <OrderedFixedDb as ManagedDb<_>>::new_batch(&db)
                .await
                .with_metadata(metadata);
            let (carried_values, staged) = carried_batch.stage(&keys).await.unwrap();
            let carried_root = staged
                .merkleize(indexed_updates.clone(), upserts.clone())
                .await
                .unwrap()
                .root();
            assert_eq!(explicit_values, carried_values);
            assert_eq!(explicit_root, carried_root);
        });
    }

    #[test]
    fn ordered_variable_managed_db_finalizes_batch_and_proves_exclusion() {
        deterministic::Runner::default().start(|context| async move {
            let config = variable_config("ordered-variable-managed-db", &context);
            let db = <OrderedVariableDb as ManagedDb<_>>::init(context.child("db"), config)
                .await
                .unwrap();
            let db = Arc::new(TracedAsyncRwLock::new("test", db));
            let key = Sha256::hash(b"key");
            let value = Sha256::hash(b"value");
            let metadata = Sha256::hash(b"metadata");
            let missing = Sha256::hash(b"missing");

            let batch = <OrderedVariableDb as ManagedDb<_>>::new_batch(&db)
                .await
                .write(key, Some(value))
                .with_metadata(metadata);
            let merkleized = crate::stateful::db::Unmerkleized::merkleize(batch)
                .await
                .unwrap();
            let expected_root = merkleized.root();

            {
                let mut guard = db.write().await;
                <OrderedVariableDb as ManagedDb<_>>::finalize(&mut *guard, merkleized)
                    .await
                    .unwrap();
            }

            let guard = db.read().await;
            assert_eq!(guard.root(), expected_root);
            assert_eq!(guard.get(&key).await.unwrap(), Some(value));

            let proof = guard.exclusion_proof(&missing).await.unwrap();
            assert!(OrderedVariableDb::verify_exclusion_proof(
                &missing,
                &proof,
                &guard.root(),
            ));
        });
    }

    #[test]
    fn ordered_managed_db_matches_sync_target_rejects_wrong_ops_root_and_range() {
        deterministic::Runner::default().start(|context| async move {
            let config = fixed_config("ordered-matches-sync-target", &context);
            let db = <OrderedFixedDb as ManagedDb<_>>::init(context.child("db"), config.clone())
                .await
                .unwrap();
            let db = Arc::new(TracedAsyncRwLock::new("test", db));

            let key = Sha256::hash(b"key");
            let value = Sha256::hash(b"value");
            let metadata = Sha256::hash(b"metadata");

            let batch = <OrderedFixedDb as ManagedDb<_>>::new_batch(&db)
                .await
                .write(key, Some(value))
                .with_metadata(metadata);
            let merkleized = crate::stateful::db::Unmerkleized::merkleize(batch)
                .await
                .unwrap();

            let mut verification_db =
                <OrderedFixedDb as ManagedDb<_>>::init(context.child("verification_db"), config)
                    .await
                    .unwrap();
            verification_db
                .apply_batch(merkleized.inner.clone())
                .await
                .unwrap();
            verification_db.sync().await.unwrap();

            let valid_target = <OrderedFixedDb as ManagedDb<_>>::sync_target(&verification_db);
            assert!(<OrderedFixedDb as ManagedDb<_>>::matches_sync_target(
                &merkleized,
                &valid_target,
            ));

            let mut wrong_root = valid_target.clone();
            wrong_root.root = Sha256::hash(b"wrong ops root");
            assert!(!<OrderedFixedDb as ManagedDb<_>>::matches_sync_target(
                &merkleized,
                &wrong_root,
            ));

            let mut wrong_range = valid_target.clone();
            wrong_range.range = non_empty_range!(
                mmr::Location::new(*valid_target.range.start()),
                mmr::Location::new(*valid_target.range.end() + 1)
            );
            assert!(!<OrderedFixedDb as ManagedDb<_>>::matches_sync_target(
                &merkleized,
                &wrong_range,
            ));
        });
    }

    #[test]
    fn ordered_managed_db_rewind_to_target_round_trips() {
        deterministic::Runner::default().start(|context| async move {
            let config = fixed_config("ordered-rewind-round-trip", &context);
            let db = <OrderedFixedDb as ManagedDb<_>>::init(context.child("db"), config)
                .await
                .unwrap();
            let db = Arc::new(TracedAsyncRwLock::new("test", db));

            let key1 = Sha256::hash(b"key1");
            let value1 = Sha256::hash(b"value1");
            let metadata1 = Sha256::hash(b"metadata1");
            let batch1 = <OrderedFixedDb as ManagedDb<_>>::new_batch(&db)
                .await
                .write(key1, Some(value1))
                .with_metadata(metadata1);
            let merkleized1 = crate::stateful::db::Unmerkleized::merkleize(batch1)
                .await
                .unwrap();
            {
                let mut guard = db.write().await;
                <OrderedFixedDb as ManagedDb<_>>::finalize(&mut *guard, merkleized1)
                    .await
                    .unwrap();
            }
            let target_after_first = {
                let guard = db.read().await;
                <OrderedFixedDb as ManagedDb<_>>::sync_target(&*guard)
            };

            let key2 = Sha256::hash(b"key2");
            let value2 = Sha256::hash(b"value2");
            let metadata2 = Sha256::hash(b"metadata2");
            let batch2 = <OrderedFixedDb as ManagedDb<_>>::new_batch(&db)
                .await
                .write(key2, Some(value2))
                .with_metadata(metadata2);
            let merkleized2 = crate::stateful::db::Unmerkleized::merkleize(batch2)
                .await
                .unwrap();
            {
                let mut guard = db.write().await;
                <OrderedFixedDb as ManagedDb<_>>::finalize(&mut *guard, merkleized2)
                    .await
                    .unwrap();
            }

            {
                let mut guard = db.write().await;
                <OrderedFixedDb as ManagedDb<_>>::rewind_to_target(
                    &mut *guard,
                    target_after_first.clone(),
                )
                .await
                .unwrap();
            }
            let target_after_rewind = {
                let guard = db.read().await;
                <OrderedFixedDb as ManagedDb<_>>::sync_target(&*guard)
            };
            assert_eq!(target_after_rewind, target_after_first);
        });
    }

    #[test]
    fn managed_db_matches_sync_target_rejects_wrong_ops_root_and_range() {
        deterministic::Runner::default().start(|context| async move {
            let config = fixed_config("matches-sync-target", &context);
            let db = FixedDb::init(context.child("db"), config.clone())
                .await
                .unwrap();
            let db = Arc::new(TracedAsyncRwLock::new("test", db));

            let key = Sha256::hash(b"key");
            let value = Sha256::hash(b"value");
            let metadata = Sha256::hash(b"metadata");

            let batch = <FixedDb as ManagedDb<_>>::new_batch(&db)
                .await
                .write(key, Some(value))
                .with_metadata(metadata);
            let merkleized = crate::stateful::db::Unmerkleized::merkleize(batch)
                .await
                .unwrap();

            let mut verification_db = FixedDb::init(context.child("verification_db"), config)
                .await
                .unwrap();
            verification_db
                .apply_batch(merkleized.inner.clone())
                .await
                .unwrap();
            verification_db.sync().await.unwrap();

            let valid_target = <FixedDb as ManagedDb<_>>::sync_target(&verification_db);
            assert!(<FixedDb as ManagedDb<_>>::matches_sync_target(
                &merkleized,
                &valid_target,
            ));

            let mut wrong_root = valid_target.clone();
            wrong_root.root = Sha256::hash(b"wrong ops root");
            assert!(!<FixedDb as ManagedDb<_>>::matches_sync_target(
                &merkleized,
                &wrong_root,
            ));

            let mut wrong_range = valid_target.clone();
            wrong_range.range = non_empty_range!(
                mmr::Location::new(*valid_target.range.start()),
                mmr::Location::new(*valid_target.range.end() + 1)
            );
            assert!(!<FixedDb as ManagedDb<_>>::matches_sync_target(
                &merkleized,
                &wrong_range,
            ));
        });
    }
}