use crate::{
    dkg::{
        ParticipantsProvider, Registrar, ReshareBlock, SecretStore,
        fence::Fence,
        network::{
            AddressableManager, Addresses, Directory, Manager as DkgManager, MissingAddress,
        },
        orchestrator, probe as dkg_probe,
        reshare::{self, Input as ReshareInput},
        state_sync::{Config as StateSyncConfig, Plan as StateSyncPlan, StateSync},
        tests::{
            max_supported_mode,
            mocks::{FilteredReceiver, MemorySecretStore},
        },
        types::*,
    },
    simulate::{
        engine::{EngineDefinition, InitContext},
        processed::ProcessedHeight,
        reporter::MonitorReporter,
    },
    stateful::{
        Application, Config as StatefulConfig, Input, Proposed, Stateful as StatefulActor,
        SyncPlan,
        db::{
            DatabaseSet, Merkleized as _, Shared, SyncEngineConfig, Unmerkleized as _,
            p2p as qmdb_resolver,
        },
    },
};
use commonware_broadcast::buffered;
use commonware_codec::{
    Encode, EncodeSize, Error as CodecError, RangeCfg, Read, ReadExt as _, Write,
};
use commonware_consensus::{
    Block as ConsensusBlock, CertifiableBlock, Heightable, Reporters,
    marshal::{
        self,
        ancestry::Ancestry,
        core::{Actor as MarshalActor, CommitmentFallback, Mailbox as MarshalMailbox},
        resolver::p2p as marshal_resolver,
        standard::{Deferred, Standard},
    },
    simplex::{
        self,
        config::{ForwardPolicy, SkipPolicy},
        elector::RoundRobin,
        types::{Context, Finalization},
    },
    types::{Epoch, Epocher as _, FixedEpocher, Height, Round, View, ViewDelta},
};
use commonware_cryptography::{
    Digest as _, Digestible, Hasher, Sha256, Signer as _,
    bls12381::{
        dkg::feldman_desmedt::{Reveal, deal},
        primitives::{group::Share, sharing::Mode, variant::MinPk},
    },
    certificate::{ConstantProvider, Provider as CertificateProvider, Scoped},
    ed25519,
    sha256::{self, Digest as Sha256Digest},
};
use commonware_formatting::hex;
use commonware_math::algebra::Random;
use commonware_p2p::{Address, Provider, TrackedPeers, simulated};
use commonware_parallel::Sequential;
use commonware_runtime::{
    Buf, BufMut, BufferPooler, Clock, Handle, Metrics, Quota, Spawner, Storage, Supervisor as _,
    buffer::paged::CacheRef, deterministic::Context as DeterministicContext,
};
use commonware_storage::{
    archive::prunable,
    journal::contiguous::fixed::Config as FixedLogConfig,
    mmr::{self, Location, full::Config as MmrJournalConfig},
    qmdb::{
        any::{FixedConfig, unordered::fixed},
        sync::Target,
    },
    translator::TwoCap,
};
use commonware_utils::{
    N3f1, NZDuration, NZU16, NZU32, NZU64, NZUsize, TestRng, non_empty_range,
    ordered::{Map, Set},
    range::NonEmptyRange,
    sequence::Unit,
    sync::Mutex,
    test_rng,
};
use rand::Rng;
use std::{
    collections::{BTreeMap, HashMap, HashSet, btree_map::Entry},
    marker::PhantomData,
    net::{IpAddr, Ipv4Addr, SocketAddr},
    num::{NonZeroU16, NonZeroU32, NonZeroU64, NonZeroUsize},
    sync::Arc,
    time::Duration,
};

type Qmdb<E> =
    fixed::Db<mmr::Family, E, sha256::Digest, sha256::Digest, Sha256, TwoCap, Sequential>;
type Database<E> = Shared<Qmdb<E>>;
type Scheme = simplex::scheme::bls12381_threshold::vrf::Scheme<ed25519::PublicKey, MinPk>;
type MarshalVariant = Standard<Block>;
type Marshal = MarshalMailbox<Scheme, MarshalVariant>;
type PublicKey = ed25519::PublicKey;
type DiscoveryManager = simulated::Manager<PublicKey, DeterministicContext>;
type LookupManager = AddressableManager<simulated::SocketManager<PublicKey, DeterministicContext>>;

pub(super) const EPOCH_LENGTH: NonZeroU64 = NZU64!(32);
const NAMESPACE: &[u8] = b"_COMMONWARE_GLUE_DKG_RESHARE_E2E";
const PAGE_SIZE: NonZeroU16 = NZU16!(1024);
const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(10);
const IO_BUFFER_SIZE: NonZeroUsize = NZUsize!(2048);
const TEST_QUOTA: Quota = Quota::per_second(NZU32!(1_000_000));
const MAX_PARTICIPANTS: NonZeroU32 = NZU32!(16);

const VOTE_CHANNEL: u64 = 0;
const CERTIFICATE_CHANNEL: u64 = 1;
const RESOLVER_CHANNEL: u64 = 2;
const BACKFILL_CHANNEL: u64 = 3;
const BROADCAST_CHANNEL: u64 = 4;
const QMDB_CHANNEL: u64 = 5;
const DKG_CHANNEL: u64 = 6;
const DKG_PROBE_CHANNEL: u64 = 7;

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum Network {
    Discovery,
    Lookup,
}

impl Network {
    fn directory(
        self,
        peers: &Set<PublicKey>,
        addresses: &Map<PublicKey, Address>,
    ) -> TestDirectory {
        match self {
            Self::Discovery => TestDirectory(None),
            Self::Lookup => TestDirectory(Some(
                peers
                    .iter()
                    .map(|peer| {
                        let address = addresses
                            .get_value(peer)
                            .expect("participant must have an address")
                            .clone();
                        (peer.clone(), address)
                    })
                    .collect(),
            )),
        }
    }

