iota_graphql_rpc/types/

object.rs

// Copyright (c) Mysten Labs, Inc.
// Modifications Copyright (c) 2024 IOTA Stiftung
// SPDX-License-Identifier: Apache-2.0

use std::{
    collections::{BTreeMap, BTreeSet, HashMap},
    fmt::Write,
};

use async_graphql::{
    connection::{Connection, CursorType, Edge},
    dataloader::Loader,
    *,
};
use diesel::{BoolExpressionMethods, ExpressionMethods, QueryDsl, SelectableHelper, sql_types};
use iota_indexer::{
    ingestion::common::CommitterTables,
    models::objects::{StoredHistoryObject, StoredObject},
    schema::objects,
    types::{ObjectStatus as NativeObjectStatus, OwnerType},
};
use iota_sdk_types::{
    MoveStruct as NativeMoveStruct, ObjectId, Owner as NativeOwner, StructTag, TypeTag, Version,
};
use iota_types::object::{Object as NativeObject, bounded_visitor::BoundedVisitor};
use move_core_types::annotated_value::{MoveStruct, MoveTypeLayout};
use serde::{Deserialize, Serialize};

use crate::{
    backward_view::{HistoricalFilter, consistent, historical},
    config::DEFAULT_PAGE_SIZE,
    connection::ScanConnection,
    consistency::Checkpointed,
    data::{DataLoader, Db, DbConnection, QueryExecutor, package_resolver::PackageResolver},
    error::Error,
    filter, or_filter,
    raw_query::RawQuery,
    types::{
        available_range::AvailableRange,
        balance::{self, Balance},
        base64::Base64,
        big_int::BigInt,
        coin::Coin,
        coin_metadata::CoinMetadata,
        cursor::{self, Page, RawPaginated, ScanLimited, Target},
        digest::Digest,
        display::{Display, DisplayEntry},
        dynamic_field::{DynamicField, DynamicFieldName},
        intersect,
        iota_address::{IotaAddress, addr},
        iota_names_registration::{NameFormat, NameRegistration},
        move_object::MoveObject,
        move_package::MovePackage,
        owner::{Owner, OwnerImpl},
        stake::StakedIota,
        transaction_block,
        transaction_block::{TransactionBlock, TransactionBlockFilter},
        type_filter::{ExactTypeFilter, TypeFilter},
        uint53::UInt53,
    },
};

#[derive(Clone, Debug)]
pub(crate) struct Object {
    pub address: IotaAddress,
    pub inner: ActiveObject,
    /// The checkpoint sequence number at which this was viewed at.
    pub checkpoint_viewed_at: u64,
    /// Root parent object version for dynamic fields.
    ///
    /// This enables consistent dynamic field reads in the case of chained
    /// dynamic object fields, e.g., `Parent -> DOF1 -> DOF2`. In such
    /// cases, the object versions may end up like `Parent >= DOF1, DOF2`
    /// but `DOF1 < DOF2`. Thus, database queries for dynamic fields must
    /// bound the object versions by the version of the root object of the tree.
    ///
    /// Essentially, lamport timestamps of objects are updated for all top-level
    /// mutable objects provided as inputs to a transaction as well as any
    /// mutated dynamic child objects. However, any dynamic child objects
    /// that were loaded but not actually mutated don't end up having
    /// their versions updated.
    root_version: u64,
}

/// Type to implement GraphQL fields that are shared by all Objects.
pub(crate) struct ObjectImpl<'o>(pub &'o Object);

#[derive(Clone, Debug)]
pub(crate) struct ActiveObject {
    /// The deserialized object.
    native: NativeObject,
    /// Where the object's state was read from.
    status: ObjectStatus,
    /// The serialized object as stored in the index. `None` for `NotIndexed`
    /// objects.
    ///
    /// Avoids the re-serialization of `native` for `Indexed` objects.
    bcs: Option<Vec<u8>>,
}

#[derive(Enum, Copy, Clone, Eq, PartialEq, Debug)]
#[graphql(name = "ObjectKind")]
pub enum ObjectStatus {
    /// The object is loaded from serialized data, such as the contents of a
    /// transaction that hasn't been indexed yet.
    NotIndexed,
    /// The object is fetched from the index.
    Indexed,
}

#[derive(Clone, Debug, PartialEq, Eq, InputObject)]
pub(crate) struct ObjectRef {
    /// ID of the object.
    pub address: IotaAddress,
    /// Version or sequence number of the object.
    pub version: UInt53,
    /// Digest of the object.
    pub digest: Digest,
}

/// Constrains the set of objects returned. All filters are optional, and the
/// resulting set of objects are ones whose
///
/// - Type matches the `type` filter,
/// - AND, whose owner matches the `owner` filter,
/// - AND, whose ID is in `objectIds` OR whose ID and version is in
///   `objectKeys`.
#[derive(InputObject, Default, Debug, Clone, Eq, PartialEq)]
pub(crate) struct ObjectFilter {
    /// Filter objects by their type's `package`, `package::module`, or their
    /// fully qualified type name.
    ///
    /// Generic types can be queried by either the generic type name, e.g.
    /// `0x2::coin::Coin`, or by the full type name, such as
    /// `0x2::coin::Coin<0x2::iota::IOTA>`.
    pub type_: Option<TypeFilter>,

    /// Filter for live objects by their current owners.
    pub owner: Option<IotaAddress>,

    /// Filter for live objects by their IDs.
    pub object_ids: Option<Vec<IotaAddress>>,

    /// Filter for live or potentially historical objects by their ID and
    /// version.
    pub object_keys: Option<Vec<ObjectKey>>,
}

#[derive(InputObject, Debug, Clone, Eq, PartialEq)]
pub(crate) struct ObjectKey {
    pub object_id: IotaAddress,
    pub version: UInt53,
}

/// The object's owner type: Immutable, Shared, Parent, or Address.
#[derive(Union, Clone)]
pub(crate) enum ObjectOwner {
    Immutable(Immutable),
    Shared(Shared),
    Parent(Box<Parent>),
    Address(AddressOwner),
}

/// An immutable object is an object that can't be mutated, transferred, or
/// deleted. Immutable objects have no owner, so anyone can use them.
#[derive(SimpleObject, Clone)]
pub(crate) struct Immutable {
    #[graphql(name = "_")]
    dummy: Option<bool>,
}

/// A shared object is an object that is shared using the
/// 0x2::transfer::share_object function. Unlike owned objects, once an object
/// is shared, it stays mutable and is accessible by anyone.
#[derive(SimpleObject, Clone)]
pub(crate) struct Shared {
    initial_shared_version: UInt53,
}

/// If the object's owner is a Parent, this object is part of a dynamic field
/// (it is the value of the dynamic field, or the intermediate Field object
/// itself). Also note that if the owner is a parent, then it's guaranteed to be
/// an object.
#[derive(SimpleObject, Clone)]
pub(crate) struct Parent {
    parent: Option<Object>,
}

/// An address-owned object is owned by a specific 32-byte address that is
/// either an account address (derived from a particular signature scheme) or
/// an object ID. An address-owned object is accessible only to its owner and no
/// others.
#[derive(SimpleObject, Clone)]
pub(crate) struct AddressOwner {
    owner: Option<Owner>,
}

/// Filter for a point query of an Object.
pub(crate) enum ObjectLookup {
    LatestAt {
        /// The checkpoint sequence number at which this was viewed at.
        checkpoint_viewed_at: u64,
    },

    UnderParent {
        /// The parent version to be used as an upper bound for the query. Look
        /// for the latest version of a child object whose version is
        /// less than or equal to this upper bound.
        parent_version: u64,
        /// The checkpoint sequence number at which this was viewed at.
        checkpoint_viewed_at: u64,
    },

    VersionAt {
        /// The exact version of the object to be fetched.
        version: u64,
        /// The checkpoint sequence number at which this was viewed at.
        checkpoint_viewed_at: u64,
    },

    /// Variant analogous to [`VersionAt`](Self::VersionAt) but for optimistic
    /// transactions, using the most recent not checkpointed data,
    /// not bound by any checkpoint sequence number.
    OptimisticVersion {
        /// The exact version of the object to be fetched.
        version: u64,
    },
}

pub(crate) type Cursor = cursor::BcsCursor<HistoricalObjectCursor>;

/// The inner struct for the `Object`'s cursor. The `object_id` is used as the
/// cursor, while the `checkpoint_viewed_at` sets the consistent upper bound for
/// the cursor.
#[derive(Serialize, Deserialize, Clone, PartialEq, Eq, Debug)]
pub(crate) struct HistoricalObjectCursor {
    #[serde(rename = "o")]
    object_id: Vec<u8>,
    /// The checkpoint sequence number this was viewed at.
    #[serde(rename = "c")]
    checkpoint_viewed_at: u64,
}

