cat_gateway/db/index/queries/purge/
mod.rs

1//! Queries for purging volatile data.
2
3pub(crate) mod catalyst_id_for_public_key;
4pub(crate) mod catalyst_id_for_stake_address;
5pub(crate) mod catalyst_id_for_txn_id;
6pub(crate) mod cip36_registration;
7pub(crate) mod cip36_registration_for_vote_key;
8pub(crate) mod cip36_registration_invalid;
9pub(crate) mod rbac509_invalid_registration;
10pub(crate) mod rbac509_registration;
11pub(crate) mod stake_registration;
12pub(crate) mod txi_by_hash;
13pub(crate) mod txo_ada;
14pub(crate) mod txo_assets;
15pub(crate) mod unstaked_txo_ada;
16pub(crate) mod unstaked_txo_assets;
17
18use std::{fmt::Debug, sync::Arc};
19
20use scylla::{
21    client::{pager::QueryPager, session::Session},
22    serialize::row::SerializeRow,
23    statement::prepared::PreparedStatement,
24};
25
26use super::{FallibleQueryResults, SizedBatch};
27use crate::settings::cassandra_db;
28
29/// No parameters
30const NO_PARAMS: () = ();
31
32/// All prepared DELETE query statements (purge DB table rows).
33#[derive(strum_macros::Display)]
34pub(crate) enum PreparedDeleteQuery {
35    /// TXO Delete query.
36    TxoAda,
37    /// TXO Assets Delete query.
38    TxoAssets,
39    /// Unstaked TXO Delete query.
40    UnstakedTxoAda,
41    /// Unstaked TXO Asset Delete query.
42    UnstakedTxoAsset,
43    /// TXI by TXN Hash Delete query.
44    Txi,
45    /// Stake Registration Delete query.
46    StakeRegistration,
47    /// CIP 36 Registration Delete Query.
48    Cip36Registration,
49    /// CIP 36 Registration Invalid Delete query.
50    Cip36RegistrationInvalid,
51    /// CIP 36 Registration for vote key Delete query.
52    Cip36RegistrationForVoteKey,
53    /// RBAC 509 Registration Delete query.
54    Rbac509,
55    /// Invalid RBAC 509 Registration Delete query.
56    Rbac509Invalid,
57    /// Catalyst ID for transaction ID delete query.
58    CatalystIdForTxnId,
59    /// Catalyst ID for stake address delete query.
60    CatalystIdForStakeAddress,
61    /// Catalyst ID for public key delete query.
62    CatalystIdForPublicKey,
63}
64
65/// All prepared SELECT query statements (primary keys from table).
66#[derive(strum_macros::Display)]
67pub(crate) enum PreparedSelectQuery {
68    /// TXO Select query.
69    TxoAda,
70    /// TXO Asset Select query.
71    TxoAssets,
72    /// Unstaked TXO Select query.
73    UnstakedTxoAda,
74    /// Unstaked TXO Asset Select query.
75    UnstakedTxoAsset,
76    /// TXI by TXN Hash Select query.
77    Txi,
78    /// Stake Registration Select query.
79    StakeRegistration,
80    /// CIP 36 Registration Select Query.
81    Cip36Registration,
82    /// CIP 36 Registration Invalid Select query.
83    Cip36RegistrationInvalid,
84    /// CIP 36 Registration for vote key Select query.
85    Cip36RegistrationForVoteKey,
86    /// RBAC 509 Registration Select query.
87    Rbac509,
88    /// Invalid RBAC 509 Registration Select query.
89    Rbac509Invalid,
90    /// Catalyst ID for transaction ID select query.
91    CatalystIdForTxnId,
92    /// Catalyst ID for stake address select query.
93    CatalystIdForStakeAddress,
94    /// Catalyst ID for public key select query.
95    CatalystIdForPublicKey,
96}
97
98/// All prepared purge queries for a session.
99pub(crate) struct PreparedQueries {
100    /// TXO ADA Primary Key Query.
101    select_txo_ada: PreparedStatement,
102    /// TXO Delete Query.
103    delete_txo_ada: SizedBatch,
104    /// TXO Asset Primary Key Query.
105    select_txo_assets: PreparedStatement,
106    /// TXO Assets Delete Query.
107    delete_txo_assets: SizedBatch,
108    /// Unstaked TXO ADA Primary Key Query.
109    select_unstaked_txo_ada: PreparedStatement,
110    /// Unstaked TXO ADA Delete Query.
111    delete_unstaked_txo_ada: SizedBatch,
112    /// Unstaked TXO Assets Primary Key Query.
113    select_unstaked_txo_assets: PreparedStatement,
114    /// Unstaked TXO Asset Delete Query.
115    delete_unstaked_txo_assets: SizedBatch,
116    /// TXI by TXN Hash by TXN Hash Primary Key Query.
117    select_txi_by_hash: PreparedStatement,
118    /// TXI by TXN Hash Delete Query.
119    delete_txi_by_hash: SizedBatch,
120    /// Stake Registration Primary Key Query.
121    select_stake_registration: PreparedStatement,
122    /// Stake Registration Delete Query.
123    delete_stake_registration: SizedBatch,
124    /// CIP36 Registrations Primary Key Query.
125    select_cip36_registration: PreparedStatement,
126    /// CIP36 Registrations Delete Query.
127    delete_cip36_registration: SizedBatch,
128    /// CIP36 Registration Invalid Primary Key Query.
129    select_cip36_registration_invalid: PreparedStatement,
130    /// CIP36 Registration Invalid Delete Query.
131    delete_cip36_registration_invalid: SizedBatch,
132    /// CIP36 Registration for Vote Key Primary Key Query.
133    select_cip36_registration_for_vote_key: PreparedStatement,
134    /// CIP36 Registration for Vote Key Delete Query.
135    delete_cip36_registration_for_vote_key: SizedBatch,
136    /// RBAC 509 Registrations Primary Key Query.
137    select_rbac509_registration: PreparedStatement,
138    /// RBAC 509 Registrations Delete Query.
139    delete_rbac509_registration: SizedBatch,
140    /// RBAC 509 invalid registrations Primary Key Query.
141    select_rbac509_invalid_registration: PreparedStatement,
142    /// RBAC 509 invalid registrations Delete Query.
143    delete_rbac509_invalid_registration: SizedBatch,
144    /// A Catalyst ID for transaction ID primary key query.
145    select_catalyst_id_for_txn_id: PreparedStatement,
146    /// A Catalyst ID for transaction ID delete query.
147    delete_catalyst_id_for_txn_id: SizedBatch,
148    /// A Catalyst ID for stake address primary key query.
149    select_catalyst_id_for_stake_address: PreparedStatement,
150    /// A Catalyst ID for stake address delete query.
151    delete_catalyst_id_for_stake_address: SizedBatch,
152    /// A Catalyst ID for public key primary key query.
153    select_catalyst_id_for_public_key: PreparedStatement,
154    /// A Catalyst ID for public key delete query.
155    delete_catalyst_id_for_public_key: SizedBatch,
156}
157
158impl PreparedQueries {
159    /// Create new prepared queries for a given session.
160    pub(crate) async fn new(
161        session: Arc<Session>, cfg: &cassandra_db::EnvVars,
162    ) -> anyhow::Result<Self> {
163        // We initialize like this, so that all errors preparing querys get shown before aborting.
164        Ok(Self {
165            select_txo_ada: txo_ada::PrimaryKeyQuery::prepare(&session).await?,
166            delete_txo_ada: txo_ada::DeleteQuery::prepare_batch(&session, cfg).await?,
167            select_txo_assets: txo_assets::PrimaryKeyQuery::prepare(&session).await?,
168            delete_txo_assets: txo_assets::DeleteQuery::prepare_batch(&session, cfg).await?,
169            select_unstaked_txo_ada: unstaked_txo_ada::PrimaryKeyQuery::prepare(&session).await?,
170            delete_unstaked_txo_ada: unstaked_txo_ada::DeleteQuery::prepare_batch(&session, cfg)
171                .await?,
172            select_unstaked_txo_assets: unstaked_txo_assets::PrimaryKeyQuery::prepare(&session)
173                .await?,
174            delete_unstaked_txo_assets: unstaked_txo_assets::DeleteQuery::prepare_batch(
175                &session, cfg,
176            )
177            .await?,
178            select_txi_by_hash: txi_by_hash::PrimaryKeyQuery::prepare(&session).await?,
179            delete_txi_by_hash: txi_by_hash::DeleteQuery::prepare_batch(&session, cfg).await?,
180            select_stake_registration: stake_registration::PrimaryKeyQuery::prepare(&session)
181                .await?,
182            delete_stake_registration: stake_registration::DeleteQuery::prepare_batch(
183                &session, cfg,
184            )
185            .await?,
186            select_cip36_registration: cip36_registration::PrimaryKeyQuery::prepare(&session)
187                .await?,
188            delete_cip36_registration: cip36_registration::DeleteQuery::prepare_batch(
189                &session, cfg,
190            )
191            .await?,
192            select_cip36_registration_invalid:
193                cip36_registration_invalid::PrimaryKeyQuery::prepare(&session).await?,
194            delete_cip36_registration_invalid:
195                cip36_registration_invalid::DeleteQuery::prepare_batch(&session, cfg).await?,
196            select_cip36_registration_for_vote_key:
197                cip36_registration_for_vote_key::PrimaryKeyQuery::prepare(&session).await?,
198            delete_cip36_registration_for_vote_key:
199                cip36_registration_for_vote_key::DeleteQuery::prepare_batch(&session, cfg).await?,
200            select_rbac509_registration: rbac509_registration::PrimaryKeyQuery::prepare(&session)
201                .await?,
202            delete_rbac509_registration: rbac509_registration::DeleteQuery::prepare_batch(
203                &session, cfg,
204            )
205            .await?,
206            select_rbac509_invalid_registration:
207                rbac509_invalid_registration::PrimaryKeyQuery::prepare(&session).await?,
208            delete_rbac509_invalid_registration:
209                rbac509_invalid_registration::DeleteQuery::prepare_batch(&session, cfg).await?,
210            select_catalyst_id_for_txn_id: catalyst_id_for_txn_id::PrimaryKeyQuery::prepare(
211                &session,
212            )
213            .await?,
214            delete_catalyst_id_for_txn_id: catalyst_id_for_txn_id::DeleteQuery::prepare_batch(
215                &session, cfg,
216            )
217            .await?,
218            select_catalyst_id_for_stake_address:
219                catalyst_id_for_stake_address::PrimaryKeyQuery::prepare(&session).await?,
220            delete_catalyst_id_for_stake_address:
221                catalyst_id_for_stake_address::DeleteQuery::prepare_batch(&session, cfg).await?,
222            select_catalyst_id_for_public_key:
223                catalyst_id_for_public_key::PrimaryKeyQuery::prepare(&session).await?,
224            delete_catalyst_id_for_public_key:
225                catalyst_id_for_public_key::DeleteQuery::prepare_batch(&session, cfg).await?,
226        })
227    }
228
229    /// Prepares a statement.
230    pub(crate) async fn prepare(
231        session: Arc<Session>, query: &str, consistency: scylla::statement::Consistency,
232        idempotent: bool,
233    ) -> anyhow::Result<PreparedStatement> {
234        super::PreparedQueries::prepare(session, query, consistency, idempotent).await
235    }
236
237    /// Prepares all permutations of the batch from 1 to max.
238    /// It is necessary to do this because batches are pre-sized, they can not be dynamic.
239    /// Preparing the batches in advance is a very larger performance increase.
240    pub(crate) async fn prepare_batch(
241        session: Arc<Session>, query: &str, cfg: &cassandra_db::EnvVars,
242        consistency: scylla::statement::Consistency, idempotent: bool, logged: bool,
243    ) -> anyhow::Result<SizedBatch> {
244        super::PreparedQueries::prepare_batch(session, query, cfg, consistency, idempotent, logged)
245            .await
246    }
247
248    /// Executes a select query with the given parameters.
249    ///
250    /// Returns an iterator that iterates over all the result pages that the query
251    /// returns.
252    pub(crate) async fn execute_iter(
253        &self, session: Arc<Session>, select_query: PreparedSelectQuery,
254    ) -> anyhow::Result<QueryPager> {
255        let prepared_stmt = match select_query {
256            PreparedSelectQuery::TxoAda => &self.select_txo_ada,
257            PreparedSelectQuery::TxoAssets => &self.select_txo_assets,
258            PreparedSelectQuery::UnstakedTxoAda => &self.select_unstaked_txo_ada,
259            PreparedSelectQuery::UnstakedTxoAsset => &self.select_unstaked_txo_assets,
260            PreparedSelectQuery::Txi => &self.select_txi_by_hash,
261            PreparedSelectQuery::StakeRegistration => &self.select_stake_registration,
262            PreparedSelectQuery::Cip36Registration => &self.select_cip36_registration,
263            PreparedSelectQuery::Cip36RegistrationInvalid => {
264                &self.select_cip36_registration_invalid
265            },
266            PreparedSelectQuery::Cip36RegistrationForVoteKey => {
267                &self.select_cip36_registration_for_vote_key
268            },
269            PreparedSelectQuery::Rbac509 => &self.select_rbac509_registration,
270            PreparedSelectQuery::Rbac509Invalid => &self.select_rbac509_invalid_registration,
271            PreparedSelectQuery::CatalystIdForTxnId => &self.select_catalyst_id_for_txn_id,
272            PreparedSelectQuery::CatalystIdForStakeAddress => {
273                &self.select_catalyst_id_for_stake_address
274            },
275            PreparedSelectQuery::CatalystIdForPublicKey => &self.select_catalyst_id_for_public_key,
276        };
277
278        super::session_execute_iter(session, prepared_stmt, NO_PARAMS).await
279    }
280
281    /// Execute a purge query with the given parameters.
282    pub(crate) async fn execute_batch<T: SerializeRow + Debug>(
283        &self, session: Arc<Session>, cfg: Arc<cassandra_db::EnvVars>, query: PreparedDeleteQuery,
284        values: Vec<T>,
285    ) -> FallibleQueryResults {
286        let query_map = match query {
287            PreparedDeleteQuery::TxoAda => &self.delete_txo_ada,
288            PreparedDeleteQuery::TxoAssets => &self.delete_txo_assets,
289            PreparedDeleteQuery::UnstakedTxoAda => &self.delete_unstaked_txo_ada,
290            PreparedDeleteQuery::UnstakedTxoAsset => &self.delete_unstaked_txo_assets,
291            PreparedDeleteQuery::Txi => &self.delete_txi_by_hash,
292            PreparedDeleteQuery::StakeRegistration => &self.delete_stake_registration,
293            PreparedDeleteQuery::Cip36Registration => &self.delete_cip36_registration,
294            PreparedDeleteQuery::Cip36RegistrationInvalid => {
295                &self.delete_cip36_registration_invalid
296            },
297            PreparedDeleteQuery::Cip36RegistrationForVoteKey => {
298                &self.delete_cip36_registration_for_vote_key
299            },
300            PreparedDeleteQuery::Rbac509 => &self.delete_rbac509_registration,
301            PreparedDeleteQuery::Rbac509Invalid => &self.delete_rbac509_invalid_registration,
302            PreparedDeleteQuery::CatalystIdForTxnId => &self.delete_catalyst_id_for_txn_id,
303            PreparedDeleteQuery::CatalystIdForStakeAddress => {
304                &self.delete_catalyst_id_for_stake_address
305            },
306            PreparedDeleteQuery::CatalystIdForPublicKey => &self.delete_catalyst_id_for_public_key,
307        };
308
309        super::session_execute_batch(session, query_map, cfg, query, values).await
310    }
311}
cat_gateway/db/index/queries/purge/mod.rs

cat_gateway/db/index/queries/purge/
mod.rs
