sui_indexer_alt/handlers/
ev_struct_inst.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::{collections::BTreeSet, ops::Range, sync::Arc};
5
6use anyhow::{Context, Result};
7use diesel::{ExpressionMethods, QueryDsl};
8use diesel_async::RunQueryDsl;
9use sui_indexer_alt_framework::{
10    pipeline::Processor,
11    postgres::{Connection, handler::Handler},
12    types::full_checkpoint_content::Checkpoint,
13};
14use sui_indexer_alt_schema::{events::StoredEvStructInst, schema::ev_struct_inst};
15
16use crate::handlers::cp_sequence_numbers::tx_interval;
17use async_trait::async_trait;
18
19pub(crate) struct EvStructInst;
20
21#[async_trait]
22impl Processor for EvStructInst {
23    const NAME: &'static str = "ev_struct_inst";
24
25    type Value = StoredEvStructInst;
26
27    async fn process(&self, checkpoint: &Arc<Checkpoint>) -> Result<Vec<Self::Value>> {
28        let Checkpoint {
29            transactions,
30            summary,
31            ..
32        } = checkpoint.as_ref();
33
34        let mut values = BTreeSet::new();
35        let first_tx = summary.network_total_transactions as usize - transactions.len();
36
37        for (i, tx) in transactions.iter().enumerate() {
38            let tx_sequence_number = (first_tx + i) as i64;
39            for (j, ev) in tx.events.iter().flat_map(|evs| evs.data.iter().enumerate()) {
40                values.insert(StoredEvStructInst {
41                    package: ev.type_.address.to_vec(),
42                    module: ev.type_.module.to_string(),
43                    name: ev.type_.name.to_string(),
44                    instantiation: bcs::to_bytes(&ev.type_.type_params)
45                        .with_context(|| format!(
46                            "Failed to serialize type parameters for event ({tx_sequence_number}, {j})"
47                        ))?,
48                    tx_sequence_number: (first_tx + i) as i64,
49                    sender: ev.sender.to_vec(),
50                });
51            }
52        }
53
54        Ok(values.into_iter().collect())
55    }
56}
57
58#[async_trait]
59impl Handler for EvStructInst {
60    const MIN_EAGER_ROWS: usize = 100;
61    const MAX_PENDING_ROWS: usize = 10000;
62
63    async fn commit<'a>(values: &[Self::Value], conn: &mut Connection<'a>) -> Result<usize> {
64        Ok(diesel::insert_into(ev_struct_inst::table)
65            .values(values)
66            .on_conflict_do_nothing()
67            .execute(conn)
68            .await?)
69    }
70
71    async fn prune<'a>(
72        &self,
73        from: u64,
74        to_exclusive: u64,
75        conn: &mut Connection<'a>,
76    ) -> Result<usize> {
77        let Range {
78            start: from_tx,
79            end: to_tx,
80        } = tx_interval(conn, from..to_exclusive).await?;
81
82        let filter = ev_struct_inst::table
83            .filter(ev_struct_inst::tx_sequence_number.between(from_tx as i64, to_tx as i64 - 1));
84
85        Ok(diesel::delete(filter).execute(conn).await?)
86    }
87}
88
89#[cfg(test)]
90mod tests {
91    use super::*;
92    use diesel_async::RunQueryDsl;
93    use sui_indexer_alt_framework::{
94        Indexer,
95        types::{event::Event, test_checkpoint_data_builder::TestCheckpointBuilder},
96    };
97    use sui_indexer_alt_schema::MIGRATIONS;
98
99    use crate::handlers::cp_sequence_numbers::CpSequenceNumbers;
100
101    async fn get_all_ev_struct_inst(conn: &mut Connection<'_>) -> Result<Vec<StoredEvStructInst>> {
102        let query = ev_struct_inst::table
103            .order_by((
104                ev_struct_inst::tx_sequence_number,
105                ev_struct_inst::sender,
106                ev_struct_inst::package,
107                ev_struct_inst::module,
108                ev_struct_inst::name,
109                ev_struct_inst::instantiation,
110            ))
111            .load(conn)
112            .await?;
113        Ok(query)
114    }
115
116    #[tokio::test]
117    async fn test_ev_struct_inst_pruning_complains_if_no_mapping() {
118        let (indexer, _db) = Indexer::new_for_testing(&MIGRATIONS).await;
119        let mut conn = indexer.store().connect().await.unwrap();
120
121        let result = EvStructInst.prune(0, 2, &mut conn).await;
122
123        assert!(result.is_err());
124        assert_eq!(
125            result.unwrap_err().to_string(),
126            "No checkpoint mapping found for checkpoint 0"
127        );
128    }
129
130    #[tokio::test]
131    async fn test_ev_struct_inst_process_no_events() {
132        let (indexer, _db) = Indexer::new_for_testing(&MIGRATIONS).await;
133        let mut conn = indexer.store().connect().await.unwrap();
134
135        let checkpoint: Arc<Checkpoint> = Arc::new(
136            TestCheckpointBuilder::new(0)
137                .start_transaction(0)
138                .finish_transaction()
139                .build_checkpoint(),
140        );
141
142        let values = EvStructInst.process(&checkpoint).await.unwrap();
143        EvStructInst::commit(&values, &mut conn).await.unwrap();
144
145        assert_eq!(values.len(), 0);
146    }
147
148    #[tokio::test]
149    async fn test_ev_struct_inst_process_single_event() {
150        let (indexer, _db) = Indexer::new_for_testing(&MIGRATIONS).await;
151        let mut conn = indexer.store().connect().await.unwrap();
152
153        let checkpoint: Arc<Checkpoint> = Arc::new(
154            TestCheckpointBuilder::new(0)
155                .start_transaction(0)
156                .with_events(vec![Event::random_for_testing()])
157                .finish_transaction()
158                .build_checkpoint(),
159        );
160
161        // Process checkpoint with one event
162        let values = EvStructInst.process(&checkpoint).await.unwrap();
163        EvStructInst::commit(&values, &mut conn).await.unwrap();
164
165        let events = get_all_ev_struct_inst(&mut conn).await.unwrap();
166        assert_eq!(events.len(), 1);
167    }
168
169    #[tokio::test]
170    async fn test_ev_struct_inst_prune_events() {
171        let (indexer, _db) = Indexer::new_for_testing(&MIGRATIONS).await;
172        let mut conn = indexer.store().connect().await.unwrap();
173
174        // 0th checkpoint has no events
175        let mut builder = TestCheckpointBuilder::new(0);
176        builder = builder.start_transaction(0).finish_transaction();
177        let checkpoint = Arc::new(builder.build_checkpoint());
178        let values = EvStructInst.process(&checkpoint).await.unwrap();
179
180        EvStructInst::commit(&values, &mut conn).await.unwrap();
181        let values = CpSequenceNumbers.process(&checkpoint).await.unwrap();
182        CpSequenceNumbers::commit(&values, &mut conn).await.unwrap();
183
184        // 1st checkpoint has 1 event
185        builder = builder
186            .start_transaction(0)
187            .with_events(vec![Event::random_for_testing()])
188            .finish_transaction();
189        let checkpoint = Arc::new(builder.build_checkpoint());
190        let values = EvStructInst.process(&checkpoint).await.unwrap();
191
192        EvStructInst::commit(&values, &mut conn).await.unwrap();
193        let values = CpSequenceNumbers.process(&checkpoint).await.unwrap();
194        CpSequenceNumbers::commit(&values, &mut conn).await.unwrap();
195
196        // 2nd checkpoint has 2 events
197        builder = builder
198            .start_transaction(0)
199            .with_events(vec![
200                Event::random_for_testing(),
201                Event::random_for_testing(),
202            ])
203            .finish_transaction();
204        let checkpoint = Arc::new(builder.build_checkpoint());
205        let values = EvStructInst.process(&checkpoint).await.unwrap();
206
207        EvStructInst::commit(&values, &mut conn).await.unwrap();
208        let values = CpSequenceNumbers.process(&checkpoint).await.unwrap();
209        CpSequenceNumbers::commit(&values, &mut conn).await.unwrap();
210
211        // Prune checkpoints from `[0, 2)`, expect 2 events remaining
212        let rows_pruned = EvStructInst.prune(0, 2, &mut conn).await.unwrap();
213        assert_eq!(rows_pruned, 1);
214
215        let remaining_events = get_all_ev_struct_inst(&mut conn).await.unwrap();
216        assert_eq!(remaining_events.len(), 2);
217    }
218}
sui_indexer_alt/handlers/ev_struct_inst.rs

sui_indexer_alt/handlers/
ev_struct_inst.rs