/// Interface implemented by on-chain values that are addressable by an ID (also
/// referred to as its address). This includes Move objects and packages.
#[expect(clippy::duplicated_attributes)]
#[derive(Interface)]
#[graphql(
    name = "IObject",
    field(name = "version", ty = "UInt53"),
    field(
        name = "status",
        ty = "ObjectStatus",
        desc = r#"
            The current status of the object as read from the off-chain store. The
            possible states are:
            - NOT_INDEXED: The object is loaded from serialized data, such as the
            contents of a genesis or system package upgrade transaction.
            - INDEXED: The object is retrieved from the off-chain index and
            represents the most recent or historical state of the object.
        "#
    ),
    field(
        name = "digest",
        ty = "Option<String>",
        desc = "32-byte hash that identifies the object's current contents, encoded as a Base58 \
                string."
    ),
    field(
        name = "owner",
        ty = "Option<ObjectOwner>",
        desc = "The owner type of this object: Immutable, Shared, Parent, Address\n\
                Immutable and Shared Objects do not have owners."
    ),
    field(
        name = "previous_transaction_block",
        ty = "Option<TransactionBlock>",
        desc = "The transaction block that created this version of the object."
    ),
    field(name = "storage_rebate", ty = "Option<BigInt>", desc = "",),
    field(
        name = "received_transaction_blocks",
        arg(name = "first", ty = "Option<u64>"),
        arg(name = "after", ty = "Option<transaction_block::Cursor>"),
        arg(name = "last", ty = "Option<u64>"),
        arg(name = "before", ty = "Option<transaction_block::Cursor>"),
        arg(name = "filter", ty = "Option<TransactionBlockFilter>"),
        arg(
            name = "scan_limit",
            ty = "Option<u64>",
            deprecation = "`scanLimit` will be removed with v1.38, along with the support for combining complex filters."
        ),
        ty = "ScanConnection<String, TransactionBlock>",
        desc = "The transaction blocks that sent objects to this object."
    ),
    field(
        name = "bcs",
        ty = "Option<Base64>",
        desc = "The Base64-encoded BCS serialization of the object's content."
    )
)]
pub(crate) enum IObject {
    Object(Object),
    MovePackage(MovePackage),
    MoveObject(MoveObject),
    Coin(Coin),
    CoinMetadata(CoinMetadata),
    StakedIota(StakedIota),
}

/// `DataLoader` key for fetching an `Object` at a specific version, constrained
/// by a consistency cursor (if that version was created after the checkpoint
/// the query is viewing at, then it will fail).
#[derive(Copy, Clone, Hash, Eq, PartialEq, Debug)]
struct HistoricalKey {
    id: IotaAddress,
    version: u64,
    checkpoint_viewed_at: u64,
}

/// `DataLoader` key for fetching objects that haven't been checkpointed yet.
/// This is used specifically for loading objects
/// that are part of optimistic transaction effects.
#[derive(Copy, Clone, Hash, Eq, PartialEq, Debug)]
struct OptimisticKey {
    id: IotaAddress,
    version: u64,
}

/// `DataLoader` key for fetching the latest version of an object whose parent
/// object has version `parent_version`, as of `checkpoint_viewed_at`. This
/// look-up can fail to find a valid object if the key is not self-consistent,
/// for example if the `parent_version` is set to a higher version
/// than the object's actual parent as of `checkpoint_viewed_at`.
#[derive(Copy, Clone, Hash, Eq, PartialEq, Debug)]
struct ParentVersionKey {
    id: IotaAddress,
    parent_version: u64,
    checkpoint_viewed_at: u64,
}

/// `DataLoader` key for fetching the latest version of an object as of a given
/// checkpoint.
#[derive(Copy, Clone, Hash, Eq, PartialEq, Debug)]
struct LatestAtKey {
    id: IotaAddress,
    checkpoint_viewed_at: u64,
}

/// An object in IOTA is a package (set of Move bytecode modules) or object
/// (typed data structure with fields) with additional metadata detailing its
/// id, version, transaction digest, owner field indicating how this object can
/// be accessed.
#[Object]
impl Object {
    pub(crate) async fn address(&self) -> IotaAddress {
        OwnerImpl::from(self).address().await
    }

    /// Objects owned by this object, optionally `filter`-ed.
    pub(crate) async fn objects(
        &self,
        ctx: &Context<'_>,
        first: Option<u64>,
        after: Option<Cursor>,
        last: Option<u64>,
        before: Option<Cursor>,
        filter: Option<ObjectFilter>,
    ) -> Result<Connection<String, MoveObject>> {
        OwnerImpl::from(self)
            .objects(ctx, first, after, last, before, filter)
            .await
    }

    /// Total balance of all coins with marker type owned by this object. If
    /// type is not supplied, it defaults to `0x2::iota::IOTA`.
    pub(crate) async fn balance(
        &self,
        ctx: &Context<'_>,
        type_: Option<ExactTypeFilter>,
    ) -> Result<Option<Balance>> {
        OwnerImpl::from(self).balance(ctx, type_).await
    }

    /// The balances of all coin types owned by this object.
    pub(crate) async fn balances(
        &self,
        ctx: &Context<'_>,
        first: Option<u64>,
        after: Option<balance::Cursor>,
        last: Option<u64>,
        before: Option<balance::Cursor>,
    ) -> Result<Connection<String, Balance>> {
        OwnerImpl::from(self)
            .balances(ctx, first, after, last, before)
            .await
    }

    /// The coin objects for this object.
    ///
    /// `type` is a filter on the coin's type parameter, defaulting to
    /// `0x2::iota::IOTA`.
    pub(crate) async fn coins(
        &self,
        ctx: &Context<'_>,
        first: Option<u64>,
        after: Option<Cursor>,
        last: Option<u64>,
        before: Option<Cursor>,
        type_: Option<ExactTypeFilter>,
    ) -> Result<Connection<String, Coin>> {
        OwnerImpl::from(self)
            .coins(ctx, first, after, last, before, type_)
            .await
    }

    /// The `0x3::staking_pool::StakedIota` objects owned by this object.
    pub(crate) async fn staked_iotas(
        &self,
        ctx: &Context<'_>,
        first: Option<u64>,
        after: Option<Cursor>,
        last: Option<u64>,
        before: Option<Cursor>,
    ) -> Result<Connection<String, StakedIota>> {
        OwnerImpl::from(self)
            .staked_iotas(ctx, first, after, last, before)
            .await
    }

    /// The name explicitly configured as the default name pointing to this
    /// address.
    pub(crate) async fn iota_names_default_name(
        &self,
        ctx: &Context<'_>,
        format: Option<NameFormat>,
    ) -> Result<Option<String>> {
        OwnerImpl::from(self)
            .iota_names_default_name(ctx, format)
            .await
    }

    /// The NameRegistration NFTs owned by this address. These grant the
    /// owner the capability to manage the associated name.
    pub(crate) async fn iota_names_registrations(
        &self,
        ctx: &Context<'_>,
        first: Option<u64>,
        after: Option<Cursor>,
        last: Option<u64>,
        before: Option<Cursor>,
    ) -> Result<Connection<String, NameRegistration>> {
        OwnerImpl::from(self)
            .iota_names_registrations(ctx, first, after, last, before)
            .await
    }

    pub(crate) async fn version(&self) -> UInt53 {
        ObjectImpl(self).version().await
    }

    /// The current status of the object as read from the off-chain store. The
    /// possible states are:
    /// - NOT_INDEXED: The object is loaded from serialized data, such as the
    ///   contents of a genesis or system package upgrade transaction.
    /// - INDEXED: The object is retrieved from the off-chain index and
    ///   represents the most recent or historical state of the object.
    pub(crate) async fn status(&self) -> ObjectStatus {
        ObjectImpl(self).status().await
    }

    /// 32-byte hash that identifies the object's current contents, encoded as a
    /// Base58 string.
    pub(crate) async fn digest(&self) -> Option<String> {
        ObjectImpl(self).digest().await
    }

    /// The owner type of this object: Immutable, Shared, Parent, Address
    /// Immutable and Shared Objects do not have owners.
    pub(crate) async fn owner(&self, ctx: &Context<'_>) -> Option<ObjectOwner> {
        ObjectImpl(self).owner(ctx).await
    }

    /// The transaction block that created this version of the object.
    pub(crate) async fn previous_transaction_block(
        &self,
        ctx: &Context<'_>,
    ) -> Result<Option<TransactionBlock>> {
        ObjectImpl(self).previous_transaction_block(ctx).await
    }

    /// The amount of IOTA we would rebate if this object gets deleted or
    /// mutated. This number is recalculated based on the present storage
    /// gas price.
    pub(crate) async fn storage_rebate(&self) -> Option<BigInt> {
        ObjectImpl(self).storage_rebate().await
    }

