//! Compact [`ManagedDb`] and [`StateSyncDb`] implementations for QMDB
//! [`immutable`](commonware_storage::qmdb::immutable) databases.
//!
//! Compact databases retain only the current Merkle peaks. Batches support `set` and
//! merkleization but no historical reads.

use crate::stateful::db::{
    BatchContext, InitError, ManagedDb, Merkleized as MerkleizedTrait, Shared, StateSyncDb,
    SyncEngineConfig, Unmerkleized as UnmerkleizedTrait, sync_compact_db, validate_initialization,
};
use commonware_codec::{EncodeShared, Read as CodecRead};
use commonware_cryptography::Hasher;
use commonware_parallel::Strategy;
use commonware_runtime::{Handle, Spawner};
use commonware_storage::{
    Context,
    merkle::{Family, Location},
    qmdb::{
        Error,
        any::value::{FixedEncoding, FixedValue, ValueEncoding, VariableEncoding, VariableValue},
        immutable::{
            CompactDb, CompactMerkleizedBatch, CompactUnmerkleizedBatch, Operation, fixed,
            initial_root, variable,
        },
        operation::Key,
        sync,
    },
};
use commonware_utils::{
    Array,
    channel::{mpsc, ring},
};
use std::{ops::Deref, sync::Arc};

/// A speculative batch of new keyed values over a shared compact immutable database.
pub struct ImmutableUnjournaledUnmerkleized<F, E, K, V, H, S>
where
    F: Family,
    E: Context,
    K: Key,
    V: ValueEncoding,
    H: Hasher,
    Operation<F, K, V>: EncodeShared,
    Operation<F, K, V>: CodecRead,
    S: Strategy,
{
    batch: CompactUnmerkleizedBatch<F, H, K, V, S>,
    db: Shared<CompactDb<F, E, K, V, H, S>>,
    metadata: Option<V::Value>,
    inactivity_floor: Option<Location<F>>,
}

impl<F, E, K, V, H, S> Deref for ImmutableUnjournaledUnmerkleized<F, E, K, V, H, S>
where
    F: Family,
    E: Context,
    K: Key,
    V: ValueEncoding,
    H: Hasher,
    Operation<F, K, V>: EncodeShared,
    Operation<F, K, V>: CodecRead,
    S: Strategy,
{
    type Target = CompactUnmerkleizedBatch<F, H, K, V, S>;

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

impl<F, E, K, V, H, S> ImmutableUnjournaledUnmerkleized<F, E, K, V, H, S>
where
    F: Family,
    E: Context,
    K: Key,
    V: ValueEncoding,
    H: Hasher,
    Operation<F, K, V>: EncodeShared,
    Operation<F, K, V>: CodecRead,
    S: Strategy,
{
    /// Sets the metadata committed by [`merkleize`](UnmerkleizedTrait::merkleize).
    pub fn with_metadata(mut self, metadata: V::Value) -> Self {
        self.metadata = Some(metadata);
        self
    }

    /// Sets the inactivity floor committed by [`merkleize`](UnmerkleizedTrait::merkleize)
    /// (location 0 when unset).
    pub const fn with_inactivity_floor(mut self, floor: Location<F>) -> Self {
        self.inactivity_floor = Some(floor);
        self
    }

    /// Sets `key` to `value` in the batch.
    pub fn set(mut self, key: K, value: V::Value) -> Self {
        self.batch = self.batch.set(key, value);
        self
    }
}

/// A sealed compact immutable batch with a computed root.
pub struct ImmutableUnjournaledMerkleized<F, E, K, V, H, S>
where
    F: Family,
    E: Context,
    K: Key,
    V: ValueEncoding,
    H: Hasher,
    Operation<F, K, V>: EncodeShared,
    Operation<F, K, V>: CodecRead,
    S: Strategy,
{
    inner: Arc<CompactMerkleizedBatch<F, H::Digest, K, V, S>>,
    db: Shared<CompactDb<F, E, K, V, H, S>>,
}

impl<F, E, K, V, H, S> Clone for ImmutableUnjournaledMerkleized<F, E, K, V, H, S>
where
    F: Family,
    E: Context,
    K: Key,
    V: ValueEncoding,
    H: Hasher,
    Operation<F, K, V>: EncodeShared,
    Operation<F, K, V>: CodecRead,
    S: Strategy,
{
    fn clone(&self) -> Self {
        Self {
            inner: Arc::clone(&self.inner),
            db: self.db.clone(),
        }
    }
}

impl<F, E, K, V, H, S> Deref for ImmutableUnjournaledMerkleized<F, E, K, V, H, S>
where
    F: Family,
    E: Context,
    K: Key,
    V: ValueEncoding,
    H: Hasher,
    Operation<F, K, V>: EncodeShared,
    Operation<F, K, V>: CodecRead,
    S: Strategy,
{
    type Target = CompactMerkleizedBatch<F, H::Digest, K, V, S>;

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

impl<F, E, K, V, H, S> UnmerkleizedTrait for ImmutableUnjournaledUnmerkleized<F, E, K, V, H, S>
where
    F: Family,
    E: Context,
    K: Key,
    V: ValueEncoding,
    H: Hasher,
    Operation<F, K, V>: EncodeShared,
    Operation<F, K, V>: CodecRead,
    S: Strategy,
{
    type Merkleized = ImmutableUnjournaledMerkleized<F, E, K, V, H, 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,
                self.inactivity_floor.unwrap_or_default(),
            )
            .await?;
        Ok(ImmutableUnjournaledMerkleized {
            inner: merkleized,
            db: self.db.clone(),
        })
    }
}

