rgb-ops 0.11.1

// RGB ops library for working with smart contracts on Bitcoin & Lightning
//
// SPDX-License-Identifier: Apache-2.0
//
// Written in 2019-2024 by
//     Dr Maxim Orlovsky <orlovsky@lnp-bp.org>
//
// Copyright (C) 2019-2024 LNP/BP Standards Association. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::collections::{btree_map, BTreeMap, BTreeSet, HashMap, HashSet};
use std::convert::Infallible;
use std::error::Error;
use std::fmt::Debug;
use std::num::NonZeroU32;

use amplify::confinement::{Confined, LargeOrdSet};
use nonasync::persistence::{CloneNoPersistence, PersistenceError, PersistenceProvider};
use rgb::bitcoin::{OutPoint as Outpoint, Txid};
use rgb::dbc::{Anchor, Proof};
use rgb::validation::{
    OpoutsDagData, OpoutsDagInfo, ResolveWitness, UnsafeHistoryMap, WitnessOrdProvider,
    WitnessResolverError, WitnessStatus,
};
use rgb::vm::WitnessOrd;
use rgb::{
    validation, AssignmentType, BundleId, ChainNet, ContractId, Genesis, GraphSeal, Identity,
    KnownTransition, OpId, Operation, Opout, OutputSeal, Schema, SchemaId, SecretSeal, Transition,
    TransitionType, TxoSeal, UnrelatedTransition,
};
use strict_types::FieldName;

use super::{
    ContractStateRead, Index, IndexError, IndexInconsistency, IndexProvider, IndexReadProvider,
    IndexWriteProvider, MemIndex, MemStash, MemState, Stash, StashDataError, StashError,
    StashInconsistency, StashProvider, StashReadProvider, StashWriteProvider, State, StateError,
    StateInconsistency, StateProvider, StateReadProvider, StateWriteProvider, StoreTransaction,
};
use crate::containers::{
    Consignment, ConsignmentExt, ContainerVer, Contract, Fascia, Kit, SealWitness, SecretSeals,
    ToWitnessId, Transfer, ValidConsignment, ValidContract, ValidKit, ValidTransfer, WitnessBundle,
};
use crate::contract::{
    AllocatedState, BuilderError, ContractBuilder, ContractData, IssuerWrapper, LinkError,
    LinkableIssuerWrapper, LinkableSchemaWrapper, SchemaWrapper, TransitionBuilder,
};
use crate::info::{ContractInfo, SchemaInfo};
use crate::MergeRevealError;

pub type ContractAssignments = HashMap<OutputSeal, HashMap<Opout, AllocatedState>>;

type SortedBundlesWithDag = (Vec<WitnessBundle>, Option<OpoutsDagData>);

type ConsignmentWithOptDag<const TRANSFER: bool> = (Consignment<TRANSFER>, Option<OpoutsDagData>);

/// Consignment and its operations DAG
pub type ConsignmentWithDag<const TRANSFER: bool> = (Consignment<TRANSFER>, OpoutsDagData);

#[derive(Debug, Display, Error, From)]
#[display(inner)]
pub enum StockError<
    S: StashProvider = MemStash,
    H: StateProvider = MemState,
    P: IndexProvider = MemIndex,
    E: Error = Infallible,
> {
    InvalidInput(E),
    Resolver(String),
    StashRead(<S as StashReadProvider>::Error),
    StashWrite(<S as StashWriteProvider>::Error),
    IndexRead(<P as IndexReadProvider>::Error),
    IndexWrite(<P as IndexWriteProvider>::Error),
    StateRead(<H as StateReadProvider>::Error),
    StateWrite(<H as StateWriteProvider>::Error),

    #[from]
    #[display(doc_comments)]
    /// {0}
    ///
    /// It may happen due to RGB ops library bug, or indicate internal
    /// stash inconsistency and compromised stash data storage.
    StashInconsistency(StashInconsistency),

    #[from]
    #[display(doc_comments)]
    /// state for contract {0} is not known.
    ///
    /// It may happen due to RGB ops library bug, or indicate internal
    /// stash inconsistency and compromised stash data storage.
    StateInconsistency(StateInconsistency),

    #[from]
    #[display(doc_comments)]
    /// {0}
    ///
    /// It may happen due to RGB ops library bug, or indicate internal
    /// stash inconsistency and compromised stash data storage.
    IndexInconsistency(IndexInconsistency),

    #[from]
    StashData(StashDataError),

    /// valid (non-archived) witness is absent in the list of witnesses for a
    /// state transition bundle.
    AbsentValidWitness,

    /// Unable to sort bundles because of data inconsistency.
    BundlesInconsistency,

    /// witness {0} can't be resolved: {1}
    WitnessUnresolved(Txid, WitnessResolverError),

    #[from]
    /// contract link is not valid: {1}
    ContractLinkError(LinkError),
}

impl<S: StashProvider, H: StateProvider, P: IndexProvider, E: Error> From<StashError<S>>
    for StockError<S, H, P, E>
{
    fn from(err: StashError<S>) -> Self {
        match err {
            StashError::ReadProvider(err) => Self::StashRead(err),
            StashError::WriteProvider(err) => Self::StashWrite(err),
            StashError::Data(e) => Self::StashData(e),
            StashError::Inconsistency(e) => Self::StashInconsistency(e),
        }
    }
}

impl<S: StashProvider, H: StateProvider, P: IndexProvider, E: Error> From<StateError<H>>
    for StockError<S, H, P, E>
{
    fn from(err: StateError<H>) -> Self {
        match err {
            StateError::ReadProvider(err) => Self::StateRead(err),
            StateError::WriteProvider(err) => Self::StateWrite(err),
            StateError::Inconsistency(e) => Self::StateInconsistency(e),
            StateError::Resolver(id, e) => Self::WitnessUnresolved(id, e),
            StateError::AbsentValidWitness => Self::AbsentValidWitness,
        }
    }
}
impl<S: StashProvider, H: StateProvider, P: IndexProvider, E: Error> From<IndexError<P>>
    for StockError<S, H, P, E>
{
    fn from(err: IndexError<P>) -> Self {
        match err {
            IndexError::ReadProvider(err) => Self::IndexRead(err),
            IndexError::WriteProvider(err) => Self::IndexWrite(err),
            IndexError::Inconsistency(e) => Self::IndexInconsistency(e),
        }
    }
}

#[derive(Clone, PartialEq, Eq, Debug, Display, Error, From)]
#[display(doc_comments)]
pub enum ConsignError {
    /// unable to construct consignment: too many terminals provided.
    TooManyTerminals,

    /// unable to construct consignment: invalid number of secret seals.
    InvalidSecretSealsNumber,

    /// unable to construct consignment: history size too large, resulting in
    /// too many transitions.
    TooManyBundles,

    #[from]
    #[display(inner)]
    MergeReveal(MergeRevealError),

    #[from]
    #[display(inner)]
    Transition(UnrelatedTransition),

    /// the spent state from transition {1} inside bundle {0} is concealed.
    Concealed(BundleId, OpId),

    /// the requested contract is unrelated to other inputs.
    UnrelatedContract(ContractId),

    /// the transition {1} inside bundle {0} is concealed.
    ConcealedTransition(BundleId, OpId),

    /// the transition {1} inside bundle {0} appears after its child.
    UnorderedTransition(BundleId, OpId),
}

impl<S: StashProvider, H: StateProvider, P: IndexProvider> From<ConsignError>
    for StockError<S, H, P, ConsignError>
{
    fn from(err: ConsignError) -> Self { Self::InvalidInput(err) }
}

impl<S: StashProvider, H: StateProvider, P: IndexProvider> From<MergeRevealError>
    for StockError<S, H, P, ConsignError>
{
    fn from(err: MergeRevealError) -> Self { Self::InvalidInput(err.into()) }
}

impl<S: StashProvider, H: StateProvider, P: IndexProvider> From<UnrelatedTransition>
    for StockError<S, H, P, ConsignError>
{
    fn from(err: UnrelatedTransition) -> Self { Self::InvalidInput(err.into()) }
}

#[derive(Clone, PartialEq, Eq, Debug, Display, Error, From)]
#[display(doc_comments)]
pub enum ComposeError {
    /// no outputs available to store state of type {0}
    NoExtraOrChange(AssignmentType),

    /// the provided PSBT doesn't pay any sats to the RGB beneficiary address.
    NoBeneficiaryOutput,

    /// beneficiary output number is given when secret seal is used.
    BeneficiaryVout,

    /// expired invoice.
    InvoiceExpired,

    /// the invoice contains no contract information.
    NoContract,