    /// The transaction blocks that sent objects to this object.
    ///
    /// `scanLimit` restricts the number of candidate transactions scanned when
    /// gathering a page of results. It is required for queries that apply two
    /// or more complex filters (on function, affected address, recipient, input
    /// object, changed object, or wrapped or deleted object), and can be at
    /// most `serviceConfig.maxScanLimit`. A `kind` filter cannot be
    /// combined with any of them.
    ///
    /// When the scan limit is reached the page will be returned even if it has
    /// fewer than `first` results when paginating forward (`last` when
    /// paginating backwards). If there are more transactions to scan,
    /// `pageInfo.hasNextPage` (or `pageInfo.hasPreviousPage`) will be set to
    /// `true`, and `PageInfo.endCursor` (or `PageInfo.startCursor`) will be set
    /// to the last transaction that was scanned as opposed to the last (or
    /// first) transaction in the page.
    ///
    /// Requesting the next (or previous) page after this cursor will resume the
    /// search, scanning the next `scanLimit` many transactions in the
    /// direction of pagination, and so on until all transactions in the
    /// scanning range have been visited.
    ///
    /// By default, the scanning range includes all transactions known to
    /// GraphQL, but it can be restricted by the `after` and `before`
    /// cursors, and the `beforeCheckpoint`, `afterCheckpoint` and
    /// `atCheckpoint` filters.
    ///
    /// DEPRECATION NOTICE: Support for the combination of two or more complex
    /// filters as discussed above will stop with the v1.38 release. `scanLimit`
    /// will thus become obsolete and will be removed as well.
    #[graphql(
        complexity = "first.or(last).unwrap_or(DEFAULT_PAGE_SIZE as u64) as usize * child_complexity"
    )]
    pub(crate) async fn received_transaction_blocks(
        &self,
        ctx: &Context<'_>,
        first: Option<u64>,
        after: Option<transaction_block::Cursor>,
        last: Option<u64>,
        before: Option<transaction_block::Cursor>,
        filter: Option<TransactionBlockFilter>,
        #[graphql(
            deprecation = "`scanLimit` will be removed with v1.38, along with the support for combining complex filters."
        )]
        scan_limit: Option<u64>,
    ) -> Result<ScanConnection<String, TransactionBlock>> {
        ObjectImpl(self)
            .received_transaction_blocks(ctx, first, after, last, before, filter, scan_limit)
            .await
    }

    /// The Base64-encoded BCS serialization of the object's content.
    pub(crate) async fn bcs(&self) -> Result<Option<Base64>> {
        ObjectImpl(self).bcs().await
    }

    /// The set of named templates defined on-chain for the type of this object,
    /// to be handled off-chain. The server substitutes data from the object
    /// into these templates to generate a display string per template.
    async fn display(&self, ctx: &Context<'_>) -> Result<Option<Vec<DisplayEntry>>> {
        ObjectImpl(self).display(ctx).await
    }

    /// Access a dynamic field on an object using its name. Names are arbitrary
    /// Move values whose type have `copy`, `drop`, and `store`, and are
    /// specified using their type, and their BCS contents, Base64 encoded.
    ///
    /// Dynamic fields on wrapped objects can be accessed by using the same API
    /// under the Owner type.
    async fn dynamic_field(
        &self,
        ctx: &Context<'_>,
        name: DynamicFieldName,
    ) -> Result<Option<DynamicField>> {
        OwnerImpl::from(self)
            .dynamic_field(ctx, name, Some(self.root_version()))
            .await
    }

    /// Access a dynamic object field on an object using its name. Names are
    /// arbitrary Move values whose type have `copy`, `drop`, and `store`,
    /// and are specified using their type, and their BCS contents, Base64
    /// encoded. The value of a dynamic object field can also be accessed
    /// off-chain directly via its address (e.g. using `Query.object`).
    ///
    /// Dynamic fields on wrapped objects can be accessed by using the same API
    /// under the Owner type.
    async fn dynamic_object_field(
        &self,
        ctx: &Context<'_>,
        name: DynamicFieldName,
    ) -> Result<Option<DynamicField>> {
        OwnerImpl::from(self)
            .dynamic_object_field(ctx, name, Some(self.root_version()))
            .await
    }

    /// The dynamic fields and dynamic object fields on an object.
    ///
    /// Dynamic fields on wrapped objects can be accessed by using the same API
    /// under the Owner type.
    async fn dynamic_fields(
        &self,
        ctx: &Context<'_>,
        first: Option<u64>,
        after: Option<Cursor>,
        last: Option<u64>,
        before: Option<Cursor>,
    ) -> Result<Connection<String, DynamicField>> {
        OwnerImpl::from(self)
            .dynamic_fields(ctx, first, after, last, before, Some(self.root_version()))
            .await
    }

    /// Attempts to convert the object into a MoveObject
    async fn as_move_object(&self) -> Option<MoveObject> {
        MoveObject::try_from(self).ok()
    }

    /// Attempts to convert the object into a MovePackage
    async fn as_move_package(&self) -> Option<MovePackage> {
        MovePackage::try_from(self).ok()
    }
}

impl ObjectImpl<'_> {
    pub(crate) async fn version(&self) -> UInt53 {
        self.0.version_impl().into()
    }

    pub(crate) async fn status(&self) -> ObjectStatus {
        ObjectStatus::from(&self.0.inner)
    }

    pub(crate) async fn digest(&self) -> Option<String> {
        Some(self.0.native_impl().digest().to_base58())
    }

    pub(crate) async fn owner(&self, ctx: &Context<'_>) -> Option<ObjectOwner> {
        use NativeOwner as O;

        let native = self.0.native_impl();

        match native.owner {
            O::Address(address) => {
                let address = IotaAddress::from(address);
                Some(ObjectOwner::Address(AddressOwner {
                    owner: Some(Owner {
                        address,
                        checkpoint_viewed_at: self.0.checkpoint_viewed_at,
                        root_version: None,
                    }),
                }))
            }
            O::Immutable => Some(ObjectOwner::Immutable(Immutable { dummy: None })),
            O::Object(address) => {
                let parent = Object::query(
                    ctx,
                    address.into(),
                    Object::under_parent(self.0.root_version, self.0.checkpoint_viewed_at),
                )
                .await
                .ok()
                .flatten();

                Some(ObjectOwner::Parent(Box::new(Parent { parent })))
            }
            O::Shared(initial_shared_version) => Some(ObjectOwner::Shared(Shared {
                initial_shared_version: initial_shared_version.as_u64().into(),
            })),
            _ => unimplemented!("a new Owner enum variant was added and needs to be handled"),
        }
    }

    pub(crate) async fn previous_transaction_block(
        &self,
        ctx: &Context<'_>,
    ) -> Result<Option<TransactionBlock>> {
        let native = self.0.native_impl();
        let digest = native.previous_transaction;
        let key = transaction_block::DigestKey::new(digest.into(), self.0.checkpoint_viewed_at);

        TransactionBlock::query(ctx, key).await.extend()
    }

    pub(crate) async fn storage_rebate(&self) -> Option<BigInt> {
        Some(self.0.native_impl().storage_rebate.into())
    }

    pub(crate) async fn received_transaction_blocks(
        &self,
        ctx: &Context<'_>,
        first: Option<u64>,
        after: Option<transaction_block::Cursor>,
        last: Option<u64>,
        before: Option<transaction_block::Cursor>,
        filter: Option<TransactionBlockFilter>,
        scan_limit: Option<u64>,
    ) -> Result<ScanConnection<String, TransactionBlock>> {
        let page = Page::from_params(ctx.data_unchecked(), first, after, last, before)?;

        let Some(filter) = filter
            .unwrap_or_default()
            .intersect(TransactionBlockFilter {
                recv_address: Some(self.0.address),
                ..Default::default()
            })
        else {
            return Ok(ScanConnection::new(false, false));
        };

        TransactionBlock::paginate(ctx, page, filter, self.0.checkpoint_viewed_at, scan_limit)
            .await
            .extend()
    }

    pub(crate) async fn bcs(&self) -> Result<Option<Base64>> {
        let inner = &self.0.inner;
        Ok(match &inner.bcs {
            Some(serialized) => Some(Base64::from(serialized)),

            None => {
                let bytes = bcs::to_bytes(&inner.native)
                    .map_err(|e| {
                        Error::Internal(format!(
                            "Failed to serialize object at {}: {e}",
                            self.0.address
                        ))
                    })
                    .extend()?;
                Some(Base64::from(&bytes))
            }
        })
    }

    /// `display` is part of the `IMoveObject` interface, but is implemented on
    /// `ObjectImpl` to allow for a convenience function on `Object`.
    pub(crate) async fn display(&self, ctx: &Context<'_>) -> Result<Option<Vec<DisplayEntry>>> {
        let native = self.0.native_impl();

        let move_object = native
            .data
            .as_opt_struct()
            .ok_or_else(|| Error::Internal("Failed to convert object into MoveObject".to_string()))
            .extend()?;

        let (struct_tag, move_struct) = deserialize_move_struct(move_object, ctx.data_unchecked())
            .await
            .extend()?;

        let Some(display) = Display::query(ctx.data_unchecked(), struct_tag.into())
            .await
            .extend()?
        else {
            return Ok(None);
        };

        Ok(Some(display.render(&move_struct).extend()?))
    }
}

impl Object {
    /// Construct a GraphQL object from a native object, without its stored
    /// (indexed) counterpart.
    ///
    /// `checkpoint_viewed_at` represents the checkpoint sequence number at
    /// which this `Object` was constructed in. This is stored on `Object`
    /// so that when viewing that entity's state, it will be as if it was
    /// read at the same checkpoint.
    ///
    /// `root_version` represents the version of the root object in some nested
    /// chain of dynamic fields. This should typically be left `None`,
    /// unless the object(s) being resolved is a dynamic field, or if
    /// `root_version` has been explicitly set for this object. If None, then
    /// we use [`version_for_dynamic_fields`] to infer a root version to then
    /// propagate from this object down to its dynamic fields.
    pub(crate) fn from_native(
        address: IotaAddress,
        native: NativeObject,
        checkpoint_viewed_at: u64,
        root_version: Option<u64>,
    ) -> Object {
        let root_version = root_version.unwrap_or_else(|| version_for_dynamic_fields(&native));
        Object {
            address,
            inner: ActiveObject {
                native,
                status: ObjectStatus::NotIndexed,
                bcs: None,
            },
            checkpoint_viewed_at,
            root_version,
        }
    }

    pub(crate) fn native_impl(&self) -> &NativeObject {
        &self.inner.native
    }

    pub(crate) fn version_impl(&self) -> u64 {
        self.native_impl().version().as_u64()
    }

    /// Root parent object version for dynamic fields.
    ///
    /// Check [`Object::root_version`] for details.
    pub(crate) fn root_version(&self) -> u64 {
        self.root_version
    }

