fix(cat-gateway): Refactor and cleanup Cassandra session

catalyst-gateway/bin/Cargo.toml

@@ -64,7 +64,6 @@ num-bigint = "0.4.6"
 futures = "0.3.31"
 rand = "0.8.5"
 moka = { version = "0.12.8", features = ["future"] }
-crossbeam-skiplist = "0.1.3"
 poem = { version = "=3.1.6", features = ["embed", "prometheus", "compression"] }
 poem-openapi = { version = "=5.1.5", features = [
     "openapi-explorer",

catalyst-gateway/bin/src/db/index/block/certs.rs

@@ -1,6 +1,6 @@
 //! Index certs found in a transaction.
 
-use std::{fmt::Debug, sync::Arc};
+use std::{fmt, sync::Arc};
 
 use cardano_blockchain_types::{MultiEraBlock, Slot, StakeAddress, TxnIndex, VKeyHash};
 use ed25519_dalek::VerifyingKey;
@@ -16,6 +16,7 @@ use crate::{
         },
         types::{DbPublicKey, DbSlot, DbStakeAddress, DbTxnIndex},
     },
+    impl_query_batch,
     settings::cassandra_db,
 };
 
@@ -42,8 +43,8 @@ pub(crate) struct StakeRegistrationInsertQuery {
     pool_delegation: MaybeUnset<Vec<u8>>,
 }
 