    /// the invoice requirements can't be fulfilled using available assets or
    /// smart contract state.
    InsufficientState,

    /// the spent UTXOs contain too many seals which can't fit the state
    /// transition input limit.
    TooManyInputs,

    /// the operation produces too many extra state transitions which can't fit
    /// the container requirements.
    TooManyExtras,

    #[from]
    #[display(inner)]
    Builder(BuilderError),
}

impl<S: StashProvider, H: StateProvider, P: IndexProvider> From<ComposeError>
    for StockError<S, H, P, ComposeError>
{
    fn from(err: ComposeError) -> Self { Self::InvalidInput(err) }
}

impl<S: StashProvider, H: StateProvider, P: IndexProvider> From<BuilderError>
    for StockError<S, H, P, ComposeError>
{
    fn from(err: BuilderError) -> Self { Self::InvalidInput(err.into()) }
}

#[derive(Clone, PartialEq, Eq, Debug, Display, Error, From)]
#[display(doc_comments)]
pub enum FasciaError {
    /// bundle {1} for contract {0} contains invalid transition input map.
    InvalidBundle(ContractId, BundleId),
}

impl<S: StashProvider, H: StateProvider, P: IndexProvider> From<FasciaError>
    for StockError<S, H, P, FasciaError>
{
    fn from(err: FasciaError) -> Self { Self::InvalidInput(err) }
}

#[derive(Clone, PartialEq, Eq, Debug, Display, Error, From)]
#[display(inner)]
pub enum InputError {
    #[from]
    Compose(ComposeError),
    #[from]
    Consign(ConsignError),
    #[from]
    Fascia(FasciaError),
}

macro_rules! stock_err_conv {
    (Infallible, $err2:ty) => {
        impl<S: StashProvider, H: StateProvider, P: IndexProvider>
            From<StockError<S, H, P, Infallible>> for StockError<S, H, P, $err2>
        {
            fn from(err: StockError<S, H, P, Infallible>) -> Self {
                stock_err_conv!(@body err, e, match e {})
            }
        }
    };
    ($err1:ty, $err2:ty) => {
        impl<S: StashProvider, H: StateProvider, P: IndexProvider> From<StockError<S, H, P, $err1>>
            for StockError<S, H, P, $err2>
        {
            fn from(err: StockError<S, H, P, $err1>) -> Self {
                stock_err_conv!(@body err, e, StockError::InvalidInput(e.into()))
            }
        }
    };
    (@body $err:expr, $e:ident, $($invalid:tt)*) => {
        match $err {
            StockError::InvalidInput($e) => $($invalid)*,
            StockError::Resolver(e) => StockError::Resolver(e),
            StockError::StashRead(e) => StockError::StashRead(e),
            StockError::StashWrite(e) => StockError::StashWrite(e),
            StockError::IndexRead(e) => StockError::IndexRead(e),
            StockError::IndexWrite(e) => StockError::IndexWrite(e),
            StockError::StateRead(e) => StockError::StateRead(e),
            StockError::StateWrite(e) => StockError::StateWrite(e),
            StockError::AbsentValidWitness => StockError::AbsentValidWitness,
            StockError::BundlesInconsistency => StockError::BundlesInconsistency,
            StockError::StashData(e) => StockError::StashData(e),
            StockError::StashInconsistency(e) => StockError::StashInconsistency(e),
            StockError::StateInconsistency(e) => StockError::StateInconsistency(e),
            StockError::IndexInconsistency(e) => StockError::IndexInconsistency(e),
            StockError::WitnessUnresolved(id, e) => StockError::WitnessUnresolved(id, e),
            StockError::ContractLinkError(e) => StockError::ContractLinkError(e),
        }
    };
}

stock_err_conv!(Infallible, ComposeError);
stock_err_conv!(Infallible, ConsignError);
stock_err_conv!(Infallible, FasciaError);
stock_err_conv!(Infallible, InputError);
stock_err_conv!(ComposeError, InputError);
stock_err_conv!(ConsignError, InputError);
stock_err_conv!(FasciaError, InputError);

pub type StockErrorMem<E = Infallible> = StockError<MemStash, MemState, MemIndex, E>;
pub type StockErrorAll<S = MemStash, H = MemState, P = MemIndex> = StockError<S, H, P, InputError>;

/// Resolver serving a set of already-resolved witness statuses, falling back
/// to the wrapped resolver for the other witnesses.
struct PreresolvedWitnesses<R: ResolveWitness> {
    statuses: BTreeMap<Txid, WitnessStatus>,
    fallback: R,
}

impl<R: ResolveWitness> ResolveWitness for PreresolvedWitnesses<R> {
    fn resolve_witness(&self, witness_id: Txid) -> Result<WitnessStatus, WitnessResolverError> {
        match self.statuses.get(&witness_id) {
            Some(status) => Ok(status.clone()),
            None => self.fallback.resolve_witness(witness_id),
        }
    }

    fn check_chain_net(&self, chain_net: ChainNet) -> Result<(), WitnessResolverError> {
        self.fallback.check_chain_net(chain_net)
    }
}

#[derive(Debug)]
pub struct Stock<
    S: StashProvider = MemStash,
    H: StateProvider = MemState,
    P: IndexProvider = MemIndex,
> {
    stash: Stash<S>,
    state: State<H>,
    index: Index<P>,
}

impl<S: StashProvider, H: StateProvider, P: IndexProvider> CloneNoPersistence for Stock<S, H, P> {
    fn clone_no_persistence(&self) -> Self {
        Self {
            stash: self.stash.clone_no_persistence(),
            state: self.state.clone_no_persistence(),
            index: self.index.clone_no_persistence(),
        }
    }
}

impl<S: StashProvider, H: StateProvider, P: IndexProvider> Default for Stock<S, H, P>
where
    S: Default,
    H: Default,
    P: Default,
{
    fn default() -> Self {
        Self {
            stash: default!(),
            state: default!(),
            index: default!(),
        }
    }
}

impl Stock {
    #[inline]
    pub fn in_memory() -> Self {
        Self::with(MemStash::in_memory(), MemState::in_memory(), MemIndex::in_memory())
    }
}

impl<S: StashProvider, H: StateProvider, I: IndexProvider> Stock<S, H, I> {
    pub fn load<P>(provider: P, autosave: bool) -> Result<Self, PersistenceError>
    where P: Clone
            + PersistenceProvider<S>
            + PersistenceProvider<H>
            + PersistenceProvider<I>
            + 'static {
        let stash = S::load(provider.clone(), autosave)?;
        let state = H::load(provider.clone(), autosave)?;
        let index = I::load(provider, autosave)?;
        Ok(Self::with(stash, state, index))
    }

    pub fn make_persistent<P>(
        &mut self,
        provider: P,
        autosave: bool,
    ) -> Result<bool, PersistenceError>
    where
        P: Clone
            + PersistenceProvider<S>
            + PersistenceProvider<H>
            + PersistenceProvider<I>
            + 'static,
    {
        let a = self
            .as_stash_provider_mut()
            .make_persistent(provider.clone(), autosave)?;
        let b = self
            .as_state_provider_mut()
            .make_persistent(provider.clone(), autosave)?;
        let c = self
            .as_index_provider_mut()
            .make_persistent(provider, autosave)?;
        Ok(a && b && c)
    }

    pub fn store(&mut self) -> Result<(), PersistenceError> {
        // TODO: Revert on failure

        self.as_stash_provider_mut().store()?;
        self.as_state_provider_mut().store()?;
        self.as_index_provider_mut().store()?;

        Ok(())
    }
}

impl<S: StashProvider, H: StateProvider, P: IndexProvider> Stock<S, H, P> {
    pub fn with(stash_provider: S, state_provider: H, index_provider: P) -> Self {
        Stock {
            stash: Stash::new(stash_provider),
            state: State::new(state_provider),
            index: Index::new(index_provider),
        }
    }

    #[doc(hidden)]
    pub fn as_stash_provider(&self) -> &S { self.stash.as_provider() }
    #[doc(hidden)]
    pub fn as_state_provider(&self) -> &H { self.state.as_provider() }
    #[doc(hidden)]
    pub fn as_index_provider(&self) -> &P { self.index.as_provider() }

    #[doc(hidden)]
    pub fn as_stash_provider_mut(&mut self) -> &mut S { self.stash.as_provider_mut() }
    #[doc(hidden)]
    pub fn as_state_provider_mut(&mut self) -> &mut H { self.state.as_provider_mut() }
    #[doc(hidden)]
    pub fn as_index_provider_mut(&mut self) -> &mut P { self.index.as_provider_mut() }