    /// Query the database for a `page` of objects, optionally `filter`-ed.
    ///
    /// `checkpoint_viewed_at` represents the checkpoint sequence number at
    /// which this page was queried for. Each entity returned in the
    /// connection will inherit this checkpoint, so that when viewing that
    /// entity's state, it will be as if it was read at the same checkpoint.
    pub(crate) async fn paginate(
        db: &Db,
        page: Page<Cursor>,
        filter: ObjectFilter,
        checkpoint_viewed_at: u64,
    ) -> Result<Connection<String, Object>, Error> {
        Self::paginate_subtype(db, page, filter, checkpoint_viewed_at, Ok).await
    }

    /// Query the database for a `page` of some sub-type of Object. The page
    /// uses the bytes of an Object ID and the checkpoint when the query was
    /// made as the cursor, and can optionally be further `filter`-ed. The
    /// subtype is created using the `downcast` function, which is allowed
    /// to fail, if the downcast has failed.
    ///
    /// `checkpoint_viewed_at` represents the checkpoint sequence number at
    /// which this page was queried for. Each entity returned in the
    /// connection will inherit this checkpoint, so that when viewing that
    /// entity's state, it will be as if it was read at the same checkpoint.
    ///
    /// If a `Page<Cursor>` is also provided, then this function will defer to
    /// the `checkpoint_viewed_at` in the cursors. Otherwise, use the value
    /// from the parameter, or set to None. This is so that paginated
    /// queries are consistent with the previous query that created the
    /// cursor.
    pub(crate) async fn paginate_subtype<T: OutputType>(
        db: &Db,
        page: Page<Cursor>,
        filter: ObjectFilter,
        checkpoint_viewed_at: u64,
        downcast: impl Fn(Object) -> Result<T, Error>,
    ) -> Result<Connection<String, T>, Error> {
        // If cursors are provided, defer to the `checkpoint_viewed_at` in the cursor if
        // they are consistent. Otherwise, use the value from the parameter, or
        // set to None. This is so that paginated queries are consistent with
        // the previous query that created the cursor.
        let cursor_viewed_at = page.validate_cursor_consistency()?;
        let checkpoint_viewed_at = cursor_viewed_at.unwrap_or(checkpoint_viewed_at);

        let max_available_range = db.max_available_range;

        let Some((prev, next, results)) = db
            .execute_repeatable(move |conn| {
                if !AvailableRange::is_checkpoint_in_backward_history_range(
                    conn,
                    checkpoint_viewed_at,
                    max_available_range,
                )? {
                    return Ok::<_, diesel::result::Error>(None);
                };

                let (prev, next, results_iter) = page.paginate_raw_query::<StoredBackwardObject>(
                    conn,
                    checkpoint_viewed_at,
                    backward_objects_query(&filter, checkpoint_viewed_at, &page),
                )?;
                let results: Vec<StoredBackwardObject> = results_iter.collect();
                Ok(Some((prev, next, results)))
            })
            .await?
        else {
            return Err(Error::Client(
                "Requested data is outside the available range".to_string(),
            ));
        };

        let mut conn: Connection<String, T> = Connection::new(prev, next);

        for stored in results {
            // To maintain consistency, the returned cursor should have the same upper-bound
            // as the checkpoint found on the cursor.
            let cursor = stored.cursor(checkpoint_viewed_at).encode_cursor();
            let stored_history = stored.into_stored_history(checkpoint_viewed_at);
            let active_object = ActiveObject::try_from(stored_history)?;
            let object = Object::from_active_object(active_object, checkpoint_viewed_at, None);
            conn.edges.push(Edge::new(cursor, downcast(object)?));
        }

        Ok(conn)
    }

    /// Look-up the latest version of the object as of a given checkpoint.
    pub(crate) fn latest_at(checkpoint_viewed_at: u64) -> ObjectLookup {
        ObjectLookup::LatestAt {
            checkpoint_viewed_at,
        }
    }

    /// Look-up the latest version of an object whose version is less than or
    /// equal to its parent's version, as of a given checkpoint.
    pub(crate) fn under_parent(parent_version: u64, checkpoint_viewed_at: u64) -> ObjectLookup {
        ObjectLookup::UnderParent {
            parent_version,
            checkpoint_viewed_at,
        }
    }

    /// Look-up a specific version of the object, as of a given checkpoint.
    pub(crate) fn at_version(version: u64, checkpoint_viewed_at: u64) -> ObjectLookup {
        ObjectLookup::VersionAt {
            version,
            checkpoint_viewed_at,
        }
    }

    /// Look-up a specific version of the object from optimistic transactions.
    pub(crate) fn at_optimistic_version(version: u64) -> ObjectLookup {
        ObjectLookup::OptimisticVersion { version }
    }

    pub(crate) async fn query(
        ctx: &Context<'_>,
        id: IotaAddress,
        key: ObjectLookup,
    ) -> Result<Option<Self>, Error> {
        let DataLoader(loader) = &ctx.data_unchecked();

        match key {
            ObjectLookup::VersionAt {
                version,
                checkpoint_viewed_at,
            } => loader
                .load_one(HistoricalKey {
                    id,
                    version,
                    checkpoint_viewed_at,
                })
                .await?
                .transpose(),

            ObjectLookup::OptimisticVersion { version } => loader
                .load_one(OptimisticKey { id, version })
                .await?
                .transpose(),

            ObjectLookup::UnderParent {
                parent_version,
                checkpoint_viewed_at,
            } => loader
                .load_one(ParentVersionKey {
                    id,
                    parent_version,
                    checkpoint_viewed_at,
                })
                .await?
                .transpose(),

            ObjectLookup::LatestAt {
                checkpoint_viewed_at,
            } => {
                loader
                    .load_one(LatestAtKey {
                        id,
                        checkpoint_viewed_at,
                    })
                    .await
            }
        }
    }

    /// Query for a singleton object identified by its type. Note: the object is
    /// assumed to be a singleton (we either find at least one object with
    /// this type and then return it, or return nothing).
    pub(crate) async fn query_singleton(
        db: &Db,
        type_: StructTag,
        checkpoint_viewed_at: u64,
    ) -> Result<Option<Object>, Error> {
        let filter = ObjectFilter {
            type_: Some(TypeFilter::ByType(type_)),
            ..Default::default()
        };

        let connection = Self::paginate(db, Page::bounded(1), filter, checkpoint_viewed_at).await?;

        Ok(connection.edges.into_iter().next().map(|edge| edge.node))
    }

    /// Builds an `Object` from an active object, viewed at
    /// `checkpoint_viewed_at`.
    ///
    /// `checkpoint_viewed_at` represents the checkpoint sequence number at
    /// which this `Object` was constructed in. This is stored on `Object`
    /// so that when viewing that entity's state, it will be as if it was
    /// read at the same checkpoint.
    ///
    /// `root_version` represents the version of the root object in some nested
    /// chain of dynamic fields. This should typically be left `None`,
    /// unless the object(s) being resolved is a dynamic field, or if
    /// `root_version` has been explicitly set for this object. If None, then
    /// we use [`version_for_dynamic_fields`] to infer a root version to then
    /// propagate from this object down to its dynamic fields.
    pub(crate) fn from_active_object(
        active_object: ActiveObject,
        checkpoint_viewed_at: u64,
        root_version: Option<u64>,
    ) -> Self {
        let address = IotaAddress::from(active_object.native.id());
        let root_version =
            root_version.unwrap_or_else(|| version_for_dynamic_fields(&active_object.native));
        Self {
            address,
            inner: active_object,
            checkpoint_viewed_at,
            root_version,
        }
    }

    pub(crate) fn try_from_stored_object(
        stored_object: StoredObject,
        checkpoint_viewed_at: u64,
        root_version: Option<u64>,
    ) -> Result<Self, Error> {
        let active_object = ActiveObject {
            native: NativeObject::try_from(&stored_object)?,
            status: ObjectStatus::Indexed,
            bcs: Some(stored_object.serialized_object),
        };
        Ok(Self::from_active_object(
            active_object,
            checkpoint_viewed_at,
            root_version,
        ))
    }
}

impl TryFrom<StoredHistoryObject> for ActiveObject {
    type Error = Error;

    /// Builds a value from an active stored history row by
    /// deserializing its native object.
    ///
    /// # Errors
    ///
    /// Fails in the following cases:
    ///
    /// - The row is not active (a wrapped or deleted tombstone, or an
    ///   unrecognized status).
    /// - The row has no serialized object, or it fails to deserialize.
    fn try_from(stored: StoredHistoryObject) -> Result<Self, Self::Error> {
        if !matches!(
            NativeObjectStatus::try_from(stored.object_status),
            Ok(NativeObjectStatus::Active)
        ) {
            return Err(Error::Internal(format!(
                "Expected active object 0x{} at version {}, but found status {}",
                hex::encode(&stored.object_id),
                stored.object_version,
                stored.object_status,
            )));
        }

        let native = NativeObject::try_from(&stored)?;
        Ok(ActiveObject {
            native,
            status: ObjectStatus::Indexed,
            bcs: stored.serialized_object,
        })
    }
}

/// We're deliberately choosing to use a child object's version as the root
/// here, and letting the caller override it with the actual root object's
/// version if it has access to it.
///
/// Using the child object's version as the root means that we're seeing the
/// dynamic field tree under this object at the state resulting from the
/// transaction that produced this version.
///
/// See [`Object::root_version`] for more details on parent/child object version
/// mechanics.
fn version_for_dynamic_fields(native: &NativeObject) -> u64 {
    native.as_inner().version().as_u64()
}

