Introducing db-sync-explorer service which expose db-sync database as…

… as REST service. Added some helper endpoint to check meta information about db-sync and validate sync process (we can query sync process expressed as percentage or difference between current date and last registered block). Finally added api/v0/tx/hash/{...} endpoint which can return transaction information if exists in db sync. This can be useful for any clients for example GVC to check if registration transaction is confirmed

Cargo.toml

@@ -47,6 +47,7 @@ members = [
   "src/vit-testing/integration-tests",
   "src/vit-testing/mainnet-tools",
   "src/vit-testing/mainnet-lib",
+  "src/vit-testing/db-sync-explorer",
   "src/vit-testing/snapshot-trigger-service",
   "src/vit-testing/signals-handler",
   "src/vit-testing/scheduler-service-lib",

src/vit-testing/db-sync-explorer/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "db-sync-explorer"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+color-eyre = "0.6"  # error handling
+voting_tools_rs = { path = "../../voting-tools-rs"} # db sync schema
+microtype = { version = "0.7.5", features = ["serde"] } # defining secrecy and db internals
+diesel =  { version = "2", features = ["postgres", "64-column-tables", "numeric", "serde_json", "r2d2"]} # connector lib
+diesel-derive-enum = "2.0.0-rc.0" #extension to diesel
+chrono = "0.4.23"
+tracing = "0.1"
+bigdecimal = "0.3.0"
+warp = { version = "0.3.2", features = ["tls"] }
+vit-servicing-station-lib  = { path = "../../vit-servicing-station/vit-servicing-station-lib" }
+futures-util = "0.3.25"
+thiserror = "1.0.37"
+http-zipkin = "0.3.0"
+jortestkit = { path = "../../jortestkit" }
+tokio = { version = "1.18.0", features = ["macros", "signal", "rt", "fs", "sync"] }
+tracing-subscriber = "0.3"
+clap = { version = "3.2", features = ["derive"] }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+

src/vit-testing/db-sync-explorer/config-snapshot.json

@@ -0,0 +1,16 @@
+{
+	"port": 8080,
+	"result-dir": "./data",
+	"voting-tools": {
+		"bin": "voting-tools-mock",
+		"network": {
+			"testnet": 1234231321
+		},
+		"db": "cexplorer",
+		"db-user": "cexplorer",
+		"db-pass": "pass",
+		"db-host": "/alloc",
+		"scale": 1000000
+	},
+	"token":"RBj0OfHw5jT87A"
+}

src/vit-testing/db-sync-explorer/src/bin/db-sync-explorer.rs

@@ -0,0 +1,34 @@
+use std::path::Path;
+
+use clap::Parser;
+use color_eyre::Result;
+use db_sync_explorer::{connect, rest, Args, Config};
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    color_eyre::install()?;
+    tracing_subscriber::fmt().init();
+
+    let Args { config, .. } = Args::parse();
+
+    let configuration: Config = read_config(&config)?;
+    let db_pool = connect(configuration.db)?;
+
+    let address = configuration.address;
+    let context = rest::v0::context::new_shared_context(db_pool, configuration.token);
+
+    let app = rest::v0::filter(context).await;
+
+    tracing::debug!("listening on {}", address);
+    let (_, server) = warp::serve(app).bind_with_graceful_shutdown(
+        address,
+        vit_servicing_station_lib::server::signals::watch_signal_for_shutdown(),
+    );
+    server.await;
+    Ok(())
+}
+
+pub fn read_config<P: AsRef<Path>>(config: P) -> Result<Config, color_eyre::Report> {
+    let contents = std::fs::read_to_string(&config)?;
+    serde_json::from_str(&contents).map_err(Into::into)
+}

src/vit-testing/db-sync-explorer/src/cli.rs

