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, JoinOnDsl, QueryDsl, SelectableHelper, sql_types,
};
use iota_indexer::{
    models::objects::{StoredHistoryObject, StoredObject},
    schema::{objects, objects_history, objects_version},
    types::{ObjectStatus as NativeObjectStatus, OwnerType},
};
use iota_types::{
    base_types::{StructTag, TypeTag},
    object::{
        MoveObject as NativeMoveObject, Object as NativeObject, Owner as NativeOwner,
        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 kind: ObjectKind,
    /// 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)]
#[expect(clippy::large_enum_variant)]
pub(crate) enum ObjectKind {
    /// An object loaded from serialized data, such as the contents of a
    /// transaction that hasn't been indexed yet.
    NotIndexed(NativeObject),
    /// An object fetched from the index.
    Indexed(NativeObject, StoredHistoryObject),
    /// The object is wrapped or deleted and only partial information can be
    /// loaded from the indexer. The `u64` is the version of the object.
    WrappedOrDeleted(u64),
}

#[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,
    /// The object is deleted or wrapped and only partial information can be
    /// loaded from the indexer.
    #[graphql(
        deprecation = "will be removed with v1.26, as such objects can be considered non-existent"
    )]
    WrappedOrDeleted,
}

#[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.
            - WRAPPED_OR_DELETED: The object is deleted or wrapped and only partial
            information can be loaded.
        "#
    ),
    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>"),
        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.
    /// - WRAPPED_OR_DELETED: The object is deleted or wrapped and only partial
    ///   information can be loaded.
    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
    /// more than two complex filters (on function, kind, sender, recipient,
    /// input object, changed object, or ids), and can be at most
    /// `serviceConfig.maxScanLimit`.
    ///
    /// 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.
    #[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>,
        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.kind)
    }

    pub(crate) async fn digest(&self) -> Option<String> {
        self.0
            .native_impl()
            .map(|native| native.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::latest_at(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 Some(native) = self.0.native_impl() else {
            return Ok(None);
        };
        let digest = native.previous_transaction;
        let key = transaction_block::DigestKey::new(digest.into(), self.0.checkpoint_viewed_at);

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

    pub(crate) async fn storage_rebate(&self) -> Option<BigInt> {
        self.0
            .native_impl()
            .map(|native| BigInt::from(native.storage_rebate))
    }

    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>> {
        use ObjectKind as K;
        Ok(match &self.0.kind {
            K::WrappedOrDeleted(_) => None,
            // WrappedOrDeleted objects are also read from the historical objects table, and they do
            // not have a serialized object, so the column is also nullable for stored historical
            // objects.
            K::Indexed(_, stored) => stored.serialized_object.as_ref().map(Base64::from),

            K::NotIndexed(native) => {
                let bytes = bcs::to_bytes(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 Some(native) = self.0.native_impl() else {
            return Ok(None);
        };

        let move_object = native
            .data
            .as_struct_opt()
            .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,
            kind: ObjectKind::NotIndexed(native),
            checkpoint_viewed_at,
            root_version,
        }
    }

    pub(crate) fn native_impl(&self) -> Option<&NativeObject> {
        use ObjectKind as K;

        match &self.kind {
            K::NotIndexed(native) | K::Indexed(native, _) => Some(native),
            K::WrappedOrDeleted(_) => None,
        }
    }

    pub(crate) fn version_impl(&self) -> u64 {
        use ObjectKind as K;

        match &self.kind {
            K::NotIndexed(native) | K::Indexed(native, _) => native.version().as_u64(),
            K::WrappedOrDeleted(object_version) => *object_version,
        }
    }

    /// 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 = results_iter.collect();
                let results = resolve_tombstone_versions(conn, results)?;
                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 object =
                Object::try_from_stored_history_object(stored_history, 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
            }

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

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

            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))
    }

    /// `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 try_from_stored_history_object(
        history_object: StoredHistoryObject,
        checkpoint_viewed_at: u64,
        root_version: Option<u64>,
    ) -> Result<Self, Error> {
        let address = addr(&history_object.object_id)?;

        let object_status =
            NativeObjectStatus::try_from(history_object.object_status).map_err(|_| {
                Error::Internal(format!(
                    "Unknown object status {} for object {} at version {}",
                    history_object.object_status, address, history_object.object_version
                ))
            })?;

        match object_status {
            NativeObjectStatus::Active => {
                let Some(serialized_object) = &history_object.serialized_object else {
                    return Err(Error::Internal(format!(
                        "Live object {} at version {} cannot have missing serialized_object field",
                        address, history_object.object_version
                    )));
                };

                let native_object = bcs::from_bytes(serialized_object).map_err(|_| {
                    Error::Internal(format!("Failed to deserialize object {address}"))
                })?;

                let root_version =
                    root_version.unwrap_or_else(|| version_for_dynamic_fields(&native_object));
                Ok(Self {
                    address,
                    kind: ObjectKind::Indexed(native_object, history_object),
                    checkpoint_viewed_at,
                    root_version,
                })
            }
            NativeObjectStatus::WrappedOrDeleted => Ok(Self {
                address,
                kind: ObjectKind::WrappedOrDeleted(history_object.object_version as u64),
                checkpoint_viewed_at,
                root_version: history_object.object_version as u64,
            }),
        }
    }

    pub(crate) fn try_from_stored_object(
        stored_object: StoredObject,
        checkpoint_viewed_at: u64,
    ) -> Result<Self, Error> {
        let address = addr(&stored_object.object_id)?;

        let native_object = bcs::from_bytes(&stored_object.serialized_object)
            .map_err(|_| Error::Internal(format!("Failed to deserialize object {address}")))?;

        let root_version = version_for_dynamic_fields(&native_object);

        let stored_history_like = StoredHistoryObject {
            object_id: stored_object.object_id,
            object_version: stored_object.object_version,
            object_digest: Some(stored_object.object_digest),
            object_status: NativeObjectStatus::Active as i16,
            checkpoint_sequence_number: checkpoint_viewed_at as i64,
            serialized_object: Some(stored_object.serialized_object),
            object_type: stored_object.object_type,
            object_type_package: stored_object.object_type_package,
            object_type_module: stored_object.object_type_module,
            object_type_name: stored_object.object_type_name,
            owner_type: Some(stored_object.owner_type),
            owner_id: stored_object.owner_id,
            coin_type: stored_object.coin_type,
            coin_balance: stored_object.coin_balance,
            df_kind: stored_object.df_kind,
        };

        Ok(Self {
            address,
            kind: ObjectKind::Indexed(native_object, stored_history_like),
            checkpoint_viewed_at,
            root_version,
        })
    }
}

/// 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>,
    /// `TRUE` when the row came from `objects_backward_history`, `FALSE`
    /// otherwise (from `checkpointed_objects` or `objects_version`). Used to
    /// decide whether version correction is needed for `WrappedOrDeleted`
    /// entries, since backward history stores a lamport-1 approximation.
    #[diesel(sql_type = sql_types::Bool)]
    pub from_backward_history: bool,
}

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,
        }
    }
}