impl ObjectFilter {
    /// Try to create a filter whose results are the intersection of objects in
    /// `self`'s results and objects in `other`'s results. This may not be
    /// possible if the resulting filter is inconsistent in some way (e.g. a
    /// filter that requires one field to be two different values
    /// simultaneously).
    pub(crate) fn intersect(self, other: ObjectFilter) -> Option<Self> {
        macro_rules! intersect {
            ($field:ident, $body:expr) => {
                intersect::field(self.$field, other.$field, $body)
            };
        }

        // Treat `object_ids` and `object_keys` as a single filter on IDs, and
        // optionally versions, and compute the intersection of that.
        let keys = intersect::field(self.keys(), other.keys(), |k, l| {
            let mut combined = BTreeMap::new();

            for (id, v) in k {
                if let Some(w) = l.get(&id).copied() {
                    combined.insert(id, intersect::field(v, w, intersect::by_eq)?);
                }
            }

            // If the intersection is empty, it means, there were some ID or Key filters in
            // both `self` and `other`, but they don't overlap, so the final
            // result is inconsistent.
            (!combined.is_empty()).then_some(combined)
        })?;

        // Extract the ID and Key filters back out. At this point, we know that if there
        // were ID/Key filters in both `self` and `other`, then they intersected
        // to form a consistent set of constraints, so it is safe to interpret
        // the lack of any ID/Key filters respectively as a lack of that kind of
        // constraint, rather than a constraint on the empty set.

        let object_ids = {
            let partition: Vec<_> = keys
                .iter()
                .flatten()
                .filter_map(|(id, v)| v.is_none().then_some(*id))
                .collect();

            (!partition.is_empty()).then_some(partition)
        };

        let object_keys = {
            let partition: Vec<_> = keys
                .iter()
                .flatten()
                .filter_map(|(id, v)| {
                    Some(ObjectKey {
                        object_id: *id,
                        version: (*v)?.into(),
                    })
                })
                .collect();

            (!partition.is_empty()).then_some(partition)
        };

        Some(Self {
            type_: intersect!(type_, TypeFilter::intersect)?,
            owner: intersect!(owner, intersect::by_eq)?,
            object_ids,
            object_keys,
        })
    }

    /// Extract the Object ID and Key filters into one combined map from Object
    /// IDs in this filter, to the versions they should have (or None if the
    /// filter mentions the ID but no version for it).
    fn keys(&self) -> Option<BTreeMap<IotaAddress, Option<u64>>> {
        if self.object_keys.is_none() && self.object_ids.is_none() {
            return None;
        }

        Some(BTreeMap::from_iter(
            self.object_keys
                .iter()
                .flatten()
                .map(|key| (key.object_id, Some(key.version.into())))
                // Chain ID filters after Key filters so if there is overlap, we overwrite the key
                // filter with the ID filter.
                .chain(self.object_ids.iter().flatten().map(|id| (*id, None))),
        ))
    }

    /// Applies ObjectFilter to the input `RawQuery` and returns a new
    /// `RawQuery`.
    pub(crate) fn apply(&self, mut query: RawQuery) -> RawQuery {
        // Start by applying the filters on IDs and/or keys because they are combined as
        // a disjunction, while the remaining queries are conjunctions.
        if let Some(object_ids) = &self.object_ids {
            // Maximally strict - match a vec of 0 elements
            if object_ids.is_empty() {
                query = or_filter!(query, "1=0");
            } else {
                let mut inner = String::new();
                let mut prefix = "object_id IN (";
                for id in object_ids {
                    // SAFETY: Writing to a `String` cannot fail.
                    write!(
                        &mut inner,
                        "{prefix}'\\x{}'::bytea",
                        hex::encode(id.into_vec())
                    )
                    .unwrap();
                    prefix = ", ";
                }
                inner.push(')');
                query = or_filter!(query, inner);
            }
        }

        if let Some(object_keys) = &self.object_keys {
            // Maximally strict - match a vec of 0 elements
            if object_keys.is_empty() {
                query = or_filter!(query, "1=0");
            } else {
                let mut inner = String::new();
                let mut prefix = "(";
                for ObjectKey { object_id, version } in object_keys {
                    // SAFETY: Writing to a `String` cannot fail.
                    write!(
                        &mut inner,
                        "{prefix}(object_id = '\\x{}'::bytea AND object_version = {})",
                        hex::encode(object_id.into_vec()),
                        version
                    )
                    .unwrap();
                    prefix = " OR ";
                }
                inner.push(')');
                query = or_filter!(query, inner);
            }
        }

        if let Some(owner) = self.owner {
            query = filter!(
                query,
                format!(
                    "owner_id = '\\x{}'::bytea AND owner_type = {}",
                    hex::encode(owner.into_vec()),
                    OwnerType::Address as i16
                )
            );
        }

        if let Some(type_) = &self.type_ {
            return type_.apply_raw(
                query,
                "object_type",
                "object_type_package",
                "object_type_module",
                "object_type_name",
            );
        }

        query
    }
}

impl HistoricalObjectCursor {
    pub(crate) fn new(object_id: Vec<u8>, checkpoint_viewed_at: u64) -> Self {
        Self {
            object_id,
            checkpoint_viewed_at,
        }
    }
}

impl Checkpointed for Cursor {
    fn checkpoint_viewed_at(&self) -> u64 {
        self.checkpoint_viewed_at
    }
}

impl ScanLimited for Cursor {}

impl RawPaginated<Cursor> for StoredHistoryObject {
    fn filter_ge(cursor: &Cursor, query: RawQuery) -> RawQuery {
        filter!(
            query,
            format!(
                "candidates.object_id >= '\\x{}'::bytea",
                hex::encode(cursor.object_id.clone())
            )
        )
    }

    fn filter_le(cursor: &Cursor, query: RawQuery) -> RawQuery {
        filter!(
            query,
            format!(
                "candidates.object_id <= '\\x{}'::bytea",
                hex::encode(cursor.object_id.clone())
            )
        )
    }

    fn order(asc: bool, query: RawQuery) -> RawQuery {
        if asc {
            query.order_by("candidates.object_id ASC")
        } else {
            query.order_by("candidates.object_id DESC")
        }
    }
}

impl Target<Cursor> for StoredHistoryObject {
    fn cursor(&self, checkpoint_viewed_at: u64) -> Cursor {
        Cursor::new(HistoricalObjectCursor::new(
            self.object_id.clone(),
            checkpoint_viewed_at,
        ))
    }
}

/// Query-result struct for the backward diff read path. Used by
/// `build_backward_objects_query` which unions `checkpointed_objects` and
/// `objects_backward_history`. Has explicit `sql_type` annotations because
/// there is no single backing Diesel table definition.
#[derive(diesel::QueryableByName, Clone, Debug)]
pub(crate) struct StoredBackwardObject {
    #[diesel(sql_type = sql_types::Binary)]
    pub object_id: Vec<u8>,
    #[diesel(sql_type = sql_types::BigInt)]
    pub object_version: i64,
    #[diesel(sql_type = sql_types::SmallInt)]
    pub object_status: i16,
    #[diesel(sql_type = sql_types::Nullable<sql_types::Binary>)]
    pub object_digest: Option<Vec<u8>>,
    #[diesel(sql_type = sql_types::Nullable<sql_types::SmallInt>)]
    pub owner_type: Option<i16>,
    #[diesel(sql_type = sql_types::Nullable<sql_types::Binary>)]
    pub owner_id: Option<Vec<u8>>,
    #[diesel(sql_type = sql_types::Nullable<sql_types::Text>)]
    pub object_type: Option<String>,
    #[diesel(sql_type = sql_types::Nullable<sql_types::Binary>)]
    pub object_type_package: Option<Vec<u8>>,
    #[diesel(sql_type = sql_types::Nullable<sql_types::Text>)]
    pub object_type_module: Option<String>,
    #[diesel(sql_type = sql_types::Nullable<sql_types::Text>)]
    pub object_type_name: Option<String>,
    #[diesel(sql_type = sql_types::Nullable<sql_types::Binary>)]
    pub serialized_object: Option<Vec<u8>>,
    #[diesel(sql_type = sql_types::Nullable<sql_types::Text>)]
    pub coin_type: Option<String>,
    #[diesel(sql_type = sql_types::Nullable<sql_types::BigInt>)]
    pub coin_balance: Option<i64>,
    #[diesel(sql_type = sql_types::Nullable<sql_types::SmallInt>)]
    pub df_kind: Option<i16>,
}

impl StoredBackwardObject {
    /// Convert into a `StoredHistoryObject`, filling in
    /// `checkpoint_sequence_number` with the checkpoint the object was
    /// viewed at. The backward diff query guarantees the result is valid at
    /// this checkpoint, so it's the most accurate value we have.
    pub(crate) fn into_stored_history(self, checkpoint_viewed_at: u64) -> StoredHistoryObject {
        StoredHistoryObject {
            object_id: self.object_id,
            object_version: self.object_version,
            object_status: self.object_status,
            object_digest: self.object_digest,
            checkpoint_sequence_number: checkpoint_viewed_at as i64,
            owner_type: self.owner_type,
            owner_id: self.owner_id,
            object_type: self.object_type,
            object_type_package: self.object_type_package,
            object_type_module: self.object_type_module,
            object_type_name: self.object_type_name,
            serialized_object: self.serialized_object,
            coin_type: self.coin_type,
            coin_balance: self.coin_balance,
            df_kind: self.df_kind,
        }
    }
}