    pub fn schemata(&self) -> Result<impl Iterator<Item = SchemaInfo> + '_, StockError<S, H, P>> {
        Ok(self.stash.schemata()?.map(SchemaInfo::with))
    }
    pub fn schema(&self, schema_id: SchemaId) -> Result<&Schema, StockError<S, H, P>> {
        Ok(self.stash.schema(schema_id)?)
    }

    pub fn contracts(
        &self,
    ) -> Result<impl Iterator<Item = ContractInfo> + '_, StockError<S, H, P>> {
        Ok(self.stash.geneses()?.map(ContractInfo::with))
    }

    /// Iterates over ids of all contract assigning state to the provided set of
    /// output seals.
    pub fn contracts_assigning(
        &self,
        outputs: impl IntoIterator<Item = impl Into<Outpoint>>,
    ) -> Result<impl Iterator<Item = ContractId> + '_, StockError<S, H, P>> {
        let outputs = outputs
            .into_iter()
            .map(|o| o.into())
            .collect::<BTreeSet<_>>();
        Ok(self.index.contracts_assigning(outputs)?)
    }

    #[allow(clippy::type_complexity)]
    fn contract_raw(
        &self,
        contract_id: ContractId,
    ) -> Result<(&Schema, H::ContractRead<'_>, ContractInfo), StockError<S, H, P>> {
        let state = self.state.contract_state(contract_id)?;
        let schema_id = state.schema_id();
        let schema = self.stash.schema(schema_id)?;
        Ok((schema, state, self.contract_info(contract_id)?))
    }

    pub fn contract_info(
        &self,
        contract_id: ContractId,
    ) -> Result<ContractInfo, StockError<S, H, P>> {
        Ok(ContractInfo::with(self.stash.genesis(contract_id)?))
    }

    pub fn contract_state(
        &self,
        contract_id: ContractId,
    ) -> Result<H::ContractRead<'_>, StockError<S, H, P>> {
        self.state
            .contract_state(contract_id)
            .map_err(StockError::from)
    }

    pub fn contract_wrapper<C: IssuerWrapper>(
        &self,
        contract_id: ContractId,
    ) -> Result<C::Wrapper<H::ContractRead<'_>>, StockError<S, H, P>> {
        self.schema_wrapper::<C::Wrapper<_>>(contract_id)
    }

    fn schema_wrapper<'a, C: SchemaWrapper<H::ContractRead<'a>>>(
        &'a self,
        contract_id: ContractId,
    ) -> Result<C, StockError<S, H, P>> {
        let contract_data = self.contract_data(contract_id)?;
        Ok(C::with(contract_data))
    }

    /// Returns the contract data for the given contract ID
    pub fn contract_data(
        &self,
        contract_id: ContractId,
    ) -> Result<ContractData<H::ContractRead<'_>>, StockError<S, H, P>> {
        let (schema, state, info) = self.contract_raw(contract_id)?;

        let (types, _) = self.stash.extract(schema)?;

        Ok(ContractData {
            state,
            schema: schema.clone(),
            types,
            info,
        })
    }

    pub fn contract_assignments_for(
        &self,
        contract_id: ContractId,
        outpoints: impl IntoIterator<Item = impl Into<Outpoint>>,
    ) -> Result<ContractAssignments, StockError<S, H, P>> {
        let outputs: BTreeSet<Outpoint> = outpoints.into_iter().map(|o| o.into()).collect();

        let state = self.contract_state(contract_id)?;

        let mut res =
            HashMap::<OutputSeal, HashMap<Opout, AllocatedState>>::with_capacity(outputs.len());

        for item in state.fungible_all() {
            let outpoint = item.seal.into();
            if outputs.contains::<Outpoint>(&outpoint) {
                res.entry(item.seal)
                    .or_default()
                    .insert(item.opout, AllocatedState::Amount(item.state));
            }
        }

        for item in state.data_all() {
            let outpoint = item.seal.into();
            if outputs.contains::<Outpoint>(&outpoint) {
                res.entry(item.seal)
                    .or_default()
                    .insert(item.opout, AllocatedState::Data(item.state.clone()));
            }
        }

        for item in state.rights_all() {
            let outpoint = item.seal.into();
            if outputs.contains::<Outpoint>(&outpoint) {
                res.entry(item.seal)
                    .or_default()
                    .insert(item.opout, AllocatedState::Void);
            }
        }

        Ok(res)
    }

    pub fn contract_builder(
        &self,
        issuer: impl Into<Identity>,
        schema_id: SchemaId,
        chain_net: ChainNet,
    ) -> Result<ContractBuilder, StockError<S, H, P>> {
        Ok(self
            .stash
            .contract_builder(issuer.into(), schema_id, chain_net)?)
    }

    pub fn transition_builder(
        &self,
        contract_id: ContractId,
        transition_name: impl Into<FieldName>,
    ) -> Result<TransitionBuilder, StockError<S, H, P>> {
        Ok(self
            .stash
            .transition_builder(contract_id, transition_name)?)
    }

    pub fn transition_builder_raw(
        &self,
        contract_id: ContractId,
        transition_type: TransitionType,
    ) -> Result<TransitionBuilder, StockError<S, H, P>> {
        Ok(self
            .stash
            .transition_builder_raw(contract_id, transition_type)?)
    }

    pub fn export_schema(&self, schema_id: SchemaId) -> Result<ValidKit, StockError<S, H, P>> {
        let mut kit = Kit::default();
        let schema = self.schema(schema_id)?;
        kit.schemata.push(schema.clone()).expect("single item");
        let (types, scripts) = self.stash.extract(schema)?;
        kit.scripts
            .extend(scripts.into_values())
            .expect("type guarantees");
        kit.types = types;
        Ok(kit.validate().expect("stock produced invalid kit"))
    }

    pub fn export_contract(
        &self,
        contract_id: ContractId,
    ) -> Result<Contract, StockError<S, H, P, ConsignError>> {
        self.consign::<false>(contract_id, [], vec![], [], None, false)
            .map(|(c, _)| c)
    }

    pub fn transfer(
        &self,
        contract_id: ContractId,
        outputs: impl AsRef<[OutputSeal]>,
        secret_seals: impl AsRef<[SecretSeal]>,
        opids: impl IntoIterator<Item = OpId>,
        witness_id: Option<Txid>,
    ) -> Result<Transfer, StockError<S, H, P, ConsignError>> {
        self.consign(contract_id, outputs, secret_seals, opids, witness_id, false)
            .map(|(c, _)| c)
    }

    pub fn transfer_with_dag(
        &self,
        contract_id: ContractId,
        outputs: impl AsRef<[OutputSeal]>,
        secret_seals: impl AsRef<[SecretSeal]>,
        opids: impl IntoIterator<Item = OpId>,
        witness_id: Option<Txid>,
    ) -> Result<ConsignmentWithDag<true>, StockError<S, H, P, ConsignError>> {
        self.consign(contract_id, outputs, secret_seals, opids, witness_id, true)
            .map(|(c, d)| (c, d.unwrap()))
    }

    fn sort_bundles(
        &self,
        bundles: BTreeMap<BundleId, (WitnessBundle, u32)>,
        contract_id: ContractId,
        build_opouts_dag: bool,
        genesis: &Genesis,
    ) -> Result<SortedBundlesWithDag, StockError<S, H, P, ConsignError>> {
        let mut dag_info = None;
        if build_opouts_dag {
            dag_info = Some(OpoutsDagInfo::new());
        }
        if let Some(ref mut dag_info) = dag_info {
            dag_info.register_outputs(genesis, &genesis.id());
        }

        let bundles_len = bundles.len();
        if bundles_len <= 1 {
            let bundles = bundles.into_values().map(|(b, _)| b).collect::<Vec<_>>();
            if let Some(ref mut dag_info) = dag_info {
                dag_info.build_dag(
                    &bundles
                        .iter()
                        .flat_map(|wb| wb.bundle.known_transitions.iter())
                        .collect::<Vec<_>>(),
                );
            }
            return Ok((bundles, dag_info.map(|d| d.to_opouts_dag_data())));
        }

        // Pre-sort by witness height for efficiency
        let mut bundles_with_height = bundles.into_iter().collect::<Vec<_>>();
        bundles_with_height.sort_by_key(|(_, (_, num))| *num);

        // Dependency violation detection
        let mut needs_reordering = false;
        let bundle_positions = bundles_with_height
            .iter()
            .enumerate()
            .map(|(i, (bundle_id, (_, _)))| (*bundle_id, i))
            .collect::<HashMap<_, _>>();
        'outer: for (i, (_, (witness_bundle, _))) in bundles_with_height.iter().enumerate() {
            for KnownTransition { transition, opid } in &witness_bundle.bundle.known_transitions {
                if let Some(ref mut dag_info) = dag_info {
                    dag_info.register_outputs(transition, opid);
                }
                for input in &transition.inputs {
                    if let Some(ref mut dag_info) = dag_info {
                        dag_info.connect_input_to_outputs_by_opid(input, opid);
                    }
                    if input.op != contract_id {
                        let input_bundle_id = self.index.bundle_id_for_op(input.op)?;
                        // ignore missing input bundles (e.g. can happen in case of replace)
                        if let Some(&input_pos) = bundle_positions.get(&input_bundle_id) {
                            if input_pos > i {
                                needs_reordering = true;
                                break 'outer;
                            }
                        }
                    }
                }
            }
        }
        if !needs_reordering {
            let bundles = bundles_with_height
                .into_iter()
                .map(|(_, (wb, _))| wb)
                .collect::<Vec<_>>();
            return Ok((bundles, dag_info.map(|d| d.to_opouts_dag_data())));
        }

        // Topological sort
        let mut known_bundle_dependencies: HashMap<BundleId, HashSet<BundleId>> =
            HashMap::with_capacity(bundles_len);
        for (bundle_id, (witness_bundle, _)) in &bundles_with_height {
            for KnownTransition { transition, opid } in &witness_bundle.bundle.known_transitions {
                if let Some(ref mut dag_info) = dag_info {
                    dag_info.register_outputs(transition, opid);
                }
                for input in &transition.inputs {
                    if let Some(ref mut dag_info) = dag_info {
                        dag_info.connect_input_to_outputs_by_opid(input, opid);
                    }
                    if input.op != contract_id {
                        let input_bundle_id = self.index.bundle_id_for_op(input.op)?;
                        if bundle_positions.contains_key(&input_bundle_id)
                            && input_bundle_id != *bundle_id
                        {
                            known_bundle_dependencies
                                .entry(*bundle_id)
                                .or_default()
                                .insert(input_bundle_id);
                        }
                    }
                }
            }
        }
        let mut sorted_bundles: Vec<WitnessBundle> = Vec::with_capacity(bundles_len);
        let mut remaining = bundles_with_height
            .into_iter()
            .map(|(id, (wb, _))| (id, wb))
            .collect::<Vec<_>>();
        while !remaining.is_empty() {
            let processed_ids = sorted_bundles
                .iter()
                .map(|wb| wb.bundle.bundle_id())
                .collect::<HashSet<_>>();
            let mut found = false;
            let mut i = 0;
            while i < remaining.len() {
                let (bundle_id, _) = &remaining[i];
                let dependencies = known_bundle_dependencies
                    .get(bundle_id)
                    .cloned()
                    .unwrap_or_default();
                if dependencies.is_subset(&processed_ids) {
                    let (_, witness_bundle) = remaining.remove(i);
                    sorted_bundles.push(witness_bundle);
                    found = true;
                    break;
                }
                i += 1;
            }
            if !found {
                return Err(StockError::BundlesInconsistency);
            }
        }
        Ok((sorted_bundles, dag_info.map(|d| d.to_opouts_dag_data())))
    }

    fn consign<const TRANSFER: bool>(
        &self,
        contract_id: ContractId,
        outputs: impl AsRef<[OutputSeal]>,
        secret_seals: impl AsRef<[SecretSeal]>,
        opids: impl IntoIterator<Item = OpId>,
        witness_id: Option<Txid>,
        build_opouts_dag: bool,
    ) -> Result<ConsignmentWithOptDag<TRANSFER>, StockError<S, H, P, ConsignError>> {
        let outputs = outputs.as_ref();
        let secret_seals = secret_seals.as_ref();

        // Collect initial set of opids to include
        let mut opids = opids.into_iter().collect::<HashSet<_>>();
        opids.extend(
            self.index
                .public_opouts(contract_id)?
                .into_iter()
                .chain(
                    self.index
                        .opouts_by_outputs(contract_id, outputs.iter().copied())?,
                )
                .chain(
                    self.index
                        .opouts_by_terminals(secret_seals.iter().copied())?,
                )
                .map(|opout| opout.op),
        );

        self.consign_operations(contract_id, opids, secret_seals, witness_id, build_opouts_dag)
    }

    fn consign_operations<const TRANSFER: bool>(
        &self,
        contract_id: ContractId,
        opids: impl IntoIterator<Item = OpId>,
        secret_seals: &[SecretSeal],
        witness_id: Option<Txid>,
        build_opouts_dag: bool,
    ) -> Result<ConsignmentWithOptDag<TRANSFER>, StockError<S, H, P, ConsignError>> {
        // 1.3. Collect all state transitions assigning state to the provided outpoints
        let mut bundles = BTreeMap::<BundleId, (WitnessBundle, u32)>::new();
        let mut parent_opids = Vec::<OpId>::new();
        let mut bundle_sec_seals: BTreeMap<BundleId, BTreeSet<SecretSeal>> = BTreeMap::new();
        for opid in opids {
            if opid == contract_id {
                continue; // we skip genesis since it will be present anywhere
            }

            let transition = self.transition(opid)?;

            let bundle_id = self.index.bundle_id_for_op(transition.id())?;

            // skip bundles not associated to the terminals witness
            if let Some(witness_id) = witness_id {
                let (mut witness_ids, _) = self.index.bundle_info(bundle_id)?;
                if !witness_ids.any(|w| w == witness_id) {
                    continue;
                }
            }

            parent_opids.extend(transition.inputs().iter().map(|input| input.op));

            // 1.4. Collect secret seals for this bundle to add to the consignment terminals
            for typed_assignments in transition.assignments.values() {
                for seal in typed_assignments.to_confidential_seals() {
                    if secret_seals.contains(&seal) {
                        bundle_sec_seals.entry(bundle_id).or_default().insert(seal);
                    }
                }
            }

            if let Some((wbundle, _)) = bundles.get_mut(&bundle_id) {
                wbundle.bundle.reveal_transition(transition.clone())?;
            } else {
                bundles.insert(bundle_id, self.witness_bundle(bundle_id, opid)?);
            };
        }
        self.consign_bundles(contract_id, bundles, parent_opids, bundle_sec_seals, build_opouts_dag)
    }

    fn consign_bundles<const TRANSFER: bool>(
        &self,
        contract_id: ContractId,
        mut bundles: BTreeMap<BundleId, (WitnessBundle, u32)>,
        mut parent_opids: Vec<OpId>,
        bundle_sec_seals: BTreeMap<BundleId, BTreeSet<SecretSeal>>,
        build_opouts_dag: bool,
    ) -> Result<ConsignmentWithOptDag<TRANSFER>, StockError<S, H, P, ConsignError>> {
        // 2. Collect all state transitions between terminals and genesis
        let mut seen_ids = HashSet::new();
        while let Some(id) = parent_opids.pop() {
            if id == contract_id {
                continue; // we skip genesis since it will be present anywhere
            }
            if !seen_ids.insert(id) {
                continue; // we skip seen IDs to avoid re-processing duplicates
            }
            let transition = self.transition(id)?;
            parent_opids.extend(transition.inputs().iter().map(|input| input.op));
            let bundle_id = self.index.bundle_id_for_op(transition.id())?;
            if let Some((wbundle, _)) = bundles.get_mut(&bundle_id) {
                wbundle.bundle.reveal_transition(transition.clone())?;
            } else {
                bundles.insert(bundle_id, self.witness_bundle(bundle_id, id)?);
            };
        }

        let genesis = self.stash.genesis(contract_id)?.clone();

        let schema = self.stash.schema(genesis.schema_id)?.clone();

        let (sorted_bundles, dag) =
            self.sort_bundles(bundles, contract_id, build_opouts_dag, &genesis)?;

        let bundles =
            Confined::try_from_iter(sorted_bundles).map_err(|_| ConsignError::TooManyBundles)?;
        let terminals = Confined::try_from(
            bundle_sec_seals
                .into_iter()
                .map(|(bundle_id, seals)| {
                    Confined::try_from(seals)
                        .map(|confined| (bundle_id, SecretSeals::from(confined)))
                        .map_err(|_| ConsignError::InvalidSecretSealsNumber)
                })
                .collect::<Result<BTreeMap<_, _>, _>>()?,
        )
        .map_err(|_| ConsignError::TooManyTerminals)?;

        let (types, scripts) = self.stash.extract(&schema)?;
        let scripts = Confined::from_iter_checked(scripts.into_values());
        // TODO: Conceal everything we do not need

        let consignment = Consignment {
            version: ContainerVer::V0,
            transfer: TRANSFER,

            schema,
            genesis,
            terminals,
            bundles,

            types,
            scripts,
        };

        Ok((consignment, dag))
    }

    pub fn transfer_from_fascia(
        &self,
        contract_id: ContractId,
        outputs: impl AsRef<[OutputSeal]>,
        secret_seals: impl AsRef<[SecretSeal]>,
        opids: impl IntoIterator<Item = OpId>,
        fascia: &Fascia,
    ) -> Result<Consignment<true>, StockError<S, H, P, ConsignError>> {
        self.consign_from_fascia(contract_id, outputs, secret_seals, opids, fascia, false)
            .map(|(c, _)| c)
    }

    pub fn transfer_from_fascia_with_dag(
        &self,
        contract_id: ContractId,
        outputs: impl AsRef<[OutputSeal]>,
        secret_seals: impl AsRef<[SecretSeal]>,
        opids: impl IntoIterator<Item = OpId>,
        fascia: &Fascia,
    ) -> Result<ConsignmentWithDag<true>, StockError<S, H, P, ConsignError>> {
        self.consign_from_fascia(contract_id, outputs, secret_seals, opids, fascia, true)
            .map(|(c, d)| (c, d.expect("build_opouts_dag=true")))
    }

    fn consign_from_fascia(
        &self,
        contract_id: ContractId,
        outputs: impl AsRef<[OutputSeal]>,
        secret_seals: impl AsRef<[SecretSeal]>,
        opids: impl IntoIterator<Item = OpId>,
        fascia: &Fascia,
        build_opouts_dag: bool,
    ) -> Result<ConsignmentWithOptDag<true>, StockError<S, H, P, ConsignError>> {
        let mut contract_bundle = fascia
            .bundles()
            .get(&contract_id)
            .ok_or(ConsignError::UnrelatedContract(contract_id))?
            .clone();
        let bundle_id = contract_bundle.bundle_id();
        let all_bundle_opids = contract_bundle.input_map_opids();
        let bundle_revealed_opids = contract_bundle.known_transitions_opids();
        let opids = opids.into_iter().collect::<HashSet<_>>();
        let secret_seals = secret_seals
            .as_ref()
            .iter()
            .cloned()
            .collect::<BTreeSet<_>>();
        let outputs = outputs.as_ref().iter().collect::<HashSet<_>>();
        let witness_id = fascia.witness_id();
        let is_requested_transition = |kt: &KnownTransition| {
            if opids.contains(&kt.opid) {
                return true; // 1. explicitly requested opids
            }
            for typed_assigns in kt.transition.assignments.values() {
                for index in 0..typed_assigns.len_u16() {
                    match typed_assigns
                        .revealed_seal_at(index)
                        .expect("cycling indexes")
                    {
                        Some(s) => {
                            if outputs.contains(&OutputSeal::with(witness_id, s.vout())) {
                                return true; // 2. outputs (witness)
                            }
                        }
                        None => {
                            if secret_seals.contains(
                                &typed_assigns
                                    .confidential_seal_at(index)
                                    .expect("cycling indexes"),
                            ) {
                                return true; // 3. secret seals (blinded)
                            }
                        }
                    }
                }
            }
            false
        };
        // filter only required transitions in the bundle
        // process transitions in reverse order since children must appear after parents
        let mut required_opids = bset![];
        let mut rev_bundle_transitions = vec![];
        for known_transition in contract_bundle.known_transitions.into_iter().rev() {
            if required_opids.contains(&known_transition.opid)
                || is_requested_transition(&known_transition)
            {
                required_opids.remove(&known_transition.opid);
                required_opids.extend(known_transition.transition.inputs.iter().map(|o| o.op));
                rev_bundle_transitions.push(known_transition);
            }
        }
        if let Some(opid) = required_opids.intersection(&all_bundle_opids).next() {
            if let Some(opid) = required_opids.intersection(&bundle_revealed_opids).next() {
                return Err(ConsignError::ConcealedTransition(bundle_id, *opid).into());
            }
            return Err(ConsignError::UnorderedTransition(bundle_id, *opid).into());
        }
        rev_bundle_transitions.reverse();
        contract_bundle.known_transitions = Confined::from_checked(rev_bundle_transitions);
        let SealWitness {
            public: pub_witness,
            merkle_block,
            dbc_proof,
        } = fascia.seal_witness().clone();
        let anchor = Anchor::new(
            merkle_block
                .into_merkle_proof(contract_id.into())
                .map_err(|_| ConsignError::UnrelatedContract(contract_id))?,
            dbc_proof,
        );
        let bundle_sec_seals = if !secret_seals.is_empty() {
            bmap! {bundle_id => secret_seals}
        } else {
            bmap! {}
        };
        self.consign_bundles(
            contract_id,
            bmap! {bundle_id => (WitnessBundle::with(pub_witness, anchor, contract_bundle), u32::MAX)},
            required_opids.into_iter().collect::<Vec<_>>(),
            bundle_sec_seals,
            build_opouts_dag,
        )
    }

    fn store_transaction<E: Error>(
        &mut self,
        f: impl FnOnce(
            &mut Stash<S>,
            &mut State<H>,
            &mut Index<P>,
        ) -> Result<(), StockError<S, H, P, E>>,
    ) -> Result<(), StockError<S, H, P, E>> {
        self.state.begin_transaction()?;
        self.stash
            .begin_transaction()
            .inspect_err(|_| self.stash.rollback_transaction())?;
        self.index.begin_transaction().inspect_err(|_| {
            self.state.rollback_transaction();
            self.stash.rollback_transaction();
        })?;
        f(&mut self.stash, &mut self.state, &mut self.index)?;
        self.index
            .commit_transaction()
            .map_err(StockError::from)
            .and_then(|_| self.state.commit_transaction().map_err(StockError::from))
            .and_then(|_| self.stash.commit_transaction().map_err(StockError::from))
            .inspect_err(|_| {
                self.state.rollback_transaction();
                self.stash.rollback_transaction();
                self.index.rollback_transaction();
            })
    }

    pub fn import_kit(&mut self, kit: ValidKit) -> Result<validation::Status, StockError<S, H, P>> {
        let (kit, status) = kit.split();
        self.stash.begin_transaction()?;
        self.stash.consume_kit(kit)?;
        self.stash.commit_transaction()?;
        Ok(status)
    }

    pub fn import_contract<R: ResolveWitness>(
        &mut self,
        contract: ValidContract,
        resolver: R,
    ) -> Result<(), StockError<S, H, P>> {
        self.consume_consignment(contract, resolver)
    }

    pub fn accept_transfer<R: ResolveWitness>(
        &mut self,
        contract: ValidTransfer,
        resolver: R,
    ) -> Result<(), StockError<S, H, P>> {
        self.consume_consignment(contract, resolver)
    }

    /// Consumes a validated consignment.
    ///
    /// The consignment witnesses are re-resolved before consuming, since
    /// their ords may have changed after the validation, e.g. if a reorg
    /// happened in the meantime. If that leaves a consignment bundle without
    /// any valid witness, the consignment is NOT consumed and
    /// [`StockError::AbsentValidWitness`] is returned; in that case, as a
    /// side effect, the fresh ords of the already-known witnesses are stored
    /// and the known operations of the bundles left without a valid witness
    /// are set as invalid, together with all their descendants.
    fn consume_consignment<R: ResolveWitness, const TRANSFER: bool>(
        &mut self,
        consignment: ValidConsignment<TRANSFER>,
        resolver: R,
    ) -> Result<(), StockError<S, H, P>> {
        let consignment = self.stash.resolve_secrets(consignment.into_consignment())?;
        let consignment_bundles: Vec<(Txid, BundleId, BTreeSet<OpId>)> = consignment
            .bundled_witnesses()
            .map(|wb| {
                let bundle = wb.bundle();
                (
                    wb.pub_witness.to_witness_id(),
                    bundle.bundle_id(),
                    bundle.known_transitions_opids(),
                )
            })
            .collect();

        // resolve the consignment witnesses with accept-time resolutions,
        // which may differ from the ones seen at validation time if a reorg
        // happened in the meantime
        let mut statuses: BTreeMap<Txid, WitnessStatus> = bmap![];
        for (witness_id, _, _) in &consignment_bundles {
            if !statuses.contains_key(witness_id) {
                let status = resolver
                    .resolve_witness(*witness_id)
                    .map_err(|e| StockError::WitnessUnresolved(*witness_id, e))?;
                statuses.insert(*witness_id, status);
            }
        }

        // witness ords as they will be once the consignment is consumed:
        // the stored ones overlaid with the fresh resolutions
        let mut witnesses = self.as_state_provider().witnesses().release();
        let known_witness_ids: BTreeSet<Txid> = witnesses.keys().copied().collect();
        for (witness_id, status) in &statuses {
            witnesses.insert(*witness_id, status.witness_ord());
        }

        // collect the bundles left without any valid witness, re-resolving
        // the alternative witnesses known for a bundle before giving up on it
        let mut bundles_without_witness: Vec<BundleId> = vec![];
        for (witness_id, bundle_id, _) in &consignment_bundles {
            if witnesses.get(witness_id).is_some_and(|ord| ord.is_valid()) {
                continue;
            }
            let alt_witness_ids: BTreeSet<Txid> = match self.index.bundle_info(*bundle_id) {
                Ok((witness_ids, _)) => witness_ids.collect(),
                // the bundle is not known yet, so it has no other witnesses
                Err(IndexError::Inconsistency(IndexInconsistency::BundleWitnessUnknown(_))) => {
                    bset![]
                }
                Err(e) => return Err(e.into()),
            };
            let mut has_valid_witness = false;
            for alt_witness_id in alt_witness_ids {
                if let btree_map::Entry::Vacant(e) = statuses.entry(alt_witness_id) {
                    let status = resolver
                        .resolve_witness(alt_witness_id)
                        .map_err(|e| StockError::WitnessUnresolved(alt_witness_id, e))?;
                    witnesses.insert(alt_witness_id, status.witness_ord());
                    e.insert(status);
                }
                if witnesses
                    .get(&alt_witness_id)
                    .is_some_and(|ord| ord.is_valid())
                {
                    has_valid_witness = true;
                    break;
                }
            }
            if !has_valid_witness {
                bundles_without_witness.push(*bundle_id);
            }
        }

        // the consignment is stale: store the chain knowledge acquired while
        // checking it, then refuse it
        if !bundles_without_witness.is_empty() {
            let mut ops_to_invalidate = vec![];
            for bundle_id in &bundles_without_witness {
                if let Ok(bundle) = self.stash.bundle(*bundle_id) {
                    ops_to_invalidate.extend(bundle.known_transitions_opids());
                }
            }
            self.state.begin_transaction()?;
            // store the fresh ords of the already-known witnesses
            for (witness_id, status) in &statuses {
                if known_witness_ids.contains(witness_id) {
                    self.state
                        .upsert_witness(*witness_id, status.witness_ord())?;
                }
            }
            // set the known operations of the bundles left without a valid
            // witness as invalid, together with all their descendants
            let mut visited = bset!();
            for opid in ops_to_invalidate {
                self.set_ops_as_invalid(opid, &mut visited)?;
            }
            self.state.commit_transaction()?;
            return Err(StockError::AbsentValidWitness);
        }

        // serve the accept-time resolutions to the consignment consumption,
        // so the stored ords cannot diverge from the ones checked above
        let resolver = PreresolvedWitnesses {
            statuses,
            fallback: resolver,
        };
        self.store_transaction(move |stash, state, index| {
            state.update_from_consignment(&consignment, &resolver)?;
            index.index_consignment(&consignment)?;
            stash.consume_consignment(consignment)?;
            Ok(())
        })?;

        // the consignment was validated and all its bundles have a valid
        // witness: revalidate any of its operations that a reorg had
        // previously set as invalid, together with their descendants
        let mut invalid_ops = self.as_state_provider().invalid_ops();
        if consignment_bundles
            .iter()
            .any(|(_, _, opids)| opids.iter().any(|opid| invalid_ops.contains(opid)))
        {
            self.state.begin_transaction()?;
            let witnesses = self.as_state_provider().witnesses().release();
            let mut maybe_became_valid_opids: BTreeSet<OpId> = consignment_bundles
                .iter()
                .flat_map(|(_, _, opids)| opids.iter().copied())
                .collect();
            for (_, bundle_id, opids) in &consignment_bundles {
                for opid in opids {
                    self.maybe_update_ops_as_valid(
                        *opid,
                        *bundle_id,
                        &mut invalid_ops,
                        &mut maybe_became_valid_opids,
                        &witnesses,
                    )?;
                }
            }
            self.state.commit_transaction()?;
        }

        Ok(())
    }

    /// Imports fascia into the stash, index and inventory.
    ///
    /// Part of the transfer workflow. Called once PSBT is completed and an RGB
    /// fascia containing anchor and all state transitions is exported from
    /// it.
    ///
    /// Must be called before the consignment is created, when witness
    /// transaction is not yet mined.
    pub fn consume_fascia<WP: WitnessOrdProvider>(
        &mut self,
        fascia: Fascia,
        witness_ord_provider: WP,
    ) -> Result<(), StockError<S, H, P, FasciaError>> {
        self.store_transaction(move |stash, state, index| {
            let witness_id = fascia.witness_id();
            stash.consume_witness(fascia.seal_witness())?;

            for (contract_id, bundle) in fascia.into_bundles() {
                bundle
                    .check_opid_commitments()
                    .map_err(|_| FasciaError::InvalidBundle(contract_id, bundle.bundle_id()))?;

                index.index_bundle(contract_id, &bundle, witness_id)?;
                state.update_from_bundle(
                    contract_id,
                    &bundle,
                    witness_id,
                    &witness_ord_provider,
                )?;
                stash.consume_bundle(bundle)?;
            }
            Ok(())
        })
    }

    fn transition(&self, opid: OpId) -> Result<&Transition, StockError<S, H, P, ConsignError>> {
        let bundle_id = self.index.bundle_id_for_op(opid)?;
        let bundle = self.stash.bundle(bundle_id)?;
        bundle
            .get_transition(opid)
            .ok_or(ConsignError::Concealed(bundle_id, opid).into())
    }

    fn witness_bundle(
        &self,
        bundle_id: BundleId,
        opid: OpId,
    ) -> Result<(WitnessBundle, u32), StockError<S, H, P, ConsignError>> {
        let (witness_ids, contract_id) = self.index.bundle_info(bundle_id)?;
        let bundle = self
            .stash
            .bundle(bundle_id)?
            .to_concealed_except(opid)
            .map_err(|e| StockError::from(ConsignError::Transition(e)))?;
        let (witness_id, witness_ord) = self.state.select_valid_witness(witness_ids)?;
        let witness = self.stash.witness(witness_id)?;
        let pub_witness = witness.public.clone();
        let Ok(mpc_proof) = witness.merkle_block.to_merkle_proof(contract_id.into()) else {
            return Err(StashInconsistency::WitnessMissesContract(
                witness_id,
                bundle_id,
                contract_id,
                witness.dbc_proof.method(),
            )
            .into());
        };
        let anchor = Anchor::new(mpc_proof, witness.dbc_proof.clone());

        let height = match witness_ord {
            WitnessOrd::Mined(pos) => pos.height().into(),
            WitnessOrd::Tentative => u32::MAX - 1,
            WitnessOrd::Ignored => u32::MAX,
            WitnessOrd::Archived => unreachable!("select_valid_witness prevents this"),
        };

        Ok((WitnessBundle::with(pub_witness, anchor, bundle), height))
    }

    pub fn store_secret_seal(&mut self, seal: GraphSeal) -> Result<bool, StockError<S, H, P>> {
        Ok(self.stash.store_secret_seal(seal)?)
    }

    fn op_children(&self, opid: OpId) -> Result<Vec<(OpId, BundleId)>, StockError<S, H, P>> {
        // collect all bundle ids of the children of the operation
        let children_bundle_ids = match self.index.bundle_ids_children_of_op(opid) {
            Ok(bundle_ids) => bundle_ids,
            Err(IndexError::Inconsistency(IndexInconsistency::BundleAbsent(_))) => {
                // this transition has no children yet
                small_bset![]
            }
            Err(e) => return Err(e.into()),
        };
        // collect all opids of transitions consuming outputs of the operation,
        // together with their bundle ids
        let mut children = vec![];
        for child_bundle_id in children_bundle_ids {
            let child_bundle = self.stash.bundle(child_bundle_id)?;
            for kt in &child_bundle.known_transitions {
                if kt.transition.inputs.iter().any(|input| input.op == opid) {
                    children.push((kt.opid, child_bundle_id));
                }
            }
        }
        Ok(children)
    }

    fn set_ops_as_invalid(
        &mut self,
        opid: OpId,
        visited: &mut BTreeSet<OpId>,
    ) -> Result<(), StockError<S, H, P>> {
        // descendant trees of different operations can overlap and converge;
        // visit each operation only once per update
        // the visited set is local to the update on purpose: operations
        // already invalid from previous updates must still be re-visited,
        // since new descendants may have been added in the meantime
        if !visited.insert(opid) {
            return Ok(());
        }
        // add operation to set of invalid operations
        self.state.update_op(opid, false)?;
        // recursively set all descendant operations as invalid
        for (child_opid, _) in self.op_children(opid)? {
            self.set_ops_as_invalid(child_opid, visited)?;
        }
        Ok(())
    }

    fn maybe_update_ops_as_valid(
        &mut self,
        opid: OpId,
        bundle_id: BundleId,
        invalid_ops: &mut LargeOrdSet<OpId>,
        maybe_became_valid_opids: &mut BTreeSet<OpId>,
        witnesses: &BTreeMap<Txid, WitnessOrd>,
    ) -> Result<bool, StockError<S, H, P>> {
        let bundle = self.stash.bundle(bundle_id)?;
        let transition = bundle
            .get_transition(opid)
            .ok_or(StashInconsistency::OperationAbsent(opid))?
            .clone();

        // a valid operation needs a valid witness for its bundle
        let bundle_witness_ids = self.index.bundle_info(bundle_id)?.0;
        let mut valid = bundle_witness_ids
            .into_iter()
            .any(|id| witnesses.get(&id).is_some_and(|ord| ord.is_valid()));

        // recursively visit operation ancestors
        if valid {
            for input in &transition.inputs {
                let input_opid = input.op;
                // process parent first if its status is also uncertain
                if maybe_became_valid_opids.contains(&input_opid) {
                    let input_bundle_id = self.index.bundle_id_for_op(input_opid)?;
                    if !self.maybe_update_ops_as_valid(
                        input_opid,
                        input_bundle_id,
                        invalid_ops,
                        maybe_became_valid_opids,
                        witnesses,
                    )? {
                        valid = false;
                        break;
                    }
                // a single invalid parent is enough to consider the operation as invalid
                } else if invalid_ops.contains(&input_opid) {
                    valid = false;
                    break;
                }
            }
        }

        // remove operation since at this point we are sure about its status
        maybe_became_valid_opids.remove(&opid);

        if valid {
            // remove operation from set of invalid operations
            self.state.update_op(opid, true)?;
            invalid_ops.remove(&opid).unwrap();
            // recursively visit operation descendants to check if they became valid as well
            for (child_opid, child_bundle_id) in self.op_children(opid)? {
                // a child may have already been settled as valid earlier in
                // this update, when reached through another revalidated
                // parent; don't re-walk its subtree
                if !invalid_ops.contains(&child_opid)
                    && !maybe_became_valid_opids.contains(&child_opid)
                {
                    continue;
                }
                self.maybe_update_ops_as_valid(
                    child_opid,
                    child_bundle_id,
                    invalid_ops,
                    maybe_became_valid_opids,
                    witnesses,
                )?;
            }
        }

        Ok(valid)
    }

    fn update_witness_ord(
        &mut self,
        resolver: impl ResolveWitness,
        id: &Txid,
        ord: &mut WitnessOrd,
        became_invalid_witnesses: &mut BTreeMap<Txid, BTreeSet<BundleId>>,
        became_valid_witnesses: &mut BTreeMap<Txid, BTreeSet<BundleId>>,
    ) -> Result<(), StockError<S, H, P>> {
        let new = resolver
            .resolve_witness(*id)
            .map_err(|e| StockError::WitnessUnresolved(*id, e))?
            .witness_ord();
        let changed = *ord != new;
        if changed {
            let bundle_valid = match (*ord, new) {
                (WitnessOrd::Archived, _) => Some(true),
                (_, WitnessOrd::Archived) => Some(false),
                _ => None,
            };
            // save witnesses that became valid or invalid
            if let Some(valid) = bundle_valid {
                let seal_witness = self.stash.witness(*id)?;
                let bundle_ids: BTreeSet<_> = seal_witness.known_bundle_ids().collect();
                if valid {
                    became_valid_witnesses.insert(*id, bundle_ids);
                } else {
                    became_invalid_witnesses.insert(*id, bundle_ids);
                }
            }
            // save the changed witness ord
            self.state.upsert_witness(*id, new)?;
            *ord = new
        }
        Ok(())
    }

    pub fn update_witnesses(
        &mut self,
        resolver: impl ResolveWitness,
        after_height: u32,
        force_witnesses: Vec<Txid>,
    ) -> Result<UpdateRes, StockError<S, H, P>> {
        let after_height = NonZeroU32::new(after_height).unwrap_or(NonZeroU32::MIN);
        let mut succeeded = 0;
        let mut failed = map![];
        self.state.begin_transaction()?;
        let witnesses = self.as_state_provider().witnesses();
        let mut witnesses = witnesses.release();
        let mut became_invalid_witnesses = bmap!();
        let mut became_valid_witnesses = bmap!();
        // 1. update witness ord of all witnesses
        for (id, ord) in &mut witnesses {
            if matches!(ord, WitnessOrd::Ignored) && !force_witnesses.contains(id) {
                continue;
            }
            if matches!(ord, WitnessOrd::Mined(pos) if pos.height() < after_height) {
                continue;
            }
            match self.update_witness_ord(
                &resolver,
                id,
                ord,
                &mut became_invalid_witnesses,
                &mut became_valid_witnesses,
            ) {
                Ok(()) => {
                    succeeded += 1;
                }
                Err(err) => {
                    failed.insert(*id, err.to_string());
                }
            }
        }

        // 2. set invalidity of operations
        let mut visited = bset!();
        for bundle_ids in became_invalid_witnesses.values() {
            for bundle_id in bundle_ids {
                let bundle_witness_ids: BTreeSet<Txid> =
                    self.index.bundle_info(*bundle_id)?.0.collect();
                // set the bundle operations as invalid only if there are no valid witnesses
                // associated to the bundle
                if bundle_witness_ids
                    .iter()
                    .all(|id| !witnesses.get(id).unwrap().is_valid())
                {
                    // set all the bundle operations and their descendants as invalid
                    for opid in self.stash.bundle(*bundle_id)?.known_transitions_opids() {
                        self.set_ops_as_invalid(opid, &mut visited)?;
                    }
                }
            }
        }

        // 3. set validity of operations
        let mut maybe_became_valid_opids = bset!();
        // get all operations that became invalid and ones that were already invalid
        let mut invalid_ops_pre = self.as_state_provider().invalid_ops();
        for bundle_ids in became_valid_witnesses.values() {
            for bundle_id in bundle_ids {
                // store operations that may become valid (to be sure their ancestors are
                // checked)
                maybe_became_valid_opids
                    .extend(self.stash.bundle(*bundle_id)?.known_transitions_opids());
            }
        }
        for bundle_ids in became_valid_witnesses.values() {
            for bundle_id in bundle_ids {
                // check if the bundle operations and their descendants are now valid
                for opid in self.stash.bundle(*bundle_id)?.known_transitions_opids() {
                    self.maybe_update_ops_as_valid(
                        opid,
                        *bundle_id,
                        &mut invalid_ops_pre,
                        &mut maybe_became_valid_opids,
                        &witnesses,
                    )?;
                }
            }
        }

        self.state.commit_transaction()?;
        Ok(UpdateRes { succeeded, failed })
    }

    pub fn upsert_witness(
        &mut self,
        witness_id: Txid,
        witness_ord: WitnessOrd,
    ) -> Result<(), StockError<S, H, P>> {
        self.store_transaction(move |_stash, state, _index| {
            Ok(state.upsert_witness(witness_id, witness_ord)?)
        })
    }

    fn _check_bundle_history(
        &self,
        bundle_id: &BundleId,
        safe_height: NonZeroU32,
        contract_history: &mut HashMap<ContractId, HashMap<u32, HashSet<Txid>>>,
    ) -> Result<(), StockError<S, H, P>> {
        let (bundle_witness_ids, contract_id) = self.index.bundle_info(*bundle_id)?;
        let (witness_id, ord) = self.state.select_valid_witness(bundle_witness_ids)?;
        match ord {
            WitnessOrd::Mined(witness_pos) => {
                let witness_height = witness_pos.height();
                if witness_height > safe_height {
                    contract_history
                        .entry(contract_id)
                        .or_default()
                        .entry(witness_height.into())
                        .or_default()
                        .insert(witness_id);
                }
            }
            WitnessOrd::Tentative | WitnessOrd::Ignored | WitnessOrd::Archived => {
                contract_history
                    .entry(contract_id)
                    .or_default()
                    .entry(0)
                    .or_default()
                    .insert(witness_id);
            }
        }

        // recursively check bundle ancestors
        let bundle = self.stash.bundle(*bundle_id)?.clone();
        for KnownTransition { transition, .. } in bundle.known_transitions {
            for input in &transition.inputs {
                let input_opid = input.op;
                let input_bundle_id = match self.index.bundle_id_for_op(input_opid) {
                    Ok(id) => Some(id),
                    Err(IndexError::Inconsistency(IndexInconsistency::BundleAbsent(_))) => {
                        // reached genesis
                        None
                    }
                    Err(e) => return Err(e.into()),
                };

                if let Some(input_bundle_id) = input_bundle_id {
                    self._check_bundle_history(&input_bundle_id, safe_height, contract_history)?;
                }
            }
        }

        Ok(())
    }

    pub fn get_outpoint_unsafe_history(
        &self,
        outpoint: Outpoint,
        safe_height: NonZeroU32,
    ) -> Result<HashMap<ContractId, UnsafeHistoryMap>, StockError<S, H, P>> {
        let mut contract_history: HashMap<ContractId, HashMap<u32, HashSet<Txid>>> = HashMap::new();

        for id in self.contracts_assigning([outpoint])? {
            let state = self.contract_assignments_for(id, [outpoint])?;
            for opid in state
                .values()
                .flat_map(|assigns| assigns.keys().map(|opout| opout.op))
            {
                let bundle_id = self.index.bundle_id_for_op(opid)?;
                self._check_bundle_history(&bundle_id, safe_height, &mut contract_history)?;
            }
        }

        Ok(contract_history)
    }

    pub fn validate_contracts_link<Parent: LinkableIssuerWrapper, Child: LinkableIssuerWrapper>(
        &self,
        parent_contract_id: ContractId,
        child_contract_id: ContractId,
    ) -> Result<(), StockError<S, H, P>> {
        let parent_links_to_child = self
            .schema_wrapper::<<Parent as LinkableIssuerWrapper>::Wrapper<_>>(parent_contract_id)?
            .link_to()?
            .ok_or(LinkError::NoValue)?
            == child_contract_id;
        let child_links_to_parent = self
            .schema_wrapper::<<Child as LinkableIssuerWrapper>::Wrapper<_>>(child_contract_id)?
            .link_from()?
            .ok_or(LinkError::NoValue)?
            == parent_contract_id;
        if parent_links_to_child && child_links_to_parent {
            Ok(())
        } else {
            Err(LinkError::ValueMismatch.into())
        }
    }
}

