sui_indexer_alt/handlers/
sum_displays.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::collections::BTreeMap;
5use std::sync::Arc;
6
7use anyhow::Result;
8use anyhow::anyhow;
9use async_trait::async_trait;
10use diesel::ExpressionMethods;
11use diesel::upsert::excluded;
12use diesel_async::RunQueryDsl;
13use sui_indexer_alt_framework::FieldCount;
14use sui_indexer_alt_framework::pipeline::Processor;
15use sui_indexer_alt_framework::pipeline::sequential::Handler;
16use sui_indexer_alt_framework::postgres::Connection;
17use sui_indexer_alt_framework::postgres::Db;
18use sui_indexer_alt_framework::types::display::DisplayVersionUpdatedEvent;
19use sui_indexer_alt_framework::types::full_checkpoint_content::Checkpoint;
20use sui_indexer_alt_schema::displays::StoredDisplay;
21use sui_indexer_alt_schema::schema::sum_displays;
22
23const MAX_INSERT_CHUNK_ROWS: usize = i16::MAX as usize / StoredDisplay::FIELD_COUNT;
24
25pub(crate) struct SumDisplays;
26
27#[async_trait]
28impl Processor for SumDisplays {
29    const NAME: &'static str = "sum_displays";
30
31    type Value = StoredDisplay;
32
33    async fn process(&self, checkpoint: &Arc<Checkpoint>) -> Result<Vec<Self::Value>> {
34        let Checkpoint { transactions, .. } = checkpoint.as_ref();
35
36        let mut values = vec![];
37        for tx in transactions {
38            let Some(events) = &tx.events else {
39                continue;
40            };
41
42            for event in &events.data {
43                let Some((object_type, update)) = DisplayVersionUpdatedEvent::try_from_event(event)
44                else {
45                    continue;
46                };
47
48                values.push(StoredDisplay {
49                    object_type: bcs::to_bytes(&object_type).map_err(|e| {
50                        anyhow!(
51                            "Error serializing object type {}: {e}",
52                            object_type.to_canonical_display(/* with_prefix */ true)
53                        )
54                    })?,
55
56                    display_id: update.id.bytes.to_vec(),
57                    display_version: update.version as i16,
58                    display: event.contents.clone(),
59                })
60            }
61        }
62
63        Ok(values)
64    }
65}
66
67#[async_trait]
68impl Handler for SumDisplays {
69    type Store = Db;
70    type Batch = BTreeMap<Vec<u8>, Self::Value>;
71
72    fn batch(&self, batch: &mut Self::Batch, values: std::vec::IntoIter<Self::Value>) {
73        for value in values {
74            batch.insert(value.object_type.clone(), value);
75        }
76    }
77
78    async fn commit<'a>(&self, batch: &Self::Batch, conn: &mut Connection<'a>) -> Result<usize> {
79        let values: Vec<_> = batch.values().cloned().collect();
80        let mut updates = 0;
81        for chunk in values.chunks(MAX_INSERT_CHUNK_ROWS) {
82            updates += diesel::insert_into(sum_displays::table)
83                .values(chunk)
84                .on_conflict(sum_displays::object_type)
85                .do_update()
86                .set((
87                    sum_displays::display_id.eq(excluded(sum_displays::display_id)),
88                    sum_displays::display_version.eq(excluded(sum_displays::display_version)),
89                    sum_displays::display.eq(excluded(sum_displays::display)),
90                ))
91                .execute(conn)
92                .await?;
93        }
94
95        Ok(updates)
96    }
97}
sui_indexer_alt/handlers/sum_displays.rs

sui_indexer_alt/handlers/
sum_displays.rs
