indexer-alt: sum_displays pipeline

crates/sui-indexer-alt/migrations/2024-10-31-174742_sum_displays/down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS sum_displays;

crates/sui-indexer-alt/migrations/2024-10-31-174742_sum_displays/up.sql

@@ -0,0 +1,13 @@
+-- This table tracks the latest versions of `Display`, keyed by Object type.
+CREATE TABLE IF NOT EXISTS sum_displays
+(
+    -- BCS-encoded StructTag of the object that this Display belongs to.
+    object_type                 BYTEA         PRIMARY KEY,
+    -- Object ID of the Display object
+    display_id                  BYTEA         NOT NULL,
+    -- Version of the Display object (In the VersionUpdate event this is stored as a u16)
+    display_version             SMALLINT      NOT NULL,
+    -- BCS-encoded content of DisplayVersionUpdatedEvent that was indexed into
+    -- this record.
+    display                     BYTEA         NOT NULL
+);

crates/sui-indexer-alt/src/handlers/mod.rs

@@ -8,6 +8,7 @@ pub mod kv_objects;
 pub mod kv_transactions;
 pub mod obj_versions;
 pub mod sum_coin_balances;
+pub mod sum_displays;
 pub mod sum_obj_types;
 pub mod sum_packages;
 pub mod tx_affected_addresses;

crates/sui-indexer-alt/src/handlers/sum_displays.rs

@@ -0,0 +1,89 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::{collections::BTreeMap, sync::Arc};
+
+use anyhow::{anyhow, Result};
+use diesel::{upsert::excluded, ExpressionMethods};
+use diesel_async::RunQueryDsl;
+use futures::future::try_join_all;
+use sui_types::{display::DisplayVersionUpdatedEvent, full_checkpoint_content::CheckpointData};
+
+use crate::{
+    db,
+    models::displays::StoredDisplay,
+    pipeline::{sequential::Handler, Processor},
+    schema::sum_displays,
+};
+
+const CHUNK_ROWS: usize = i16::MAX as usize / 4;
+
+pub struct SumDisplays;
+
+impl Processor for SumDisplays {
+    const NAME: &'static str = "sum_displays";
+
+    type Value = StoredDisplay;
+
+    fn process(checkpoint: &Arc<CheckpointData>) -> Result<Vec<Self::Value>> {
+        let CheckpointData { transactions, .. } = checkpoint.as_ref();
+
+        let mut values = vec![];
+        for tx in transactions {
+            let Some(events) = &tx.events else {
+                continue;
+            };
+
+            for event in &events.data {
+                let Some((object_type, update)) = DisplayVersionUpdatedEvent::try_from_event(event)
+                else {
+                    continue;
+                };
+
+                values.push(StoredDisplay {
+                    object_type: bcs::to_bytes(&object_type).map_err(|e| {
+                        anyhow!(
+                            "Error serializing object type {}: {e}",
+                            object_type.to_canonical_display(/* with_prefix */ true)
+                        )
+                    })?,
+
+                    display_id: update.id.bytes.to_vec(),
+                    display_version: update.version as i16,
+                    display: event.contents.clone(),
+                })
+            }
+        }
+
+        Ok(values)
+    }
+}
+
+#[async_trait::async_trait]
+impl Handler for SumDisplays {
+    type Batch = BTreeMap<Vec<u8>, Self::Value>;
+
+    fn batch(batch: &mut Self::Batch, values: Vec<Self::Value>) {
+        for value in values {
+            batch.insert(value.object_type.clone(), value);
+        }
+    }
+
+    async fn commit(batch: &Self::Batch, conn: &mut db::Connection<'_>) -> Result<usize> {
+        let values: Vec<_> = batch.values().cloned().collect();
+        let updates = values.chunks(CHUNK_ROWS).map(|chunk| {
+            diesel::insert_into(sum_displays::table)
+                .values(chunk)
+                .on_conflict(sum_displays::object_type)
+                .do_update()
+                .set((
+                    sum_displays::display_id.eq(excluded(sum_displays::display_id)),
+                    sum_displays::display_version.eq(excluded(sum_displays::display_version)),
+                    sum_displays::display.eq(excluded(sum_displays::display)),
+                ))
+                .execute(conn)
+        });
+
+        Ok(try_join_all(updates).await?.into_iter().sum())
+    }
+}

crates/sui-indexer-alt/src/main.rs

@@ -10,10 +10,11 @@ use sui_indexer_alt::{
     handlers::{
         ev_emit_mod::EvEmitMod, ev_struct_inst::EvStructInst, kv_checkpoints::KvCheckpoints,
         kv_objects::KvObjects, kv_transactions::KvTransactions, obj_versions::ObjVersions,
-        sum_coin_balances::SumCoinBalances, sum_obj_types::SumObjTypes, sum_packages::SumPackages,
-        tx_affected_addresses::TxAffectedAddress, tx_affected_objects::TxAffectedObjects,
-        tx_balance_changes::TxBalanceChanges, tx_calls_fun::TxCallsFun, tx_digests::TxDigests,
-        tx_kinds::TxKinds, wal_coin_balances::WalCoinBalances, wal_obj_types::WalObjTypes,
+        sum_coin_balances::SumCoinBalances, sum_displays::SumDisplays, sum_obj_types::SumObjTypes,
+        sum_packages::SumPackages, tx_affected_addresses::TxAffectedAddress,
+        tx_affected_objects::TxAffectedObjects, tx_balance_changes::TxBalanceChanges,
+        tx_calls_fun::TxCallsFun, tx_digests::TxDigests, tx_kinds::TxKinds,
+        wal_coin_balances::WalCoinBalances, wal_obj_types::WalObjTypes,
     },
     Indexer,
 };
@@ -53,6 +54,7 @@ async fn main() -> Result<()> {
             indexer.concurrent_pipeline::<WalCoinBalances>().await?;
             indexer.concurrent_pipeline::<WalObjTypes>().await?;
             indexer.sequential_pipeline::<SumCoinBalances>(lag).await?;
+            indexer.sequential_pipeline::<SumDisplays>(None).await?;
             indexer.sequential_pipeline::<SumObjTypes>(lag).await?;
             indexer.sequential_pipeline::<SumPackages>(None).await?;
 

crates/sui-indexer-alt/src/models/displays.rs

@@ -0,0 +1,15 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use diesel::prelude::*;
+
+use crate::schema::sum_displays;
+
+#[derive(Insertable, Debug, Clone)]
+#[diesel(table_name = sum_displays, primary_key(object_type))]
+pub struct StoredDisplay {
+    pub object_type: Vec<u8>,
+    pub display_id: Vec<u8>,
+    pub display_version: i16,
+    pub display: Vec<u8>,
+}

crates/sui-indexer-alt/src/models/mod.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 pub mod checkpoints;
+pub mod displays;
 pub mod events;
 pub mod objects;
 pub mod packages;

crates/sui-indexer-alt/src/schema.rs

@@ -68,6 +68,15 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    sum_displays (object_type) {
+        object_type -> Bytea,
+        display_id -> Bytea,
+        display_version -> Int2,
+        display -> Bytea,
+    }
+}
+
 diesel::table! {
     sum_obj_types (object_id) {
         object_id -> Bytea,
@@ -185,6 +194,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     kv_transactions,
     obj_versions,
     sum_coin_balances,
+    sum_displays,
     sum_obj_types,
     sum_packages,
     tx_affected_addresses,