indexer-alt: sum_displays pipeline

## Description

Pipeline for indexing Display VersionUpdated events, so readers can
render Display for Objects.

## Test plan

Run the indexer on the first ~1.6M checkpoints (the first display events
happen around checkpoint 1.5M), and then inspect the database:

```
sui$ cargo run -p sui-indexer-alt --release --                                   \
  --database-url "postgres://postgres:postgrespw@localhost:5432/sui_indexer_alt" \
  indexer --remote-store-url https://checkpoints.mainnet.sui.io                  \
  --consistent-range 3600
```

```
sui_indexer_alt=# SELECT object_type, display_id FROM sum_displays LIMIT 10;
                                                                                                              object_type                                                                                                               |                             display_id
----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+--------------------------------------------------------------------
 \x72f9c76421170b5a797432ba9e1b3b2e2b7cf6faa26eb955396c773af2479e1e0967616d655f383139320847616d653831393200                                                                                                                             | \xb06e4607642f6c08e196f8bd84b653c784864a6badca2066e63db42e846fdd4c
 \x72ad8f46d4fc3bbc24df15f6c62a98044c2649a9a31895a957254b2cf16e0cb905646d656e7305446d656e7300                                                                                                                                           | \x831cd05a3de12ae3dcea4382e9f56eadded8efd83c742036acd789fc41f1a03f
 \x72ad8f46d4fc3bbc24df15f6c62a98044c2649a9a31895a957254b2cf16e0cb905646d656e7309446d656e734d65746100                                                                                                                                   | \x3e8acb139222c9e335980f8b427270e92d276152140545bfde46c700472a2e11
 \xee496a0cc04d06a345982ba6697c90c619020de9e274408c7819f787ff66e1a1087375696672656e73075375694672656e0107ee496a0cc04d06a345982ba6697c90c619020de9e274408c7819f787ff66e1a10463617079044361707900                                         | \x3d53effd472191435e7735ff9ed0b3083ef10dee4793db51df92038cea48b139
 \x5b45da03d42b064f5e051741b6fed3b29eb817c7923b83b92f37a1d2abf4fbab036e6674034e667400                                                                                                                                                   | \x23453afb61ba4e0c9f040839c0d620c2cab774491656c086e10871aae8f8b3e9
 \xbb1531504c9c3235d3cd637ed9573cbe18461255b4175a1cb1e1b07b8aa8e11b0873686f77636173650853686f776361736500                                                                                                                               | \xa306db0c6d3f5cbb8169143c22da5b61caf3576f1b3f3e77537aef8d4ca4b515
 \xbb1531504c9c3235d3cd637ed9573cbe18461255b4175a1cb1e1b07b8aa8e11b0473756961054d6564616c00                                                                                                                                             | \x70e764f567b976d8e654064c14a7fbd94f0994a229fb0a7e119d80b4f1e4b760
 \xbb1531504c9c3235d3cd637ed9573cbe18461255b4175a1cb1e1b07b8aa8e11b04737569610d506572736f6e616c4d6564616c00                                                                                                                             | \xb438c4cbda18b0890de8ca2a1813be70fc3b64fedf171ea7c658d43605a11ece
 \xbb1531504c9c3235d3cd637ed9573cbe18461255b4175a1cb1e1b07b8aa8e11b08746573745f6e66740b53756961546573744e465400                                                                                                                         | \xa37afca51a40bc90e30991631515f2ab8b9ecef90e51fbf4af12aa0ea5413250
 \xf1e7b4093872df432b2adb50533a1fa6f9e4b20856217657242fe7d9d4ac42e80b6574686f735f73717561640a4574686f7353717561640107f1e7b4093872df432b2adb50533a1fa6f9e4b20856217657242fe7d9d4ac42e80b6574686f735f73717561640b4574686f7353717561643100 | \x7c330d61f99af7983e8714b4e63437d93813316ac54e8f83103b2ab2a021997e
(10 rows)
```

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_objects;

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,9 +10,10 @@ 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_objects::TxAffectedObjects, tx_balance_changes::TxBalanceChanges,
-        wal_coin_balances::WalCoinBalances, wal_obj_types::WalObjTypes,
+        sum_coin_balances::SumCoinBalances, sum_displays::SumDisplays, sum_obj_types::SumObjTypes,
+        sum_packages::SumPackages, tx_affected_objects::TxAffectedObjects,
+        tx_balance_changes::TxBalanceChanges, wal_coin_balances::WalCoinBalances,
+        wal_obj_types::WalObjTypes,
     },
     Indexer,
 };
@@ -47,6 +48,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,
@@ -153,6 +162,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_objects,