impl<F, E, K, V, H, S> MerkleizedTrait for ImmutableUnjournaledMerkleized<F, E, K, V, H, S>
where
    F: Family,
    E: Context,
    K: Key,
    V: ValueEncoding,
    H: Hasher,
    Operation<F, K, V>: EncodeShared,
    Operation<F, K, V>: CodecRead,
    S: Strategy,
{
    type Digest = H::Digest;
    type Unmerkleized = ImmutableUnjournaledUnmerkleized<F, E, K, V, H, S>;

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

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

impl<F, E, K, V, H, S> ManagedDb<E> for fixed::CompactDb<F, E, K, V, H, S>
where
    F: Family,
    E: Context,
    K: Array,
    V: FixedValue,
    H: Hasher,
    S: Strategy,
    Operation<F, K, FixedEncoding<V>>: EncodeShared + CodecRead<Cfg = ()>,
{
    type Unmerkleized = ImmutableUnjournaledUnmerkleized<F, E, K, FixedEncoding<V>, H, S>;
    type Merkleized = ImmutableUnjournaledMerkleized<F, E, K, FixedEncoding<V>, H, S>;
    type Error = Error<F>;
    type Config = fixed::CompactConfig<S>;
    type SyncTarget = sync::CompactTarget<F, H::Digest>;

    async fn init(
        context: E,
        config: Self::Config,
        expected: Option<Self::SyncTarget>,
    ) -> Result<Self, InitError<Error<F>, Self::SyncTarget>> {
        let db = <Self>::init(context, config, expected.as_ref().map(|target| target.size))
            .await
            .map_err(InitError::Database)?;
        validate_initialization(db, expected)
    }

    fn initial_sync_target() -> Self::SyncTarget {
        sync::CompactTarget {
            root: initial_root::<F, K, FixedEncoding<V>, H>(),
            size: Location::new(1),
        }
    }

    fn new_batch(database: BatchContext<'_, Self>) -> Self::Unmerkleized {
        let (database, shared) = database.into_parts();
        ImmutableUnjournaledUnmerkleized {
            batch: database.new_batch(),
            db: shared,
            metadata: None,
            inactivity_floor: None,
        }
    }

    fn matches_sync_target(batch: &Self::Merkleized, target: &Self::SyncTarget) -> bool {
        batch.root() == target.root && target.size == batch.bounds().tip.size
    }

    async fn apply(self, batch: Self::Merkleized) -> Result<Self, Error<F>> {
        self.apply_batch(batch.inner).await
    }

    async fn finalize(self) -> Result<(Self, Handle<()>), Error<F>> {
        self.start_sync().await
    }

    async fn prune(self, target: &Self::SyncTarget) -> Result<Self, Error<F>> {
        Self::prune(self, target.size).await
    }

    fn sync_target(&self) -> Self::SyncTarget {
        self.target()
    }
}

impl<F, E, K, V, H, S> ManagedDb<E> for variable::CompactDb<F, E, K, V, H, S>
where
    F: Family,
    E: Context,
    K: Key,
    V: VariableValue,
    H: Hasher,
    Operation<F, K, VariableEncoding<V>>: EncodeShared + CodecRead,
    S: Strategy,
{
    type Unmerkleized = ImmutableUnjournaledUnmerkleized<F, E, K, VariableEncoding<V>, H, S>;
    type Merkleized = ImmutableUnjournaledMerkleized<F, E, K, VariableEncoding<V>, H, S>;
    type Error = Error<F>;
    type Config =
        variable::CompactConfig<<Operation<F, K, VariableEncoding<V>> as CodecRead>::Cfg, S>;
    type SyncTarget = sync::CompactTarget<F, H::Digest>;

    async fn init(
        context: E,
        config: Self::Config,
        expected: Option<Self::SyncTarget>,
    ) -> Result<Self, InitError<Error<F>, Self::SyncTarget>> {
        let db = <Self>::init(context, config, expected.as_ref().map(|target| target.size))
            .await
            .map_err(InitError::Database)?;
        validate_initialization(db, expected)
    }

    fn initial_sync_target() -> Self::SyncTarget {
        sync::CompactTarget {
            root: initial_root::<F, K, VariableEncoding<V>, H>(),
            size: Location::new(1),
        }
    }

    fn new_batch(database: BatchContext<'_, Self>) -> Self::Unmerkleized {
        let (database, shared) = database.into_parts();
        ImmutableUnjournaledUnmerkleized {
            batch: database.new_batch(),
            db: shared,
            metadata: None,
            inactivity_floor: None,
        }
    }

    fn matches_sync_target(batch: &Self::Merkleized, target: &Self::SyncTarget) -> bool {
        batch.root() == target.root && target.size == batch.bounds().tip.size
    }

    async fn apply(self, batch: Self::Merkleized) -> Result<Self, Error<F>> {
        self.apply_batch(batch.inner).await
    }

    async fn finalize(self) -> Result<(Self, Handle<()>), Error<F>> {
        self.start_sync().await
    }

    async fn prune(self, target: &Self::SyncTarget) -> Result<Self, Error<F>> {
        Self::prune(self, target.size).await
    }

    fn sync_target(&self) -> Self::SyncTarget {
        self.target()
    }
}

impl<F, E, K, V, H, R, S> StateSyncDb<E, R> for fixed::CompactDb<F, E, K, V, H, S>
where
    F: Family,
    E: Context + Spawner,
    K: Array,
    V: FixedValue,
    H: Hasher,
    S: Strategy,
    Operation<F, K, FixedEncoding<V>>: EncodeShared + CodecRead<Cfg = ()>,
    R: sync::SourceFor<Self>,
{
    type SyncError = sync::Error<F, R::Error, H::Digest>;

    async fn sync_db(
        context: E,
        config: Self::Config,
        source: R,
        target: Self::SyncTarget,
        tip_updates: ring::Receiver<Self::SyncTarget>,
        finish: Option<mpsc::Receiver<()>>,
        reached_target: Option<ring::Sender<Self::SyncTarget>>,
        sync_config: SyncEngineConfig,
    ) -> Result<Self, Self::SyncError> {
        sync_compact_db(
            context,
            config,
            source,
            target,
            tip_updates,
            finish,
            reached_target,
            sync_config,
        )
        .await
    }
}

impl<F, E, K, V, H, R, S> StateSyncDb<E, R> for variable::CompactDb<F, E, K, V, H, S>
where
    F: Family,
    E: Context + Spawner,
    K: Key,
    V: VariableValue,
    H: Hasher,
    Operation<F, K, VariableEncoding<V>>: EncodeShared + CodecRead,
    S: Strategy,
    R: sync::SourceFor<Self>,
{
    type SyncError = sync::Error<F, R::Error, H::Digest>;

    async fn sync_db(
        context: E,
        config: Self::Config,
        source: R,
        target: Self::SyncTarget,
        tip_updates: ring::Receiver<Self::SyncTarget>,
        finish: Option<mpsc::Receiver<()>>,
        reached_target: Option<ring::Sender<Self::SyncTarget>>,
        sync_config: SyncEngineConfig,
    ) -> Result<Self, Self::SyncError> {
        sync_compact_db(
            context,
            config,
            source,
            target,
            tip_updates,
            finish,
            reached_target,
            sync_config,
        )
        .await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use commonware_cryptography::{Sha256, sha256::Digest};
    use commonware_macros::select;
    use commonware_parallel::Sequential;
    use commonware_runtime::{
        Clock as _, Metrics as _, Runner as _, Supervisor as _, buffer::paged::CacheRef,
        deterministic,
    };
    use commonware_storage::{
        journal::contiguous::fixed::Config as FixedJournalConfig,
        merkle::{full::Config as MerkleConfig, mmr},
        qmdb::sync::source,
        translator::TwoCap,
    };
    use commonware_utils::{NZU16, NZU64, NZUsize};
    use futures::pin_mut;
    use std::time::Duration;

    type FixedDb =
        fixed::CompactDb<mmr::Family, deterministic::Context, Digest, Digest, Sha256, Sequential>;
    type FullFixedDb =
        fixed::Db<mmr::Family, deterministic::Context, Digest, Digest, Sha256, TwoCap, Sequential>;
    type VariableDb = variable::CompactDb<
        mmr::Family,
        deterministic::Context,
        Digest,
        Vec<u8>,
        Sha256,
        Sequential,
    >;

    fn fixed_config(suffix: &str) -> fixed::CompactConfig<Sequential> {
        fixed::CompactConfig {
            strategy: Sequential,
            witness: commonware_storage::journal::contiguous::variable::Config {
                partition: format!("stateful-immutable-unjournaled-{suffix}-witness"),
                items_per_section: NZU64!(64),
                compression: None,
                codec_config: (),
                page_cache: CacheRef::new(NZU16!(101), NZUsize!(11)),
                write_buffer: NZUsize!(1024),
                replay_buffer: NZUsize!(1024),
            },
            commit_codec_config: (),
        }
    }

    fn full_fixed_config(suffix: &str) -> fixed::Config<TwoCap, Sequential> {
        let page_cache = CacheRef::new(NZU16!(101), NZUsize!(11));
        fixed::Config {
            merkle_config: MerkleConfig {
                journal_partition: format!("stateful-immutable-full-journal-{suffix}"),
                metadata_partition: format!("stateful-immutable-full-metadata-{suffix}"),
                items_per_blob: NZU64!(11),
                write_buffer: NZUsize!(1024),
                replay_buffer: NZUsize!(1024),
                strategy: Sequential,
                page_cache: page_cache.clone(),
            },
            log: FixedJournalConfig {
                partition: format!("stateful-immutable-full-log-{suffix}"),
                items_per_blob: NZU64!(7),
                page_cache,
                write_buffer: NZUsize!(1024),
                replay_buffer: NZUsize!(1024),
            },
            translator: TwoCap,
            init_buffer: NZUsize!(1 << 21),
        }
    }

    const fn sync_config() -> SyncEngineConfig {
        SyncEngineConfig {
            fetch_batch_size: NZU64!(1),
            apply_batch_size: NZU64!(1),
            max_outstanding_requests: NZUsize!(1),
        }
    }

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

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

    #[derive(Clone)]
    struct SupersedingCompactSource {
        source: Arc<FullFixedDb>,
        stale_target: sync::CompactTarget<mmr::Family, Digest>,
        stale_request_tx: mpsc::Sender<()>,
    }

    impl sync::Source for SupersedingCompactSource {
        type Family = mmr::Family;
        type Digest = Digest;
        type Op = fixed::Operation<mmr::Family, Digest, Digest>;
        type Error = <Arc<FullFixedDb> as sync::Source>::Error;

        async fn serve(&self, request: sync::Request<Self::Family>) -> source::Result<Self> {
            if request.size() == self.stale_target.size {
                let _ = self.stale_request_tx.send(()).await;
                return futures::future::pending().await;
            }

            self.source.serve(request).await
        }
    }

    #[test]
    fn immutable_unjournaled_trait_impls_compile() {
        assert_managed_db::<FixedDb>();
        assert_managed_db::<VariableDb>();
        assert_state_sync_db::<FixedDb, Arc<FixedDb>>();
        assert_state_sync_db::<VariableDb, Arc<VariableDb>>();
    }

    #[test]
    fn managed_db_apply_and_finalize_persists_fixed_immutable_unjournaled_batches() {
        deterministic::Runner::default().start(|context| async move {
            let config = fixed_config("managed-db");
            let db = FixedDb::init(context.child("db"), config, None)
                .await
                .unwrap();
            let db = Shared::new("test", db);
            let key = Sha256::hash(&[&[1]]);
            let value = Sha256::hash(&[&[2]]);
            let metadata = Sha256::hash(&[&[3]]);

            let batch = db
                .new_batch_for_test::<_>()
                .await
                .set(key, value)
                .with_inactivity_floor(mmr::Location::new(1))
                .with_metadata(metadata);
            let merkleized = crate::stateful::db::Unmerkleized::merkleize(batch)
                .await
                .unwrap();
            let expected_root = merkleized.root();

            {
                let (slot, database) = db.write().await;
                let database = <FixedDb as ManagedDb<_>>::apply(database, merkleized)
                    .await
                    .unwrap();
                let (database, sync) = <FixedDb as ManagedDb<_>>::finalize(database).await.unwrap();
                slot.put(database);
                sync.await.expect("database sync failed");
            }

            let guard = db.read().await;
            assert_eq!(guard.root(), expected_root);
            assert_eq!(guard.get_metadata(), Some(metadata));

            let target = <FixedDb as ManagedDb<_>>::sync_target(&guard);
            assert_eq!(target.root, expected_root);
            assert_eq!(target.size, mmr::Location::new(3));
        });
    }

    #[test]
    fn managed_db_apply_retains_each_immutable_bounded_initialization_target() {
        deterministic::Runner::default().start(|context| async move {
            let config = fixed_config("apply-checkpoints");
            let db = FixedDb::init(context.child("db"), config, None)
                .await
                .unwrap();
            let db = Shared::new("test", db);

            let first = db
                .new_batch_for_test::<_>()
                .await
                .set(Sha256::hash(&[&[1]]), Sha256::hash(&[&[2]]))
                .with_metadata(Sha256::hash(&[&[11]]));
            let first = crate::stateful::db::Unmerkleized::merkleize(first)
                .await
                .unwrap();
            let first_target = sync::CompactTarget {
                root: first.root(),
                size: first.bounds().tip.size,
            };
            let (slot, database) = db.write().await;
            let database = <FixedDb as ManagedDb<_>>::apply(database, first)
                .await
                .unwrap();
            slot.put(database);

            let second = db
                .new_batch_for_test::<_>()
                .await
                .set(Sha256::hash(&[&[3]]), Sha256::hash(&[&[4]]))
                .with_metadata(Sha256::hash(&[&[22]]));
            let second = crate::stateful::db::Unmerkleized::merkleize(second)
                .await
                .unwrap();
            let (slot, database) = db.write().await;
            let database = <FixedDb as ManagedDb<_>>::apply(database, second)
                .await
                .unwrap();
            let (database, sync) = <FixedDb as ManagedDb<_>>::finalize(database).await.unwrap();
            sync.await.expect("database sync failed");
            slot.put(database);
            drop(db);

            let database = <FixedDb as ManagedDb<_>>::init(
                context.child("reopen"),
                fixed_config("apply-checkpoints"),
                Some(first_target.clone()),
            )
            .await
            .unwrap();
            assert_eq!(
                <FixedDb as ManagedDb<_>>::sync_target(&database),
                first_target,
            );
        });
    }

    #[test]
    fn database_set_bounded_initialization_persists_aligned_immutable_target() {
        deterministic::Runner::default().start(|context| async move {
            let config = fixed_config("aligned-bounded-init");
            let db = FixedDb::init(context.child("db"), config.clone(), None)
                .await
                .unwrap();
            let db = Shared::new("test", db);

            let batch = db
                .new_batch_for_test::<_>()
                .await
                .set(Sha256::hash(&[&[1]]), Sha256::hash(&[&[2]]))
                .with_metadata(Sha256::hash(&[&[3]]));
            let batch = crate::stateful::db::Unmerkleized::merkleize(batch)
                .await
                .unwrap();
            crate::stateful::db::DatabaseSet::apply(&db, batch).await;
            let target = crate::stateful::db::DatabaseSet::committed_targets(&db).await;
            crate::stateful::db::DatabaseSet::finalize(&db)
                .await
                .durable()
                .await;
            drop(db);
            let db = <Shared<FixedDb> as crate::stateful::db::DatabaseSet<_>>::init(
                context.child("aligned_cap"),
                config,
                Some(target.clone()),
            )
            .await;
            drop(db);

            let database = FixedDb::init(
                context.child("reopen"),
                fixed_config("aligned-bounded-init"),
                None,
            )
            .await
            .unwrap();
            assert_eq!(<FixedDb as ManagedDb<_>>::sync_target(&database), target,);
        });
    }

    #[test]
    fn state_sync_fetches_fixed_immutable_compact_state() {
        deterministic::Runner::default().start(|context| async move {
            let source = FixedDb::init(context.child("source"), fixed_config("source"), None)
                .await
                .unwrap();
            let metadata = Sha256::hash(&[&[3]]);
            let floor = source.inactivity_floor_loc();
            let batch = source
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::hash(&[&[2]]))
                .merkleize(&source, Some(metadata), floor)
                .await
                .unwrap();
            let source = source.apply_batch(batch).await.unwrap();
            let source = source.sync().await.unwrap();

            let target = source.target();
            let (_update_tx, update_rx) = ring::channel(NZUsize!(1));
            let synced = <FixedDb as StateSyncDb<_, Arc<FixedDb>>>::sync_db(
                context.child("target"),
                fixed_config("target"),
                Arc::new(source),
                target.clone(),
                update_rx,
                None,
                None,
                sync_config(),
            )
            .await
            .unwrap();

            assert_eq!(synced.target(), target);
            assert_eq!(synced.get_metadata(), Some(metadata));
        });
    }

    #[test]
    fn state_sync_reports_compact_progress() {
        deterministic::Runner::default().start(|context| async move {
            let source_context = context.child("source");
            let source_config = full_fixed_config("source");
            let source = FullFixedDb::init(source_context, source_config, None)
                .await
                .unwrap();
            let floor = source.inactivity_floor_loc();
            let batch = source
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::hash(&[&[2]]))
                .merkleize(&source, Some(Sha256::hash(&[&[3]])), floor)
                .await
                .unwrap();
            let source = source.apply_batch(batch).await.unwrap();
            let source = source.sync().await.unwrap();
            let target = sync::CompactTarget {
                root: source.root(),
                size: source.bounds().end,
            };

            // A larger target the source never serves. Its sync attempt
            // hangs so the test can observe the gauges while they diverge.
            let unservable_target = sync::CompactTarget {
                root: Sha256::hash(&[&[0xFF]]),
                size: target.size + 1,
            };
            let (stale_request_tx, mut stale_request_rx) = mpsc::channel(1);
            let superseding_source = SupersedingCompactSource {
                source: Arc::new(source),
                stale_target: unservable_target.clone(),
                stale_request_tx,
            };

            let (update_tx, update_rx) = ring::channel(NZUsize!(1));
            let (_finish_tx, finish_rx) = mpsc::channel(1);
            let (reached_tx, mut reached_rx) = ring::channel(NZUsize!(1));
            let client_context = context.child("client");
            let client_config = fixed_config("client");
            let sync = <FixedDb as StateSyncDb<_, _>>::sync_db(
                client_context,
                client_config,
                superseding_source,
                target.clone(),
                update_rx,
                Some(finish_rx),
                Some(reached_tx),
                sync_config(),
            );
            pin_mut!(sync);

            select! {
                _ = sync.as_mut() => panic!("sync completed before explicit finish signal"),
                reached = reached_rx.recv() => assert_eq!(reached, Some(target.clone())),
            }

            let synced_size = *target.size;
            let encoded = context.encode();
            assert!(
                encoded.contains(&format!("\nclient_sync_target_size {synced_size}")),
                "missing compact sync target gauge: {encoded}"
            );
            assert!(
                encoded.contains(&format!("\nclient_sync_size {synced_size}")),
                "missing compact sync progress gauge: {encoded}"
            );

            // Supersede with the unservable target and wait for its fetch to
            // start. The target gauge advances while the synced gauge still
            // reports the previously reached target.
            assert!(update_tx.send_lossy(unservable_target.clone()));
            select! {
                _ = sync.as_mut() => panic!("sync completed with an unservable target"),
                request = stale_request_rx.recv() => assert_eq!(request, Some(())),
            }

            let target_size_val = *unservable_target.size;
            let encoded = context.encode();
            assert!(
                encoded.contains(&format!("\nclient_sync_target_size {target_size_val}")),
                "target gauge should advance to the superseding target: {encoded}"
            );
            assert!(
                encoded.contains(&format!("\nclient_sync_size {synced_size}")),
                "synced gauge should still report the reached target: {encoded}"
            );
        });
    }

    #[test]
    fn state_sync_supersedes_in_flight_stale_compact_target() {
        deterministic::Runner::default().start(|context| async move {
            let source =
                FullFixedDb::init(context.child("source"), full_fixed_config("source"), None)
                    .await
                    .unwrap();

            let floor = source.inactivity_floor_loc();
            let batch = source
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::hash(&[&[2]]))
                .merkleize(&source, Some(Sha256::hash(&[&[9]])), floor)
                .await
                .unwrap();
            let source = source.apply_batch(batch).await.unwrap();
            let source = source.sync().await.unwrap();
            let stale_target = sync::CompactTarget {
                root: source.root(),
                size: source.bounds().end,
            };

            let floor = source.inactivity_floor_loc();
            let batch = source
                .new_batch()
                .set(Sha256::hash(&[&[3]]), Sha256::hash(&[&[4]]))
                .merkleize(&source, Some(Sha256::hash(&[&[10]])), floor)
                .await
                .unwrap();
            let source = source.apply_batch(batch).await.unwrap();
            let source = source.sync().await.unwrap();
            let latest_target = sync::CompactTarget {
                root: source.root(),
                size: source.bounds().end,
            };

            let (stale_request_tx, mut stale_request_rx) = mpsc::channel(1);
            let superseding_source = SupersedingCompactSource {
                source: Arc::new(source),
                stale_target: stale_target.clone(),
                stale_request_tx,
            };

            let (update_tx, update_rx) = ring::channel(NZUsize!(1));
            let sync_handle = context.child("sync").spawn(move |context| async move {
                <FixedDb as StateSyncDb<_, _>>::sync_db(
                    context.child("target"),
                    fixed_config("supersede-target"),
                    superseding_source,
                    stale_target,
                    update_rx,
                    None,
                    None,
                    sync_config(),
                )
                .await
            });

            context
                .timeout(Duration::from_secs(1), async move {
                    stale_request_rx.recv().await.unwrap();
                })
                .await
                .expect("sync should request the stale target first");
            assert!(update_tx.send_lossy(latest_target.clone()));

            let synced = context
                .timeout(Duration::from_secs(1), sync_handle)
                .await
                .expect("sync should switch to the latest target")
                .expect("spawned sync task should complete")
                .unwrap();

            assert_eq!(synced.target(), latest_target);
            assert_eq!(synced.get_metadata(), Some(Sha256::hash(&[&[10]])));
        });
    }

    #[test]
    fn managed_db_initializes_fixed_immutable_unjournaled_multiple_commit_ranges() {
        deterministic::Runner::default().start(|context| async move {
            let config = fixed_config("bounded-init");
            let db = FixedDb::init(context.child("db"), config.clone(), None)
                .await
                .unwrap();

            let floor = db.inactivity_floor_loc();
            let batch = db
                .new_batch()
                .set(Sha256::hash(&[&[1]]), Sha256::hash(&[&[2]]))
                .merkleize(&db, Some(Sha256::hash(&[&[11]])), floor)
                .await
                .unwrap();
            let mut db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            let first_target = <FixedDb as ManagedDb<_>>::sync_target(&db);

            // Add two ranges so reopening at the first target spans multiple commits.
            for i in [3u8, 5] {
                let floor = db.inactivity_floor_loc();
                let batch = db
                    .new_batch()
                    .set(Sha256::hash(&[&[i]]), Sha256::hash(&[&[i + 1]]))
                    .merkleize(&db, Some(Sha256::hash(&[&[i * 11]])), floor)
                    .await
                    .unwrap();
                db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
            }
            let third_target = <FixedDb as ManagedDb<_>>::sync_target(&db);
            assert_ne!(third_target, first_target);

            drop(db);
            let db = <FixedDb as ManagedDb<_>>::init(
                context.child("cap"),
                config.clone(),
                Some(first_target.clone()),
            )
            .await
            .unwrap();

            let recovered_target = <FixedDb as ManagedDb<_>>::sync_target(&db);
            assert_eq!(recovered_target, first_target);
            assert_eq!(db.get_metadata(), Some(Sha256::hash(&[&[11]])));
        });
    }

    #[test]
    fn managed_db_prune_bounds_fixed_immutable_unjournaled_bounded_initialization_history() {
        deterministic::Runner::default().start(|context| async move {
            // One witness entry per section so pruning takes effect at entry granularity.
            let mut config = fixed_config("prune");
            config.witness.items_per_section = NZU64!(1);
            let mut db = FixedDb::init(context.child("db"), config.clone(), None)
                .await
                .unwrap();

            // Commit three ranges, recording each target.
            let mut targets = Vec::new();
            for i in [1u8, 3, 5] {
                let floor = db.inactivity_floor_loc();
                let batch = db
                    .new_batch()
                    .set(Sha256::hash(&[&[i]]), Sha256::hash(&[&[i + 1]]))
                    .merkleize(&db, Some(Sha256::hash(&[&[i * 11]])), floor)
                    .await
                    .unwrap();
                db = db.apply_batch(batch).await.unwrap().sync().await.unwrap();
                targets.push(<FixedDb as ManagedDb<_>>::sync_target(&db));
            }

            assert_ne!(targets[0], targets[1]);

            // Pruning at the second target retains it but excludes the first.
            let db = <FixedDb as ManagedDb<_>>::prune(db, &targets[1])
                .await
                .unwrap();
            drop(db);
            let db = <FixedDb as ManagedDb<_>>::init(
                context.child("cap"),
                config.clone(),
                Some(targets[1].clone()),
            )
            .await
            .unwrap();
            assert_eq!(<FixedDb as ManagedDb<_>>::sync_target(&db), targets[1]);
            drop(db);
            assert!(matches!(
                <FixedDb as ManagedDb<_>>::init(
                    context.child("pruned_cap"),
                    config,
                    Some(targets[0].clone())
                )
                .await,
                Err(InitError::Database(Error::HistoricalFloorPruned(_)))
            ));
        });
    }
}