impl RawPaginated<Cursor> for StoredBackwardObject {
    fn filter_ge(cursor: &Cursor, query: RawQuery) -> RawQuery {
        filter!(
            query,
            format!(
                "candidates.object_id >= '\\x{}'::bytea",
                hex::encode(cursor.object_id.clone())
            )
        )
    }

    fn filter_le(cursor: &Cursor, query: RawQuery) -> RawQuery {
        filter!(
            query,
            format!(
                "candidates.object_id <= '\\x{}'::bytea",
                hex::encode(cursor.object_id.clone())
            )
        )
    }

    fn order(asc: bool, query: RawQuery) -> RawQuery {
        if asc {
            query.order_by("candidates.object_id ASC")
        } else {
            query.order_by("candidates.object_id DESC")
        }
    }
}

impl Target<Cursor> for StoredBackwardObject {
    fn cursor(&self, checkpoint_viewed_at: u64) -> Cursor {
        Cursor::new(HistoricalObjectCursor::new(
            self.object_id.clone(),
            checkpoint_viewed_at,
        ))
    }
}

/// Synthetic `object_status` for object versions that still exist in
/// `objects_version` table, but are already pruned from
/// `objects_backward_history`.
const OBJECT_STATUS_PRUNED: i16 = -1;

const OBJECT_STATUS_WRAPPED_OR_DELETED: i16 = NativeObjectStatus::WrappedOrDeleted as i16;

const OBJECT_STATUS_ACTIVE: i16 = NativeObjectStatus::Active as i16;

/// Supplies the parameters needed to build an [`Object`] from a row fetched
/// from the historical fallback: the checkpoint it is viewed at, and the root
/// version to record on it.
trait FallbackObjectKey: Eq + std::hash::Hash {
    /// The checkpoint the object is viewed at.
    fn checkpoint_viewed_at(&self) -> u64;
    /// The root version to record on the resulting object, if any.
    fn root_version(&self) -> Option<u64>;
}

impl FallbackObjectKey for HistoricalKey {
    fn checkpoint_viewed_at(&self) -> u64 {
        self.checkpoint_viewed_at
    }
    fn root_version(&self) -> Option<u64> {
        None
    }
}

impl FallbackObjectKey for ParentVersionKey {
    fn checkpoint_viewed_at(&self) -> u64 {
        self.checkpoint_viewed_at
    }
    fn root_version(&self) -> Option<u64> {
        Some(self.parent_version)
    }
}

/// Checks if `objects_version` table covers the whole chain history.
///
/// This function currently returns false for indexer restored from snapshot.
fn objects_version_complete(db: &Db) -> bool {
    db.inner
        .ensure_data_not_pruned_for_checkpoint(0, &[CommitterTables::ObjectsVersion])
        .is_ok()
}

/// Fills the missing `results` entries with objects fetched from the
/// historical fallback.
///
/// It stores nothing when the object does not exist at the requested version
/// (e.g. when the object is wrapped or deleted). It stores
/// [`Error::DataPruned`] per key when the historical fallback is not
/// configured.
async fn fill_missing_objects_from_fallback<K: FallbackObjectKey>(
    db: &Db,
    keys_with_refs: Vec<(K, (ObjectId, Version))>,
    results: &mut HashMap<K, Result<Object, Error>>,
) -> Result<(), Error> {
    if keys_with_refs.is_empty() {
        return Ok(());
    }

    let object_refs: Vec<(ObjectId, Version)> =
        keys_with_refs.iter().map(|(_, obj_ref)| *obj_ref).collect();

    let fetched: Vec<Result<Option<StoredObject>, Error>> = if db.inner.is_fallback_enabled() {
        db.inner
            .multi_get_fallback_objects(&object_refs, false)
            .await
            .map_err(Error::from)?
            .into_iter()
            .map(Ok)
            .collect()
    } else {
        object_refs
            .iter()
            .map(|(id, _)| {
                Err(Error::DataPruned(format!(
                    "data for object {id} potentially pruned"
                )))
            })
            .collect()
    };

    for ((key, _), fetched) in keys_with_refs.into_iter().zip(fetched) {
        match fetched {
            Ok(Some(stored)) => {
                let object = Object::try_from_stored_object(
                    stored,
                    key.checkpoint_viewed_at(),
                    key.root_version(),
                )?;
                results.insert(key, Ok(object));
            }
            // The object does not exist at the requested version.
            Ok(None) => {}
            // Only this key cannot be served.
            Err(e) => {
                results.insert(key, Err(e));
            }
        }
    }

    Ok(())
}

impl Loader<HistoricalKey> for Db {
    /// Error stored per key, e.g. when object version is pruned.
    type Value = Result<Object, Error>;
    type Error = Error;

    async fn load(
        &self,
        keys: &[HistoricalKey],
    ) -> Result<HashMap<HistoricalKey, Result<Object, Error>>, Error> {
        if keys.is_empty() {
            return Ok(HashMap::new());
        }

        let (ids, versions): (Vec<Vec<u8>>, Vec<i64>) = keys
            .iter()
            .map(|key| (key.id.into_vec(), key.version as i64))
            .collect::<BTreeSet<_>>()
            .into_iter()
            .unzip();

        // For each `(object_id, object_version)` pair, locate the row content
        // in `checkpointed_objects` (current state of the object) or
        // `objects_backward_history` (a superseded prior state). The
        // `objects_version` join confirms the version is real and supplies the
        // checkpoint at which it became current.
        //
        // A row is returned for every version found in `objects_version`.
        // Wrapped-or-deleted status is reported for versions without row
        // content that are in the retention window of
        // `objects_backward_history`. Versions outside of this retention
        // window are returned with the synthetic `OBJECT_STATUS_PRUNED`
        // status.
        let sql = "SELECT \
                v.object_id, \
                v.object_version, \
                v.cp_sequence_number AS checkpoint_sequence_number, \
                COALESCE(co.object_status, bh.object_status, \
                    CASE WHEN v.cp_sequence_number >= COALESCE((\
                        SELECT min_available_cp FROM watermarks \
                        WHERE entity = 'objects_backward_history'), 0) \
                    THEN $3 ELSE $4 END) AS object_status, \
                COALESCE(co.object_digest, bh.object_digest) AS object_digest, \
                COALESCE(co.owner_type, bh.owner_type) AS owner_type, \
                COALESCE(co.owner_id, bh.owner_id) AS owner_id, \
                COALESCE(co.object_type, bh.object_type) AS object_type, \
                COALESCE(co.object_type_package, bh.object_type_package) AS object_type_package, \
                COALESCE(co.object_type_module, bh.object_type_module) AS object_type_module, \
                COALESCE(co.object_type_name, bh.object_type_name) AS object_type_name, \
                COALESCE(co.serialized_object, bh.serialized_object) AS serialized_object, \
                COALESCE(co.coin_type, bh.coin_type) AS coin_type, \
                COALESCE(co.coin_balance, bh.coin_balance) AS coin_balance, \
                COALESCE(co.df_kind, bh.df_kind) AS df_kind \
            FROM unnest($1::bytea[], $2::bigint[]) AS pairs(object_id, object_version) \
            INNER JOIN objects_version v \
                    ON v.object_id = pairs.object_id \
                   AND v.object_version = pairs.object_version \
            LEFT JOIN checkpointed_objects co \
                   ON co.object_id = v.object_id \
                  AND co.object_version = v.object_version \
            LEFT JOIN objects_backward_history bh \
                   ON bh.object_id = v.object_id \
                  AND bh.object_version = v.object_version";

        let objects: Vec<StoredHistoryObject> = self
            .execute(move |conn| {
                conn.results(move || {
                    diesel::sql_query(sql)
                        .bind::<sql_types::Array<sql_types::Binary>, _>(ids.clone())
                        .bind::<sql_types::Array<sql_types::BigInt>, _>(versions.clone())
                        .bind::<sql_types::SmallInt, _>(OBJECT_STATUS_WRAPPED_OR_DELETED)
                        .bind::<sql_types::SmallInt, _>(OBJECT_STATUS_PRUNED)
                })
            })
            .await
            .map_err(|e| Error::Internal(format!("Failed to fetch objects: {e}")))?;

        let mut id_version_to_stored = BTreeMap::new();
        for stored in objects {
            let key = (addr(&stored.object_id)?, stored.object_version as u64);
            id_version_to_stored.insert(key, stored);
        }

        let mut result = HashMap::new();
        let mut fallback_keys = Vec::new();
        // Keys with no `objects_version` entry at all.
        let mut unresolved_keys = Vec::new();
        for key in keys {
            let Some(stored) = id_version_to_stored.get(&(key.id, key.version)) else {
                unresolved_keys.push(*key);
                continue;
            };

            // Filter by key's checkpoint viewed at here. Doing this in memory because it
            // should be quite rare that this query actually filters something,
            // but encoding it in SQL is complicated.
            if key.checkpoint_viewed_at < stored.checkpoint_sequence_number as u64 {
                continue;
            }

            let active_object = match stored.object_status {
                OBJECT_STATUS_PRUNED => {
                    fallback_keys.push(*key);
                    continue;
                }
                OBJECT_STATUS_ACTIVE => ActiveObject::try_from(stored.clone())?,
                // Wrapped, deleted, or not yet created: resolve as non-existent.
                _ => continue,
            };
            // This conversion will use the object's own version as the
            // `Object::root_version`.
            let object = Object::from_active_object(active_object, key.checkpoint_viewed_at, None);
            result.insert(*key, Ok(object));
        }

        // When version history is complete, unresolved key = nonexisting object
        // version. When history is incomplete we have to query the KV to know
        // if object exists or not.
        if !objects_version_complete(self) {
            fallback_keys.append(&mut unresolved_keys);
        }

        let keys_with_refs = fallback_keys
            .into_iter()
            .map(|key| (key, (key.id.into(), Version::from(key.version))))
            .collect();
        fill_missing_objects_from_fallback(self, keys_with_refs, &mut result).await?;

        Ok(result)
    }
}