#[derive(Clone, Eq, PartialEq, Debug)]
pub struct UpdateRes {
    pub succeeded: usize,
    pub failed: HashMap<Txid, String>,
}

#[cfg(test)]
mod test {
    use baid64::FromBaid64Str;
    use rgb::commit_verify::{Conceal, DigestExt, Sha256};
    use rgb::Vout;

    use super::*;
    use crate::containers::ConsignmentExt;

    #[test]
    fn test_consign() {
        let mut stock = Stock::in_memory();
        let seal = GraphSeal::new_random_vout(Vout::from_u32(0));
        let secret_seal = seal.conceal();

        stock.store_secret_seal(seal).unwrap();
        let contract_id =
            ContractId::from_baid64_str("rgb:qFuT6DN8-9AuO95M-7R8R8Mc-AZvs7zG-obum1Va-BRnweKk")
                .unwrap();
        if let Ok(transfer) =
            stock.consign::<true>(contract_id, [], vec![secret_seal], [], None, false)
        {
            println!("{transfer:?}")
        }
    }

    #[test]
    fn test_export_contract() {
        let stock = Stock::in_memory();
        let contract_id =
            ContractId::from_baid64_str("rgb:qFuT6DN8-9AuO95M-7R8R8Mc-AZvs7zG-obum1Va-BRnweKk")
                .unwrap();
        if let Ok(contract) = stock.export_contract(contract_id) {
            println!("{:?}", contract.contract_id())
        }
    }

