Merge pull request #4242 from nymtech/feature/basic-scraper

Feature/basic scraper

Cargo.toml

@@ -67,6 +67,7 @@ members = [
     "common/nymsphinx/params",
     "common/nymsphinx/routing",
     "common/nymsphinx/types",
+    "common/nyxd-scraper",
     "common/pemstore",
     "common/socks5-client-core",
     "common/socks5/proxy-helpers",
@@ -159,6 +160,7 @@ reqwest = "0.11.22"
 schemars = "0.8.1"
 serde = "1.0.152"
 serde_json = "1.0.91"
+sqlx = "0.6.3"
 tap = "1.0.1"
 time = "0.3.30"
 thiserror = "1.0.48"
@@ -200,8 +202,8 @@ cw-controllers = { version = "=1.1.0" }
 # cosmrs-related
 bip32 = "0.5.1"
 cosmrs = "=0.15.0"
-tendermint-rpc = "0.34" # same version as used by cosmrs
 tendermint = "0.34" # same version as used by cosmrs
+tendermint-rpc = "0.34" # same version as used by cosmrs
 prost = "0.12"
 
 # wasm-related dependencies

common/client-core/Cargo.toml

@@ -59,7 +59,7 @@ features = ["time"]
 version = "0.20.1"
 
 [target."cfg(not(target_arch = \"wasm32\"))".dependencies.sqlx]
-version = "0.6.2"
+workspace = true
 features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"]
 optional = true
 
@@ -90,7 +90,7 @@ tempfile = "3.1.0"
 
 [build-dependencies]
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
-sqlx = { version = "0.6.2", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }
 
 [features]
 default = []

common/credential-storage/Cargo.toml

@@ -14,14 +14,14 @@ thiserror = { workspace = true }
 tokio = { version = "1.24.1", features = ["sync"]}
 
 [target."cfg(not(target_arch = \"wasm32\"))".dependencies.sqlx]
-version = "0.5"
+workspace = true
 features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"]
 
 [target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio]
-version = "1.24.1"
+workspace = true
 features = [ "rt-multi-thread", "net", "signal", "fs" ]
 
 
 [build-dependencies]
-sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }
 tokio = { version = "1.24.1", features = ["rt-multi-thread", "macros"] }

common/nyxd-scraper/Cargo.toml

@@ -0,0 +1,37 @@
+[package]
+name = "nyxd-scraper"
+version = "0.1.0"
+authors.workspace = true
+repository.workspace = true
+homepage.workspace = true
+documentation.workspace = true
+edition.workspace = true
+license.workspace = true
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+async-trait.workspace = true
+const_format = "0.2.32"
+cosmrs.workspace = true
+eyre = "0.6.9"
+futures.workspace = true
+sha2 = "0.10.8"
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate", "time"] }
+tendermint.workspace = true
+tendermint-rpc = { workspace = true, features = ["websocket-client", "http-client"] }
+thiserror.workspace = true
+tokio = { workspace = true, features = ["full"] }
+tokio-stream = "0.1.14"
+tokio-util = { version = "0.7.10", features = ["rt"]}
+tracing.workspace = true
+url.workspace = true
+
+
+# TEMP
+nym-bin-common = { path = "../bin-common", features = ["basic_tracing"]}
+
+
+[build-dependencies]
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }

common/nyxd-scraper/build.rs

@@ -0,0 +1,28 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+#[tokio::main]
+async fn main() {
+    use sqlx::{Connection, SqliteConnection};
+    use std::env;
+
+    let out_dir = env::var("OUT_DIR").unwrap();
+    let database_path = format!("{out_dir}/scraper-example.sqlite");
+
+    let mut conn = SqliteConnection::connect(&format!("sqlite://{database_path}?mode=rwc"))
+        .await
+        .expect("Failed to create SQLx database connection");
+
+    sqlx::migrate!("./sql_migrations")
+        .run(&mut conn)
+        .await
+        .expect("Failed to perform SQLx migrations");
+
+    #[cfg(target_family = "unix")]
+    println!("cargo:rustc-env=DATABASE_URL=sqlite://{}", &database_path);
+
+    #[cfg(target_family = "windows")]
+    // for some strange reason we need to add a leading `/` to the windows path even though it's
+    // not a valid windows path... but hey, it works...
+    println!("cargo:rustc-env=DATABASE_URL=sqlite:///{}", &database_path);
+}

common/nyxd-scraper/sql_migrations/01_metadata.sql

@@ -0,0 +1,10 @@
+/*
+ * Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+CREATE TABLE METADATA
+(
+    id                    INTEGER PRIMARY KEY CHECK (id = 0),
+    last_processed_height INTEGER NOT NULL
+);

common/nyxd-scraper/sql_migrations/02_cosmos.sql

@@ -0,0 +1,76 @@
+CREATE TABLE validator
+(
+    consensus_address TEXT NOT NULL PRIMARY KEY, /* Validator consensus address */
+    consensus_pubkey  TEXT NOT NULL UNIQUE /* Validator consensus public key */
+);
+
+CREATE TABLE pre_commit
+(
+    validator_address TEXT                        NOT NULL REFERENCES validator (consensus_address),
+    height            BIGINT                      NOT NULL,
+    timestamp         TIMESTAMP WITHOUT TIME ZONE NOT NULL,
+    voting_power      BIGINT                      NOT NULL,
+    proposer_priority BIGINT                      NOT NULL,
+    UNIQUE (validator_address, timestamp)
+);
+CREATE INDEX pre_commit_validator_address_index ON pre_commit (validator_address);
+CREATE INDEX pre_commit_height_index ON pre_commit (height);
+
+CREATE TABLE block
+(
+    height           BIGINT  UNIQUE PRIMARY KEY,
+    hash             TEXT    NOT NULL UNIQUE,
+    num_txs          INTEGER DEFAULT 0,
+    total_gas        BIGINT  DEFAULT 0,
+    proposer_address TEXT REFERENCES validator (consensus_address),
+    timestamp        TIMESTAMP WITHOUT TIME ZONE NOT NULL
+);
+CREATE INDEX block_height_index ON block (height);
+CREATE INDEX block_hash_index ON block (hash);
+CREATE INDEX block_proposer_address_index ON block (proposer_address);
+
+-- no JSONB in sqlite (yet) : )
+CREATE TABLE "transaction"
+(
+    hash         TEXT    UNIQUE NOT NULL,
+    height       BIGINT  NOT NULL REFERENCES block (height),
+    "index"      INTEGER NOT NULL,
+    success      BOOLEAN NOT NULL,
+
+    /* Body */
+    num_messages INTEGER NOT NULL,
+--     messages     JSONB   NOT NULL,-- DEFAULT '[]'::JSONB,
+    memo         TEXT,
+--     signatures   TEXT[]  NOT NULL,
+
+    /* AuthInfo */
+--     signer_infos JSONB   NOT NULL,-- DEFAULT '[]'::JSONB,
+--     fee          JSONB   NOT NULL,-- DEFAULT '{}'::JSONB,
+
+    /* Tx response */
+    gas_wanted   BIGINT           DEFAULT 0,
+    gas_used     BIGINT           DEFAULT 0,
+    raw_log      TEXT
+--     logs         JSONB
+);
+CREATE INDEX transaction_hash_index ON "transaction" (hash);
+CREATE INDEX transaction_height_index ON "transaction" (height);
+
+CREATE TABLE message
+(
+    transaction_hash            TEXT   NOT NULL REFERENCES "transaction" (hash),
+    "index"                       BIGINT NOT NULL,
+    type                        TEXT   NOT NULL,
+--     value                       JSONB  NOT NULL,
+--     involved_accounts_addresses TEXT[] NOT NULL,
+
+    height                      BIGINT NOT NULL,
+    CONSTRAINT unique_message_per_tx UNIQUE (transaction_hash, "index")
+);
+CREATE INDEX message_transaction_hash_index ON message (transaction_hash);
+CREATE INDEX message_type_index ON message (type);
+
+CREATE TABLE pruning
+(
+    last_pruned_height BIGINT NOT NULL
+);

common/nyxd-scraper/src/block_processor/helpers.rs

@@ -0,0 +1,65 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::block_processor::MAX_RANGE_SIZE;
+use std::cmp::min;
+use std::collections::VecDeque;
+use std::ops::Range;
+
+pub(crate) fn split_request_range(request_range: Range<u32>) -> VecDeque<Range<u32>> {
+    let mut requests = VecDeque::new();
+
+    let mut start = request_range.start;
+    let mut end = min(request_range.end, start + MAX_RANGE_SIZE as u32);
+
+    loop {
+        requests.push_back(start..end);
+        start = min(start + MAX_RANGE_SIZE as u32, request_range.end);
+        end = min(end + MAX_RANGE_SIZE as u32, request_range.end);
+
+        if start == end {
+            break;
+        }
+    }
+
+    requests
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn splitting_request_range() {
+        let range = 0..100;
+        let mut expected = VecDeque::new();
+        expected.push_back(0..30);
+        expected.push_back(30..60);
+        expected.push_back(60..90);
+        expected.push_back(90..100);
+        assert_eq!(expected, split_request_range(range));
+
+        let range = 0..30;
+        let mut expected = VecDeque::new();
+        expected.push_back(0..30);
+        assert_eq!(expected, split_request_range(range));
+
+        let range = 0..60;
+        let mut expected = VecDeque::new();
+        expected.push_back(0..30);
+        expected.push_back(30..60);
+        assert_eq!(expected, split_request_range(range));
+
+        let range = 0..5;
+        let mut expected = VecDeque::new();
+        expected.push_back(0..5);
+        assert_eq!(expected, split_request_range(range));
+
+        let range = 123..200;
+        let mut expected = VecDeque::new();
+        expected.push_back(123..153);
+        expected.push_back(153..183);
+        expected.push_back(183..200);
+        assert_eq!(expected, split_request_range(range));
+    }
+}