    fn manager(self, oracle: &simulated::Oracle<PublicKey, DeterministicContext>) -> TestManager {
        match self {
            Self::Discovery => TestManager::Discovery(oracle.manager()),
            Self::Lookup => TestManager::Lookup(AddressableManager::new(oracle.socket_manager())),
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) struct TestDirectory(Option<Addresses<PublicKey>>);

impl Write for TestDirectory {
    fn write(&self, buf: &mut impl BufMut) {
        self.0.write(buf);
    }
}

impl EncodeSize for TestDirectory {
    fn encode_size(&self) -> usize {
        self.0.encode_size()
    }
}

impl Read for TestDirectory {
    type Cfg = RangeCfg<usize>;

    fn read_cfg(buf: &mut impl Buf, cfg: &Self::Cfg) -> Result<Self, CodecError> {
        Option::<Addresses<PublicKey>>::read_cfg(buf, cfg).map(Self)
    }
}

impl Directory<PublicKey> for TestDirectory {
    fn codec_config(peers: &Set<PublicKey>) -> Self::Cfg {
        RangeCfg::exact(peers.len())
    }

    fn matches(&self, peers: &Set<PublicKey>) -> bool {
        self.0
            .as_ref()
            .is_none_or(|addresses| addresses.matches(peers))
    }
}

#[derive(Clone, Debug)]
enum TestManager {
    Discovery(DiscoveryManager),
    Lookup(LookupManager),
}

impl Provider for TestManager {
    type PublicKey = PublicKey;

    async fn peer_set(&mut self, id: u64) -> Option<TrackedPeers<Self::PublicKey>> {
        match self {
            Self::Discovery(manager) => manager.peer_set(id).await,
            Self::Lookup(manager) => manager.peer_set(id).await,
        }
    }

    async fn subscribe(&mut self) -> commonware_p2p::PeerSetSubscription<Self::PublicKey> {
        match self {
            Self::Discovery(manager) => manager.subscribe().await,
            Self::Lookup(manager) => manager.subscribe().await,
        }
    }
}

impl DkgManager for TestManager {
    type Directory = TestDirectory;
    type Error = MissingAddress<PublicKey>;

    fn track(
        &mut self,
        epoch: Epoch,
        peers: TrackedPeers<Self::PublicKey>,
        directory: &Self::Directory,
    ) -> Result<(), Self::Error> {
        match (self, &directory.0) {
            (Self::Discovery(manager), None) => {
                DkgManager::track(manager, epoch, peers, &Unit).map_err(|error| match error {})
            }
            (Self::Lookup(manager), Some(addresses)) => {
                DkgManager::track(manager, epoch, peers, addresses)
            }
            _ => panic!("network and directory must use the same transport"),
        }
    }
}

#[derive(Clone, PartialEq, Eq)]
pub(super) struct Block {
    context: Context<sha256::Digest, ed25519::PublicKey>,
    parent: sha256::Digest,
    height: Height,
    state_root: sha256::Digest,
    range: NonEmptyRange<Location>,
    payload: Option<Payload<MinPk, ed25519::PrivateKey, TestDirectory>>,
}

impl Write for Block {
    fn write(&self, buf: &mut impl BufMut) {
        self.context.write(buf);
        self.parent.write(buf);
        self.height.write(buf);
        self.state_root.write(buf);
        self.range.write(buf);
        self.payload.write(buf);
    }
}

impl EncodeSize for Block {
    fn encode_size(&self) -> usize {
        self.context.encode_size()
            + self.parent.encode_size()
            + self.height.encode_size()
            + self.state_root.encode_size()
            + self.range.encode_size()
            + self.payload.encode_size()
    }
}

impl Read for Block {
    type Cfg = ();

    fn read_cfg(buf: &mut impl Buf, _: &Self::Cfg) -> Result<Self, CodecError> {
        Ok(Self {
            context: Context::read(buf)?,
            parent: sha256::Digest::read(buf)?,
            height: Height::read(buf)?,
            state_root: sha256::Digest::read(buf)?,
            range: NonEmptyRange::read(buf)?,
            payload: Option::<Payload<MinPk, ed25519::PrivateKey, TestDirectory>>::read_cfg(
                buf,
                &(MAX_PARTICIPANTS, max_supported_mode()),
            )?,
        })
    }
}

impl Digestible for Block {
    type Digest = sha256::Digest;

    fn digest(&self) -> sha256::Digest {
        Sha256::hash(&[&self.encode()])
    }
}

impl Heightable for Block {
    fn height(&self) -> Height {
        self.height
    }
}

impl ConsensusBlock for Block {
    fn parent(&self) -> sha256::Digest {
        self.parent
    }
}

impl CertifiableBlock for Block {
    type Context = Context<sha256::Digest, ed25519::PublicKey>;

    fn context(&self) -> Self::Context {
        self.context.clone()
    }
}

impl ReshareBlock for Block {
    type Variant = MinPk;
    type Signer = ed25519::PrivateKey;
    type Directory = TestDirectory;

    fn payload(&self) -> Option<Payload<Self::Variant, Self::Signer, Self::Directory>> {
        self.payload.clone()
    }
}

impl Block {
    fn genesis(
        leader: ed25519::PublicKey,
        info: EpochInfo<MinPk, ed25519::PublicKey, TestDirectory>,
    ) -> Self {
        Self {
            context: Context {
                round: Round::new(Epoch::zero(), View::zero()),
                leader,
                parent: (View::zero(), sha256::Digest::EMPTY),
            },
            parent: sha256::Digest::EMPTY,
            height: Height::zero(),
            state_root: empty_db_root(),
            range: non_empty_range!(Location::new(0), Location::new(1)),
            payload: Some(Payload::EpochInfo(info)),
        }
    }
}

#[derive(Clone)]
struct App {
    genesis: Block,
    processed: Arc<Mutex<BTreeMap<ed25519::PublicKey, u64>>>,
    public_key: ed25519::PublicKey,
    // Participant and processed height at which its app parks until crashed
    // (cleared on restart), so a processed-height crash window cannot be
    // raced past.
    hold: Arc<Mutex<Option<(ed25519::PublicKey, u64)>>>,
}

impl App {
    async fn execute<E: Rng + Spawner + Metrics + Clock + Storage + BufferPooler>(
        height: Height,
        mut batches: <Database<E> as DatabaseSet<E>>::Unmerkleized,
    ) -> <Database<E> as DatabaseSet<E>>::Merkleized {
        let key = Sha256::hash(&[b"height"]);
        batches = batches.write(key, Some(u64_to_digest(height.get())));
        batches.merkleize().await.unwrap()
    }
}

impl<E: Rng + Spawner + Metrics + Clock + Storage + BufferPooler> Application<E> for App {
    type SigningScheme = Scheme;
    type Context = Context<sha256::Digest, ed25519::PublicKey>;
    type Block = Block;
    type Databases = Database<E>;
    type Captured = ();
    type Provider = ();
    type Input = ReshareInput<(), MinPk, ed25519::PrivateKey, TestDirectory>;

    async fn genesis(&mut self) -> Self::Block {
        self.genesis.clone()
    }

    async fn propose(
        &mut self,
        context: (E, Self::Context),
        ancestry: impl Ancestry<Self::Block>,
        batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
        input: Input<Self::Input, Self::Provider>,
    ) -> Option<Proposed<Self, E>> {
        let parent = ancestry.peek()?.clone();
        let height = Height::new(parent.height().get() + 1);
        // The reshare::Application wrapper selected and fetched the payload.
        let payload = input.upstream.payload;
        let merkleized = Self::execute(height, batches).await;
        let bounds = merkleized.bounds();
        let block = Block {
            context: context.1,
            parent: parent.digest(),
            height,
            state_root: merkleized.root(),
            range: non_empty_range!(bounds.inactivity_floor, bounds.tip.size),
            payload,
        };
        Some(Proposed { block, merkleized })
    }

    async fn verify(
        &mut self,
        _context: (E, Self::Context),
        ancestry: impl Ancestry<Self::Block>,
        batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
    ) -> Option<<Self::Databases as DatabaseSet<E>>::Merkleized> {
        // Reshare final-block payload validation is enforced by the surrounding
        // reshare::Application wrapper; this inner app only executes state.
        let tip = ancestry.peek()?.clone();
        let merkleized = Self::execute(tip.height(), batches).await;
        Some(merkleized)
    }

    async fn apply(
        &mut self,
        _context: (E, Self::Context),
        block: &Self::Block,
        batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
    ) -> Option<<Self::Databases as DatabaseSet<E>>::Merkleized> {
        Some(Self::execute(block.height(), batches).await)
    }

    async fn capture(
        &mut self,
        _context: (E, Self::Context),
        _block: &Self::Block,
        _batches: &<Self::Databases as DatabaseSet<E>>::Merkleized,
        _readers: <Self::Databases as DatabaseSet<E>>::Readers,
    ) {
    }

    async fn finalized(
        &mut self,
        context: (E, Self::Context),
        block: &Self::Block,
        _captured: Self::Captured,
        _readers: <Self::Databases as DatabaseSet<E>>::Readers,
    ) {
        self.processed
            .lock()
            .insert(self.public_key.clone(), block.height().get());
        // Park once the held height is reached so the simulator's
        // processed-height crash trigger observes it deterministically.
        while self.hold.lock().as_ref().is_some_and(|(held, height)| {
            *held == self.public_key && block.height().get() >= *height
        }) {
            context.0.sleep(Duration::from_millis(25)).await;
        }
    }

    fn sync_targets(block: &Self::Block) -> <Self::Databases as DatabaseSet<E>>::SyncTargets {
        Target::new(block.state_root, block.range.clone())
    }
}

#[derive(Clone)]
struct DynamicProvider {
    schemes: Arc<Mutex<HashMap<Epoch, Arc<Scheme>>>>,
}

impl DynamicProvider {
    pub(super) fn new() -> Self {
        Self {
            schemes: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    fn register(&self, epoch: Epoch, scheme: Scheme) {
        self.schemes.lock().insert(epoch, Arc::new(scheme));
    }
}

impl CertificateProvider for DynamicProvider {
    type Scope = Epoch;
    type Scheme = Scheme;

    fn scoped(&self, scope: Self::Scope) -> Option<Scoped<Self::Scheme>> {
        self.schemes.lock().get(&scope).cloned().map(Scoped::scheme)
    }

    fn scheme(&self, scope: Self::Scope) -> Option<Arc<Self::Scheme>> {
        self.schemes.lock().get(&scope).cloned()
    }
}

#[derive(Clone)]
struct TestRegistrar {
    provider: DynamicProvider,
    events: Arc<Mutex<BTreeMap<ed25519::PublicKey, Vec<Registration>>>>,
    public_key: ed25519::PublicKey,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum RegistrationRole {
    Signer,
    Verifier,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) struct Registration {
    pub(super) epoch: Epoch,
    pub(super) role: RegistrationRole,
}

impl Registrar for TestRegistrar {
    type Variant = MinPk;
    type PublicKey = ed25519::PublicKey;

    async fn register(&self, epoch: Epoch, info: SchemeInfo<Self::Variant, Self::PublicKey>) {
        let (scheme, role) = match info {
            SchemeInfo::Verifier {
                participants,
                sharing,
            } => (
                Scheme::verifier(NAMESPACE, participants, sharing),
                RegistrationRole::Verifier,
            ),
            SchemeInfo::Signer {
                participants,
                sharing,
                share,
            } => (
                Scheme::signer(NAMESPACE, participants, sharing, share)
                    .expect("share must match participant set"),
                RegistrationRole::Signer,
            ),
        };
        self.provider.register(epoch, scheme);
        self.events
            .lock()
            .entry(self.public_key.clone())
            .or_default()
            .push(Registration { epoch, role });
    }
}

#[derive(Clone)]
struct ScheduleProvider {
    pub(super) schedule: Arc<CommitteeSchedule>,
    network: Network,
    addresses: Arc<Map<PublicKey, Address>>,
}

impl ParticipantsProvider for ScheduleProvider {
    type PublicKey = PublicKey;
    type Directory = TestDirectory;

    async fn participants(&mut self, epoch: Epoch) -> Set<Self::PublicKey> {
        self.schedule.players(epoch)
    }

    async fn directory(&mut self, _: Epoch, peers: Set<Self::PublicKey>) -> Self::Directory {
        self.network.directory(&peers, &self.addresses)
    }
}

#[derive(Clone)]
pub(super) struct CommitteeSchedule {
    participants: Vec<ed25519::PublicKey>,
    committee_sizes: Vec<usize>,
}

impl CommitteeSchedule {
    pub(super) fn players(&self, epoch: Epoch) -> Set<ed25519::PublicKey> {
        let offset = epoch.get() as usize % self.participants.len();
        let committee_size =
            self.committee_sizes[epoch.get() as usize % self.committee_sizes.len()];
        let players = (0..committee_size)
            .map(|i| self.participants[(offset + i) % self.participants.len()].clone());
        Set::from_iter_dedup(players)
    }
}

#[derive(Clone)]
pub(super) struct ReshareEngine {
    network: Network,
    signers: Vec<ed25519::PrivateKey>,
    pub(super) participants: Vec<ed25519::PublicKey>,
    addresses: Arc<Map<PublicKey, Address>>,
    pub(super) schedule: Arc<CommitteeSchedule>,
    initial: Arc<InitialState>,
    sharing_mode: Mode,
    stores: Arc<Mutex<BTreeMap<ed25519::PublicKey, MemorySecretStore>>>,
    pub(super) registrations: Arc<Mutex<BTreeMap<ed25519::PublicKey, Vec<Registration>>>>,
    pub(super) state_syncs: Arc<Mutex<BTreeMap<ed25519::PublicKey, u64>>>,
    pub(super) state_sync_starts: Arc<Mutex<BTreeMap<ed25519::PublicKey, u64>>>,
    state_sync_floor: Option<Height>,
    processed_hold: Arc<Mutex<Option<(ed25519::PublicKey, u64)>>>,
    epoch_cross_during_sync: bool,
    processed: Arc<Mutex<BTreeMap<ed25519::PublicKey, u64>>>,
    marshals: Arc<Mutex<BTreeMap<ed25519::PublicKey, Marshal>>>,
    failures: Arc<HashSet<u64>>,
}

pub(super) struct ValidatorEngine {
    context: DeterministicContext,
    handles: [Handle<()>; 6],
}

#[derive(Clone)]
struct InitialState {
    info: EpochInfo<MinPk, ed25519::PublicKey, TestDirectory>,
    shares: Map<ed25519::PublicKey, Share>,
}

impl InitialState {
    async fn register_epoch_zero(
        &self,
        provider: &DynamicProvider,
        store: &MemorySecretStore,
    ) -> RegistrationRole {
        let mut store = store.clone();
        let participants = self.info.output.players();
        let sharing = self.info.output.public();
        store.get_share(Epoch::zero()).await.map_or_else(
            || {
                provider.register(
                    Epoch::zero(),
                    Scheme::verifier(NAMESPACE, participants.clone(), sharing.clone()),
                );
                RegistrationRole::Verifier
            },
            |share| {
                provider.register(
                    Epoch::zero(),
                    Scheme::signer(NAMESPACE, participants.clone(), sharing.clone(), share)
                        .expect("initial signer share"),
                );
                RegistrationRole::Signer
            },
        )
    }
}

impl ReshareEngine {
    pub(super) fn new(network: Network) -> Self {
        Self::with_committee(network, 5, 4)
    }

    pub(super) fn with_committee(network: Network, total: u32, committee_size: usize) -> Self {
        Self::with_committees(network, total, vec![committee_size])
    }

    pub(super) fn with_committees(
        network: Network,
        total: u32,
        committee_sizes: Vec<usize>,
    ) -> Self {
        assert!(!committee_sizes.is_empty());
        for committee_size in &committee_sizes {
            assert!(*committee_size > 0);
            assert!(*committee_size <= total as usize);
        }
        let mut rng = test_rng();
        let signers = (0..total)
            .map(|_| ed25519::PrivateKey::random(&mut rng))
            .collect::<Vec<_>>();
        let participants = signers.iter().map(|s| s.public_key()).collect::<Vec<_>>();
        let addresses = Arc::new(Map::from_iter_dedup(participants.iter().enumerate().map(
            |(index, participant)| {
                let socket =
                    SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 10_000 + index as u16);
                (participant.clone(), Address::Symmetric(socket))
            },
        )));
        let schedule = Arc::new(CommitteeSchedule {
            participants: participants.clone(),
            committee_sizes,
        });
        let players = schedule.players(Epoch::zero());
        let next_players = schedule.players(Epoch::new(1));
        let (output, shares) =
            deal::<MinPk, _, N3f1>(TestRng::new(10), Mode::NonZeroCounter, players.clone())
                .expect("trusted initial deal");
        let initial_peers = Set::from_iter_dedup(
            output
                .players()
                .iter()
                .chain(players.iter())
                .chain(next_players.iter())
                .cloned(),
        );
        let info = EpochInfo {
            outcome: EpochOutcome::Success,
            epoch: Epoch::zero(),
            output,
            players,
            next_players,
            directory: network.directory(&initial_peers, &addresses),
        };
        Self {
            network,
            signers,
            participants,
            addresses,
            schedule,
            initial: Arc::new(InitialState { info, shares }),
            sharing_mode: Mode::NonZeroCounter,
            stores: Arc::new(Mutex::new(BTreeMap::new())),
            registrations: Arc::new(Mutex::new(BTreeMap::new())),
            state_syncs: Arc::new(Mutex::new(BTreeMap::new())),
            state_sync_starts: Arc::new(Mutex::new(BTreeMap::new())),
            state_sync_floor: None,
            processed_hold: Arc::new(Mutex::new(None)),
            epoch_cross_during_sync: false,
            processed: Arc::new(Mutex::new(BTreeMap::new())),
            marshals: Arc::new(Mutex::new(BTreeMap::new())),
            failures: Arc::new(HashSet::new()),
        }
    }

    pub(super) fn with_failures(mut self, failures: impl IntoIterator<Item = u64>) -> Self {
        self.failures = Arc::new(failures.into_iter().collect());
        self
    }

    pub(super) const fn with_sharing_mode(mut self, sharing_mode: Mode) -> Self {
        self.sharing_mode = sharing_mode;
        self
    }

    pub(super) const fn with_state_sync_floor(mut self, height: Height) -> Self {
        self.state_sync_floor = Some(height);
        self
    }

    /// Parks `participant`'s application once it processes `height`, until it
    /// is crashed and restarted. Combine with a processed-height crash so the
    /// crash window is entered deterministically rather than sampled.
    pub(super) fn with_processed_hold(
        self,
        participant: ed25519::PublicKey,
        height: Height,
    ) -> Self {
        *self.processed_hold.lock() = Some((participant, height.get()));
        self
    }

    /// Hold state-sync bootstrap after the probe artifact resolves until
    /// every member of the sampled committee has processed into the next
    /// epoch, so the network crosses an epoch boundary while the node is
    /// still syncing to its floor.
    pub(super) const fn with_epoch_cross_during_sync(mut self) -> Self {
        self.epoch_cross_during_sync = true;
        self
    }

    pub(super) const fn state_sync_floor(&self) -> Option<Height> {
        self.state_sync_floor
    }

    fn store(&self, public_key: &ed25519::PublicKey) -> MemorySecretStore {
        let mut stores = self.stores.lock();
        match stores.entry(public_key.clone()) {
            Entry::Occupied(entry) => entry.get().clone(),
            Entry::Vacant(entry) => {
                let store = MemorySecretStore::default();
                if let Some(share) = self.initial.shares.get_value(public_key).cloned() {
                    store.seed_share(Epoch::zero(), share);
                }
                entry.insert(store.clone());
                store
            }
        }
    }
}

impl EngineDefinition for ReshareEngine {
    type PublicKey = ed25519::PublicKey;
    type Engine = ValidatorEngine;
    type State = ValidatorState;

    fn participants(&self) -> Vec<Self::PublicKey> {
        self.participants.clone()
    }

    fn channels(&self) -> Vec<(u64, Quota)> {
        vec![
            (VOTE_CHANNEL, TEST_QUOTA),
            (CERTIFICATE_CHANNEL, TEST_QUOTA),
            (RESOLVER_CHANNEL, TEST_QUOTA),
            (BACKFILL_CHANNEL, TEST_QUOTA),
            (BROADCAST_CHANNEL, TEST_QUOTA),
            (QMDB_CHANNEL, TEST_QUOTA),
            (DKG_CHANNEL, TEST_QUOTA),
            (DKG_PROBE_CHANNEL, TEST_QUOTA),
        ]
    }

    async fn init(&self, ctx: InitContext<'_, Self::PublicKey>) -> (Self::Engine, Self::State) {
        let InitContext {
            context,
            index,
            delayed,
            public_key,
            oracle,
            channels,
            participants: _,
            monitor,
        } = ctx;

        let signer = self.signers[index].clone();
        let partition_prefix = format!("reshare-e2e-{index}");
        let page_cache = CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE);
        // A node restarting means its configured processed hold has served its
        // purpose: release it so the restarted node can progress.
        {
            let mut hold = self.processed_hold.lock();
            if hold.as_ref().is_some_and(|(held, _)| held == public_key)
                && self.stores.lock().contains_key(public_key)
            {
                hold.take();
            }
        }

        let mut channels = channels.into_iter();
        let vote_network = channels.next().unwrap();
        let certificate_network = channels.next().unwrap();
        let resolver_network = channels.next().unwrap();
        let backfill_network = channels.next().unwrap();
        let broadcast_network = channels.next().unwrap();
        let qmdb_network = channels.next().unwrap();
        let dkg_network = channels.next().unwrap();
        let probe_boundary_network = channels.next().unwrap();

        let provider = DynamicProvider::new();
        let store = self.store(public_key);
        self.initial.register_epoch_zero(&provider, &store).await;
        let dkg_manager = self.network.manager(oracle);

        let resolver = marshal_resolver::init(
            context.child("marshal_resolver"),
            marshal_resolver::Config {
                public_key: public_key.clone(),
                peer_provider: oracle.manager(),
                blocker: oracle.control(public_key.clone()),
                mailbox_size: NZUsize!(100),
                timeout: Duration::from_secs(2),
                fetch_retry_timeout: Duration::from_millis(100),
                priority_requests: false,
                priority_responses: false,
            },
            backfill_network,
        );

        let broadcast_config = buffered::Config {
            public_key: public_key.clone(),
            mailbox_size: NZUsize!(100),
            deque_size: 10,
            priority: false,
            codec_config: (),
            peer_provider: oracle.manager(),
        };
        let (broadcast_engine, buffer) =
            buffered::Engine::new(context.child("broadcast"), broadcast_config);
        broadcast_engine.start(broadcast_network);

        let finalizations_by_height = prunable::Archive::init(
            context.child("finalizations_by_height"),
            archive_config(&partition_prefix, "finalizations", page_cache.clone(), ()),
        )
        .await
        .expect("finalizations archive");
        let finalized_blocks = prunable::Archive::init(
            context.child("finalized_blocks"),
            archive_config(&partition_prefix, "blocks", page_cache.clone(), ()),
        )
        .await
        .expect("blocks archive");

        let genesis = Block::genesis(self.participants[0].clone(), self.initial.info.clone());
        let (probe_actor, probe_mailbox) = dkg_probe::Actor::new(dkg_probe::Config {
            context: context.child("dkg_probe"),
            manager: dkg_manager.clone(),
            bootstrap: dkg_probe::Bootstrap {
                epoch: Epoch::zero(),
                participants: self.initial.info.participants(),
                directory: self.initial.info.directory.clone(),
            },
            verifier: Scheme::certificate_verifier(
                NAMESPACE,
                *self.initial.info.output.public().public(),
            ),
            genesis: self.initial.info.clone(),
            strategy: Sequential,
            blocker: oracle.control(public_key.clone()),
            blocks_per_epoch: EPOCH_LENGTH,
            retry_timeout: NZDuration!(Duration::from_millis(500)),
            mailbox_size: NZUsize!(100),
            block_codec_config: (),
        });
        let probe_handle = probe_actor.start(probe_boundary_network);

        let stateful_startup_context = context.child("stateful_startup");
        let mut plan = SyncPlan::init(&stateful_startup_context, partition_prefix.clone()).await;
        let should_state_sync = plan.should_state_sync(delayed);
        if should_state_sync {
            *self
                .state_sync_starts
                .lock()
                .entry(public_key.clone())
                .or_default() += 1;
        }
        let probe_artifact = if should_state_sync {
            let artifact = probe_mailbox.subscribe().await.expect("probe stopped");
            provider.register(
                artifact.info.epoch,
                Scheme::verifier(
                    NAMESPACE,
                    artifact.info.output.players().clone(),
                    artifact.info.output.public().clone(),
                ),
            );
            Some(artifact)
        } else {
            None
        };
        // Optionally hold bootstrap after the probe artifact resolves until
        // the whole sampled committee has processed into the next epoch, so
        // the network crosses an epoch boundary while the node is still
        // syncing to its (already fixed) floor.
        if self.epoch_cross_during_sync && should_state_sync {
            let artifact = probe_artifact
                .as_ref()
                .expect("epoch cross hold requires state-sync bootstrap");
            let committee = artifact.info.output.players().clone();
            assert!(
                committee.position(public_key).is_none(),
                "delayed node must not be a sampled-committee member for the epoch-cross hold"
            );
            let target = FixedEpocher::new(EPOCH_LENGTH)
                .first(artifact.info.epoch.next())
                .expect("next epoch must be supported")
                .get();
            loop {
                let crossed = {
                    let processed = self.processed.lock();
                    committee
                        .iter()
                        .all(|member| processed.get(member).is_some_and(|height| *height > target))
                };
                if crossed {
                    break;
                }
                context.sleep(Duration::from_millis(50)).await;
            }
        }
        if should_state_sync {
            let finalization = match self.state_sync_floor {
                Some(height) => {
                    let source = self
                        .marshals
                        .lock()
                        .values()
                        .next()
                        .cloned()
                        .expect("state-sync floor source must be available");
                    source
                        .get_finalization(height)
                        .await
                        .expect("configured state-sync floor must be finalized")
                }
                None => probe_artifact
                    .as_ref()
                    .expect("state-sync startup must have a probe artifact")
                    .floor
                    .clone(),
            };
            plan = plan.with_floor(finalization);
        }
        let (marshal_actor, marshal, floor) = MarshalActor::init(
            context.child("marshal"),
            finalizations_by_height,
            finalized_blocks,
            marshal::Config {
                provider: ConstantProvider::<_, Epoch>::new(Scheme::certificate_verifier(
                    NAMESPACE,
                    *self.initial.info.output.public().public(),
                )),
                epocher: FixedEpocher::new(EPOCH_LENGTH),
                start: plan.marshal_start(genesis.clone()),
                partition_prefix: partition_prefix.clone(),
                mailbox_size: NZUsize!(100),
                view_retention: ViewDelta::new(10),
                prunable_items_per_section: NZU64!(10),
                page_cache: page_cache.clone(),
                replay_buffer: IO_BUFFER_SIZE,
                key_write_buffer: IO_BUFFER_SIZE,
                value_write_buffer: IO_BUFFER_SIZE,
                block_codec_config: (),
                max_repair: NZUsize!(10),
                max_pending_acks: NZUsize!(1),
                strategy: Sequential,
            },
        )
        .await;
        self.marshals
            .lock()
            .insert(public_key.clone(), marshal.clone());

        let db_config = FixedConfig {
            merkle_config: MmrJournalConfig {
                journal_partition: format!("{partition_prefix}-qmdb-mmr-journal"),
                metadata_partition: format!("{partition_prefix}-qmdb-mmr-metadata"),
                items_per_blob: NZU64!(11),
                write_buffer: IO_BUFFER_SIZE,
                replay_buffer: IO_BUFFER_SIZE,
                strategy: Sequential,
                page_cache: page_cache.clone(),
            },
            journal_config: FixedLogConfig {
                partition: format!("{partition_prefix}-qmdb-log-journal"),
                items_per_blob: NZU64!(7),
                page_cache: page_cache.clone(),
                write_buffer: IO_BUFFER_SIZE,
                replay_buffer: IO_BUFFER_SIZE,
            },
            translator: TwoCap,
            init_cache_size: Some(NZUsize!(1024)),
            init_buffer: NZUsize!(1 << 21),
            init_concurrency: (),
        };

        let (qmdb_resolver_actor, qmdb_sync_resolver) = qmdb_resolver::Actor::new(
            context.child("qmdb_resolver"),
            qmdb_resolver::Config {
                peer_provider: oracle.manager(),
                blocker: oracle.control(public_key.clone()),
                database: None,
                mailbox_size: NZUsize!(100),
                me: Some(public_key.clone()),
                timeout: Duration::from_secs(2),
                fetch_retry_timeout: Duration::from_millis(100),
                max_serve_ops: NZU64!(16),
                priority_requests: false,
                priority_responses: false,
            },
        );
        let qmdb_handle = qmdb_resolver_actor.start(qmdb_network);

        let fence_epoch = probe_artifact
            .as_ref()
            .map_or_else(Epoch::zero, |artifact| artifact.info.epoch);
        let state_sync = probe_artifact.map(|artifact| {
            let floor = plan
                .floor()
                .cloned()
                .expect("state-sync startup must have a probe floor");
            StateSync {
                info: artifact.info,
                floor,
            }
        });
        let registrar = TestRegistrar {
            provider: provider.clone(),
            events: self.registrations.clone(),
            public_key: public_key.clone(),
        };
        let (fence, gate) = Fence::new(fence_epoch);
        let sync_floor: Option<Finalization<Scheme, sha256::Digest>> = state_sync
            .as_ref()
            .map(|state_sync| state_sync.floor.clone());
        let state_sync = StateSyncPlan::init(
            context.child("dkg_state_sync_plan"),
            StateSyncConfig {
                partition_prefix: partition_prefix.clone(),
                max_participants: MAX_PARTICIPANTS,
                max_supported_mode: max_supported_mode(),
            },
            state_sync,
        )
        .await;
        let (reshare_actor, reshare_mailbox) = reshare::Actor::new(
            context.child("reshare"),
            reshare::Config {
                signer: signer.clone(),
                manager: dkg_manager.clone(),
                blocker: oracle.control(public_key.clone()),
                participants_provider: ScheduleProvider {
                    schedule: self.schedule.clone(),
                    network: self.network,
                    addresses: self.addresses.clone(),
                },
                secret_store: store,
                strategy: Sequential,
                registrar,
                marshal: marshal.clone(),
                state_sync: state_sync.clone(),
                fence,
                namespace: NAMESPACE,
                sharing_mode: self.sharing_mode,
                reveal: Reveal::V1,
                mailbox_size: NZUsize!(100),
                partition_prefix: format!("{partition_prefix}-reshare"),
                max_participants: MAX_PARTICIPANTS,
                blocks_per_epoch: EPOCH_LENGTH,
                batch_verifier: PhantomData::<ed25519::Batch>,
            },
        );
        let dkg_network = (
            dkg_network.0,
            FilteredReceiver::epochs(dkg_network.1, self.failures.clone()),
        );
        let reshare_handle = reshare_actor.start(dkg_network);

        let (stateful_actor, stateful_mailbox) = StatefulActor::init(
            context.child("stateful"),
            StatefulConfig {
                application: App {
                    genesis: genesis.clone(),
                    processed: self.processed.clone(),
                    public_key: public_key.clone(),
                    hold: self.processed_hold.clone(),
                },
                db_config,
                provider: (),
                marshal: (marshal.clone(), floor),
                mailbox_size: NZUsize!(100),
                plan,
                resolvers: qmdb_sync_resolver,
                sync_config: SyncEngineConfig {
                    fetch_batch_size: NZU64!(16),
                    apply_batch_size: NZU64!(64),
                    max_outstanding_requests: 8,
                    update_channel_size: NZUsize!(256),
                    max_retained_roots: 8,
                },
                prune_config: None,
            },
        );

        // The reshare wrapper drives the payload for the stateful application.
        let deferred = Deferred::new(
            context.child("deferred"),
            reshare::Application::new(
                stateful_mailbox.clone(),
                reshare_mailbox.clone(),
                EPOCH_LENGTH,
            ),
            marshal.clone(),
            FixedEpocher::new(EPOCH_LENGTH),
        );

        let (orchestrator_actor, orchestrator_mailbox) = orchestrator::Actor::new(
            context.child("orchestrator"),
            orchestrator::Config {
                oracle: oracle.control(public_key.clone()),
                manager: dkg_manager,
                provider: provider.clone(),
                marshal: marshal.clone(),
                application: deferred,
                strategy: Sequential,
                simplex: orchestrator::SimplexConfig {
                    elector: RoundRobin::<Sha256>::default(),
                    mailbox_size: NZUsize!(3),
                    replay_buffer: IO_BUFFER_SIZE,
                    write_buffer: IO_BUFFER_SIZE,
                    page_cache_page_size: PAGE_SIZE,
                    page_cache_pages: PAGE_CACHE_SIZE,
                    leader_timeout: Duration::from_secs(1),
                    certification_timeout: Duration::from_secs(2),
                    timeout_retry: Duration::from_millis(500),
                    fetch_timeout: Duration::from_secs(2),
                    view_retention: ViewDelta::new(10),
                    skip: SkipPolicy::Enabled {
                        timeout: Duration::from_secs(5),
                        budget: simplex::SkipBudget::Participants,
                    },
                    forward: ForwardPolicy::Disabled,
                    track_historical_votes: false,
                },
                gate,
                state_sync,
                blocks_per_epoch: EPOCH_LENGTH,
                muxer_size: 128,
                mailbox_size: NZUsize!(100),
                partition_prefix: format!("{partition_prefix}-orchestrator"),
            },
        );
        let orchestrator_handle =
            orchestrator_actor.start(vote_network, certificate_network, resolver_network);

        let reporters = Reporters::from((
            stateful_mailbox,
            Reporters::from((orchestrator_mailbox, reshare_mailbox)),
        ));
        let marshal_handle = marshal_actor.start(
            MonitorReporter::new(public_key.clone(), monitor, reporters),
            buffer,
            resolver,
        );
        probe_mailbox.attach(marshal.clone());
        // Record the synced height without blocking init: awaiting the floor
        // block here would stall the simulator's restart arm (and with it the
        // crash monitor) until the block is fetched.
        if let Some(finalization) = sync_floor {
            let marshal = marshal.clone();
            let state_syncs = self.state_syncs.clone();
            let public_key = public_key.clone();
            context
                .child("sync_floor_recorder")
                .spawn(move |_| async move {
                    let Ok(block) = marshal
                        .subscribe_by_commitment(
                            finalization.proposal.payload,
                            CommitmentFallback::Wait,
                        )
                        .await
                    else {
                        return;
                    };
                    state_syncs.lock().insert(public_key, block.height().get());
                });
        }
        let stateful_handle = stateful_actor.start();

        (
            ValidatorEngine {
                context,
                handles: [
                    probe_handle,
                    qmdb_handle,
                    reshare_handle,
                    orchestrator_handle,
                    marshal_handle,
                    stateful_handle,
                ],
            },
            ValidatorState {
                marshal,
                processed: self.processed.clone(),
                registrations: self.registrations.clone(),
                state_syncs: self.state_syncs.clone(),
                public_key: public_key.clone(),
            },
        )
    }

    fn start(engine: Self::Engine) -> Handle<()> {
        let ValidatorEngine { context, handles } = engine;
        context.spawn(move |_| async move {
            Handle::select(handles)
                .await
                .expect("validator actor failed");
        })
    }
}

#[derive(Clone)]
pub(super) struct ValidatorState {
    pub(super) marshal: Marshal,
    processed: Arc<Mutex<BTreeMap<ed25519::PublicKey, u64>>>,
    registrations: Arc<Mutex<BTreeMap<ed25519::PublicKey, Vec<Registration>>>>,
    pub(super) state_syncs: Arc<Mutex<BTreeMap<ed25519::PublicKey, u64>>>,
    public_key: ed25519::PublicKey,
}

impl PartialEq for ValidatorState {
    fn eq(&self, other: &Self) -> bool {
        self.public_key == other.public_key
    }
}

impl ProcessedHeight for ValidatorState {
    async fn processed_height(&self) -> u64 {
        self.processed
            .lock()
            .get(&self.public_key)
            .copied()
            .unwrap_or_default()
    }
}

impl ValidatorState {
    pub(super) fn public_key(&self) -> &ed25519::PublicKey {
        &self.public_key
    }

    pub(super) fn registrations(&self) -> Vec<Registration> {
        self.registrations
            .lock()
            .get(&self.public_key)
            .cloned()
            .unwrap_or_default()
    }

    pub(super) fn state_sync_height(&self) -> Option<u64> {
        self.state_syncs.lock().get(&self.public_key).copied()
    }
}

#[test]
fn restart_after_epoch_zero_pruning_does_not_reseed_share() {
    let engine = ReshareEngine::new(Network::Discovery);
    let public_key = engine
        .initial
        .shares
        .get(0)
        .cloned()
        .expect("epoch-zero player");

    let store = engine.store(&public_key);
    assert!(store.has_share(Epoch::zero()));

    futures::executor::block_on(async {
        let mut pruned = store.clone();
        pruned.prune(Epoch::new(1)).await;
        assert!(pruned.get_share(Epoch::zero()).await.is_none());

        let restarted = engine.store(&public_key);
        let mut restarted_view = restarted.clone();
        assert!(restarted_view.get_share(Epoch::zero()).await.is_none());

        let provider = DynamicProvider::new();
        let role = engine
            .initial
            .register_epoch_zero(&provider, &restarted)
            .await;
        assert_eq!(role, RegistrationRole::Verifier);
    });
}

fn archive_config<C>(
    prefix: &str,
    name: &str,
    page_cache: CacheRef,
    codec_config: C,
) -> prunable::Config<TwoCap, C> {
    prunable::Config {
        translator: TwoCap,
        metadata_partition: format!("{prefix}-{name}-metadata"),
        key_partition: format!("{prefix}-{name}-key"),
        key_page_cache: page_cache,
        value_partition: format!("{prefix}-{name}-value"),
        compression: None,
        codec_config,
        items_per_section: NZU64!(10),
        key_write_buffer: IO_BUFFER_SIZE,
        value_write_buffer: IO_BUFFER_SIZE,
        replay_buffer: IO_BUFFER_SIZE,
    }
}

fn empty_db_root() -> sha256::Digest {
    Sha256Digest::from(hex!(
        "ea6e0567a525372add5e4ef4d0600c18ed47fa5dd041a0ab0d25b60ea8c35978"
    ))
}

fn u64_to_digest(v: u64) -> sha256::Digest {
    let mut bytes = [0u8; 32];
    bytes[..8].copy_from_slice(&v.to_be_bytes());
    sha256::Digest::from(bytes)
}

pub(super) fn final_height(epoch: u64) -> Height {
    FixedEpocher::new(EPOCH_LENGTH)
        .last(Epoch::new(epoch))
        .expect("test epoch should be supported")
}

pub(super) fn height_round(height: Height) -> Round {
    let info = FixedEpocher::new(EPOCH_LENGTH)
        .containing(height)
        .expect("test height should be supported");
    Round::new(info.epoch(), View::new(info.relative().get()))
}