feat(cat-gateway): Add autogenerated cassandra schema version and sync status tables.

.config/dictionaries/project.dic

@@ -189,6 +189,7 @@ Precertificate
 preprod
 projectcatalyst
 Prokhorenko
+proptest
 psql
 Ptarget
 pubkey

catalyst-gateway/bin/Cargo.toml

@@ -21,7 +21,7 @@ pallas = { version = "0.30.1", git = "https://github.com/input-output-hk/catalys
 pallas-traverse = { version = "0.30.1", git = "https://github.com/input-output-hk/catalyst-pallas.git", rev = "9b5183c8b90b90fe2cc319d986e933e9518957b3" }
 #pallas-crypto = { version = "0.30.1", git = "https://github.com/input-output-hk/catalyst-pallas.git", rev = "9b5183c8b90b90fe2cc319d986e933e9518957b3" }
 
-clap = { version = "4.5.17", features = ["derive", "env"] }
+clap = { version = "4.5.18", features = ["derive", "env"] }
 tracing = { version = "0.1.40", features = ["log"] }
 tracing-subscriber = { version = "0.3.18", features = [
     "fmt",
@@ -31,25 +31,25 @@ tracing-subscriber = { version = "0.3.18", features = [
     "time",
     "env-filter",
 ] }
-serde = { version = "1.0.204", features = ["derive"] }
+serde = { version = "1.0.210", features = ["derive"] }
 serde_json = "1.0.128"
-thiserror = "1.0.63"
+thiserror = "1.0.64"
 chrono = "0.4.38"
 # async-trait = "0.1.82"
 bb8 = "0.8.5"
 bb8-postgres = "0.8.1"
-tokio-postgres = { version = "0.7.11", features = [
+tokio-postgres = { version = "0.7.12", features = [
     "with-chrono-0_4",
     "with-serde_json-1",
     "with-time-0_3",
 ] }
-tokio = { version = "1.39.2", features = ["rt", "macros", "rt-multi-thread"] }
+tokio = { version = "1.40.0", features = ["rt", "macros", "rt-multi-thread"] }
 dotenvy = "0.15.7"
-local-ip-address = "0.6.2"
+local-ip-address = "0.6.3"
 gethostname = "0.5.0"
 hex = "0.4.3"
-handlebars = "6.0.0"
-anyhow = "1.0.86"
+handlebars = "6.1.0"
+anyhow = "1.0.89"
 #cddl = "0.9.4"
 #ciborium = "0.2.2"
 # stringzilla = "3.9.3"
@@ -69,8 +69,8 @@ rust_decimal = { version = "1.36.0", features = [
     "serde-with-float",
     "db-tokio-postgres",
 ] }
-poem = { version = "3.0.4", features = ["embed", "prometheus", "compression"] }
-poem-openapi = { version = "5.0.3", features = [
+poem = { version = "3.1.0", features = ["embed", "prometheus", "compression"] }
+poem-openapi = { version = "5.1.1", features = [
     "openapi-explorer",
     "rapidoc",
     "redoc",
@@ -81,15 +81,18 @@ poem-openapi = { version = "5.0.3", features = [
 ] }
 uuid = { version = "1.10.0", features = ["v4", "serde"] }
 ulid = { version = "1.1.3", features = ["serde", "uuid"] }
-cryptoxide = "0.4.4" # TODO: For blake2b replace with blake2b_simd.
+blake2b_simd = "1.0.2"
 url = "2.5.2"
 panic-message = "0.3.0"
 cpu-time = "1.0.0"
 prometheus = "0.13.4"
 rust-embed = "8.5.0"
 num-traits = "0.2.19"
 base64 = "0.22.1"
-dashmap = "6.0.1"
+dashmap = "6.1.0"
+
+[dev-dependencies]
+proptest = "1.5.0"
 
 [build-dependencies]
 build-info-build = "0.0.38"

catalyst-gateway/bin/src/cardano/util.rs

@@ -1,5 +1,4 @@
 //! Block stream parsing and filtering utils
-use cryptoxide::{blake2b::Blake2b, digest::Digest};
 use pallas::ledger::{
     primitives::conway::StakeCredential,
     traverse::{Era, MultiEraAsset, MultiEraCert, MultiEraPolicyAssets},
@@ -18,21 +17,6 @@ pub type StakeCredentialHash = String;
 /// Correct stake credential key in hex
 pub type StakeCredentialKey = String;
 
-/// Hash size
-#[allow(dead_code)]
-pub(crate) const BLAKE_2B_256_HASH_SIZE: usize = 256 / 8;
-
-/// Helper function to generate the `blake2b_256` hash of a byte slice
-#[allow(dead_code)]
-pub(crate) fn hash(bytes: &[u8]) -> [u8; BLAKE_2B_256_HASH_SIZE] {
-    let mut digest = [0u8; BLAKE_2B_256_HASH_SIZE];
-    let mut context = Blake2b::new(BLAKE_2B_256_HASH_SIZE);
-    context.input(bytes);
-    context.result(&mut digest);
-
-    digest
-}
-
 #[derive(Default, Debug, Serialize)]
 /// Assets
 pub struct Asset {

catalyst-gateway/bin/src/db/index/schema/cql/sync_status.cql

@@ -0,0 +1,11 @@
+-- Most recent completed 
+-- Can also be used to convert a known stake key hash back to a full stake address.
+CREATE TABLE IF NOT EXISTS sync_status (
+    -- Primary Key Data
+    end_slot        varint,    -- The slot that has been indexed up-to (inclusive).
+    start_slot      varint,    -- The slot the sync block started at (inclusive).
+    sync_time       timestamp, -- The time we finished the sync.
+    node_id         uuid,      -- The node that synced this data.
+
+    PRIMARY KEY (end_slot, start_slot, sync_time, node_id)
+);

catalyst-gateway/bin/src/db/index/schema/mod.rs

@@ -2,23 +2,33 @@
 
 use std::sync::Arc;
 
-use anyhow::Context;
+use anyhow::{bail, Context};
 use handlebars::Handlebars;
 use scylla::Session;
 use serde_json::json;
 use tracing::error;
 
-use crate::settings::cassandra_db;
+use crate::{settings::cassandra_db, utilities::blake2b_hash::generate_uuid_string_from_data};
+
+/// The version of the Index DB Schema we SHOULD BE using.
+/// DO NOT change this unless you are intentionally changing the Schema.
+///
+/// This constant is ONLY used by Unit tests to identify when the schema version will
+/// change accidentally, and is NOT to be used directly to set the schema version of the
+/// table namespaces.
+#[allow(dead_code)]
+const SCHEMA_VERSION: &str = "a0e54866-1f30-8ad2-9ac7-df1cfaf9c634";
 
 /// Keyspace Create (Templated)
 const CREATE_NAMESPACE_CQL: &str = include_str!("./cql/namespace.cql");
 
-/// The version of the Schema we are using.
-/// Must be incremented if there is a breaking change in any schema tables below.
-pub(crate) const SCHEMA_VERSION: u64 = 1;
-
 /// All Schema Creation Statements
 const SCHEMAS: &[(&str, &str)] = &[
+    (
+        // Sync Status Table Schema
+        include_str!("./cql/sync_status.cql"),
+        "Create Sync Status Table",
+    ),
     (
         // TXO by Stake Address Table Schema
         include_str!("./cql/txo_by_stake_table.cql"),
@@ -66,10 +76,72 @@ const SCHEMAS: &[(&str, &str)] = &[
     ),
 ];
 
+/// Removes all comments from each line in the input query text and joins the remaining
+/// lines into a single string, reducing consecutive whitespace characters to a single
+/// space. Comments are defined as any text following `--` on a line.
+///
+/// # Arguments
+///
+/// * `text`: A string slice that holds the query to be cleaned.
+///
+/// # Returns
+///
+/// A new string with comments removed and whitespace reduced, where each remaining line
+/// from the original text is separated by a newline character.
+fn remove_comments_and_join_query_lines(text: &str) -> String {
+    // Split the input text into lines, removing any trailing empty lines
+    let raw_lines: Vec<&str> = text.lines().collect();
+    let mut clean_lines: Vec<String> = Vec::new();
+
+    // Filter out comments from each line
+    for line in raw_lines {
+        let mut clean_line = line.to_string();
+        if let Some(no_comment) = line.split_once("--") {
+            clean_line = no_comment.0.to_string();
+        }
+        clean_line = clean_line
+            .split_whitespace()
+            .collect::<Vec<&str>>()
+            .join(" ")
+            .trim()
+            .to_string();
+        if !clean_line.is_empty() {
+            clean_lines.push(clean_line);
+        }
+    }
+    clean_lines.join("\n")
+}
+
+/// Comment
+fn generate_cql_schema_version() -> String {
+    // Where we will actually store the bytes we derive the UUID from.
+    let mut clean_schemas: Vec<String> = Vec::new();
+
+    // Iterate through each CQL schema and add it to the list of clean schemas documents.
+    for (schema, _) in SCHEMAS {
+        let schema = remove_comments_and_join_query_lines(schema);
+        if !schema.is_empty() {
+            clean_schemas.push(schema);
+        }
+    }
+
+    // make sure any re-ordering of the schemas in the list does not effect the generated
+    // schema version
+    clean_schemas.sort();
+
+    // Generate a unique hash of the clean schemas,
+    // and use it to form a UUID to identify the schema version.
+    generate_uuid_string_from_data("Catalyst-Gateway Index Database Schema", clean_schemas)
+}
+
 /// Get the namespace for a particular db configuration
 pub(crate) fn namespace(cfg: &cassandra_db::EnvVars) -> String {
     // Build and set the Keyspace to use.
-    format!("{}_V{}", cfg.namespace.as_str(), SCHEMA_VERSION)
+    format!(
+        "{}_{}",
+        cfg.namespace.as_str(),
+        generate_cql_schema_version().replace('-', "_")
+    )
 }
 
 /// Create the namespace we will use for this session
@@ -83,11 +155,19 @@ async fn create_namespace(
     // disable default `html_escape` function
     // which transforms `<`, `>` symbols to `&lt`, `&gt`
     reg.register_escape_fn(|s| s.into());
-    let query = reg.render_template(CREATE_NAMESPACE_CQL, &json!({"keyspace": keyspace}))?;
+    let query = reg
+        .render_template(CREATE_NAMESPACE_CQL, &json!({"keyspace": keyspace}))
+        .context(format!("Keyspace: {keyspace}"))?;
 
     // Create the Keyspace if it doesn't exist already.
-    let stmt = session.prepare(query).await?;
-    session.execute_unpaged(&stmt, ()).await?;
+    let stmt = session
+        .prepare(query)
+        .await
+        .context(format!("Keyspace: {keyspace}"))?;
+    session
+        .execute_unpaged(&stmt, ())
+        .await
+        .context(format!("Keyspace: {keyspace}"))?;
 
     // Wait for the Schema to be ready.
     session.await_schema_agreement().await?;
@@ -104,22 +184,90 @@ async fn create_namespace(
 pub(crate) async fn create_schema(
     session: &mut Arc<Session>, cfg: &cassandra_db::EnvVars,
 ) -> anyhow::Result<()> {
-    create_namespace(session, cfg).await?;
-
-    for schema in SCHEMAS {
-        let stmt = session
-            .prepare(schema.0)
-            .await
-            .context(format!("{} : Prepared", schema.1))?;
-
-        session
-            .execute_unpaged(&stmt, ())
-            .await
-            .context(format!("{} : Executed", schema.1))?;
+    create_namespace(session, cfg)
+        .await
+        .context("Creating Namespace")?;
+
+    let mut failed = false;
+
+    for (schema, schema_name) in SCHEMAS {
+        match session.prepare(*schema).await {
+            Ok(stmt) => {
+                if let Err(err) = session.execute_unpaged(&stmt, ()).await {
+                    failed = true;
+                    error!(schema=schema_name, error=%err, "Failed to Execute Create Schema Query");
+                };
+            },
+            Err(err) => {
+                failed = true;
+                error!(schema=schema_name, error=%err, "Failed to Prepare Create Schema Query");
+            },
+        }
+    }
+
+    if failed {
+        bail!("Failed to Create Schema");
     }
 
     // Wait for the Schema to be ready.
     session.await_schema_agreement().await?;
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    /// This test is designed to fail if the schema version has changed.
+    /// It is used to help detect inadvertent schema version changes.
+    /// If you did NOT intend to change the index db schema and this test fails,
+    /// then revert or fix your changes to the schema files.
+    fn check_schema_version_has_not_changed() {
+        let calculated_version = generate_cql_schema_version();
+        assert_eq!(SCHEMA_VERSION, calculated_version);
+    }
+
+    #[test]
+    fn test_no_comments() {
+        let input = "SELECT * FROM table1;";
+        let expected_output = "SELECT * FROM table1;";
+        assert_eq!(remove_comments_and_join_query_lines(input), expected_output);
+    }
+
+    #[test]
+    fn test_single_line_comment() {
+        let input = "SELECT -- some comment * FROM table1;";
+        let expected_output = "SELECT";
+        assert_eq!(remove_comments_and_join_query_lines(input), expected_output);
+    }
+
+    #[test]
+    fn test_multi_line_comment() {
+        let input = "SELECT -- some comment\n* FROM table1;";
+        let expected_output = "SELECT\n* FROM table1;";
+        assert_eq!(remove_comments_and_join_query_lines(input), expected_output);
+    }
+
+    #[test]
+    fn test_multiple_lines() {
+        let input = "SELECT * FROM table1;\n-- another comment\nSELECT * FROM table2;";
+        let expected_output = "SELECT * FROM table1;\nSELECT * FROM table2;";
+        assert_eq!(remove_comments_and_join_query_lines(input), expected_output);
+    }
+
+    #[test]
+    fn test_empty_lines() {
+        let input = "\n\nSELECT * FROM table1;\n-- comment here\n\n";
+        let expected_output = "SELECT * FROM table1;";
+        assert_eq!(remove_comments_and_join_query_lines(input), expected_output);
+    }
+
+    #[test]
+    fn test_whitespace_only() {
+        let input = "   \n  -- comment here\n   ";
+        let expected_output = "";
+        assert_eq!(remove_comments_and_join_query_lines(input), expected_output);
+    }
+}

catalyst-gateway/bin/src/main.rs

@@ -8,6 +8,7 @@ mod db;
 mod logger;
 mod service;
 mod settings;
+mod utilities;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {

catalyst-gateway/bin/src/service/api/cardano/types.rs

@@ -4,9 +4,10 @@
 //! Event DB logic for chain-sync.  They should be replaced with proper types in a better
 //! place.
 
-use cryptoxide::{blake2b::Blake2b, digest::Digest};
 use serde::{Deserialize, Serialize};
 
+use crate::utilities::blake2b_hash::blake2b_224;
+
 /// Pub key
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
 pub(crate) struct PubKey(Vec<u8>);
@@ -15,11 +16,7 @@ impl PubKey {
     /// Get credentials, a blake2b 28 bytes hash of the pub key
     #[allow(dead_code)]
     pub(crate) fn get_credentials(&self) -> [u8; 28] {
-        let mut digest = [0u8; 28];
-        let mut context = Blake2b::new(28);
-        context.input(&self.0);
-        context.result(&mut digest);
-        digest
+        blake2b_224(&self.0)
     }
 
     /// Get bytes