common/nyxd-scraper/src/block_processor/mod.rs

@@ -0,0 +1,323 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::block_processor::helpers::split_request_range;
+use crate::block_processor::types::BlockToProcess;
+use crate::block_requester::BlockRequest;
+use crate::error::ScraperError;
+use crate::modules::{BlockModule, MsgModule, TxModule};
+use crate::rpc_client::RpcClient;
+use crate::storage::{persist_block, ScraperStorage};
+use futures::StreamExt;
+use std::collections::{BTreeMap, HashSet, VecDeque};
+use std::ops::{Add, Range};
+use std::time::Duration;
+use tokio::sync::mpsc::{Sender, UnboundedReceiver};
+use tokio::time::{interval_at, Instant};
+use tokio_stream::wrappers::UnboundedReceiverStream;
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, error, info, warn};
+
+mod helpers;
+pub(crate) mod types;
+
+const MISSING_BLOCKS_CHECK_INTERVAL: Duration = Duration::from_secs(30);
+const MAX_MISSING_BLOCKS_DELAY: Duration = Duration::from_secs(15);
+const MAX_RANGE_SIZE: usize = 30;
+
+#[derive(Debug, Default)]
+struct PendingSync {
+    request_in_flight: HashSet<u32>,
+    queued_requests: VecDeque<Range<u32>>,
+}
+
+impl PendingSync {
+    fn is_empty(&self) -> bool {
+        self.request_in_flight.is_empty() && self.queued_requests.is_empty()
+    }
+}
+
+pub struct BlockProcessor {
+    cancel: CancellationToken,
+    last_processed_height: u32,
+    last_processed_at: Instant,
+    pending_sync: PendingSync,
+    queued_blocks: BTreeMap<u32, BlockToProcess>,
+
+    rpc_client: RpcClient,
+    incoming: UnboundedReceiverStream<BlockToProcess>,
+    block_requester: Sender<BlockRequest>,
+    storage: ScraperStorage,
+
+    // future work: rather than sending each msg to every msg module,
+    // let them subscribe based on `type_url` inside the message itself
+    // (like "/cosmwasm.wasm.v1.MsgExecuteContract")
+    block_modules: Vec<Box<dyn BlockModule + Send>>,
+    tx_modules: Vec<Box<dyn TxModule + Send>>,
+    msg_modules: Vec<Box<dyn MsgModule + Send>>,
+}
+
+impl BlockProcessor {
+    pub async fn new(
+        cancel: CancellationToken,
+        incoming: UnboundedReceiver<BlockToProcess>,
+        block_requester: Sender<BlockRequest>,
+        storage: ScraperStorage,
+        rpc_client: RpcClient,
+    ) -> Result<Self, ScraperError> {
+        let last_processed = storage.get_last_processed_height().await?;
+
+        Ok(BlockProcessor {
+            cancel,
+            last_processed_height: last_processed.try_into().unwrap_or_default(),
+            last_processed_at: Instant::now(),
+            pending_sync: Default::default(),
+            queued_blocks: Default::default(),
+            rpc_client,
+            incoming: incoming.into(),
+            block_requester,
+            storage,
+            block_modules: vec![],
+            tx_modules: vec![],
+            msg_modules: vec![],
+        })
+    }
+
+    async fn process_block(&mut self, block: BlockToProcess) -> Result<(), ScraperError> {
+        info!("processing block at height {}", block.height);
+
+        let full_info = self.rpc_client.try_get_full_details(block).await?;
+
+        debug!(
+            "this block has {} transaction(s)",
+            full_info.transactions.len()
+        );
+        for tx in &full_info.transactions {
+            debug!("{} has {} message(s)", tx.hash, tx.tx.body.messages.len());
+            for (index, msg) in tx.tx.body.messages.iter().enumerate() {
+                debug!("{index}: {:?}", msg.type_url)
+            }
+        }
+
+        // process the entire block as a transaction so that if anything fails,
+        // we won't end up with a corrupted storage.
+        let mut tx = self.storage.begin_processing_tx().await?;
+
+        persist_block(&full_info, &mut tx).await?;
+
+        // let the modules do whatever they want
+        // the ones wanting the full block:
+        for block_module in &mut self.block_modules {
+            block_module.handle_block(&full_info, &mut tx).await?;
+        }
+
+        // the ones wanting transactions:
+        for block_tx in full_info.transactions {
+            for tx_module in &mut self.tx_modules {
+                tx_module.handle_tx(&block_tx, &mut tx).await?;
+            }
+            // the ones concerned with individual messages
+            for (index, msg) in block_tx.tx.body.messages.iter().enumerate() {
+                for msg_module in &mut self.msg_modules {
+                    msg_module
+                        .handle_msg(index, msg, &block_tx, &mut tx)
+                        .await?
+                }
+            }
+        }
+
+        tx.commit()
+            .await
+            .map_err(|source| ScraperError::StorageTxCommitFailure { source })?;
+
+        self.last_processed_height = full_info.block.header.height.value() as u32;
+        self.last_processed_at = Instant::now();
+
+        Ok(())
+    }
+
+    pub fn set_block_modules(&mut self, modules: Vec<Box<dyn BlockModule + Send>>) {
+        self.block_modules = modules;
+    }
+
+    pub fn set_tx_modules(&mut self, modules: Vec<Box<dyn TxModule + Send>>) {
+        self.tx_modules = modules;
+    }
+
+    pub fn set_msg_modules(&mut self, modules: Vec<Box<dyn MsgModule + Send>>) {
+        self.msg_modules = modules;
+    }
+
+    async fn maybe_request_missing_blocks(&mut self) -> Result<(), ScraperError> {
+        // we're still processing, so we're good
+        if self.last_processed_at.elapsed() < MAX_MISSING_BLOCKS_DELAY {
+            debug!("no need to request missing blocks");
+            return Ok(());
+        }
+
+        if self.try_request_pending().await {
+            return Ok(());
+        }
+
+        // TODO: properly fill in the gaps later with BlockRequest::Specific,
+        let request_range = if let Some((next_available, _)) = self.queued_blocks.first_key_value()
+        {
+            self.last_processed_height + 1..*next_available
+        } else {
+            let current_height = self.rpc_client.current_block_height().await? as u32;
+            self.last_processed_height + 1..current_height + 1
+        };
+
+        self.request_missing_blocks(request_range).await?;
+
+        Ok(())
+    }
+
+    async fn request_missing_blocks(
+        &mut self,
+        request_range: Range<u32>,
+    ) -> Result<(), ScraperError> {
+        let request_range = if request_range.len() > MAX_RANGE_SIZE {
+            let mut split = split_request_range(request_range);
+
+            // SAFETY: we know that after the split of a non-empty range we have AT LEAST one value
+            #[allow(clippy::unwrap_used)]
+            let first = split.pop_front().unwrap();
+            self.pending_sync.queued_requests = split;
+            self.pending_sync.request_in_flight = first.clone().collect();
+
+            first
+        } else {
+            request_range
+        };
+
+        self.send_blocks_request(request_range).await
+    }
+
+    // technically we're not mutating self here,
+    // but we need it to help the compiler figure out the future is `Send`
+    async fn send_blocks_request(&mut self, request_range: Range<u32>) -> Result<(), ScraperError> {
+        debug!("requesting missing blocks: {request_range:?}");
+
+        self.block_requester
+            .send(BlockRequest::Range(request_range))
+            .await?;
+        Ok(())
+    }
+
+    async fn next_incoming(&mut self, block: BlockToProcess) {
+        let height = block.height;
+
+        self.pending_sync.request_in_flight.remove(&height);
+
+        if self.last_processed_height == 0 {
+            // this is the first time we've started up the process
+            debug!("setting up initial processing height");
+            self.last_processed_height = height - 1
+        }
+
+        if height <= self.last_processed_height {
+            warn!("we have already processed block for height {height}");
+            return;
+        }
+
+        if self.last_processed_height + 1 != height {
+            if self.queued_blocks.insert(height, block).is_some() {
+                warn!("we have already queued up block for height {height}");
+            }
+            return;
+        }
+
+        if let Err(err) = self.process_block(block).await {
+            error!("failed to process block at height {height}: {err}");
+            return;
+        }
+
+        // process as much as we can from the queue
+        let mut next = height + 1;
+        while let Some(next_block) = self.queued_blocks.remove(&next) {
+            if let Err(err) = self.process_block(next_block).await {
+                error!("failed to process queued-up block at height {next}: {err}")
+            }
+            next += 1;
+        }
+
+        self.try_request_pending().await;
+    }
+
+    async fn try_request_pending(&mut self) -> bool {
+        if self.pending_sync.request_in_flight.is_empty() {
+            if let Some(next_sync) = self.pending_sync.queued_requests.pop_front() {
+                debug!(
+                    "current request range has been resolved. requesting another bunch of blocks"
+                );
+                if let Err(err) = self.send_blocks_request(next_sync.clone()).await {
+                    error!("failed to request resync blocks: {err}");
+                    self.pending_sync.queued_requests.push_front(next_sync);
+                } else {
+                    self.pending_sync.request_in_flight = next_sync.collect()
+                }
+
+                return true;
+            }
+        }
+
+        false
+    }
+
+    // technically we're not mutating self here,
+    // but we need it to help the compiler figure out the future is `Send`
+    async fn startup_resync(&mut self) -> Result<(), ScraperError> {
+        assert!(self.pending_sync.is_empty());
+
+        let latest_block = self.rpc_client.current_block_height().await? as u32;
+        if latest_block > self.last_processed_height && self.last_processed_height != 0 {
+            let request_range = self.last_processed_height + 1..latest_block + 1;
+            info!("we need to request {request_range:?} to resync");
+            self.request_missing_blocks(request_range).await?;
+        }
+
+        Ok(())
+    }
+
+    pub(crate) async fn run(&mut self) {
+        info!("starting processing loop");
+
+        // sure, we could be more efficient and reset it on every processed block,
+        // but the overhead is so minimal that it doesn't matter
+        let mut missing_check_interval = interval_at(
+            Instant::now().add(MISSING_BLOCKS_CHECK_INTERVAL),
+            MISSING_BLOCKS_CHECK_INTERVAL,
+        );
+
+        if let Err(err) = self.startup_resync().await {
+            error!("failed to perform startup sync: {err}");
+            self.cancel.cancel();
+            return;
+        }
+
+        loop {
+            tokio::select! {
+                _ = self.cancel.cancelled() => {
+                    info!("received cancellation token");
+                    break
+                }
+                _ = missing_check_interval.tick() => {
+                    if let Err(err) = self.maybe_request_missing_blocks().await {
+                        error!("failed to request missing blocks: {err}")
+                    }
+                }
+                block = self.incoming.next() => {
+                    match block {
+                        Some(block) => self.next_incoming(block).await,
+                        None => {
+                            warn!("stopped receiving new blocks");
+                            self.cancel.cancel();
+                            break
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

common/nyxd-scraper/src/block_processor/types.rs

@@ -0,0 +1,121 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::error::ScraperError;
+use crate::helpers;
+use tendermint::{abci, block, tx, Block, Hash};
+use tendermint_rpc::endpoint::{block as block_endpoint, block_results, validators};
+use tendermint_rpc::event::{Event, EventData};
+
+// just get all everything out of tx::Response, but parse raw `tx` bytes
+#[derive(Clone, Debug)]
+pub struct ParsedTransactionResponse {
+    /// The hash of the transaction.
+    ///
+    /// Deserialized from a hex-encoded string (there is a discrepancy between
+    /// the format used for the request and the format used for the response in
+    /// the Tendermint RPC).
+    pub hash: Hash,
+
+    pub height: block::Height,
+
+    pub index: u32,
+
+    pub tx_result: abci::types::ExecTxResult,
+
+    pub tx: cosmrs::tx::Tx,
+
+    pub proof: Option<tx::Proof>,
+}
+
+#[derive(Debug)]
+pub struct FullBlockInformation {
+    /// Basic block information, including its signers.
+    pub block: Block,
+
+    /// All of the emitted events alongside any tx results.
+    pub results: block_results::Response,
+
+    /// Validator set for this particular block
+    pub validators: validators::Response,
+
+    /// Transaction results from this particular block
+    pub transactions: Vec<ParsedTransactionResponse>,
+}
+
+impl FullBlockInformation {
+    pub fn ensure_proposer(&self) -> Result<(), ScraperError> {
+        let block_proposer = self.block.header.proposer_address;
+        if !self
+            .validators
+            .validators
+            .iter()
+            .any(|v| v.address == block_proposer)
+        {
+            let proposer = helpers::validator_consensus_address(block_proposer)?;
+            return Err(ScraperError::BlockProposerNotInValidatorSet {
+                height: self.block.header.height.value() as u32,
+                proposer: proposer.to_string(),
+            });
+        }
+        Ok(())
+    }
+}
+
+pub(crate) struct BlockToProcess {
+    pub(crate) height: u32,
+    pub(crate) block: Block,
+}
+
+impl From<Block> for BlockToProcess {
+    fn from(block: Block) -> Self {
+        BlockToProcess {
+            height: block.header.height.value() as u32,
+            block,
+        }
+    }
+}
+
+impl TryFrom<Event> for BlockToProcess {
+    type Error = ScraperError;
+
+    fn try_from(event: Event) -> Result<Self, Self::Error> {
+        let query = event.query.clone();
+
+        // TODO: we're losing `result_begin_block` and `result_end_block` here but maybe that's fine?
+        let maybe_block = match event.data {
+            // we don't care about `NewBlock` until CometBFT 0.38, i.e. until we upgrade to wasmd 0.50
+            EventData::NewBlock { .. } => {
+                return Err(ScraperError::InvalidSubscriptionEvent {
+                    query,
+                    kind: "NewBlock".to_string(),
+                })
+            }
+            EventData::LegacyNewBlock { block, .. } => block,
+            EventData::Tx { .. } => {
+                return Err(ScraperError::InvalidSubscriptionEvent {
+                    query,
+                    kind: "Tx".to_string(),
+                })
+            }
+            EventData::GenericJsonEvent(_) => {
+                return Err(ScraperError::InvalidSubscriptionEvent {
+                    query,
+                    kind: "GenericJsonEvent".to_string(),
+                })
+            }
+        };
+
+        let Some(block) = maybe_block else {
+            return Err(ScraperError::EmptyBlockData { query });
+        };
+
+        Ok((*block).into())
+    }
+}
+
+impl From<block_endpoint::Response> for BlockToProcess {
+    fn from(value: block_endpoint::Response) -> Self {
+        value.block.into()
+    }
+}

common/nyxd-scraper/src/block_requester/mod.rs

@@ -0,0 +1,91 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::block_processor::types::BlockToProcess;
+use crate::error::ScraperError;
+use crate::rpc_client::RpcClient;
+use futures::StreamExt;
+use std::ops::Range;
+use tokio::sync::mpsc::{Receiver, UnboundedSender};
+use tokio_stream::wrappers::ReceiverStream;
+use tokio_util::sync::CancellationToken;
+use tracing::{error, info, instrument, warn};
+
+#[derive(Debug)]
+pub enum BlockRequest {
+    Range(Range<u32>),
+
+    // UNIMPLEMENTED:
+    #[allow(dead_code)]
+    Specific(Vec<u32>),
+}
+
+pub(crate) struct BlockRequester {
+    cancel: CancellationToken,
+    rpc_client: RpcClient,
+    requests: ReceiverStream<BlockRequest>,
+    blocks: UnboundedSender<BlockToProcess>,
+}
+
+impl BlockRequester {
+    pub(crate) fn new(
+        cancel: CancellationToken,
+        rpc_client: RpcClient,
+        requests: Receiver<BlockRequest>,
+        blocks: UnboundedSender<BlockToProcess>,
+    ) -> Self {
+        BlockRequester {
+            cancel,
+            rpc_client,
+            requests: requests.into(),
+            blocks,
+        }
+    }
+
+    async fn request_and_send(&self, height: u32) -> Result<(), ScraperError> {
+        let block = self.rpc_client.get_basic_block_details(height).await?;
+        self.blocks.send(block.into())?;
+        Ok(())
+    }
+
+    async fn request_blocks<I: IntoIterator<Item = u32>>(&self, heights: I) {
+        futures::stream::iter(heights)
+            .for_each_concurrent(4, |height| async move {
+                if let Err(err) = self.request_and_send(height).await {
+                    error!("failed to request block data: {err}")
+                }
+            })
+            .await
+    }
+
+    #[instrument(skip(self))]
+    async fn handle_blocks_request(&self, request: BlockRequest) {
+        info!("received request for missed blocks");
+
+        match request {
+            BlockRequest::Range(range) => self.request_blocks(range).await,
+            BlockRequest::Specific(heights) => self.request_blocks(heights).await,
+        }
+    }
+
+    pub(crate) async fn run(&mut self) {
+        loop {
+            tokio::select! {
+               _ = self.cancel.cancelled() => {
+                    info!("received cancellation token");
+                    break
+                }
+                maybe_request = self.requests.next() => {
+                    match maybe_request {
+                        Some(request) => self.handle_blocks_request(request).await,
+                        None => {
+                            warn!("stopped receiving new requests");
+                            self.cancel.cancel();
+                            break
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

common/nyxd-scraper/src/constants.rs

@@ -0,0 +1,47 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use const_format::concatcp;
+
+// TODO: make those configurable via 'NymNetworkDetails'
+
+// BECH32_PREFIX defines the main SDK Bech32 prefix of an account's address
+pub const BECH32_PREFIX: &str = "n";
+
+// ACCOUNT_PREFIX is the prefix for account keys
+pub const ACCOUNT_PREFIX: &str = "acc";
+// VALIDATOR_PREFIX is the prefix for validator keys
+pub const VALIDATOR_PREFIX: &str = "val";
+// CONSENSUS_PREFIX is the prefix for consensus keys
+pub const CONSENSUS_PREFIX: &str = "cons";
+// PUBKEY_PREFIX is the prefix for public keys
+pub const PUBKEY_PREFIX: &str = "pub";
+// OPERATOR_PREFIX is the prefix for operator keys
+pub const OPERATOR_PREFIX: &str = "oper";
+// ADDRESS_PREFIX is the prefix for addresses
+pub const ADDRESS_PREFIX: &str = "addr";
+
+// BECH32_ACCOUNT_ADDRESS_PREFIX defines the Bech32 prefix of an account's address
+pub const BECH32_ACCOUNT_ADDRESS_PREFIX: &str = BECH32_PREFIX;
+// BECH32_ACCOUNT_PUBKEY_PREFIX defines the Bech32 prefix of an account's public key
+pub const BECH32_ACCOUNT_PUBKEY_PREFIX: &str = concatcp!(BECH32_PREFIX, PUBKEY_PREFIX);
+// BECH32_VALIDATOR_ADDRESS_PREFIX defines the Bech32 prefix of a validator's operator address
+pub const BECH32_VALIDATOR_ADDRESS_PREFIX: &str =
+    concatcp!(BECH32_PREFIX, VALIDATOR_PREFIX, OPERATOR_PREFIX);
+// BECH32_VALIDATOR_PUBKEY_PREFIX defines the Bech32 prefix of a validator's operator public key
+pub const BECH32_VALIDATOR_PUBKEY_PREFIX: &str = concatcp!(
+    BECH32_PREFIX,
+    VALIDATOR_PREFIX,
+    OPERATOR_PREFIX,
+    PUBKEY_PREFIX
+);
+// BECH32_CONSENSUS_ADDRESS_PREFIX defines the Bech32 prefix of a consensus node address
+pub const BECH32_CONSENSUS_ADDRESS_PREFIX: &str =
+    concatcp!(BECH32_PREFIX, VALIDATOR_PREFIX, CONSENSUS_PREFIX);
+// BECH32_CONESNSUS_PUBKEY_PREFIX defines the Bech32 prefix of a consensus node public key
+pub const BECH32_CONESNSUS_PUBKEY_PREFIX: &str = concatcp!(
+    BECH32_PREFIX,
+    VALIDATOR_PREFIX,
+    CONSENSUS_PREFIX,
+    PUBKEY_PREFIX
+);

common/nyxd-scraper/src/error.rs

@@ -0,0 +1,131 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use tendermint::Hash;
+use thiserror::Error;
+use tokio::sync::mpsc::error::SendError;
+
+#[derive(Debug, Error)]
+pub enum ScraperError {
+    #[error("experienced internal database error: {0}")]
+    InternalDatabaseError(#[from] sqlx::Error),
+
+    #[error("failed to perform startup SQL migration: {0}")]
+    StartupMigrationFailure(#[from] sqlx::migrate::MigrateError),
+
+    #[error("can't add any modules to the scraper as it's already running")]
+    ScraperAlreadyRunning,
+
+    #[error("failed to establish websocket connection to {url}: {source}")]
+    WebSocketConnectionFailure {
+        url: String,
+        #[source]
+        source: tendermint_rpc::Error,
+    },
+
+    #[error("failed to establish rpc connection to {url}: {source}")]
+    HttpConnectionFailure {
+        url: String,
+        #[source]
+        source: tendermint_rpc::Error,
+    },
+
+    #[error("failed to create chain subscription: {source}")]
+    ChainSubscriptionFailure {
+        #[source]
+        source: tendermint_rpc::Error,
+    },
+
+    #[error("could not obtain basic block information at height: {height}: {source}")]
+    BlockQueryFailure {
+        height: u32,
+        #[source]
+        source: tendermint_rpc::Error,
+    },
+
+    #[error("could not obtain block results information at height: {height}: {source}")]
+    BlockResultsQueryFailure {
+        height: u32,
+        #[source]
+        source: tendermint_rpc::Error,
+    },
+
+    #[error("could not obtain validators information at height: {height}: {source}")]
+    ValidatorsQueryFailure {
+        height: u32,
+        #[source]
+        source: tendermint_rpc::Error,
+    },
+
+    #[error("could not obtain tx results for tx: {hash}: {source}")]
+    TxResultsQueryFailure {
+        hash: Hash,
+        #[source]
+        source: tendermint_rpc::Error,
+    },
+
+    #[error("could not obtain current abci info: {source}")]
+    AbciInfoQueryFailure {
+        #[source]
+        source: tendermint_rpc::Error,
+    },
+
+    #[error("could not parse tx {hash}: {source}")]
+    TxParseFailure {
+        hash: Hash,
+        #[source]
+        source: cosmrs::ErrorReport,
+    },
+
+    #[error("received an invalid chain subscription event of kind {kind} while we were waiting for new block data (query: '{query}')")]
+    InvalidSubscriptionEvent { query: String, kind: String },
+
+    #[error("received block data was empty (query: '{query}')")]
+    EmptyBlockData { query: String },
+
+    #[error("reached maximum number of allowed errors for subscription events")]
+    MaximumSubscriptionFailures,
+
+    #[error("failed to begin storage tx: {source}")]
+    StorageTxBeginFailure {
+        #[source]
+        source: sqlx::Error,
+    },
+
+    #[error("failed to commit storage tx: {source}")]
+    StorageTxCommitFailure {
+        #[source]
+        source: sqlx::Error,
+    },
+
+    #[error("failed to send on a closed channel")]
+    ClosedChannelError,
+
+    #[error("failed to parse validator's address: {source}")]
+    MalformedValidatorAddress {
+        #[source]
+        source: eyre::Report,
+    },
+
+    #[error("failed to parse validator's address: {source}")]
+    MalformedValidatorPubkey {
+        #[source]
+        source: eyre::Report,
+    },
+
+    #[error(
+        "could not find the block proposer ('{proposer}') for height {height} in the validator set"
+    )]
+    BlockProposerNotInValidatorSet { height: u32, proposer: String },
+
+    #[error(
+        "could not find validator information for {address}; the validator has signed a commit"
+    )]
+    MissingValidatorInfoCommitted { address: String },
+}
+
+impl<T> From<SendError<T>> for ScraperError {
+    fn from(_: SendError<T>) -> Self {
+        ScraperError::ClosedChannelError
+    }
+}

common/nyxd-scraper/src/helpers.rs

@@ -0,0 +1,46 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::block_processor::types::ParsedTransactionResponse;
+use crate::constants::{BECH32_CONESNSUS_PUBKEY_PREFIX, BECH32_CONSENSUS_ADDRESS_PREFIX};
+use crate::error::ScraperError;
+use cosmrs::AccountId;
+use sha2::{Digest, Sha256};
+use tendermint::{account, PublicKey};
+use tendermint::{validator, Hash};
+use tendermint_rpc::endpoint::validators;
+
+pub(crate) fn tx_hash<M: AsRef<[u8]>>(raw_tx: M) -> Hash {
+    Hash::Sha256(Sha256::digest(raw_tx).into())
+}
+
+pub(crate) fn validator_pubkey_to_bech32(pubkey: PublicKey) -> Result<AccountId, ScraperError> {
+    // TODO: this one seem to attach additional prefix to they pubkeys, is that what we want instead maybe?
+    // Ok(pubkey.to_bech32(BECH32_CONESNSUS_PUBKEY_PREFIX))
+    AccountId::new(BECH32_CONESNSUS_PUBKEY_PREFIX, &pubkey.to_bytes())
+        .map_err(|source| ScraperError::MalformedValidatorPubkey { source })
+}
+
+pub(crate) fn validator_consensus_address(id: account::Id) -> Result<AccountId, ScraperError> {
+    AccountId::new(BECH32_CONSENSUS_ADDRESS_PREFIX, id.as_ref())
+        .map_err(|source| ScraperError::MalformedValidatorAddress { source })
+}
+
+pub(crate) fn tx_gas_sum(txs: &[ParsedTransactionResponse]) -> i64 {
+    txs.iter().map(|tx| tx.tx_result.gas_used).sum()
+}
+
+pub(crate) fn validator_info(
+    id: account::Id,
+    validators: &validators::Response,
+) -> Result<&validator::Info, ScraperError> {
+    match validators.validators.iter().find(|v| v.address == id) {
+        Some(info) => Ok(info),
+        None => {
+            let addr = validator_consensus_address(id)?;
+            Err(ScraperError::MissingValidatorInfoCommitted {
+                address: addr.to_string(),
+            })
+        }
+    }
+}

common/nyxd-scraper/src/lib.rs

@@ -0,0 +1,18 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+#![warn(clippy::expect_used)]
+#![warn(clippy::unwrap_used)]
+
+pub(crate) mod block_processor;
+pub(crate) mod block_requester;
+pub mod constants;
+pub mod error;
+pub(crate) mod helpers;
+pub mod modules;
+pub(crate) mod rpc_client;
+pub(crate) mod scraper;
+pub mod storage;
+
+pub use modules::{BlockModule, MsgModule, TxModule};
+pub use scraper::{Config, NyxdScraper};

common/nyxd-scraper/src/modules/block_module.rs

@@ -0,0 +1,16 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::block_processor::types::FullBlockInformation;
+use crate::error::ScraperError;
+use crate::storage::StorageTransaction;
+use async_trait::async_trait;
+
+#[async_trait]
+pub trait BlockModule {
+    async fn handle_block(
+        &mut self,
+        block: &FullBlockInformation,
+        storage_tx: &mut StorageTransaction,
+    ) -> Result<(), ScraperError>;
+}

common/nyxd-scraper/src/modules/mod.rs

@@ -0,0 +1,10 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+mod block_module;
+mod msg_module;
+mod tx_module;
+
+pub use block_module::BlockModule;
+pub use msg_module::MsgModule;
+pub use tx_module::TxModule;

common/nyxd-scraper/src/modules/msg_module.rs

@@ -0,0 +1,19 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::block_processor::types::ParsedTransactionResponse;
+use crate::error::ScraperError;
+use crate::storage::StorageTransaction;
+use async_trait::async_trait;
+use cosmrs::Any;
+
+#[async_trait]
+pub trait MsgModule {
+    async fn handle_msg(
+        &mut self,
+        index: usize,
+        msg: &Any,
+        tx: &ParsedTransactionResponse,
+        storage_tx: &mut StorageTransaction,
+    ) -> Result<(), ScraperError>;
+}

common/nyxd-scraper/src/modules/tx_module.rs

@@ -0,0 +1,16 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::block_processor::types::ParsedTransactionResponse;
+use crate::error::ScraperError;
+use crate::storage::StorageTransaction;
+use async_trait::async_trait;
+
+#[async_trait]
+pub trait TxModule {
+    async fn handle_tx(
+        &mut self,
+        tx: &ParsedTransactionResponse,
+        storage_tx: &mut StorageTransaction,
+    ) -> Result<(), ScraperError>;
+}

common/nyxd-scraper/src/rpc_client.rs

@@ -0,0 +1,175 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::block_processor::types::{
+    BlockToProcess, FullBlockInformation, ParsedTransactionResponse,
+};
+use crate::error::ScraperError;
+use crate::helpers::tx_hash;
+use futures::future::join3;
+use futures::StreamExt;
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use tendermint::Hash;
+use tendermint_rpc::endpoint::{block, block_results, tx, validators};
+use tendermint_rpc::{Client, HttpClient, Paging};
+use tokio::sync::Mutex;
+use tracing::{debug, instrument};
+use url::Url;
+
+#[derive(Clone)]
+pub struct RpcClient {
+    // right now I don't care about anything nym specific, so a simple http client is sufficient,
+    // once this is inadequate, we can switch to a NyxdClient
+    inner: Arc<HttpClient>,
+}
+
+impl RpcClient {
+    pub fn new(url: &Url) -> Result<Self, ScraperError> {
+        let http_client = HttpClient::new(url.as_str()).map_err(|source| {
+            ScraperError::HttpConnectionFailure {
+                url: url.to_string(),
+                source,
+            }
+        })?;
+
+        Ok(RpcClient {
+            inner: Arc::new(http_client),
+        })
+    }
+
+    #[instrument(skip(self, block), fields(height = block.height))]
+    pub async fn try_get_full_details(
+        &self,
+        block: BlockToProcess,
+    ) -> Result<FullBlockInformation, ScraperError> {
+        debug!("getting complete block details");
+        let height = block.height;
+
+        // make all the http requests concurrently
+        let (results, validators, raw_transactions) = join3(
+            self.get_block_results(height),
+            self.get_validators_details(height),
+            self.get_transaction_results(&block.block.data),
+        )
+        .await;
+
+        let raw_transactions = raw_transactions?;
+        let mut transactions = Vec::with_capacity(raw_transactions.len());
+        for tx in raw_transactions {
+            transactions.push(ParsedTransactionResponse {
+                hash: tx.hash,
+                height: tx.height,
+                index: tx.index,
+                tx_result: tx.tx_result,
+                tx: cosmrs::Tx::from_bytes(&tx.tx).map_err(|source| {
+                    ScraperError::TxParseFailure {
+                        hash: tx.hash,
+                        source,
+                    }
+                })?,
+                proof: tx.proof,
+            })
+        }
+
+        Ok(FullBlockInformation {
+            block: block.block,
+            results: results?,
+            validators: validators?,
+            transactions,
+        })
+    }
+
+    #[instrument(skip(self), err(Display))]
+    pub async fn get_basic_block_details(
+        &self,
+        height: u32,
+    ) -> Result<block::Response, ScraperError> {
+        debug!("getting basic block details");
+
+        self.inner
+            .block(height)
+            .await
+            .map_err(|source| ScraperError::BlockQueryFailure { height, source })
+    }
+
+    #[instrument(skip(self), err(Display))]
+    pub async fn get_block_results(
+        &self,
+        height: u32,
+    ) -> Result<block_results::Response, ScraperError> {
+        debug!("getting block results");
+
+        self.inner
+            .block_results(height)
+            .await
+            .map_err(|source| ScraperError::BlockResultsQueryFailure { height, source })
+    }
+
+    pub(crate) async fn current_block_height(&self) -> Result<u64, ScraperError> {
+        debug!("getting current block height");
+
+        let info = self
+            .inner
+            .abci_info()
+            .await
+            .map_err(|source| ScraperError::AbciInfoQueryFailure { source })?;
+        Ok(info.last_block_height.value())
+    }
+
+    async fn get_transaction_results(
+        &self,
+        raw: &[Vec<u8>],
+    ) -> Result<Vec<tx::Response>, ScraperError> {
+        let ordered_results = Arc::new(Mutex::new(BTreeMap::new()));
+
+        // "Data is just a wrapper for a list of transactions, where transactions are arbitrary byte arrays"
+        // source: https://github.com/tendermint/spec/blob/d46cd7f573a2c6a2399fcab2cde981330aa63f37/spec/core/data_structures.md#data
+        //
+        // I hate that zip as much as you, dear reader, but for some reason the compiler didn't let me remove the `move`
+        futures::stream::iter(
+            raw.iter()
+                .map(tx_hash)
+                .enumerate()
+                .zip(std::iter::repeat(ordered_results.clone())),
+        )
+        .for_each_concurrent(4, |((id, tx_hash), ordered_results)| async move {
+            let res = self.get_transaction_result(tx_hash).await;
+            ordered_results.lock().await.insert(id, res);
+        })
+        .await;
+
+        // safety the futures have completed so we MUST have the only arc reference
+        #[allow(clippy::unwrap_used)]
+        let inner = Arc::into_inner(ordered_results).unwrap().into_inner();
+
+        // BTreeMap is ordered by its keys so we're guaranteed to get txs in correct order
+        inner.into_values().collect()
+    }
+
+    #[instrument(skip(self, tx_hash), fields(tx_hash = %tx_hash), err(Display))]
+    async fn get_transaction_result(&self, tx_hash: Hash) -> Result<tx::Response, ScraperError> {
+        debug!("getting tx results");
+
+        self.inner
+            .tx(tx_hash, false)
+            .await
+            .map_err(|source| ScraperError::TxResultsQueryFailure {
+                hash: tx_hash,
+                source,
+            })
+    }
+
+    #[instrument(skip(self))]
+    pub async fn get_validators_details(
+        &self,
+        height: u32,
+    ) -> Result<validators::Response, ScraperError> {
+        debug!("getting validators set");
+
+        self.inner
+            .validators(height, Paging::All)
+            .await
+            .map_err(|source| ScraperError::ValidatorsQueryFailure { height, source })
+    }
+}

common/nyxd-scraper/src/scraper/mod.rs

@@ -0,0 +1,197 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::block_processor::BlockProcessor;
+use crate::block_requester::BlockRequester;
+use crate::error::ScraperError;
+use crate::modules::{BlockModule, MsgModule, TxModule};
+use crate::rpc_client::RpcClient;
+use crate::scraper::subscriber::{run_websocket_driver, ChainSubscriber};
+use crate::storage::ScraperStorage;
+use std::path::PathBuf;
+use tendermint_rpc::WebSocketClientDriver;
+use tokio::sync::mpsc::{channel, unbounded_channel};
+use tokio_util::sync::CancellationToken;
+use tokio_util::task::TaskTracker;
+use tracing::info;
+use url::Url;
+
+mod subscriber;
+
+pub struct Config {
+    /// Url to the websocket endpoint of a validator, for example `wss://rpc.nymtech.net/websocket`
+    pub websocket_url: Url,
+
+    /// Url to the rpc endpoint of a validator, for example `https://rpc.nymtech.net/`
+    pub rpc_url: Url,
+
+    pub database_path: PathBuf,
+}
+
+pub struct NyxdScraperBuilder {
+    config: Config,
+
+    block_modules: Vec<Box<dyn BlockModule + Send>>,
+    tx_modules: Vec<Box<dyn TxModule + Send>>,
+    msg_modules: Vec<Box<dyn MsgModule + Send>>,
+}
+
+impl NyxdScraperBuilder {
+    pub async fn build_and_start(self) -> Result<NyxdScraper, ScraperError> {
+        let scraper = NyxdScraper::new(self.config).await?;
+
+        let (processing_tx, processing_rx) = unbounded_channel();
+        let (req_tx, req_rx) = channel(5);
+
+        let rpc_client = RpcClient::new(&scraper.config.rpc_url)?;
+
+        // create the tasks
+        let block_requester = BlockRequester::new(
+            scraper.cancel_token.clone(),
+            rpc_client.clone(),
+            req_rx,
+            processing_tx.clone(),
+        );
+        let mut block_processor = BlockProcessor::new(
+            scraper.cancel_token.clone(),
+            processing_rx,
+            req_tx,
+            scraper.storage.clone(),
+            rpc_client,
+        )
+        .await?;
+        block_processor.set_block_modules(self.block_modules);
+        block_processor.set_tx_modules(self.tx_modules);
+        block_processor.set_msg_modules(self.msg_modules);
+
+        let mut chain_subscriber = ChainSubscriber::new(
+            &scraper.config.websocket_url,
+            scraper.cancel_token.clone(),
+            processing_tx,
+        )
+        .await?;
+        let ws_driver = chain_subscriber.ws_driver();
+
+        scraper.start_tasks(
+            block_requester,
+            block_processor,
+            chain_subscriber,
+            ws_driver,
+        );
+
+        Ok(scraper)
+    }
+
+    pub fn new(config: Config) -> Self {
+        NyxdScraperBuilder {
+            config,
+            block_modules: vec![],
+            tx_modules: vec![],
+            msg_modules: vec![],
+        }
+    }
+
+    pub fn with_block_module<M: BlockModule + Send + 'static>(mut self, module: M) -> Self {
+        self.block_modules.push(Box::new(module));
+        self
+    }
+
+    pub fn with_tx_module<M: TxModule + Send + 'static>(mut self, module: M) -> Self {
+        self.tx_modules.push(Box::new(module));
+        self
+    }
+
+    pub fn with_msg_module<M: MsgModule + Send + 'static>(mut self, module: M) -> Self {
+        self.msg_modules.push(Box::new(module));
+        self
+    }
+}
+
+pub struct NyxdScraper {
+    config: Config,
+
+    task_tracker: TaskTracker,
+    cancel_token: CancellationToken,
+    storage: ScraperStorage,
+}
+
+impl NyxdScraper {
+    pub fn builder(config: Config) -> NyxdScraperBuilder {
+        NyxdScraperBuilder::new(config)
+    }
+
+    pub async fn new(config: Config) -> Result<Self, ScraperError> {
+        let storage = ScraperStorage::init(&config.database_path).await?;
+
+        Ok(NyxdScraper {
+            config,
+            task_tracker: TaskTracker::new(),
+            cancel_token: CancellationToken::new(),
+            storage,
+        })
+    }
+
+    fn start_tasks(
+        &self,
+        mut block_requester: BlockRequester,
+        mut block_processor: BlockProcessor,
+        mut chain_subscriber: ChainSubscriber,
+        ws_driver: WebSocketClientDriver,
+    ) {
+        self.task_tracker
+            .spawn(async move { block_requester.run().await });
+        self.task_tracker
+            .spawn(async move { block_processor.run().await });
+        self.task_tracker
+            .spawn(async move { chain_subscriber.run().await });
+        self.task_tracker
+            .spawn(run_websocket_driver(ws_driver, self.cancel_token.clone()));
+        self.task_tracker.close();
+    }
+
+    pub async fn start(&self) -> Result<(), ScraperError> {
+        let (processing_tx, processing_rx) = unbounded_channel();
+        let (req_tx, req_rx) = channel(5);
+
+        let rpc_client = RpcClient::new(&self.config.rpc_url)?;
+
+        // create the tasks
+        let block_requester = BlockRequester::new(
+            self.cancel_token.clone(),
+            rpc_client.clone(),
+            req_rx,
+            processing_tx.clone(),
+        );
+        let block_processor = BlockProcessor::new(
+            self.cancel_token.clone(),
+            processing_rx,
+            req_tx,
+            self.storage.clone(),
+            rpc_client,
+        )
+        .await?;
+        let mut chain_subscriber = ChainSubscriber::new(
+            &self.config.websocket_url,
+            self.cancel_token.clone(),
+            processing_tx,
+        )
+        .await?;
+        let ws_driver = chain_subscriber.ws_driver();
+
+        // spawn them
+        self.start_tasks(
+            block_requester,
+            block_processor,
+            chain_subscriber,
+            ws_driver,
+        );
+
+        Ok(())
+    }
+
+    pub async fn stop(self) {
+        info!("stopping the chain scrapper");
+        self.cancel_token.cancel();
+        self.task_tracker.wait().await
+    }
+}

common/nyxd-scraper/src/scraper/subscriber.rs

@@ -0,0 +1,129 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::block_processor::types::BlockToProcess;
+use crate::error::ScraperError;
+use tendermint_rpc::event::Event;
+use tendermint_rpc::query::EventType;
+use tendermint_rpc::{SubscriptionClient, WebSocketClient, WebSocketClientDriver};
+use tokio::sync::mpsc::UnboundedSender;
+use tokio_stream::StreamExt;
+use tokio_util::sync::CancellationToken;
+use tracing::{error, info, warn};
+use url::Url;
+
+const MAX_FAILURES: usize = 10;
+
+pub struct ChainSubscriber {
+    cancel: CancellationToken,
+    block_sender: UnboundedSender<BlockToProcess>,
+
+    websocket_client: WebSocketClient,
+    websocket_driver: Option<WebSocketClientDriver>,
+}
+
+impl ChainSubscriber {
+    pub async fn new(
+        websocket_endpoint: &Url,
+        cancel: CancellationToken,
+        block_sender: UnboundedSender<BlockToProcess>,
+    ) -> Result<Self, ScraperError> {
+        // sure, we could have just used websocket client entirely, but let's keep the logic for
+        // getting current blocks and historical blocks completely separate with the dual connection
+        let (client, driver) = WebSocketClient::new(websocket_endpoint.as_str())
+            .await
+            .map_err(|source| ScraperError::WebSocketConnectionFailure {
+                url: websocket_endpoint.to_string(),
+                source,
+            })?;
+
+        Ok(ChainSubscriber {
+            cancel,
+            block_sender,
+            websocket_client: client,
+            websocket_driver: Some(driver),
+        })
+    }
+
+    fn handle_new_event(&mut self, event: Event) -> Result<(), ScraperError> {
+        if let Err(err) = self.block_sender.send(event.try_into()?) {
+            // this error has nothing to do with the websocket or chain
+            error!("failed to send block for processing: {err} - are we shutting down?")
+        }
+        Ok(())
+    }
+
+    pub(crate) async fn run(&mut self) -> Result<(), ScraperError> {
+        let _drop_guard = self.cancel.clone().drop_guard();
+
+        info!("creating chain subscription");
+        let mut subs = self
+            .websocket_client
+            .subscribe(EventType::NewBlock.into())
+            .await
+            .map_err(|source| ScraperError::ChainSubscriptionFailure { source })?;
+
+        let mut failures = 0;
+
+        info!("starting processing loop");
+        loop {
+            tokio::select! {
+                _ = self.cancel.cancelled() => {
+                    info!("received cancellation token");
+                    break
+                }
+                maybe_event = subs.next() => {
+                    let Some(maybe_event) = maybe_event else {
+                        warn!("stopped receiving new events");
+                        break;
+                    };
+                    match maybe_event {
+                        Ok(event) => {
+                            if let Err(err) = self.handle_new_event(event) {
+                                error!("failed to process received block: {err}");
+                                failures += 1
+                            } else {
+                                failures = 0;
+                            }
+                        }
+                        Err(err) => {
+                            error!("failed to receive a valid subscription event: {err}");
+                            failures += 1
+                        }
+                    }
+                    if failures >= MAX_FAILURES {
+                        // note: the drop_guard will get dropped and thus cause a shutdown
+                        return Err(ScraperError::MaximumSubscriptionFailures);
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+
+    pub(crate) fn ws_driver(&mut self) -> WebSocketClientDriver {
+        #[allow(clippy::expect_used)]
+        self.websocket_driver
+            .take()
+            .expect("websocket driver has already been started!")
+    }
+}
+
+pub async fn run_websocket_driver(driver: WebSocketClientDriver, cancel: CancellationToken) {
+    info!("starting websocket driver");
+    tokio::select! {
+        _ = cancel.cancelled() => {
+            info!("received cancellation token")
+        }
+        res = driver.run() => {
+            match res {
+                Ok(_) => info!("our websocket driver has finished execution"),
+                Err(err) => {
+                    // TODO: in the future just attempt to reconnect
+                    error!("our websocket driver has errored out: {err}")
+                }
+            }
+            cancel.cancel()
+        }
+    }
+}

common/nyxd-scraper/src/storage/helpers.rs

@@ -0,0 +1,2 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0

common/nyxd-scraper/src/storage/manager.rs

@@ -0,0 +1,328 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::storage::models::Validator;
+use sqlx::types::time::OffsetDateTime;
+use sqlx::{Executor, Sqlite};
+use tracing::{instrument, trace};
+
+#[derive(Clone)]
+pub(crate) struct StorageManager {
+    pub(crate) connection_pool: sqlx::SqlitePool,
+}
+
+impl StorageManager {
+    pub(crate) async fn set_initial_metadata(&self) -> Result<(), sqlx::Error> {
+        if sqlx::query("SELECT * from metadata")
+            .fetch_optional(&self.connection_pool)
+            .await?
+            .is_none()
+        {
+            sqlx::query("INSERT INTO metadata (id, last_processed_height) VALUES (0, 0)")
+                .execute(&self.connection_pool)
+                .await?;
+        }
+        Ok(())
+    }
+
+    pub(crate) async fn get_first_block_height_after(
+        &self,
+        time: OffsetDateTime,
+    ) -> Result<Option<i64>, sqlx::Error> {
+        let maybe_record = sqlx::query!(
+            r#"
+                SELECT height
+                FROM block
+                WHERE timestamp > ?
+                ORDER BY timestamp
+                LIMIT 1
+            "#,
+            time
+        )
+        .fetch_optional(&self.connection_pool)
+        .await?;
+
+        if let Some(row) = maybe_record {
+            Ok(row.height)
+        } else {
+            Ok(None)
+        }
+    }
+
+    pub(crate) async fn get_last_block_height_before(
+        &self,
+        time: OffsetDateTime,
+    ) -> Result<Option<i64>, sqlx::Error> {
+        let maybe_record = sqlx::query!(
+            r#"
+                SELECT height
+                FROM block
+                WHERE timestamp < ?
+                ORDER BY timestamp DESC
+                LIMIT 1
+            "#,
+            time
+        )
+        .fetch_optional(&self.connection_pool)
+        .await?;
+
+        if let Some(row) = maybe_record {
+            Ok(row.height)
+        } else {
+            Ok(None)
+        }
+    }
+
+    pub(crate) async fn get_signed_between(
+        &self,
+        consensus_address: &str,
+        start_height: i64,
+        end_height: i64,
+    ) -> Result<i32, sqlx::Error> {
+        let count = sqlx::query!(
+            r#"
+                SELECT COUNT(*) as count FROM pre_commit
+                WHERE 
+                    validator_address == ?
+                    AND height > ? 
+                    AND height < ?
+            "#,
+            consensus_address,
+            start_height,
+            end_height
+        )
+        .fetch_one(&self.connection_pool)
+        .await?
+        .count;
+
+        Ok(count)
+    }
+
+    pub(crate) async fn get_block_validators(
+        &self,
+        height: i64,
+    ) -> Result<Vec<Validator>, sqlx::Error> {
+        sqlx::query_as!(
+            Validator,
+            r#"
+                SELECT * FROM validator 
+                WHERE EXISTS (
+                    SELECT 1 FROM pre_commit
+                    WHERE height == ?
+                    AND pre_commit.validator_address = validator.consensus_address
+                )
+            "#,
+            height
+        )
+        .fetch_all(&self.connection_pool)
+        .await
+    }
+
+    pub(crate) async fn get_last_processed_height(&self) -> Result<i64, sqlx::Error> {
+        let maybe_record = sqlx::query!(
+            r#"
+                SELECT last_processed_height FROM metadata
+            "#
+        )
+        .fetch_optional(&self.connection_pool)
+        .await?;
+
+        if let Some(row) = maybe_record {
+            Ok(row.last_processed_height)
+        } else {
+            Ok(-1)
+        }
+    }
+}
+
+// make those generic over executor so that they could be performed over connection pool and a tx
+
+#[instrument(skip(executor))]
+pub(crate) async fn insert_validator<'a, E>(
+    consensus_address: String,
+    consensus_pubkey: String,
+    executor: E,
+) -> Result<(), sqlx::Error>
+where
+    E: Executor<'a, Database = Sqlite>,
+{
+    trace!("insert validator");
+
+    sqlx::query!(
+        r#"
+            INSERT INTO validator (consensus_address, consensus_pubkey)
+            VALUES (?, ?)
+            ON CONFLICT DO NOTHING
+        "#,
+        consensus_address,
+        consensus_pubkey
+    )
+    .execute(executor)
+    .await?;
+
+    Ok(())
+}
+
+#[instrument(skip(executor))]
+pub(crate) async fn insert_block<'a, E>(
+    height: i64,
+    hash: String,
+    num_txs: u32,
+    total_gas: i64,
+    proposer_address: String,
+    timestamp: OffsetDateTime,
+    executor: E,
+) -> Result<(), sqlx::Error>
+where
+    E: Executor<'a, Database = Sqlite>,
+{
+    trace!("insert block");
+
+    sqlx::query!(
+        r#"
+            INSERT INTO block (height, hash, num_txs, total_gas, proposer_address, timestamp)
+            VALUES (?, ?, ?, ?, ?, ?)
+            ON CONFLICT DO NOTHING
+        "#,
+        height,
+        hash,
+        num_txs,
+        total_gas,
+        proposer_address,
+        timestamp
+    )
+    .execute(executor)
+    .await?;
+
+    Ok(())
+}
+
+#[instrument(skip(executor))]
+pub(crate) async fn insert_precommit<'a, E>(
+    validator_address: String,
+    height: i64,
+    timestamp: OffsetDateTime,
+    voting_power: i64,
+    proposer_priority: i64,
+    executor: E,
+) -> Result<(), sqlx::Error>
+where
+    E: Executor<'a, Database = Sqlite>,
+{
+    trace!("insert precommit");
+
+    sqlx::query!(
+        r#"
+            INSERT INTO pre_commit (validator_address, height, timestamp, voting_power, proposer_priority)
+            VALUES (?, ?, ?, ?, ?)
+            ON CONFLICT (validator_address, timestamp) DO NOTHING
+        "#,
+        validator_address,
+        height,
+        timestamp,
+        voting_power,
+        proposer_priority
+    )
+    .execute(executor)
+    .await?;
+
+    Ok(())
+}
+
+#[instrument(skip(executor))]
+#[allow(clippy::too_many_arguments)]
+pub(crate) async fn insert_transaction<'a, E>(
+    hash: String,
+    height: i64,
+    index: i64,
+    success: bool,
+    messages: i64,
+    memo: String,
+    gas_wanted: i64,
+    gas_used: i64,
+    raw_log: String,
+    executor: E,
+) -> Result<(), sqlx::Error>
+where
+    E: Executor<'a, Database = Sqlite>,
+{
+    trace!("insert transaction");
+
+    sqlx::query!(
+        r#"
+            INSERT INTO "transaction" (hash, height, "index", success, num_messages, memo, gas_wanted, gas_used, raw_log)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+               ON CONFLICT (hash) DO UPDATE
+               SET height = excluded.height,
+               "index" = excluded."index",
+               success = excluded.success,
+               num_messages = excluded.num_messages,
+               memo = excluded.memo,
+               gas_wanted = excluded.gas_wanted,
+               gas_used = excluded.gas_used,
+               raw_log = excluded.raw_log
+        "#,
+            hash,
+            height,
+            index,
+            success,
+            messages,
+            memo,
+            gas_wanted,
+            gas_used,
+            raw_log,
+    )
+        .execute(executor)
+        .await?;
+
+    Ok(())
+}
+
+#[instrument(skip(executor))]
+pub(crate) async fn insert_message<'a, E>(
+    transaction_hash: String,
+    index: i64,
+    typ: String,
+    height: i64,
+    executor: E,
+) -> Result<(), sqlx::Error>
+where
+    E: Executor<'a, Database = Sqlite>,
+{
+    trace!("insert message");
+
+    sqlx::query!(
+        r#"
+            INSERT INTO message (transaction_hash, "index", type, height)
+            VALUES (?, ?, ?, ?)
+            ON CONFLICT (transaction_hash, "index") DO UPDATE
+                SET height = excluded.height,
+                type = excluded.type
+        "#,
+        transaction_hash,
+        index,
+        typ,
+        height
+    )
+    .execute(executor)
+    .await?;
+
+    Ok(())
+}
+
+#[instrument(skip(executor))]
+pub(crate) async fn update_last_processed<'a, E>(
+    height: i64,
+    executor: E,
+) -> Result<(), sqlx::Error>
+where
+    E: Executor<'a, Database = Sqlite>,
+{
+    trace!("update_last_processed");
+
+    sqlx::query!("UPDATE metadata SET last_processed_height = ?", height)
+        .execute(executor)
+        .await?;
+
+    Ok(())
+}

common/nyxd-scraper/src/storage/mod.rs

@@ -0,0 +1,299 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::block_processor::types::{FullBlockInformation, ParsedTransactionResponse};
+use crate::error::ScraperError;
+use crate::storage::manager::{
+    insert_block, insert_message, insert_precommit, insert_transaction, insert_validator,
+    update_last_processed, StorageManager,
+};
+use crate::storage::models::Validator;
+use sqlx::types::time::OffsetDateTime;
+use sqlx::{ConnectOptions, Sqlite, Transaction};
+use std::fmt::Debug;
+use std::path::Path;
+use tendermint::block::{Commit, CommitSig};
+use tendermint::Block;
+use tendermint_rpc::endpoint::validators;
+use tracing::{debug, error, info, instrument, trace, warn};
+
+mod helpers;
+mod manager;
+mod models;
+
+pub type StorageTransaction = Transaction<'static, Sqlite>;
+
+#[derive(Clone)]
+pub struct ScraperStorage {
+    pub(crate) manager: StorageManager,
+}
+
+impl ScraperStorage {
+    #[instrument]
+    pub async fn init<P: AsRef<Path> + Debug>(database_path: P) -> Result<Self, ScraperError> {
+        // TODO: we can inject here more stuff based on our nym-api global config
+        // struct. Maybe different pool size or timeout intervals?
+        let mut opts = sqlx::sqlite::SqliteConnectOptions::new()
+            .filename(database_path)
+            .create_if_missing(true);
+
+        // TODO: do we want auto_vacuum ?
+
+        opts.disable_statement_logging();
+
+        let connection_pool = match sqlx::SqlitePool::connect_with(opts).await {
+            Ok(db) => db,
+            Err(err) => {
+                error!("Failed to connect to SQLx database: {err}");
+                return Err(err.into());
+            }
+        };
+
+        if let Err(err) = sqlx::migrate!("./sql_migrations")
+            .run(&connection_pool)
+            .await
+        {
+            error!("Failed to initialize SQLx database: {err}");
+            return Err(err.into());
+        }
+
+        info!("Database migration finished!");
+
+        let manager = StorageManager { connection_pool };
+        manager.set_initial_metadata().await?;
+
+        let storage = ScraperStorage { manager };
+
+        Ok(storage)
+    }
+
+    #[instrument(skip_all)]
+    pub async fn begin_processing_tx(&self) -> Result<StorageTransaction, ScraperError> {
+        debug!("starting storage tx");
+        self.manager
+            .connection_pool
+            .begin()
+            .await
+            .map_err(|source| ScraperError::StorageTxBeginFailure { source })
+    }
+
+    pub async fn get_first_block_height_after(
+        &self,
+        time: OffsetDateTime,
+    ) -> Result<Option<i64>, ScraperError> {
+        Ok(self.manager.get_first_block_height_after(time).await?)
+    }
+
+    pub async fn get_last_block_height_before(
+        &self,
+        time: OffsetDateTime,
+    ) -> Result<Option<i64>, ScraperError> {
+        Ok(self.manager.get_last_block_height_before(time).await?)
+    }
+
+    pub async fn get_signed_between(
+        &self,
+        consensus_address: &str,
+        start_height: i64,
+        end_height: i64,
+    ) -> Result<i32, ScraperError> {
+        Ok(self
+            .manager
+            .get_signed_between(consensus_address, start_height, end_height)
+            .await?)
+    }
+
+    pub async fn get_signed_between_times(
+        &self,
+        consensus_address: &str,
+        start_time: OffsetDateTime,
+        end_time: OffsetDateTime,
+    ) -> Result<i32, ScraperError> {
+        let Some(block_start) = self.get_first_block_height_after(start_time).await? else {
+            return Ok(0);
+        };
+        let Some(block_end) = self.get_last_block_height_before(end_time).await? else {
+            return Ok(0);
+        };
+
+        self.get_signed_between(consensus_address, block_start, block_end)
+            .await
+    }
+
+    pub async fn get_block_signers(&self, height: i64) -> Result<Vec<Validator>, ScraperError> {
+        Ok(self.manager.get_block_validators(height).await?)
+    }
+
+    pub async fn get_last_processed_height(&self) -> Result<i64, ScraperError> {
+        Ok(self.manager.get_last_processed_height().await?)
+    }
+}
+
+pub async fn persist_block(
+    block: &FullBlockInformation,
+    tx: &mut StorageTransaction,
+) -> Result<(), ScraperError> {
+    let total_gas = crate::helpers::tx_gas_sum(&block.transactions);
+
+    // SANITY CHECK: make sure the block proposer is present in the validator set
+    block.ensure_proposer()?;
+
+    // persist validators
+    persist_validators(&block.validators, tx).await?;
+
+    // persist block data
+    persist_block_data(&block.block, total_gas, tx).await?;
+
+    // persist commits
+    if let Some(commit) = &block.block.last_commit {
+        persist_commits(commit, &block.validators, tx).await?;
+    } else {
+        warn!("no commits for block {}", block.block.header.height)
+    }
+
+    // persist txs
+    persist_txs(&block.transactions, tx).await?;
+
+    // persist messages (inside the transactions)
+    persist_messages(&block.transactions, tx).await?;
+
+    update_last_processed(block.block.header.height.into(), tx).await?;
+
+    Ok(())
+}
+
+async fn persist_validators(
+    validators: &validators::Response,
+    tx: &mut StorageTransaction,
+) -> Result<(), ScraperError> {
+    debug!("persisting {} validators", validators.total);
+    for validator in &validators.validators {
+        let consensus_address = crate::helpers::validator_consensus_address(validator.address)?;
+        let consensus_pubkey = crate::helpers::validator_pubkey_to_bech32(validator.pub_key)?;
+
+        insert_validator(
+            consensus_address.to_string(),
+            consensus_pubkey.to_string(),
+            &mut *tx,
+        )
+        .await?;
+    }
+
+    Ok(())
+}
+
+async fn persist_block_data(
+    block: &Block,
+    total_gas: i64,
+    tx: &mut StorageTransaction,
+) -> Result<(), ScraperError> {
+    let proposer_address =
+        crate::helpers::validator_consensus_address(block.header.proposer_address)?.to_string();
+
+    insert_block(
+        block.header.height.into(),
+        block.header.hash().to_string(),
+        block.data.len() as u32,
+        total_gas,
+        proposer_address,
+        block.header.time.into(),
+        tx,
+    )
+    .await?;
+    Ok(())
+}
+
+async fn persist_commits(
+    commits: &Commit,
+    validators: &validators::Response,
+    tx: &mut StorageTransaction,
+) -> Result<(), ScraperError> {
+    debug!("persisting up to {} commits", commits.signatures.len());
+    let height: i64 = commits.height.into();
+
+    for commit_sig in &commits.signatures {
+        let (validator_id, timestamp, signature) = match commit_sig {
+            CommitSig::BlockIdFlagAbsent => {
+                trace!("absent signature");
+                continue;
+            }
+            CommitSig::BlockIdFlagCommit {
+                validator_address,
+                timestamp,
+                signature,
+            } => (validator_address, timestamp, signature),
+            CommitSig::BlockIdFlagNil {
+                validator_address,
+                timestamp,
+                signature,
+            } => (validator_address, timestamp, signature),
+        };
+
+        let validator = crate::helpers::validator_info(*validator_id, validators)?;
+        let validator_address = crate::helpers::validator_consensus_address(*validator_id)?;
+
+        if signature.is_none() {
+            warn!("empty signature for {validator_address} at height {height}");
+            continue;
+        }
+
+        insert_precommit(
+            validator_address.to_string(),
+            height,
+            (*timestamp).into(),
+            validator.power.into(),
+            validator.proposer_priority.value(),
+            &mut *tx,
+        )
+        .await?;
+    }
+
+    Ok(())
+}
+
+async fn persist_txs(
+    txs: &[ParsedTransactionResponse],
+    tx: &mut StorageTransaction,
+) -> Result<(), ScraperError> {
+    debug!("persisting {} txs", txs.len());
+
+    for chain_tx in txs {
+        insert_transaction(
+            chain_tx.hash.to_string(),
+            chain_tx.height.into(),
+            chain_tx.index as i64,
+            chain_tx.tx_result.code.is_ok(),
+            chain_tx.tx.body.messages.len() as i64,
+            chain_tx.tx.body.memo.clone(),
+            chain_tx.tx_result.gas_wanted,
+            chain_tx.tx_result.gas_used,
+            chain_tx.tx_result.log.clone(),
+            &mut *tx,
+        )
+        .await?;
+    }
+
+    Ok(())
+}
+
+async fn persist_messages(
+    txs: &[ParsedTransactionResponse],
+    tx: &mut StorageTransaction,
+) -> Result<(), ScraperError> {
+    debug!("persisting messages");
+
+    for chain_tx in txs {
+        for (index, msg) in chain_tx.tx.body.messages.iter().enumerate() {
+            insert_message(
+                chain_tx.hash.to_string(),
+                index as i64,
+                msg.type_url.clone(),
+                chain_tx.height.into(),
+                &mut *tx,
+            )
+            .await?
+        }
+    }
+
+    Ok(())
+}

common/nyxd-scraper/src/storage/models.rs

@@ -0,0 +1,30 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use sqlx::types::time::OffsetDateTime;
+use sqlx::FromRow;
+
+#[derive(Debug, Clone, FromRow)]
+pub struct Validator {
+    pub consensus_address: String,
+    pub consensus_pubkey: String,
+}
+
+#[derive(Debug, Clone, FromRow)]
+pub struct Block {
+    pub height: i64,
+    pub hash: String,
+    pub num_txs: u32,
+    pub total_gas: i64,
+    pub proposer_address: String,
+    pub timestamp: OffsetDateTime,
+}
+
+#[derive(Debug, Clone, FromRow)]
+pub struct CommitSignature {
+    pub height: i64,
+    pub validator_address: String,
+    pub voting_power: i64,
+    pub proposer_priority: i64,
+    pub timestamp: OffsetDateTime,
+}

common/statistics/Cargo.toml

@@ -15,6 +15,6 @@ log = { workspace = true }
 reqwest = { workspace = true, features = ["json"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1"
-sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "chrono"]}
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "chrono"]}
 thiserror = { workspace = true }
 tokio = { version = "1.24.1", features = [ "time" ] }

gateway/Cargo.toml

@@ -36,7 +36,7 @@ pretty_env_logger = "0.4"
 rand = "0.7"
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
-sqlx = { version = "0.5", features = [
+sqlx = { workspace = true, features = [
     "runtime-tokio-rustls",
     "sqlite",
     "macros",
@@ -89,7 +89,7 @@ hyper = "0.14.27"
 
 [build-dependencies]
 tokio = { version = "1.24.1", features = ["rt-multi-thread", "macros"] }
-sqlx = { version = "0.5", features = [
+sqlx = { workspace = true, features = [
     "runtime-tokio-rustls",
     "sqlite",
     "macros",

gateway/src/node/storage/inboxes.rs

@@ -75,7 +75,11 @@ impl InboxManager {
             sqlx::query_as!(
                 StoredMessage,
                 r#"
-                    SELECT * FROM message_store 
+                    SELECT 
+                        id as "id!",
+                        client_address_bs58 as "client_address_bs58!",
+                        content as "content!" 
+                    FROM message_store 
                     WHERE client_address_bs58 = ? AND id > ?
                     ORDER BY id ASC
                     LIMIT ?;
@@ -90,7 +94,11 @@ impl InboxManager {
             sqlx::query_as!(
                 StoredMessage,
                 r#"
-                    SELECT * FROM message_store 
+                   SELECT 
+                        id as "id!",
+                        client_address_bs58 as "client_address_bs58!",
+                        content as "content!"
+                    FROM message_store
                     WHERE client_address_bs58 = ?
                     ORDER BY id ASC
                     LIMIT ?;

nym-api/Cargo.toml

@@ -53,7 +53,7 @@ ts-rs = { workspace = true, optional = true}
 anyhow = { workspace = true }
 getset = "0.1.1"
 
-sqlx = { version = "0.6.2", features = [
+sqlx = { workspace = true, features = [
     "runtime-tokio-rustls",
     "sqlite",
     "macros",
@@ -115,7 +115,7 @@ generate-ts = ["ts-rs"]
 
 [build-dependencies]
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
-sqlx = { version = "0.6.2", features = [
+sqlx = { workspace = true, features = [
     "runtime-tokio-rustls",
     "sqlite",
     "macros",

nym-wallet/Cargo.lock

@@ -4856,9 +4856,9 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.104"
+version = "1.0.108"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "076066c5f1078eac5b722a31827a8832fe108bed65dfa75e233c89f8206e976c"
+checksum = "3d1c7e3eac408d115102c4c24ad393e0821bb3a5df4d506a80f85f7a742a526b"
 dependencies = [
  "itoa 1.0.9",
  "ryu",

service-providers/network-requester/Cargo.toml

@@ -32,7 +32,7 @@ regex = "1.8.4"
 reqwest = { workspace = true, features = ["json"] }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
-sqlx = { version = "0.6.1", features = ["runtime-tokio-rustls", "chrono"]}
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "chrono"]}
 tap = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = [ "net", "rt-multi-thread", "macros" ] }

service-providers/network-statistics/Cargo.toml

@@ -12,7 +12,7 @@ log = { workspace = true }
 pretty_env_logger = "0.4"
 rocket = { version = "0.5.0-rc.2", features = ["json"] }
 serde = { workspace = true, features = ["derive"] }
-sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate", "chrono"]}
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate", "chrono"]}
 thiserror = { workspace = true }
 tokio = { version = "1.4", features = [ "net", "rt-multi-thread", "macros", "time" ] }
 nym-bin-common = { path = "../../common/bin-common"}
@@ -20,5 +20,5 @@ nym-statistics-common = { path = "../../common/statistics" }
 nym-task = { path = "../../common/task" }
 
 [build-dependencies]
-sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }
 tokio = { version = "1.4", features = ["rt-multi-thread", "macros"] }