/// Resolves real tombstone versions for `WrappedOrDeleted` entries from
/// `objects_backward_history`.
///
/// The backward history stores a lamport-1 version approximation which may be
/// higher than the actual tombstone version. This function looks up the real
/// version from `objects_version` using a single batch query that unnests
/// bound `bytea[]` / `bigint[]` parameter arrays into `(object_id, version)`
/// pairs and joins them via `MAX(object_version) <= backward_history_version`.
/// Only entries tagged with `from_backward_history = true` are resolved;
/// entries from `checkpointed_objects` already have the correct version.
pub(crate) fn resolve_tombstone_versions(
    conn: &mut crate::data::pg::PgConnection<'_>,
    results: Vec<StoredBackwardObject>,
) -> Result<Vec<StoredBackwardObject>, diesel::result::Error> {
    let (ids, versions): (Vec<Vec<u8>>, Vec<i64>) = results
        .iter()
        .filter(|r| {
            r.from_backward_history
                && r.object_status == NativeObjectStatus::WrappedOrDeleted as i16
        })
        .map(|r| (r.object_id.clone(), r.object_version))
        .unzip();

    if ids.is_empty() {
        return Ok(results);
    }

    // Bound `unnest` arrays (rather than an inlined `VALUES` list) keep the
    // SQL text constant across calls so Postgres can reuse a cached plan,
    // and skip the parser cost of every `::bytea` / `::bigint` cast.
    let sql = "SELECT pairs.object_id, pairs.backward_history_version, \
                      MAX(ov.object_version) AS real_version \
               FROM unnest($1::bytea[], $2::bigint[]) \
                    AS pairs(object_id, backward_history_version) \
               LEFT JOIN objects_version ov \
                 ON ov.object_id = pairs.object_id \
                AND ov.object_version <= pairs.backward_history_version \
               GROUP BY pairs.object_id, pairs.backward_history_version";

    #[derive(diesel::QueryableByName)]
    struct ResolvedVersion {
        #[diesel(sql_type = sql_types::Binary)]
        object_id: Vec<u8>,
        #[diesel(sql_type = sql_types::BigInt)]
        backward_history_version: i64,
        #[diesel(sql_type = sql_types::Nullable<sql_types::BigInt>)]
        real_version: Option<i64>,
    }

    let rows: Vec<ResolvedVersion> = conn.results(|| {
        diesel::sql_query(sql)
            .bind::<sql_types::Array<sql_types::Binary>, _>(ids.clone())
            .bind::<sql_types::Array<sql_types::BigInt>, _>(versions.clone())
    })?;

    // Key by (object_id, backward_history_version) → real_version
    let resolved_map: HashMap<Vec<u8>, HashMap<i64, i64>> = rows
        .into_iter()
        .filter_map(|r| {
            r.real_version
                .map(|real| (r.object_id, r.backward_history_version, real))
        })
        .fold(HashMap::new(), |mut acc, (id, ver, real)| {
            acc.entry(id).or_default().insert(ver, real);
            acc
        });

    Ok(results
        .into_iter()
        .map(|mut r| {
            if r.from_backward_history
                && r.object_status == NativeObjectStatus::WrappedOrDeleted as i16
            {
                if let Some(&real_version) = resolved_map
                    .get(&r.object_id)
                    .and_then(|versions| versions.get(&r.object_version))
                {
                    r.object_version = real_version;
                }
            }
            r
        })
        .collect())
}

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,
        ))
    }
}

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

    async fn load(&self, keys: &[HistoricalKey]) -> Result<HashMap<HistoricalKey, Object>, Error> {
        use objects_history::dsl as h;
        use objects_version::dsl as v;

        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<StoredHistoryObject> = self
            .execute(move |conn| {
                conn.results(move || {
                    let mut query = h::objects_history
                        .inner_join(
                            v::objects_version.on(v::cp_sequence_number
                                .eq(h::checkpoint_sequence_number)
                                .and(v::object_id.eq(h::object_id))
                                .and(v::object_version.eq(h::object_version))),
                        )
                        .select(StoredHistoryObject::as_select())
                        .into_boxed();

                    for (id, version) in id_versions.iter().cloned() {
                        query =
                            query.or_filter(v::object_id.eq(id).and(v::object_version.eq(version)));
                    }

                    query
                })
            })
            .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();
        for key in keys {
            let Some(stored) = id_version_to_stored.get(&(key.id, key.version)) else {
                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 object = Object::try_from_stored_history_object(
                stored.clone(),
                key.checkpoint_viewed_at,
                // This conversion will use the object's own version as the `Object::root_version`.
                None,
            )?;
            result.insert(*key, object);
        }

        Ok(result)
    }
}

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

    async fn load(&self, keys: &[OptimisticKey]) -> Result<HashMap<OptimisticKey, Object>, 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)?;
                result.insert(*key, object);
            } else {
                missing_keys.push(*key);
            }
        }

        // For missing keys, fallback to using the objects_history table
        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, Object> =
                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 {
    type Value = Object;
    type Error = Error;

    async fn load(
        &self,
        keys: &[ParentVersionKey],
    ) -> Result<HashMap<ParentVersionKey, Object>, 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());
        }

        // Issue concurrent reads for each group of keys.
        let futures = keys_by_cursor_and_parent_version
            .into_iter()
            .map(|(group_key, ids)| {
                self.execute(move |conn| {
                    let stored: Vec<StoredHistoryObject> = conn.results(move || {
                        use objects_history::dsl as h;
                        use objects_version::dsl as v;

                        h::objects_history
                            .inner_join(
                                v::objects_version.on(v::cp_sequence_number
                                    .eq(h::checkpoint_sequence_number)
                                    .and(v::object_id.eq(h::object_id))
                                    .and(v::object_version.eq(h::object_version))),
                            )
                            .select(StoredHistoryObject::as_select())
                            .filter(v::object_id.eq_any(ids.iter().cloned()))
                            .filter(v::object_version.le(group_key.parent_version as i64))
                            .distinct_on(v::object_id)
                            .order_by(v::object_id)
                            .then_order_by(v::object_version.desc())
                            .into_boxed()
                    })?;

                    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 results = HashMap::new();
        for group in groups {
            for (group_key, stored) in
                group.map_err(|e| Error::Internal(format!("Failed to fetch objects: {e}")))?
            {
                // This particular object is invalid -- it didn't exist at the checkpoint we are
                // viewing at.
                if group_key.checkpoint_viewed_at < stored.checkpoint_sequence_number as u64 {
                    continue;
                }

                let object = Object::try_from_stored_history_object(
                    stored,
                    group_key.checkpoint_viewed_at,
                    // If `LatestAtKey::parent_version` is set, it must have been correctly
                    // propagated from the `Object::root_version` of some object.
                    Some(group_key.parent_version),
                )?;

                let key = ParentVersionKey {
                    id: object.address,
                    checkpoint_viewed_at: group_key.checkpoint_viewed_at,
                    parent_version: group_key.parent_version,
                };

                results.insert(key, object);
            }
        }

        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()
                        })?;
                        let results = resolve_tombstone_versions(conn, results)?;

                        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 object =
                    Object::try_from_stored_history_object(stored, checkpoint_viewed_at, None)?;

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

                results.insert(key, object);
            }
        }

        Ok(results)
    }
}

impl From<&ObjectKind> for ObjectStatus {
    fn from(kind: &ObjectKind) -> Self {
        match kind {
            ObjectKind::NotIndexed(_) => ObjectStatus::NotIndexed,
            ObjectKind::Indexed(_, _) => ObjectStatus::Indexed,
            ObjectKind::WrappedOrDeleted(_) => ObjectStatus::WrappedOrDeleted,
        }
    }
}

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: &NativeMoveObject,
    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(checkpoint_viewed_at, 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(checkpoint_viewed_at, 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);
    }
}