    #[test]
    fn test_export_schema() {
        let stock = Stock::in_memory();
        let hasher = Sha256::default();
        let schema_id = SchemaId::from(hasher);
        if let Ok(schema) = stock.export_schema(schema_id) {
            println!("{:?}", schema.kit_id())
        }
    }

    #[test]
    fn test_transition_builder() {
        let stock = Stock::in_memory();
        let hasher = Sha256::default();

        let bytes_hash = hasher.finish();
        let contract_id = ContractId::copy_from_slice(bytes_hash).unwrap();

        if let Ok(builder) = stock.transition_builder(contract_id, "transfer") {
            println!("{:?}", builder.transition_type())
        }
    }

    /// The descendant guard in `maybe_update_ops_as_valid` makes sure that a
    /// subtree reachable through multiple revalidated parents is walked only
    /// once. With a chain of k diamonds (op splitting to two ops merging back
    /// into one) the unguarded walk visits the tail of the chain O(2^k)
    /// times: with k = 64 it never terminates, so a removed guard shows up
    /// here as a watchdog timeout.
    #[test]
    fn maybe_update_ops_as_valid_diamond_chain() {
        use std::sync::mpsc;
        use std::time::Duration;

        use amplify::confinement::{NonEmptyOrdMap, NonEmptyOrdSet, NonEmptyVec};
        use rgb::bitcoin::hashes::Hash;
        use rgb::{Inputs, TransitionBundle};
        use strict_encoding::StrictDumb;

        const DIAMONDS: usize = 64;

        let (tx, rx) = mpsc::channel();
        let handle = std::thread::spawn(move || {
            let mut stock = Stock::in_memory();
            let contract_id =
                ContractId::from_baid64_str("rgb:qFuT6DN8-9AuO95M-7R8R8Mc-AZvs7zG-obum1Va-BRnweKk")
                    .unwrap();
            let witness_id = Txid::from_byte_array([0xCE; 32]);
            let ty = AssignmentType::strict_dumb();

            let make_op = |parents: &[OpId], nonce: u64| -> Transition {
                let mut transition = Transition::strict_dumb();
                if !parents.is_empty() {
                    transition.inputs = Inputs::from(NonEmptyOrdSet::from_checked(
                        parents.iter().map(|p| Opout::new(*p, ty, 0)).collect(),
                    ));
                }
                transition.nonce = nonce;
                transition
            };

            let mut all_opids = bset![];
            let mut register = |stock: &mut Stock, transition: Transition| -> (OpId, BundleId) {
                let opid = transition.id();
                let input_map = NonEmptyOrdMap::from_checked(
                    transition
                        .inputs
                        .iter()
                        .map(|input| (*input, opid))
                        .collect(),
                );
                let bundle = TransitionBundle {
                    input_map,
                    known_transitions: NonEmptyVec::with(KnownTransition::new(opid, transition)),
                };
                let bundle_id = bundle.bundle_id();
                stock.stash.consume_bundle(bundle.clone()).unwrap();
                stock
                    .index
                    .index_bundle(contract_id, &bundle, witness_id)
                    .unwrap();
                // everything starts as invalid
                stock.state.update_op(opid, false).unwrap();
                all_opids.insert(opid);
                (opid, bundle_id)
            };

            // root op; its dumb input plays the role of genesis
            let (root_opid, root_bundle_id) = register(&mut stock, make_op(&[], 0));
            let mut prev = root_opid;
            for _ in 0..DIAMONDS {
                let (left, _) = register(&mut stock, make_op(&[prev], 1));
                let (right, _) = register(&mut stock, make_op(&[prev], 2));
                let (join, _) = register(&mut stock, make_op(&[left, right], 0));
                prev = join;
            }

            // revalidate the whole graph starting from the root, as if all
            // the witnesses became valid again
            let mut invalid_ops = stock.as_state_provider().invalid_ops();
            let mut maybe_became_valid_opids = all_opids;
            let witnesses = bmap! { witness_id => WitnessOrd::Tentative };
            let valid = stock
                .maybe_update_ops_as_valid(
                    root_opid,
                    root_bundle_id,
                    &mut invalid_ops,
                    &mut maybe_became_valid_opids,
                    &witnesses,
                )
                .unwrap();

            assert!(valid);
            assert!(stock.as_state_provider().invalid_ops().is_empty());
            tx.send(()).unwrap();
        });

        match rx.recv_timeout(Duration::from_secs(60)) {
            Ok(()) => handle.join().unwrap(),
            Err(mpsc::RecvTimeoutError::Timeout) => {
                panic!("revalidation did not terminate: descendant guard not working")
            }
            // the worker thread panicked: propagate its panic
            Err(mpsc::RecvTimeoutError::Disconnected) => handle.join().unwrap(),
        }
    }
}