@@ -0,0 +1,12 @@
+use clap::Parser;
+use std::path::PathBuf;
+
+#[derive(Debug, Parser)]
+#[non_exhaustive]
+#[clap(about = "Create a voting power snapshot")]
+/// CLI arguments for db-sync-explorer
+pub struct Args {
+    /// configuration file
+    #[clap(long)]
+    pub config: PathBuf,
+}

src/vit-testing/db-sync-explorer/src/config.rs

@@ -0,0 +1,10 @@
+use crate::DbConfig;
+use serde::Deserialize;
+use std::net::SocketAddr;
+
+#[derive(Debug, Deserialize)]
+pub struct Config {
+    pub address: SocketAddr,
+    pub db: DbConfig,
+    pub token: Option<String>,
+}

src/vit-testing/db-sync-explorer/src/db/config.rs

@@ -0,0 +1,35 @@
+use microtype::microtype;
+use serde::Deserialize;
+
+microtype! {
+    #[derive(Debug, PartialEq,Eq, Clone)]
+    #[string]
+    pub String {
+        /// Database name
+        DbName,
+        /// Database user
+        DbUser,
+        /// Database host
+        DbHost,
+    }
+
+    #[secret]
+    #[string]
+    pub String {
+         /// Database password
+        DbPass,
+    }
+}
+
+/// Information required to connect to a database
+#[derive(Debug, Clone, Deserialize)]
+pub struct DbConfig {
+    /// The name of the database
+    pub name: DbName,
+    /// The user to connect with
+    pub user: DbUser,
+    /// The hostname to connect to
+    pub host: DbHost,
+    /// The corresponding password for this user
+    pub password: Option<DbPass>,
+}

src/vit-testing/db-sync-explorer/src/db/connector.rs

@@ -0,0 +1,30 @@
+use crate::db::config::DbConfig;
+use color_eyre::Result;
+use diesel::r2d2::{ConnectionManager, Pool};
+use diesel::PgConnection;
+use microtype::secrecy::ExposeSecret;
+use microtype::secrecy::Zeroize;
+
+pub type DbPool = Pool<ConnectionManager<PgConnection>>;
+/// Connect to the database with the provided credentials
+///
+/// # Errors
+///
+/// Returns an error if connecting to the database fails
+pub fn connect(
+    DbConfig {
+        name,
+        user,
+        host,
+        password,
+    }: DbConfig,
+) -> Result<DbPool> {
+    let mut password = password
+        .map(|p| format!(":{}", p.expose_secret()))
+        .unwrap_or_default();
+
+    let url = format!("postgres://{user}{password}@{host}/{name}");
+    let pool = Pool::builder().build(ConnectionManager::new(url))?;
+    password.zeroize();
+    Ok(pool)
+}

src/vit-testing/db-sync-explorer/src/db/mod.rs

@@ -0,0 +1,13 @@
+mod config;
+mod connector;
+mod model;
+pub(crate) mod query;
+pub mod schema;
+pub mod types;
+
+pub use config::DbConfig;
+pub use connector::{connect, DbPool};
+pub use model::{
+    BehindDuration, Meta, Progress, TransactionConfirmation, TransactionConfirmationRow,
+};
+pub use query::behind;

src/vit-testing/db-sync-explorer/src/db/model.rs

