[indexer-alt-jsonrpc] Add basic objects JSONRPC API

crates/sui-indexer-alt-jsonrpc/Cargo.toml

@@ -19,6 +19,7 @@ bcs.workspace = true
 clap.workspace = true
 diesel = { workspace = true, features = ["chrono"] }
 diesel-async = { workspace = true, features = ["bb8", "postgres", "async-connection-wrapper"] }
+hex.workspace = true
 jsonrpsee = { workspace = true, features = ["macros", "server"] }
 pin-project-lite.workspace = true
 prometheus.workspace = true
@@ -45,3 +46,7 @@ sui-types.workspace = true
 [dev-dependencies]
 reqwest.workspace = true
 serde_json.workspace = true
+tempfile.workspace = true
+
+sui-indexer-alt-framework.workspace = true
+sui-indexer-alt.workspace = true

crates/sui-indexer-alt-jsonrpc/src/api/mod.rs

@@ -2,5 +2,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
 pub(crate) mod governance;
+pub(crate) mod objects;
 pub(crate) mod rpc_module;
 pub(crate) mod transactions;

crates/sui-indexer-alt-jsonrpc/src/api/objects.rs

@@ -0,0 +1,175 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+// FIXME: Add tests.
+// TODO: Migrate to use BigTable for KV storage.
+
+use jsonrpsee::{core::RpcResult, proc_macros::rpc};
+use sui_indexer_alt_schema::objects::StoredObject;
+use sui_json_rpc_types::{
+    SuiGetPastObjectRequest, SuiObjectData, SuiObjectDataOptions, SuiObjectRef,
+    SuiPastObjectResponse,
+};
+use sui_open_rpc::Module;
+use sui_open_rpc_macros::open_rpc;
+use sui_types::{
+    base_types::{ObjectID, SequenceNumber},
+    digests::ObjectDigest,
+    object::Object,
+};
+
+use crate::{
+    context::Context,
+    error::{internal_error, invalid_params},
+};
+
+use super::rpc_module::RpcModule;
+
+#[open_rpc(namespace = "sui", tag = "Objects API")]
+#[rpc(server, namespace = "sui")]
+trait ObjectsApi {
+    /// Note there is no software-level guarantee/SLA that objects with past versions
+    /// can be retrieved by this API, even if the object and version exists/existed.
+    /// The result may vary across nodes depending on their pruning policies.
+    /// Return the object information for a specified version
+    #[method(name = "tryGetPastObject")]
+    async fn try_get_past_object(
+        &self,
+        /// the ID of the queried object
+        object_id: ObjectID,
+        /// the version of the queried object. If None, default to the latest known version
+        version: SequenceNumber,
+        /// options for specifying the content to be returned
+        options: Option<SuiObjectDataOptions>,
+    ) -> RpcResult<SuiPastObjectResponse>;
+
+    /// Note there is no software-level guarantee/SLA that objects with past versions
+    /// can be retrieved by this API, even if the object and version exists/existed.
+    /// The result may vary across nodes depending on their pruning policies.
+    /// Return the object information for a specified version
+    #[method(name = "tryMultiGetPastObjects")]
+    async fn try_multi_get_past_objects(
+        &self,
+        /// a vector of object and versions to be queried
+        past_objects: Vec<SuiGetPastObjectRequest>,
+        /// options for specifying the content to be returned
+        options: Option<SuiObjectDataOptions>,
+    ) -> RpcResult<Vec<SuiPastObjectResponse>>;
+}
+
+pub(crate) struct Objects(pub Context);
+
+#[derive(thiserror::Error, Debug)]
+pub(crate) enum Error {
+    #[error("Object not found: {0} with version {1}")]
+    NotFound(ObjectID, SequenceNumber),
+
+    #[error("Error converting to response: {0}")]
+    Conversion(anyhow::Error),
+
+    #[error("Deserialization error: {0}")]
+    Deserialization(#[from] bcs::Error),
+}
+
+#[async_trait::async_trait]
+impl ObjectsApiServer for Objects {
+    async fn try_get_past_object(
+        &self,
+        object_id: ObjectID,
+        version: SequenceNumber,
+        options: Option<SuiObjectDataOptions>,
+    ) -> RpcResult<SuiPastObjectResponse> {
+        let Self(ctx) = self;
+        let Some(stored) = ctx
+            .loader()
+            .load_one((object_id, version))
+            .await
+            .map_err(internal_error)?
+        else {
+            return Err(invalid_params(Error::NotFound(object_id, version)));
+        };
+
+        let options = options.unwrap_or_default();
+        response(ctx, &stored, &options)
+            .await
+            .map_err(internal_error)
+    }
+
+    async fn try_multi_get_past_objects(
+        &self,
+        past_objects: Vec<SuiGetPastObjectRequest>,
+        options: Option<SuiObjectDataOptions>,
+    ) -> RpcResult<Vec<SuiPastObjectResponse>> {
+        let Self(ctx) = self;
+        let stored_objects = ctx
+            .loader()
+            .load_many(past_objects.iter().map(|p| (p.object_id, p.version)))
+            .await
+            .map_err(internal_error)?;
+
+        let mut responses = Vec::with_capacity(past_objects.len());
+        let options = options.unwrap_or_default();
+        for request in past_objects {
+            if let Some(stored) = stored_objects.get(&(request.object_id, request.version)) {
+                responses.push(
+                    response(ctx, stored, &options)
+                        .await
+                        .map_err(internal_error)?,
+                );
+            } else {
+                responses.push(SuiPastObjectResponse::VersionNotFound(
+                    request.object_id,
+                    request.version,
+                ));
+            }
+        }
+
+        Ok(responses)
+    }
+}
+
+impl RpcModule for Objects {
+    fn schema(&self) -> Module {
+        ObjectsApiOpenRpc::module_doc()
+    }
+
+    fn into_impl(self) -> jsonrpsee::RpcModule<Self> {
+        self.into_rpc()
+    }
+}
+
+/// Convert the representation of an object from the database into the response format,
+/// including the fields requested in the `options`.
+/// FIXME: Actually use the options.
+pub(crate) async fn response(
+    _ctx: &Context,
+    stored: &StoredObject,
+    _options: &SuiObjectDataOptions,
+) -> Result<SuiPastObjectResponse, Error> {
+    let object_id =
+        ObjectID::from_bytes(&stored.object_id).map_err(|e| Error::Conversion(e.into()))?;
+    let version = SequenceNumber::from_u64(stored.object_version as u64);
+
+    let Some(serialized_object) = &stored.serialized_object else {
+        return Ok(SuiPastObjectResponse::ObjectDeleted(SuiObjectRef {
+            object_id,
+            version,
+            digest: ObjectDigest::OBJECT_DIGEST_DELETED,
+        }));
+    };
+    let object: Object = bcs::from_bytes(serialized_object).map_err(Error::Deserialization)?;
+    let object_data = SuiObjectData {
+        object_id,
+        version,
+        digest: object.digest(),
+        type_: None,
+        owner: None,
+        previous_transaction: None,
+        storage_rebate: None,
+        display: None,
+        content: None,
+        bcs: None,
+    };
+
+    Ok(SuiPastObjectResponse::VersionFound(object_data))
+}

crates/sui-indexer-alt-jsonrpc/src/data/mod.rs

@@ -1,6 +1,7 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
+pub(crate) mod objects;
 pub(crate) mod package_resolver;
 pub(crate) mod reader;
 pub mod system_package_task;

crates/sui-indexer-alt-jsonrpc/src/data/objects.rs

@@ -0,0 +1,147 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::{collections::HashMap, sync::Arc};
+
+use async_graphql::dataloader::Loader;
+use sui_indexer_alt_schema::objects::StoredObject;
+use sui_types::{base_types::ObjectID, base_types::SequenceNumber};
+
+use super::reader::{ReadError, Reader};
+
+/// Load objects by key (object_id, version).
+#[async_trait::async_trait]
+impl Loader<(ObjectID, SequenceNumber)> for Reader {
+    type Value = StoredObject;
+    type Error = Arc<ReadError>;
+
+    async fn load(
+        &self,
+        keys: &[(ObjectID, SequenceNumber)],
+    ) -> Result<HashMap<(ObjectID, SequenceNumber), Self::Value>, Self::Error> {
+        if keys.is_empty() {
+            return Ok(HashMap::new());
+        }
+
+        let conditions = keys
+            .iter()
+            .map(|key| {
+                format!(
+                    "(object_id = '\\x{}'::bytea AND object_version = {})",
+                    hex::encode(key.0.to_vec()),
+                    key.1.value()
+                )
+            })
+            .collect::<Vec<_>>();
+        let query = format!("SELECT * FROM kv_objects WHERE {}", conditions.join(" OR "));
+
+        let mut conn = self.connect().await.map_err(Arc::new)?;
+        let objects: Vec<StoredObject> = conn.raw_query(&query).await.map_err(Arc::new)?;
+
+        let results: HashMap<_, _> = objects
+            .into_iter()
+            .map(|stored| {
+                (
+                    (
+                        ObjectID::from_bytes(&stored.object_id).unwrap(),
+                        SequenceNumber::from_u64(stored.object_version as u64),
+                    ),
+                    stored,
+                )
+            })
+            .collect();
+
+        Ok(results)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use diesel_async::RunQueryDsl;
+    use sui_indexer_alt_schema::{objects::StoredObject, schema::kv_objects};
+    use sui_types::{
+        base_types::{ObjectID, SequenceNumber, SuiAddress},
+        object::{Object, Owner},
+    };
+
+    use crate::test_env::IndexerReaderTestEnv;
+
+    async fn insert_objects(
+        test_env: &IndexerReaderTestEnv,
+        id_versions: impl IntoIterator<Item = (ObjectID, SequenceNumber)>,
+    ) {
+        let mut conn = test_env.indexer.db().connect().await.unwrap();
+        let stored_objects = id_versions
+            .into_iter()
+            .map(|(id, version)| {
+                let object = Object::with_id_owner_version_for_testing(
+                    id,
+                    version,
+                    Owner::AddressOwner(SuiAddress::ZERO),
+                );
+                let serialized_object = bcs::to_bytes(&object).unwrap();
+                StoredObject {
+                    object_id: id.to_vec(),
+                    object_version: version.value() as i64,
+                    serialized_object: Some(serialized_object),
+                }
+            })
+            .collect::<Vec<_>>();
+        diesel::insert_into(kv_objects::table)
+            .values(stored_objects)
+            .execute(&mut conn)
+            .await
+            .unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_load_single_object() {
+        let test_env = IndexerReaderTestEnv::new().await;
+        let id_version = (ObjectID::ZERO, SequenceNumber::from_u64(1));
+        insert_objects(&test_env, vec![id_version]).await;
+        let object = test_env
+            .loader()
+            .load_one(id_version)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(object.object_id, id_version.0.to_vec());
+        assert_eq!(object.object_version, id_version.1.value() as i64);
+    }
+
+    #[tokio::test]
+    async fn test_load_multiple_objects() {
+        let test_env = IndexerReaderTestEnv::new().await;
+        let mut id_versions = vec![
+            (ObjectID::ZERO, SequenceNumber::from_u64(1)),
+            (ObjectID::ZERO, SequenceNumber::from_u64(2)),
+            (ObjectID::ZERO, SequenceNumber::from_u64(10)),
+            (ObjectID::from_single_byte(1), SequenceNumber::from_u64(1)),
+            (ObjectID::from_single_byte(1), SequenceNumber::from_u64(2)),
+            (ObjectID::from_single_byte(1), SequenceNumber::from_u64(10)),
+        ];
+        insert_objects(&test_env, id_versions.clone()).await;
+
+        let objects = test_env
+            .loader()
+            .load_many(id_versions.clone())
+            .await
+            .unwrap();
+        assert_eq!(objects.len(), id_versions.len());
+        for (id, version) in &id_versions {
+            let object = objects.get(&(*id, *version)).unwrap();
+            assert_eq!(object.object_id, id.to_vec());
+            assert_eq!(object.object_version, version.value() as i64);
+        }
+
+        // Add a ID/version that doesn't exist in the table.
+        // Query will still succeed, but will return the same set of objects as before.
+        id_versions.push((ObjectID::from_single_byte(2), SequenceNumber::from_u64(1)));
+        let objects = test_env
+            .loader()
+            .load_many(id_versions.clone())
+            .await
+            .unwrap();
+        assert_eq!(objects.len(), id_versions.len() - 1);
+    }
+}