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

1//! CIP-36 registration Queries used in purging data.
2use std::{fmt::Debug, sync::Arc};
3
4use scylla::{
5    prepared_statement::PreparedStatement, transport::iterator::TypedRowStream, SerializeRow,
6    Session,
7};
8use tracing::error;
9
10use crate::{
11    db::{
12        index::{
13            queries::{
14                purge::{PreparedDeleteQuery, PreparedQueries, PreparedSelectQuery},
15                FallibleQueryResults, SizedBatch,
16            },
17            session::CassandraSession,
18        },
19        types::{DbSlot, DbTxnIndex},
20    },
21    settings::cassandra_db,
22};
23
24pub(crate) mod result {
25    //! Return values for CIP-36 registration purge queries.
26
27    use crate::db::types::{DbSlot, DbTxnIndex};
28
29    /// Primary Key Row
30    pub(crate) type PrimaryKey = (Vec<u8>, num_bigint::BigInt, DbSlot, DbTxnIndex);
31}
32
33/// Select primary keys for CIP-36 registration.
34const SELECT_QUERY: &str = include_str!("./cql/get_cip36_registration.cql");
35
36/// Primary Key Value.
37#[derive(SerializeRow)]
38pub(crate) struct Params {
39    /// Full Stake Address (not hashed, 32 byte ED25519 Public key).
40    pub(crate) stake_public_key: Vec<u8>,
41    /// Nonce that has been slot corrected.
42    pub(crate) nonce: num_bigint::BigInt,
43    /// Block Slot Number
44    pub(crate) slot_no: DbSlot,
45    /// Transaction Offset inside the block.
46    pub(crate) txn_index: DbTxnIndex,
47}
48
49impl Debug for Params {
50    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
51        f.debug_struct("Params")
52            .field("stake_public_key", &self.stake_public_key)
53            .field("nonce", &self.nonce)
54            .field("slot_no", &self.slot_no)
55            .field("txn_index", &self.txn_index)
56            .finish()
57    }
58}
59
60impl From<result::PrimaryKey> for Params {
61    fn from(value: result::PrimaryKey) -> Self {
62        Self {
63            stake_public_key: value.0,
64            nonce: value.1,
65            slot_no: value.2,
66            txn_index: value.3,
67        }
68    }
69}
70/// Get primary key for CIP-36 registration query.
71pub(crate) struct PrimaryKeyQuery;
72
73impl PrimaryKeyQuery {
74    /// Prepares a query to get all CIP-36 registration primary keys.
75    pub(crate) async fn prepare(session: &Arc<Session>) -> anyhow::Result<PreparedStatement> {
76        PreparedQueries::prepare(
77            session.clone(),
78            SELECT_QUERY,
79            scylla::statement::Consistency::All,
80            true,
81        )
82        .await
83        .inspect_err(
84            |error| error!(error=%error, "Failed to prepare get CIP-36 registration primary key query."),
85        )
86        .map_err(|error| anyhow::anyhow!("{error}\n--\n{SELECT_QUERY}"))
87    }
88
89    /// Executes a query to get all CIP-36 registration primary keys.
90    pub(crate) async fn execute(
91        session: &CassandraSession,
92    ) -> anyhow::Result<TypedRowStream<result::PrimaryKey>> {
93        let iter = session
94            .purge_execute_iter(PreparedSelectQuery::Cip36Registration)
95            .await?
96            .rows_stream::<result::PrimaryKey>()?;
97
98        Ok(iter)
99    }
100}
101
102/// Delete CIP-36 registration
103const DELETE_QUERY: &str = include_str!("./cql/delete_cip36_registration.cql");
104
105/// Delete CIP-36 registration Query
106pub(crate) struct DeleteQuery;
107
108impl DeleteQuery {
109    /// Prepare Batch of Delete Queries
110    pub(crate) async fn prepare_batch(
111        session: &Arc<Session>, cfg: &cassandra_db::EnvVars,
112    ) -> anyhow::Result<SizedBatch> {
113        PreparedQueries::prepare_batch(
114            session.clone(),
115            DELETE_QUERY,
116            cfg,
117            scylla::statement::Consistency::Any,
118            true,
119            false,
120        )
121        .await
122        .inspect_err(
123            |error| error!(error=%error, "Failed to prepare delete CIP-36 registration primary key query."),
124        )
125        .map_err(|error| anyhow::anyhow!("{error}\n--\n{DELETE_QUERY}"))
126    }
127
128    /// Executes a DELETE Query
129    pub(crate) async fn execute(
130        session: &CassandraSession, params: Vec<Params>,
131    ) -> FallibleQueryResults {
132        let results = session
133            .purge_execute_batch(PreparedDeleteQuery::Cip36Registration, params)
134            .await?;
135
136        Ok(results)
137    }
138}
cat_gateway/db/index/queries/purge/cip36_registration.rs

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