-impl Debug for StakeRegistrationInsertQuery {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> {
+impl fmt::Debug for StakeRegistrationInsertQuery {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let stake_public_key = hex::encode(self.stake_public_key.as_ref());
         let register = match self.register {
             MaybeUnset::Unset => "UNSET",
@@ -79,6 +80,11 @@ impl Debug for StakeRegistrationInsertQuery {
 /// Insert stake registration
 const INSERT_STAKE_REGISTRATION_QUERY: &str = include_str!("./cql/insert_stake_registration.cql");
 
+impl_query_batch!(
+    StakeRegistrationInsertQuery,
+    INSERT_STAKE_REGISTRATION_QUERY
+);
+
 impl StakeRegistrationInsertQuery {
     /// Create a new Insert Query.
     #[allow(clippy::too_many_arguments, clippy::fn_params_excessive_bools)]

catalyst-gateway/bin/src/db/index/block/cip36/insert_cip36.rs

@@ -1,6 +1,6 @@
 //! Insert CIP36 Registration Query
 
-use std::{fmt::Debug, sync::Arc};
+use std::{fmt, sync::Arc};
 
 use cardano_blockchain_types::{Cip36, Slot, TxnIndex, VotingPubKey};
 use scylla::{frame::value::MaybeUnset, SerializeRow, Session};
@@ -11,6 +11,7 @@ use crate::{
         index::queries::{PreparedQueries, SizedBatch},
         types::{DbSlot, DbTxnIndex},
     },
+    impl_query_batch,
     settings::cassandra_db,
 };
 
@@ -19,7 +20,7 @@ const INSERT_CIP36_REGISTRATION_QUERY: &str = include_str!("./cql/insert_cip36.c
 
 /// Insert CIP-36 Registration Query Parameters
 #[derive(SerializeRow, Clone)]
-pub(crate) struct Params {
+pub(crate) struct Cip36Insert {
     /// Full Stake Address (not hashed, 32 byte ED25519 Public key).
     stake_public_key: Vec<u8>,
     /// Nonce value after normalization.
@@ -40,13 +41,15 @@ pub(crate) struct Params {
     cip36: bool,
 }
 
-impl Debug for Params {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+impl_query_batch!(Cip36Insert, INSERT_CIP36_REGISTRATION_QUERY);
+
+impl fmt::Debug for Cip36Insert {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let payment_address = match self.payment_address {
             MaybeUnset::Unset => "UNSET",
             MaybeUnset::Set(ref v) => &hex::encode(v),
         };
-        f.debug_struct("Params")
+        f.debug_struct("Cip36Insert")
             .field("stake_public_key", &self.stake_public_key)
             .field("nonce", &self.nonce)
             .field("slot_no", &self.slot_no)
@@ -60,7 +63,7 @@ impl Debug for Params {
     }
 }
 
-impl Params {
+impl Cip36Insert {
     /// Create a new Insert Query.
     pub fn new(vote_key: &VotingPubKey, slot_no: Slot, txn_index: TxnIndex, cip36: &Cip36) -> Self {
         let stake_public_key = cip36
@@ -73,7 +76,7 @@ impl Params {
             .payment_address()
             .map_or(MaybeUnset::Unset, |a| MaybeUnset::Set(a.to_vec()));
         let is_cip36 = cip36.is_cip36().unwrap_or_default();
-        Params {
+        Cip36Insert {
             stake_public_key,
             nonce: cip36.nonce().unwrap_or_default().into(),
             slot_no: slot_no.into(),

catalyst-gateway/bin/src/db/index/block/cip36/insert_cip36_for_vote_key.rs

@@ -11,6 +11,7 @@ use crate::{
         index::queries::{PreparedQueries, SizedBatch},
         types::{DbSlot, DbTxnIndex},
     },
+    impl_query_batch,
     settings::cassandra_db,
 };
 
@@ -20,7 +21,7 @@ const INSERT_CIP36_REGISTRATION_FOR_VOTE_KEY_QUERY: &str =
 
 /// Insert CIP-36 Registration Invalid Query Parameters
 #[derive(SerializeRow, Debug)]
-pub(crate) struct Params {
+pub(crate) struct Cip36ForVoteKeyInsert {
     /// Voting Public Key
     vote_key: Vec<u8>,
     /// Full Stake Address (not hashed, 32 byte ED25519 Public key).
@@ -33,12 +34,17 @@ pub(crate) struct Params {
     valid: bool,
 }
 
-impl Params {
+impl_query_batch!(
+    Cip36ForVoteKeyInsert,
+    INSERT_CIP36_REGISTRATION_FOR_VOTE_KEY_QUERY
+);
+
+impl Cip36ForVoteKeyInsert {
     /// Create a new Insert Query.
     pub fn new(
         vote_key: &VotingPubKey, slot_no: Slot, txn_index: TxnIndex, cip36: &Cip36, valid: bool,
     ) -> Self {
-        Params {
+        Cip36ForVoteKeyInsert {
             vote_key: vote_key
                 .voting_pk()
                 .map(|k| k.to_bytes().to_vec())

catalyst-gateway/bin/src/db/index/block/cip36/insert_cip36_invalid.rs

@@ -1,6 +1,6 @@
 //! Insert CIP36 Registration Query (Invalid Records)
 
-use std::{fmt::Debug, sync::Arc};
+use std::{fmt, sync::Arc};
 
 use cardano_blockchain_types::{Cip36, Slot, TxnIndex, VotingPubKey};
 use pallas::ledger::addresses::Address;
@@ -12,6 +12,7 @@ use crate::{
         index::queries::{PreparedQueries, SizedBatch},
         types::{DbSlot, DbTxnIndex},
     },
+    impl_query_batch,
     settings::cassandra_db,
 };
 
@@ -21,7 +22,7 @@ const INSERT_CIP36_REGISTRATION_INVALID_QUERY: &str =
 
 /// Insert CIP-36 Registration Invalid Query Parameters
 #[derive(SerializeRow, Clone)]
-pub(crate) struct Params {
+pub(crate) struct Cip36InvalidInsert {
     /// Full Stake Address (not hashed, 32 byte ED25519 Public key).
     stake_public_key: Vec<u8>,
     /// Slot Number the cert is in.
@@ -46,13 +47,15 @@ pub(crate) struct Params {
     problem_report: String,
 }
 
-impl Debug for Params {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+impl_query_batch!(Cip36InvalidInsert, INSERT_CIP36_REGISTRATION_INVALID_QUERY);
+
+impl fmt::Debug for Cip36InvalidInsert {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let cip36 = match self.cip36 {
             MaybeUnset::Unset => "UNSET",
             MaybeUnset::Set(v) => &format!("{v:?}"),
         };
-        f.debug_struct("Params")
+        f.debug_struct("Cip36InvalidInsert")
             .field("stake_public_key", &self.stake_public_key)
             .field("slot_no", &self.slot_no)
             .field("txn_index", &self.txn_index)
@@ -68,7 +71,7 @@ impl Debug for Params {
     }
 }
 
-impl Params {
+impl Cip36InvalidInsert {
     /// Create a new Insert Query.
     pub fn new(
         vote_key: Option<&VotingPubKey>, slot_no: Slot, txn_index: TxnIndex, cip36: &Cip36,
@@ -89,7 +92,7 @@ impl Params {
             String::new()
         });
 
-        Params {
+        Cip36InvalidInsert {
             stake_public_key,
             slot_no: slot_no.into(),
             txn_index: txn_index.into(),

catalyst-gateway/bin/src/db/index/block/cip36/mod.rs

@@ -22,11 +22,11 @@ use crate::{
 /// Insert CIP-36 Registration Queries
 pub(crate) struct Cip36InsertQuery {
     /// Stake Registration Data captured during indexing.
-    registrations: Vec<insert_cip36::Params>,
+    registrations: Vec<insert_cip36::Cip36Insert>,
     /// Stake Registration Data captured during indexing.
-    invalid: Vec<insert_cip36_invalid::Params>,
+    invalid: Vec<insert_cip36_invalid::Cip36InvalidInsert>,
     /// Stake Registration Data captured during indexing.
-    for_vote_key: Vec<insert_cip36_for_vote_key::Params>,
+    for_vote_key: Vec<insert_cip36_for_vote_key::Cip36ForVoteKeyInsert>,
     /// Stake Registration Data captured during indexing.
     stake_regs: Vec<certs::StakeRegistrationInsertQuery>,
 }
@@ -46,11 +46,11 @@ impl Cip36InsertQuery {
     pub(crate) async fn prepare_batch(
         session: &Arc<Session>, cfg: &cassandra_db::EnvVars,
     ) -> anyhow::Result<(SizedBatch, SizedBatch, SizedBatch)> {
-        let insert_cip36_batch = insert_cip36::Params::prepare_batch(session, cfg).await;
+        let insert_cip36_batch = insert_cip36::Cip36Insert::prepare_batch(session, cfg).await;
         let insert_cip36_invalid_batch =
-            insert_cip36_invalid::Params::prepare_batch(session, cfg).await;
+            insert_cip36_invalid::Cip36InvalidInsert::prepare_batch(session, cfg).await;
         let insert_cip36_for_vote_key_addr_batch =
-            insert_cip36_for_vote_key::Params::prepare_batch(session, cfg).await;
+            insert_cip36_for_vote_key::Cip36ForVoteKeyInsert::prepare_batch(session, cfg).await;
         // Its a hack of inserting `stake_regs` during the indexing CIP 36 registrations.
         // Its done because some of the CIP 36 registrations contains some stake addresses which
         // are not actually some how registered using cardano certs.
@@ -82,11 +82,11 @@ impl Cip36InsertQuery {
                 let stake_pk_hash = Blake2b224Hash::new(&stake_pk.to_bytes());
                 let stake_address = StakeAddress::new(block.network(), false, stake_pk_hash);
 
-                self.registrations.push(insert_cip36::Params::new(
+                self.registrations.push(insert_cip36::Cip36Insert::new(
                     voting_key, slot_no, index, &cip36,
                 ));
                 self.for_vote_key
-                    .push(insert_cip36_for_vote_key::Params::new(
+                    .push(insert_cip36_for_vote_key::Cip36ForVoteKeyInsert::new(
                         voting_key, slot_no, index, &cip36, true,
                     ));
                 self.stake_regs
@@ -107,21 +107,24 @@ impl Cip36InsertQuery {
                 // Cannot index an invalid CIP36, if there is no stake public key.
                 if let Some(stake_pk) = cip36.stake_pk() {
                     if cip36.voting_pks().is_empty() {
-                        self.invalid.push(insert_cip36_invalid::Params::new(
-                            None, slot_no, index, &cip36,
-                        ));
+                        self.invalid
+                            .push(insert_cip36_invalid::Cip36InvalidInsert::new(
+                                None, slot_no, index, &cip36,
+                            ));
                     } else {
                         for voting_key in cip36.voting_pks() {
-                            self.invalid.push(insert_cip36_invalid::Params::new(
-                                Some(voting_key),
-                                slot_no,
-                                index,
-                                &cip36,
-                            ));
-                            self.for_vote_key
-                                .push(insert_cip36_for_vote_key::Params::new(
-                                    voting_key, slot_no, index, &cip36, false,
+                            self.invalid
+                                .push(insert_cip36_invalid::Cip36InvalidInsert::new(
+                                    Some(voting_key),
+                                    slot_no,
+                                    index,
+                                    &cip36,
                                 ));
+                            self.for_vote_key.push(
+                                insert_cip36_for_vote_key::Cip36ForVoteKeyInsert::new(
+                                    voting_key, slot_no, index, &cip36, false,
+                                ),
+                            );
                         }
                     }
 

catalyst-gateway/bin/src/db/index/block/rbac509/insert_catalyst_id_for_stake_address.rs

@@ -12,6 +12,7 @@ use crate::{
         index::queries::{PreparedQueries, SizedBatch},
         types::{DbCatalystId, DbSlot, DbStakeAddress},
     },
+    impl_query_batch,
     settings::cassandra_db::EnvVars,
 };
 
@@ -20,7 +21,7 @@ const QUERY: &str = include_str!("cql/insert_catalyst_id_for_stake_address.cql")
 
 /// Insert Catalyst ID For Stake Address Query Parameters
 #[derive(SerializeRow)]
-pub(crate) struct Params {
+pub(crate) struct CatalystIdForStakeAddressInsert {
     /// A stake address.
     stake_address: DbStakeAddress,
     /// A Catalyst short identifier.
@@ -29,20 +30,22 @@ pub(crate) struct Params {
     slot_no: DbSlot,
 }
 
-impl Debug for Params {
+impl_query_batch!(CatalystIdForStakeAddressInsert, QUERY);
+
+impl Debug for CatalystIdForStakeAddressInsert {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("Params")
+        f.debug_struct("CatalystIdForStakeAddressInsert")
             .field("stake_address", &self.stake_address)
             .field("catalyst_id", &self.catalyst_id)
             .field("slot_no", &self.slot_no)
             .finish()
     }
 }
 
-impl Params {
+impl CatalystIdForStakeAddressInsert {
     /// Create a new record for this transaction.
     pub(crate) fn new(stake_address: StakeAddress, slot_no: Slot, catalyst_id: CatalystId) -> Self {
-        Params {
+        CatalystIdForStakeAddressInsert {
             stake_address: stake_address.into(),
             catalyst_id: catalyst_id.into(),
             slot_no: slot_no.into(),

catalyst-gateway/bin/src/db/index/block/rbac509/insert_catalyst_id_for_txn_id.rs

@@ -12,6 +12,7 @@ use crate::{
         index::queries::{PreparedQueries, SizedBatch},
         types::{DbCatalystId, DbTransactionId},
     },
+    impl_query_batch,
     settings::cassandra_db::EnvVars,
 };
 
@@ -20,26 +21,28 @@ const QUERY: &str = include_str!("cql/insert_catalyst_id_for_txn_id.cql");
 
 /// Insert Catalyst ID For Transaction ID Query Parameters
 #[derive(SerializeRow)]
-pub(crate) struct Params {
+pub(crate) struct CatalystIdForTxnIdInsert {
     /// A transaction hash.
     txn_id: DbTransactionId,
     /// A Catalyst short identifier.
     catalyst_id: DbCatalystId,
 }
 
-impl Debug for Params {
+impl_query_batch!(CatalystIdForTxnIdInsert, QUERY);
+
+impl Debug for CatalystIdForTxnIdInsert {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("Params")
+        f.debug_struct("CatalystIdForTxnIdInsert")
             .field("txn_id", &self.txn_id)
             .field("catalyst_id", &self.catalyst_id)
             .finish()
     }
 }
 
-impl Params {
+impl CatalystIdForTxnIdInsert {
     /// Creates a new record for this transaction.
     pub(crate) fn new(catalyst_id: CatalystId, txn_id: TransactionId) -> Self {
-        Params {
+        CatalystIdForTxnIdInsert {
             txn_id: txn_id.into(),
             catalyst_id: catalyst_id.into(),
         }