impl Loader<OptimisticKey> for Db {
    /// Error stored per key, e.g. when object version is pruned.
    type Value = Result<Object, Error>;
    type Error = Error;

    async fn load(
        &self,
        keys: &[OptimisticKey],
    ) -> Result<HashMap<OptimisticKey, Result<Object, Error>>, Error> {
        use objects::dsl as o;

        if keys.is_empty() {
            return Ok(HashMap::new());
        }

        let id_versions: BTreeSet<_> = keys
            .iter()
            .map(|key| (key.id.into_vec(), key.version as i64))
            .collect();

        let objects: Vec<StoredObject> = self
            .execute(move |conn| {
                conn.results(move || {
                    let mut query = o::objects.select(StoredObject::as_select()).into_boxed();
                    for (id, version) in id_versions.iter().cloned() {
                        query =
                            query.or_filter(o::object_id.eq(id).and(o::object_version.eq(version)));
                    }
                    query
                })
            })
            .await
            .map_err(|e| Error::Internal(format!("Failed to fetch optimistic objects: {e}")))?;

        let mut result = HashMap::new();
        let id_version_to_stored = objects
            .into_iter()
            .map(|stored| {
                addr(&stored.object_id).map(|id| ((id, stored.object_version as u64), stored))
            })
            .collect::<Result<BTreeMap<_, _>, _>>()?;

        // Collect keys that were not found in objects table
        let mut missing_keys = Vec::new();
        for key in keys {
            if let Some(stored) = id_version_to_stored.get(&(key.id, key.version)) {
                let object = Object::try_from_stored_object(stored.clone(), u64::MAX, None)?;
                result.insert(*key, Ok(object));
            } else {
                missing_keys.push(*key);
            }
        }

        // For missing keys, fallback to the backward-history loader
        if !missing_keys.is_empty() {
            let historical_keys: Vec<HistoricalKey> = missing_keys
                .iter()
                .map(|key| HistoricalKey {
                    id: key.id,
                    version: key.version,
                    checkpoint_viewed_at: u64::MAX,
                })
                .collect();

            let historical_result: HashMap<HistoricalKey, Result<Object, Error>> =
                self.load(&historical_keys).await?;

            for (historical_key, object) in historical_result {
                let optimistic_key = OptimisticKey {
                    id: historical_key.id,
                    version: historical_key.version,
                };
                result.insert(optimistic_key, object);
            }
        }

        Ok(result)
    }
}

impl Loader<ParentVersionKey> for Db {
    /// Error stored per key, e.g. when object version is pruned.
    type Value = Result<Object, Error>;
    type Error = Error;

    async fn load(
        &self,
        keys: &[ParentVersionKey],
    ) -> Result<HashMap<ParentVersionKey, Result<Object, Error>>, Error> {
        // Group keys by checkpoint viewed at and parent version -- we'll issue a
        // separate query for each group.
        #[derive(Eq, PartialEq, Ord, PartialOrd, Clone, Copy)]
        struct GroupKey {
            checkpoint_viewed_at: u64,
            parent_version: u64,
        }

        let mut keys_by_cursor_and_parent_version: BTreeMap<_, BTreeSet<_>> = BTreeMap::new();
        for key in keys {
            let group_key = GroupKey {
                checkpoint_viewed_at: key.checkpoint_viewed_at,
                parent_version: key.parent_version,
            };

            keys_by_cursor_and_parent_version
                .entry(group_key)
                .or_default()
                .insert(key.id.into_vec());
        }

        // For each id, pick the largest version `≤ parent_version` from
        // `objects_version` (versions table contains all current and past versions of
        // objects), then locate the row content in `checkpointed_objects`
        // (current state of the object) or `objects_backward_history` (a
        // superseded prior state).
        //
        // A row is returned for the largest version `≤ parent_version` found
        // in `objects_version`. Wrapped-or-deleted status is reported for
        // versions without row content that are in the retention window of
        // `objects_backward_history`. Versions outside of this retention
        // window are returned with the synthetic `OBJECT_STATUS_PRUNED`
        // status.
        let sql = "WITH ids AS (SELECT unnest($1::bytea[]) AS object_id), \
                        latest_per_id AS (\
                       SELECT i.object_id, o.object_version, o.cp_sequence_number \
                       FROM ids i \
                       JOIN LATERAL (\
                           SELECT object_version, cp_sequence_number \
                           FROM objects_version \
                           WHERE object_id = i.object_id \
                             AND object_version <= $2::bigint \
                           ORDER BY object_version DESC \
                           LIMIT 1) o ON TRUE) \
                   SELECT \
                       v.object_id, \
                       v.object_version, \
                       v.cp_sequence_number AS checkpoint_sequence_number, \
                       COALESCE(co.object_status, bh.object_status, \
                           CASE WHEN v.cp_sequence_number >= COALESCE((\
                               SELECT min_available_cp FROM watermarks \
                               WHERE entity = 'objects_backward_history'), 0) \
                           THEN $3 ELSE $4 END) AS object_status, \
                       COALESCE(co.object_digest, bh.object_digest) AS object_digest, \
                       COALESCE(co.owner_type, bh.owner_type) AS owner_type, \
                       COALESCE(co.owner_id, bh.owner_id) AS owner_id, \
                       COALESCE(co.object_type, bh.object_type) AS object_type, \
                       COALESCE(co.object_type_package, bh.object_type_package) AS object_type_package, \
                       COALESCE(co.object_type_module, bh.object_type_module) AS object_type_module, \
                       COALESCE(co.object_type_name, bh.object_type_name) AS object_type_name, \
                       COALESCE(co.serialized_object, bh.serialized_object) AS serialized_object, \
                       COALESCE(co.coin_type, bh.coin_type) AS coin_type, \
                       COALESCE(co.coin_balance, bh.coin_balance) AS coin_balance, \
                       COALESCE(co.df_kind, bh.df_kind) AS df_kind \
                   FROM latest_per_id v \
                   LEFT JOIN checkpointed_objects co \
                          ON co.object_id = v.object_id \
                         AND co.object_version = v.object_version \
                   LEFT JOIN objects_backward_history bh \
                          ON bh.object_id = v.object_id \
                         AND bh.object_version = v.object_version";

        // Issue concurrent reads for each group of keys.
        let futures = keys_by_cursor_and_parent_version
            .into_iter()
            .map(|(group_key, ids)| {
                let parent_version = group_key.parent_version as i64;
                let ids: Vec<Vec<u8>> = ids.into_iter().collect();

                self.execute(move |conn| {
                    let stored: Vec<StoredHistoryObject> = conn.results(move || {
                        diesel::sql_query(sql)
                            .bind::<sql_types::Array<sql_types::Binary>, _>(ids.clone())
                            .bind::<sql_types::BigInt, _>(parent_version)
                            .bind::<sql_types::SmallInt, _>(OBJECT_STATUS_WRAPPED_OR_DELETED)
                            .bind::<sql_types::SmallInt, _>(OBJECT_STATUS_PRUNED)
                    })?;

                    Ok::<_, diesel::result::Error>(
                        stored
                            .into_iter()
                            .map(|stored| (group_key, stored))
                            .collect::<Vec<_>>(),
                    )
                })
            });

        // Wait for the reads to all finish, and gather them into the result map.
        let groups = futures::future::join_all(futures).await;

        let mut key_to_stored = HashMap::new();
        for group in groups {
            for (group_key, stored) in
                group.map_err(|e| Error::Internal(format!("Failed to fetch objects: {e}")))?
            {
                let key = ParentVersionKey {
                    id: addr(&stored.object_id)?,
                    checkpoint_viewed_at: group_key.checkpoint_viewed_at,
                    parent_version: group_key.parent_version,
                };
                key_to_stored.insert(key, stored);
            }
        }

        let mut results = HashMap::new();
        let mut fallback_keys = Vec::new();
        // Keys with no version `≤ parent_version` in `objects_version` at all.
        let mut unresolved_keys = Vec::new();
        for key in keys {
            let Some(stored) = key_to_stored.get(key) else {
                unresolved_keys.push(*key);
                continue;
            };

            // This version didn't exist at the checkpoint we are viewing at.
            if key.checkpoint_viewed_at < stored.checkpoint_sequence_number as u64 {
                continue;
            }

            let active_object = match stored.object_status {
                OBJECT_STATUS_PRUNED => {
                    let object_ref = (key.id.into(), Version::from(stored.object_version as u64));
                    fallback_keys.push((*key, object_ref));
                    continue;
                }
                OBJECT_STATUS_ACTIVE => ActiveObject::try_from(stored.clone())?,
                // Wrapped, deleted, or not yet created: resolve as non-existent.
                _ => continue,
            };
            // If `LatestAtKey::parent_version` is set, it must have been correctly
            // propagated from the `Object::root_version` of some object.
            let object = Object::from_active_object(
                active_object,
                key.checkpoint_viewed_at,
                Some(key.parent_version),
            );

            results.insert(*key, Ok(object));
        }

        // When version history is complete, unresolved key = nonexisting object
        // version. When history is incomplete we return DataPruned for such
        // cases, since this cannot be handled by KV currently: before_version
        // query will not work properly since we don't store wrapped-or-deleted
        // version in the KV.
        if !objects_version_complete(self) {
            for key in unresolved_keys {
                let id = key.id;
                results.insert(
                    key,
                    Err(Error::DataPruned(format!(
                        "data for object {id} potentially pruned"
                    ))),
                );
            }
        }

        fill_missing_objects_from_fallback(self, fallback_keys, &mut results).await?;

        Ok(results)
    }
}