@@ -0,0 +1,46 @@
+use bigdecimal::BigDecimal;
+use diesel::sql_types::{Numeric, Timestamp};
+use diesel::{Queryable, QueryableByName};
+use serde::{Deserialize, Serialize};
+use std::time::SystemTime;
+
+#[derive(Queryable, PartialEq, Eq, Debug, Serialize, Clone)]
+pub struct Meta {
+    pub id: i64,
+    pub start_time: SystemTime,
+    pub version: String,
+    pub network_name: String,
+}
+
+#[derive(QueryableByName, Debug, Serialize, Deserialize, Clone)]
+pub struct BehindDuration {
+    #[diesel(sql_type = Timestamp)]
+    behind_by: SystemTime,
+}
+
+#[derive(QueryableByName, Debug, Serialize, Deserialize, Clone)]
+pub struct Progress {
+    #[diesel(sql_type = Numeric)]
+    sync_percentage: BigDecimal,
+}
+
+#[derive(Debug, Serialize)]
+pub struct TransactionConfirmation {
+    epoch_no: Option<i32>,
+    slot_no: Option<i64>,
+    absolute_slot: Option<i32>,
+    block_no: Option<i32>,
+}
+
+impl From<TransactionConfirmationRow> for TransactionConfirmation {
+    fn from(row: TransactionConfirmationRow) -> Self {
+        TransactionConfirmation {
+            epoch_no: row.0,
+            slot_no: row.1,
+            absolute_slot: row.2,
+            block_no: row.3,
+        }
+    }
+}
+
+pub type TransactionConfirmationRow = (Option<i32>, Option<i64>, Option<i32>, Option<i32>);

src/vit-testing/db-sync-explorer/src/db/query.rs

@@ -0,0 +1,63 @@
+use crate::db::model::Progress;
+use crate::db::{DbPool, TransactionConfirmationRow};
+use crate::rest::v0::errors::HandleError;
+use crate::BehindDuration;
+use diesel::QueryDsl;
+use diesel::{sql_function, sql_query, sql_types::Text, RunQueryDsl};
+sql_function! (fn decode(string: Text, format: Text) -> Bytea);
+
+pub async fn hash(
+    input_hash: String,
+    pool: &DbPool,
+) -> Result<Vec<TransactionConfirmationRow>, HandleError> {
+    let mut db_conn = pool.get().map_err(HandleError::Connection)?;
+
+    tokio::task::spawn_blocking(move || {
+        use crate::db::schema::{
+            block::{self, block_no, epoch_no, epoch_slot_no, id, slot_no},
+            tx::{self, block_id},
+        };
+        use diesel::ExpressionMethods;
+        use diesel::JoinOnDsl;
+
+        let inner_join = block::table.on(id.eq(block_id));
+
+        let table = tx::table.inner_join(inner_join);
+
+        table
+            .select((epoch_no, slot_no, epoch_slot_no, block_no))
+            .filter(tx::hash.eq(decode(input_hash, "hex")))
+            .load(&mut db_conn)
+            .map_err(HandleError::Database)
+    })
+    .await
+    .map_err(HandleError::Join)?
+}
+
+/// Running sql from dbsync examples:
+/// <https://github.com/input-output-hk/cardano-db-sync/blob/master/doc/interesting-queries.md#sync-progress-of-db-sync>
+pub async fn sync_progress(pool: &DbPool) -> Result<Vec<Progress>, HandleError> {
+    let mut db_conn = pool.get().map_err(HandleError::Connection)?;
+
+    let query = sql_query("select
+            100 * (extract (epoch from (max (time) at time zone 'UTC')) - extract (epoch from (min (time) at time zone 'UTC')))
+                / (extract (epoch from (now () at time zone 'UTC')) - extract (epoch from (min (time) at time zone 'UTC')))
+                as sync_percentage
+            from block");
+
+    tokio::task::spawn_blocking(move || query.load(&mut db_conn).map_err(HandleError::Database))
+        .await
+        .map_err(HandleError::Join)?
+}
+
+/// Running sql from dbsync examples:
+/// <https://github.com/input-output-hk/cardano-db-sync/blob/master/doc/interesting-queries.md#sync-progress-of-db-sync>
+pub async fn behind(pool: &DbPool) -> Result<Vec<BehindDuration>, HandleError> {
+    let mut db_conn = pool.get().map_err(HandleError::Connection)?;
+
+    let query = sql_query("select max (time) as behind_by from block");
+
+    tokio::task::spawn_blocking(move || query.load(&mut db_conn).map_err(HandleError::Database))
+        .await
+        .map_err(HandleError::Join)?
+}