impl Loader<LatestAtKey> for Db {
    type Value = Object;
    type Error = Error;

    async fn load(&self, keys: &[LatestAtKey]) -> Result<HashMap<LatestAtKey, Object>, Error> {
        // Group keys by checkpoint viewed at -- we'll issue a separate query for each
        // group.
        let mut keys_by_cursor_and_parent_version: BTreeMap<_, BTreeSet<_>> = BTreeMap::new();

        for key in keys {
            keys_by_cursor_and_parent_version
                .entry(key.checkpoint_viewed_at)
                .or_default()
                .insert(key.id);
        }

        let max_available_range = self.max_available_range;

        // Issue concurrent reads for each group of keys.
        let futures =
            keys_by_cursor_and_parent_version
                .into_iter()
                .map(|(checkpoint_viewed_at, ids)| {
                    self.execute_repeatable(move |conn| {
                        if !AvailableRange::is_checkpoint_in_backward_history_range(
                            conn,
                            checkpoint_viewed_at,
                            max_available_range,
                        )? {
                            return Ok::<Vec<(u64, StoredHistoryObject)>, diesel::result::Error>(
                                vec![],
                            );
                        };

                        let filter = ObjectFilter {
                            object_ids: Some(ids.iter().cloned().collect()),
                            ..Default::default()
                        };

                        let results: Vec<StoredBackwardObject> = conn.results(move || {
                            consistent::query(
                                checkpoint_viewed_at,
                                &Page::bounded(ids.len() as u64),
                                |q| filter.apply(q),
                            )
                            .into_boxed()
                        })?;

                        Ok(results
                            .into_iter()
                            .map(|r| {
                                (
                                    checkpoint_viewed_at,
                                    r.into_stored_history(checkpoint_viewed_at),
                                )
                            })
                            .collect())
                    })
                });

        // Wait for the reads to all finish, and gather them into the result map.
        let groups = futures::future::join_all(futures).await;

        let mut results = HashMap::new();
        for group in groups {
            for (checkpoint_viewed_at, stored) in
                group.map_err(|e| Error::Internal(format!("Failed to fetch objects: {e}")))?
            {
                let active_object = ActiveObject::try_from(stored)?;
                let object = Object::from_active_object(active_object, checkpoint_viewed_at, None);

                let key = LatestAtKey {
                    id: object.address,
                    checkpoint_viewed_at,
                };

                results.insert(key, object);
            }
        }

        Ok(results)
    }
}

impl From<&ActiveObject> for ObjectStatus {
    fn from(active_object: &ActiveObject) -> Self {
        active_object.status
    }
}

impl From<&Object> for OwnerImpl {
    fn from(object: &Object) -> Self {
        OwnerImpl {
            address: object.address,
            checkpoint_viewed_at: object.checkpoint_viewed_at,
        }
    }
}

pub(crate) async fn deserialize_move_struct(
    move_object: &NativeMoveStruct,
    resolver: &PackageResolver,
) -> Result<(StructTag, MoveStruct), Error> {
    let struct_tag = move_object.struct_tag().clone();
    let contents = move_object.contents();
    let move_type_layout = resolver
        .type_layout(TypeTag::from(struct_tag.clone()))
        .await
        .map_err(|e| {
            Error::Internal(format!(
                "Error fetching layout for type {}: {e}",
                struct_tag.to_canonical_string(/* with_prefix */ true)
            ))
        })?;

    let MoveTypeLayout::Struct(layout) = move_type_layout else {
        return Err(Error::Internal("Object is not a move struct".to_string()));
    };

    // TODO (annotated-visitor): Use custom visitors for extracting a dynamic field,
    // and for creating a GraphQL MoveValue directly (not via an annotated
    // visitor).
    let move_struct = BoundedVisitor::deserialize_struct(contents, &layout).map_err(|e| {
        Error::Internal(format!(
            "Error deserializing move struct for type {}: {e}",
            struct_tag.to_canonical_string(/* with_prefix */ true)
        ))
    })?;

    Ok((struct_tag, move_struct))
}

/// Constructs a backward diff query for objects.
///
/// Uses consistent view for most queries to ensure point-in-time correctness.
/// Falls back to historical view only for object-key lookups (specific
/// id+version pairs) which don't need consistency filtering. When both
/// `object_ids` and `object_keys` are provided, the results from both views
/// are unioned.
fn backward_objects_query(
    filter: &ObjectFilter,
    checkpoint_viewed_at: u64,
    page: &Page<Cursor>,
) -> RawQuery {
    if let (Some(_), Some(_)) = (&filter.object_ids, &filter.object_keys) {
        // If both object IDs and object keys are specified, then we need to query in
        // both historical and consistent views, and then union the results.
        let ids_only_filter = ObjectFilter {
            object_keys: None,
            ..filter.clone()
        };
        let (id_query, id_bindings) = consistent::query(checkpoint_viewed_at, page, move |query| {
            ids_only_filter.apply(query)
        })
        .finish();

        let keys_filter: HistoricalFilter = ObjectFilter {
            object_ids: None,
            ..filter.clone()
        }
        .try_into()
        .expect("object_keys is Some by match-arm guard");
        let (key_query, key_bindings) = historical::query(page, &keys_filter).finish();

        RawQuery::new(
            format!("SELECT * FROM (({id_query}) UNION ALL ({key_query})) AS candidates",),
            id_bindings.into_iter().chain(key_bindings).collect(),
        )
        .order_by("object_id")
        .limit(page.limit() as i64)
    } else if let Ok(keys_filter) = HistoricalFilter::try_from(filter.clone()) {
        historical::query(page, &keys_filter)
    } else {
        consistent::query(checkpoint_viewed_at, page, move |query| filter.apply(query))
    }
}

#[cfg(test)]
mod tests {
    use std::str::FromStr;

    use super::*;

    #[test]
    fn test_owner_filter_intersection() {
        let f0 = ObjectFilter {
            owner: Some(IotaAddress::from_str("0x1").unwrap()),
            ..Default::default()
        };

        let f1 = ObjectFilter {
            owner: Some(IotaAddress::from_str("0x2").unwrap()),
            ..Default::default()
        };

        assert_eq!(f0.clone().intersect(f0.clone()), Some(f0.clone()));
        assert_eq!(f0.intersect(f1), None);
    }

    #[test]
    fn test_key_filter_intersection() {
        let i1 = IotaAddress::from_str("0x1").unwrap();
        let i2 = IotaAddress::from_str("0x2").unwrap();
        let i3 = IotaAddress::from_str("0x3").unwrap();
        let i4 = IotaAddress::from_str("0x4").unwrap();

        let f0 = ObjectFilter {
            object_ids: Some(vec![i1, i3]),
            object_keys: Some(vec![
                ObjectKey {
                    object_id: i2,
                    version: 1.into(),
                },
                ObjectKey {
                    object_id: i4,
                    version: 2.into(),
                },
            ]),
            ..Default::default()
        };

        let f1 = ObjectFilter {
            object_ids: Some(vec![i1, i2]),
            object_keys: Some(vec![ObjectKey {
                object_id: i4,
                version: 2.into(),
            }]),
            ..Default::default()
        };

        let f2 = ObjectFilter {
            object_ids: Some(vec![i1, i3]),
            ..Default::default()
        };

        let f3 = ObjectFilter {
            object_keys: Some(vec![
                ObjectKey {
                    object_id: i2,
                    version: 2.into(),
                },
                ObjectKey {
                    object_id: i4,
                    version: 2.into(),
                },
            ]),
            ..Default::default()
        };

        assert_eq!(
            f0.clone().intersect(f1.clone()),
            Some(ObjectFilter {
                object_ids: Some(vec![i1]),
                object_keys: Some(vec![
                    ObjectKey {
                        object_id: i2,
                        version: 1.into(),
                    },
                    ObjectKey {
                        object_id: i4,
                        version: 2.into(),
                    },
                ]),
                ..Default::default()
            })
        );

        assert_eq!(
            f1.clone().intersect(f2.clone()),
            Some(ObjectFilter {
                object_ids: Some(vec![i1]),
                ..Default::default()
            })
        );

        assert_eq!(
            f1.intersect(f3.clone()),
            Some(ObjectFilter {
                object_keys: Some(vec![
                    ObjectKey {
                        object_id: i2,
                        version: 2.into(),
                    },
                    ObjectKey {
                        object_id: i4,
                        version: 2.into(),
                    },
                ]),
                ..Default::default()
            })
        );

        // i2 got a conflicting version assignment
        assert_eq!(f0.intersect(f3.clone()), None);

        // No overlap between these two.
        assert_eq!(f2.intersect(f3), None);
    }
}