add nyxd-scraper with psql support

common/nyxd-scraper-psql/Cargo.toml

@@ -0,0 +1,43 @@
+[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
+base64.workspace = true
+const_format = { workspace = true }
+cosmrs.workspace = true
+chrono = {workspace = true}
+eyre = { workspace = true }
+futures.workspace = true
+humantime = { workspace = true }
+sha2 = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+serde_json = {workspace = true}
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "postgres", "macros", "migrate", "time", "json"] }
+tendermint.workspace = true
+tendermint-rpc = { workspace = true, features = ["websocket-client", "http-client"] }
+thiserror.workspace = true
+time = { workspace = true }
+tokio = { workspace = true, features = ["full"] }
+tokio-stream = { workspace = true }
+tokio-util = { workspace = true, 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", "postgres", "macros", "migrate"] }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }

common/nyxd-scraper-psql/README.md

@@ -0,0 +1,23 @@
+# Nyxd Scraper
+
+## Pruning
+
+Similarly to cosmos-sdk, we incorporate pruning into our (scraped) chain data. We attempt to follow their strategies as
+closely as possible for convenience's sake. Therefore, the following are available:
+
+### Strategies
+
+The strategies are configured in `config.toml`, with the format `pruning = "<strategy>"` where the options are:
+
+* `default`: only the last 362,880 states(approximately 3.5 weeks worth of state) are kept; pruning at 10 block
+  intervals
+* `nothing`: all historic states will be saved, nothing will be deleted (i.e. archiving node)
+* `everything`: 2 latest states will be kept; pruning at 10 block intervals.
+* `custom`: allow pruning options to be manually specified through `pruning.keep_recent`, and `pruning.interval`
+
+### Custom Pruning
+
+These are applied if and only if the pruning strategy is `custom`:
+
+* `pruning.keep_recent`: N means to keep all of the last N blocks
+* `pruning.interval`: N means to delete old block data from disk every Nth block.

common/nyxd-scraper-psql/build.rs

@@ -0,0 +1,41 @@
+#[tokio::main]
+async fn main() {
+    use sqlx::{Connection, Executor, PgConnection};
+
+    const POSTGRES_USER: &str = "nym";
+    const POSTGRES_PASSWORD: &str = "password123";
+    const POSTGRES_DB: &str = "nyxd_scraper";
+
+    let admin_url = format!(
+        "postgres://{}:{}@localhost:5432/postgres",
+        POSTGRES_USER, POSTGRES_PASSWORD
+    );
+    // Connect to postgres to create test database
+    let database_url =
+        format!("postgres://{POSTGRES_USER}:{POSTGRES_PASSWORD}@localhost:5432/{POSTGRES_DB}");
+
+    let mut conn = PgConnection::connect(&admin_url)
+        .await
+        .expect("Failed to connect to Postgres");
+
+    conn.execute(format!(r#"DROP DATABASE IF EXISTS {}"#, POSTGRES_DB).as_str())
+        .await
+        .expect("Failed to drop test database");
+
+    conn.execute(format!(r#"CREATE DATABASE {}"#, POSTGRES_DB).as_str())
+        .await
+        .expect("Failed to create test database");
+
+    let mut test_conn = PgConnection::connect(&database_url)
+        .await
+        .expect("Failed to connect to test database");
+
+    // Run migrations
+    sqlx::migrate!("./sql_migrations")
+        .run(&mut test_conn)
+        .await
+        .expect("Failed to perform SQLx migrations");
+
+    // Set the database URL as an environment variable
+    println!("cargo:rustc-env=DATABASE_URL={}", database_url);
+}

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

@@ -0,0 +1,8 @@
+/*
+ * 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 BIGINT NOT NULL
+);

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

@@ -0,0 +1,73 @@
+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 TIMESTAMPTZ 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 TIMESTAMPTZ 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);
+
+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 '[]',
+    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
+);
+
+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-psql/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-psql/src/block_processor/mod.rs

@@ -0,0 +1,426 @@
+// 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 crate::PruningOptions;
+use futures::StreamExt;
+use std::cmp::max;
+use std::collections::{BTreeMap, HashSet, VecDeque};
+use std::ops::{Add, Range};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::mpsc::{Sender, UnboundedReceiver};
+use tokio::sync::Notify;
+use tokio::time::{interval_at, Instant};
+use tokio_stream::wrappers::UnboundedReceiverStream;
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, error, info, instrument, trace, warn};
+
+mod helpers;
+pub(crate) mod pruning;
+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 {
+    pruning_options: PruningOptions,
+    cancel: CancellationToken,
+    synced: Arc<Notify>,
+    last_processed_height: u32,
+    last_pruned_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(
+        pruning_options: PruningOptions,
+        cancel: CancellationToken,
+        synced: Arc<Notify>,
+        incoming: UnboundedReceiver<BlockToProcess>,
+        block_requester: Sender<BlockRequest>,
+        storage: ScraperStorage,
+        rpc_client: RpcClient,
+    ) -> Result<Self, ScraperError> {
+        let last_processed = storage.get_last_processed_height().await?;
+        let last_processed_height = last_processed.try_into().unwrap_or_default();
+
+        let last_pruned = storage.get_pruned_height().await?;
+        let last_pruned_height = last_pruned.try_into().unwrap_or_default();
+
+        Ok(BlockProcessor {
+            pruning_options,
+            cancel,
+            synced,
+            last_processed_height,
+            last_pruned_height,
+            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![],
+        })
+    }
+
+    pub fn with_pruning(mut self, pruning_options: PruningOptions) -> Self {
+        self.pruning_options = pruning_options;
+        self
+    }
+
+    pub(super) 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?
+                }
+            }
+        }
+
+        let commit_start = Instant::now();
+        tx.commit()
+            .await
+            .map_err(|source| ScraperError::StorageTxCommitFailure { source })?;
+        crate::storage::log_db_operation_time("committing processing tx", commit_start);
+
+        self.last_processed_height = full_info.block.header.height.value() as u32;
+        self.last_processed_at = Instant::now();
+        if let Err(err) = self.maybe_prune_storage().await {
+            error!("failed to prune the storage: {err}");
+        }
+
+        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;
+    }
+
+    pub(super) fn last_process_height(&self) -> u32 {
+        self.last_processed_height
+    }
+
+    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(())
+    }
+
+    #[instrument(skip(self))]
+    async fn prune_storage(&mut self) -> Result<(), ScraperError> {
+        let keep_recent = self.pruning_options.strategy_keep_recent();
+        let last_to_keep = self.last_processed_height - keep_recent;
+
+        info!(
+            keep_recent,
+            oldest_to_keep = last_to_keep,
+            "pruning the storage"
+        );
+
+        let lowest: u32 = self
+            .storage
+            .lowest_block_height()
+            .await?
+            .unwrap_or_default()
+            .try_into()
+            .unwrap_or_default();
+
+        let to_prune = last_to_keep.saturating_sub(lowest);
+        match to_prune {
+            v if v > 1000 => warn!("approximately {v} blocks worth of data will be pruned"),
+            v if v > 100 => info!("approximately {v} blocks worth of data will be pruned"),
+            0 => trace!("no blocks to prune"),
+            v => debug!("approximately {v} blocks worth of data will be pruned"),
+        }
+
+        if to_prune == 0 {
+            self.last_pruned_height = self.last_processed_height;
+            return Ok(());
+        }
+
+        self.storage
+            .prune_storage(last_to_keep, self.last_processed_height)
+            .await?;
+
+        self.last_pruned_height = self.last_processed_height;
+        Ok(())
+    }
+
+    async fn maybe_prune_storage(&mut self) -> Result<(), ScraperError> {
+        debug!("checking for storage pruning");
+
+        if self.pruning_options.strategy.is_nothing() {
+            trace!("the current pruning strategy is 'nothing'");
+            return Ok(());
+        }
+
+        let interval = self.pruning_options.strategy_interval();
+        if self.last_pruned_height + interval <= self.last_processed_height {
+            self.prune_storage().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;
+
+        if self.pending_sync.is_empty() {
+            self.synced.notify_one();
+        }
+    }
+
+    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());
+
+        self.maybe_prune_storage().await?;
+
+        let latest_block = self.rpc_client.current_block_height().await? as u32;
+
+        if latest_block > self.last_processed_height && self.last_processed_height != 0 {
+            // in case we were offline for a while,
+            // make sure we don't request blocks we'd have to prune anyway
+            let keep_recent = self.pruning_options.strategy_keep_recent();
+            let last_to_keep = latest_block - keep_recent;
+            self.last_processed_height = max(self.last_processed_height, last_to_keep);
+
+            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-psql/src/block_processor/pruning.rs

@@ -0,0 +1,122 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::error::ScraperError;
+use serde::{Deserialize, Serialize};
+
+pub const DEFAULT_PRUNING_KEEP_RECENT: u32 = 362880;
+pub const DEFAULT_PRUNING_INTERVAL: u32 = 10;
+pub const EVERYTHING_PRUNING_KEEP_RECENT: u32 = 2;
+pub const EVERYTHING_PRUNING_INTERVAL: u32 = 10;
+
+/// We follow cosmos-sdk pruning strategies for convenience’s sake.
+#[derive(Debug, Default, Clone, Copy, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum PruningStrategy {
+    /// 'Default' strategy defines a pruning strategy where the last 362880 heights are
+    /// kept where to-be pruned heights are pruned at every 10th height.
+    /// The last 362880 heights are kept(approximately 3.5 weeks worth of state) assuming the typical
+    /// block time is 6s. If these values do not match the applications' requirements, use the "custom" option.
+    #[default]
+    Default,
+
+    /// 'Everything' strategy defines a pruning strategy where all committed heights are
+    /// deleted, storing only the current height and last 2 states. To-be pruned heights are
+    /// pruned at every 10th height.
+    Everything,
+
+    /// 'Nothing' strategy defines a pruning strategy where all heights are kept on disk.
+    Nothing,
+
+    /// 'Custom' strategy defines a pruning strategy where the user specifies the pruning.
+    Custom,
+}
+
+impl PruningStrategy {
+    pub fn is_custom(&self) -> bool {
+        matches!(self, PruningStrategy::Custom)
+    }
+
+    pub fn is_nothing(&self) -> bool {
+        matches!(self, PruningStrategy::Nothing)
+    }
+
+    pub fn is_everything(&self) -> bool {
+        matches!(self, PruningStrategy::Everything)
+    }
+}
+
+#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
+pub struct PruningOptions {
+    /// keep_recent defines how many recent heights to keep on disk.
+    pub keep_recent: u32,
+
+    /// interval defines the frequency of removing the pruned heights from the disk.
+    pub interval: u32,
+
+    /// strategy defines the currently used kind of [PruningStrategy].
+    pub strategy: PruningStrategy,
+}
+
+impl PruningOptions {
+    pub fn validate(&self) -> Result<(), ScraperError> {
+        // if strategy is not set to custom, other options are meaningless since they won't be applied
+        if !self.strategy.is_custom() {
+            return Ok(());
+        }
+
+        if self.interval == 0 {
+            return Err(ScraperError::ZeroPruningInterval);
+        }
+
+        if self.interval < EVERYTHING_PRUNING_INTERVAL {
+            return Err(ScraperError::TooSmallPruningInterval {
+                interval: self.interval,
+            });
+        }
+
+        if self.keep_recent < EVERYTHING_PRUNING_KEEP_RECENT {
+            return Err(ScraperError::TooSmallKeepRecent {
+                keep_recent: self.keep_recent,
+            });
+        }
+
+        Ok(())
+    }
+
+    pub fn nothing() -> Self {
+        PruningOptions {
+            keep_recent: 0,
+            interval: 0,
+            strategy: PruningStrategy::Nothing,
+        }
+    }
+
+    pub fn strategy_interval(&self) -> u32 {
+        match self.strategy {
+            PruningStrategy::Default => DEFAULT_PRUNING_INTERVAL,
+            PruningStrategy::Everything => EVERYTHING_PRUNING_INTERVAL,
+            PruningStrategy::Nothing => 0,
+            PruningStrategy::Custom => self.interval,
+        }
+    }
+
+    pub fn strategy_keep_recent(&self) -> u32 {
+        match self.strategy {
+            PruningStrategy::Default => DEFAULT_PRUNING_KEEP_RECENT,
+            PruningStrategy::Everything => EVERYTHING_PRUNING_KEEP_RECENT,
+            PruningStrategy::Nothing => 0,
+            PruningStrategy::Custom => self.keep_recent,
+        }
+    }
+}
+
+impl Default for PruningOptions {
+    fn default() -> Self {
+        PruningOptions {
+            keep_recent: DEFAULT_PRUNING_KEEP_RECENT,
+            interval: DEFAULT_PRUNING_INTERVAL,
+            strategy: Default::default(),
+        }
+    }
+}

common/nyxd-scraper-psql/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-psql/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-psql/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-psql/src/error.rs

@@ -0,0 +1,143 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::block_processor::pruning::{
+    EVERYTHING_PRUNING_INTERVAL, EVERYTHING_PRUNING_KEEP_RECENT,
+};
+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("the block scraper is 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")]
+    MaximumWebSocketFailures,
+
+    #[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 },
+
+    #[error("pruning.interval must not be set to 0. If you want to disable pruning, select pruning.strategy = \"nothing\"")]
+    ZeroPruningInterval,
+
+    #[error("pruning.interval must not be smaller than {}. got: {interval}. for most aggressive pruning, select pruning.strategy = \"everything\"", EVERYTHING_PRUNING_INTERVAL)]
+    TooSmallPruningInterval { interval: u32 },
+
+    #[error("pruning.keep_recent must not be smaller than {}. got: {keep_recent}. for most aggressive pruning, select pruning.strategy = \"everything\"", EVERYTHING_PRUNING_KEEP_RECENT)]
+    TooSmallKeepRecent { keep_recent: u32 },
+}
+
+impl<T> From<SendError<T>> for ScraperError {
+    fn from(_: SendError<T>) -> Self {
+        ScraperError::ClosedChannelError
+    }
+}

common/nyxd-scraper-psql/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-psql/src/lib.rs

@@ -0,0 +1,20 @@
+// 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 block_processor::pruning::{PruningOptions, PruningStrategy};
+pub use modules::{BlockModule, MsgModule, TxModule};
+pub use scraper::{Config, NyxdScraper};
+pub use storage::models;

common/nyxd-scraper-psql/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-psql/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-psql/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-psql/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-psql/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-psql/src/scraper/mod.rs

@@ -0,0 +1,339 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::block_processor::types::BlockToProcess;
+use crate::block_processor::BlockProcessor;
+use crate::block_requester::{BlockRequest, BlockRequester};
+use crate::error::ScraperError;
+use crate::modules::{BlockModule, MsgModule, TxModule};
+use crate::rpc_client::RpcClient;
+use crate::scraper::subscriber::ChainSubscriber;
+use crate::storage::ScraperStorage;
+use crate::PruningOptions;
+use futures::future::join_all;
+use std::path::PathBuf;
+use std::sync::Arc;
+use tokio::sync::mpsc::{
+    channel, unbounded_channel, Receiver, Sender, UnboundedReceiver, UnboundedSender,
+};
+use tokio::sync::Notify;
+use tokio_util::sync::CancellationToken;
+use tokio_util::task::TaskTracker;
+use tracing::{error, 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 pruning_options: PruningOptions,
+}
+
+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.config.pruning_options,
+            scraper.cancel_token.clone(),
+            scraper.startup_sync.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 chain_subscriber = ChainSubscriber::new(
+            &scraper.config.websocket_url,
+            scraper.cancel_token.clone(),
+            scraper.task_tracker.clone(),
+            processing_tx,
+        )
+        .await?;
+
+        scraper.start_tasks(block_requester, block_processor, chain_subscriber);
+
+        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,
+    startup_sync: Arc<Notify>,
+    pub storage: ScraperStorage,
+    rpc_client: RpcClient,
+}
+
+impl NyxdScraper {
+    pub fn builder(config: Config) -> NyxdScraperBuilder {
+        NyxdScraperBuilder::new(config)
+    }
+
+    pub async fn new(config: Config) -> Result<Self, ScraperError> {
+        config.pruning_options.validate()?;
+        let storage =
+            ScraperStorage::init(&config.database_path.to_str().unwrap_or_default()).await?;
+        let rpc_client = RpcClient::new(&config.rpc_url)?;
+
+        Ok(NyxdScraper {
+            config,
+            task_tracker: TaskTracker::new(),
+            cancel_token: CancellationToken::new(),
+            startup_sync: Arc::new(Default::default()),
+            storage,
+            rpc_client,
+        })
+    }
+
+    fn start_tasks(
+        &self,
+        mut block_requester: BlockRequester,
+        mut block_processor: BlockProcessor,
+        mut chain_subscriber: ChainSubscriber,
+    ) {
+        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.close();
+    }
+
+    pub async fn process_single_block(&self, height: u32) -> Result<(), ScraperError> {
+        info!(height = height, "attempting to process a single block");
+        if !self.task_tracker.is_empty() {
+            return Err(ScraperError::ScraperAlreadyRunning);
+        }
+
+        let (_, processing_rx) = unbounded_channel();
+        let (req_tx, _) = channel(5);
+
+        let mut block_processor = self
+            .new_block_processor(req_tx.clone(), processing_rx)
+            .await?
+            .with_pruning(PruningOptions::nothing());
+
+        let block = self.rpc_client.get_basic_block_details(height).await?;
+
+        block_processor.process_block(block.into()).await
+    }
+
+    pub async fn process_block_range(
+        &self,
+        starting_height: Option<u32>,
+        end_height: Option<u32>,
+    ) -> Result<(), ScraperError> {
+        if !self.task_tracker.is_empty() {
+            return Err(ScraperError::ScraperAlreadyRunning);
+        }
+
+        let (_, processing_rx) = unbounded_channel();
+        let (req_tx, _) = channel(5);
+
+        let mut block_processor = self
+            .new_block_processor(req_tx.clone(), processing_rx)
+            .await?
+            .with_pruning(PruningOptions::nothing());
+
+        let current_height = self.rpc_client.current_block_height().await? as u32;
+        let last_processed = block_processor.last_process_height();
+
+        let starting_height = match starting_height {
+            // always attempt to use whatever the user has provided
+            Some(explicit) => explicit,
+            None => {
+                // otherwise, attempt to resume where we last stopped
+                // and if we haven't processed anything, start from the current height
+                if last_processed != 0 {
+                    last_processed
+                } else {
+                    current_height
+                }
+            }
+        };
+
+        let end_height = match end_height {
+            // always attempt to use whatever the user has provided
+            Some(explicit) => explicit,
+            None => {
+                // otherwise, attempt to either go from the start height to the height right
+                // before the final processed block held in the storage (in case there are gaps)
+                // or finally, just go to the current block height
+                if last_processed > starting_height {
+                    last_processed - 1
+                } else {
+                    current_height
+                }
+            }
+        };
+
+        info!(
+            starting_height = starting_height,
+            end_height = end_height,
+            "attempting to process block range"
+        );
+
+        let range = (starting_height..=end_height).collect::<Vec<_>>();
+
+        // the most likely bottleneck here are going to be the chain queries,
+        // so batch multiple requests
+        for batch in range.chunks(4) {
+            let batch_result = join_all(
+                batch
+                    .iter()
+                    .map(|height| self.rpc_client.get_basic_block_details(*height)),
+            )
+            .await;
+            for result in batch_result {
+                match result {
+                    Ok(block) => block_processor.process_block(block.into()).await?,
+                    Err(err) => {
+                        error!("failed to retrieve the block: {err}. stopping...");
+                        return Err(err);
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    fn new_block_requester(
+        &self,
+        req_rx: Receiver<BlockRequest>,
+        processing_tx: UnboundedSender<BlockToProcess>,
+    ) -> BlockRequester {
+        BlockRequester::new(
+            self.cancel_token.clone(),
+            self.rpc_client.clone(),
+            req_rx,
+            processing_tx.clone(),
+        )
+    }
+
+    async fn new_block_processor(
+        &self,
+        req_tx: Sender<BlockRequest>,
+        processing_rx: UnboundedReceiver<BlockToProcess>,
+    ) -> Result<BlockProcessor, ScraperError> {
+        BlockProcessor::new(
+            self.config.pruning_options,
+            self.cancel_token.clone(),
+            self.startup_sync.clone(),
+            processing_rx,
+            req_tx,
+            self.storage.clone(),
+            self.rpc_client.clone(),
+        )
+        .await
+    }
+
+    async fn new_chain_subscriber(
+        &self,
+        processing_tx: UnboundedSender<BlockToProcess>,
+    ) -> Result<ChainSubscriber, ScraperError> {
+        ChainSubscriber::new(
+            &self.config.websocket_url,
+            self.cancel_token.clone(),
+            self.task_tracker.clone(),
+            processing_tx,
+        )
+        .await
+    }
+
+    pub async fn start(&self) -> Result<(), ScraperError> {
+        let (processing_tx, processing_rx) = unbounded_channel();
+        let (req_tx, req_rx) = channel(5);
+
+        // create the tasks
+        let block_requester = self.new_block_requester(req_rx, processing_tx.clone());
+        let block_processor = self.new_block_processor(req_tx, processing_rx).await?;
+        let chain_subscriber = self.new_chain_subscriber(processing_tx).await?;
+
+        // spawn them
+        self.start_tasks(block_requester, block_processor, chain_subscriber);
+
+        Ok(())
+    }
+
+    pub async fn wait_for_startup_sync(&self) {
+        info!("awaiting startup chain sync");
+        self.startup_sync.notified().await
+    }
+
+    pub async fn stop(self) {
+        info!("stopping the chain scraper");
+        assert!(self.task_tracker.is_closed());
+
+        self.cancel_token.cancel();
+        self.task_tracker.wait().await
+    }
+
+    pub fn cancel_token(&self) -> CancellationToken {
+        self.cancel_token.clone()
+    }
+
+    pub fn is_cancelled(&self) -> bool {
+        self.cancel_token.is_cancelled()
+    }
+}

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

@@ -0,0 +1,241 @@
+// 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 time::{Duration, OffsetDateTime};
+use tokio::sync::mpsc::UnboundedSender;
+use tokio_stream::StreamExt;
+use tokio_util::sync::CancellationToken;
+use tokio_util::task::TaskTracker;
+use tracing::{error, info, warn};
+use url::Url;
+
+const MAX_FAILURES: usize = 10;
+const MAX_RECONNECTION_ATTEMPTS: usize = 8;
+const SOCKET_FAILURE_RESET: Duration = Duration::minutes(15);
+
+pub struct ChainSubscriber {
+    cancel: CancellationToken,
+    task_tracker: TaskTracker,
+
+    block_sender: UnboundedSender<BlockToProcess>,
+
+    websocket_endpoint: Url,
+    websocket_client: WebSocketClient,
+    websocket_driver: Option<WebSocketClientDriver>,
+}
+
+impl ChainSubscriber {
+    pub async fn new(
+        websocket_endpoint: &Url,
+        cancel: CancellationToken,
+        task_tracker: TaskTracker,
+        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,
+            task_tracker,
+            block_sender,
+            websocket_endpoint: websocket_endpoint.clone(),
+            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(())
+    }
+
+    async fn remake_connection(&mut self) -> Result<(), ScraperError> {
+        info!(
+            "attempting to reestablish connection to {}",
+            self.websocket_endpoint
+        );
+
+        let (client, driver) = WebSocketClient::new(self.websocket_endpoint.as_str())
+            .await
+            .map_err(|source| ScraperError::WebSocketConnectionFailure {
+                url: self.websocket_endpoint.to_string(),
+                source,
+            })?;
+        self.websocket_client = client;
+        self.websocket_driver = Some(driver);
+
+        info!(
+            "managed to reestablish the websocket connection to {}",
+            self.websocket_endpoint
+        );
+        Ok(())
+    }
+
+    /// Returns whether the method exited due to the cancellation
+    async fn run_chain_subscription(&mut self) -> Result<bool, ScraperError> {
+        let Some(ws_driver) = self.websocket_driver.take() else {
+            error!("the websocket driver hasn't been created - we probably failed to establish the connection");
+            return Ok(false);
+        };
+
+        let driver_cancel = CancellationToken::new();
+        let _driver_guard = driver_cancel.clone().drop_guard();
+
+        // spawn the websocket driver task
+        let driver_handle = {
+            self.task_tracker.reopen();
+            let handle = self
+                .task_tracker
+                .spawn(run_websocket_driver(ws_driver, driver_cancel));
+            self.task_tracker.close();
+            handle
+        };
+        tokio::pin!(driver_handle);
+
+        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");
+                    // note: `_driver_guard` will get dropped here thus causing cancellation of the driver task
+                    return Ok(true)
+                }
+                _ = &mut driver_handle => {
+                    error!("our websocket driver has finished execution");
+                    return Ok(self.cancel.is_cancelled())
+                }
+                maybe_event = subs.next() => {
+                    let Some(maybe_event) = maybe_event else {
+                        warn!("stopped receiving new events");
+                        return Ok(false)
+                    };
+                    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 {
+                        return Ok(false)
+                    }
+                }
+            }
+        }
+    }
+
+    async fn websocket_backoff(&mut self, failure_count: usize) -> bool {
+        const MINIMUM_WAIT_MS: u64 = 10_000;
+        const INCREMENTAL_WAIT_MS: u64 = 30_000;
+
+        let backoff_duration_ms = MINIMUM_WAIT_MS + INCREMENTAL_WAIT_MS * failure_count as u64;
+        info!("going to wait {backoff_duration_ms} ms before re-attempting the reconnection");
+
+        tokio::select! {
+            _ = self.cancel.cancelled() => {
+                info!("received cancellation token");
+                true
+            }
+            _ = tokio::time::sleep(std::time::Duration::from_millis(backoff_duration_ms)) => false,
+        }
+    }
+
+    pub(crate) async fn run(&mut self) -> Result<(), ScraperError> {
+        let _drop_guard = self.cancel.clone().drop_guard();
+        let mut socket_failures = 0;
+        let mut last_failure = OffsetDateTime::now_utc();
+
+        loop {
+            if self.cancel.is_cancelled() {
+                return Ok(());
+            }
+
+            match self.run_chain_subscription().await {
+                Ok(cancelled) => {
+                    if cancelled {
+                        // we're in the middle of a shutdown
+                        return Ok(());
+                    }
+                    socket_failures += 1;
+                }
+                Err(err) => {
+                    error!("failed to create chain subscription: {err}");
+                    socket_failures += 1;
+                }
+            }
+
+            warn!("current socket failure count: {socket_failures}. the last failure was at {last_failure}");
+
+            let now = OffsetDateTime::now_utc();
+
+            // if it's been a while since the last failure, reset the count
+            if now - last_failure > SOCKET_FAILURE_RESET {
+                warn!("resetting the failure count to 1");
+                socket_failures = 1;
+            }
+            last_failure = now;
+
+            if socket_failures >= MAX_RECONNECTION_ATTEMPTS {
+                error!("reached the maximum allowed failure count");
+                return Err(ScraperError::MaximumWebSocketFailures);
+            }
+
+            // BACKOFF
+            let cancelled = self.websocket_backoff(socket_failures).await;
+            if cancelled {
+                return Ok(());
+            }
+
+            if let Err(err) = self.remake_connection().await {
+                error!("failed to re-establish the websocket connection: {err}");
+            }
+        }
+    }
+}
+
+pub async fn run_websocket_driver(driver: WebSocketClientDriver, driver_cancel: CancellationToken) {
+    info!("starting websocket driver");
+    tokio::select! {
+        _ = driver_cancel.cancelled() => {
+            info!("received cancellation token")
+        }
+        res = driver.run() => {
+            match res {
+                Ok(_) => info!("our websocket driver has finished execution"),
+                Err(err) => {
+                    error!("our websocket driver has errored out: {err}");
+                }
+            }
+        }
+    }
+}

common/nyxd-scraper-psql/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-psql/src/storage/manager.rs

@@ -0,0 +1,600 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use std::vec;
+
+use crate::storage::log_db_operation_time;
+use crate::storage::models::{CommitSignature, Validator};
+use base64;
+use serde_json::Value as JsonValue;
+use sqlx::types::time::OffsetDateTime;
+use sqlx::{Executor, Postgres};
+use tokio::time::Instant;
+use tracing::{instrument, trace};
+
+#[derive(Clone)]
+pub(crate) struct StorageManager {
+    pub(crate) connection_pool: sqlx::PgPool,
+}
+
+impl StorageManager {
+    pub(crate) async fn set_initial_metadata(&self) -> Result<(), sqlx::Error> {
+        if sqlx::query::<Postgres>("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_lowest_block(&self) -> Result<Option<i64>, sqlx::Error> {
+        trace!("get_lowest_block");
+        let start = Instant::now();
+
+        let maybe_record = sqlx::query!(
+            r#"
+                SELECT height
+                FROM block
+                ORDER BY height ASC
+                LIMIT 1
+            "#,
+        )
+        .fetch_optional(&self.connection_pool)
+        .await?;
+        log_db_operation_time("get_lowest_block", start);
+
+        if let Some(row) = maybe_record {
+            Ok(Some(row.height))
+        } else {
+            Ok(None)
+        }
+    }
+
+    pub(crate) async fn get_first_block_height_after(
+        &self,
+        time: OffsetDateTime,
+    ) -> Result<Option<i64>, sqlx::Error> {
+        trace!("get_first_block_height_after");
+        let start = Instant::now();
+
+        let maybe_record = sqlx::query!(
+            r#"
+                SELECT height
+                FROM block
+                WHERE timestamp > $1
+                ORDER BY timestamp
+                LIMIT 1
+            "#,
+            time
+        )
+        .fetch_optional(&self.connection_pool)
+        .await?;
+        log_db_operation_time("get_first_block_height_after", start);
+
+        if let Some(row) = maybe_record {
+            Ok(Some(row.height))
+        } else {
+            Ok(None)
+        }
+    }
+
+    pub(crate) async fn get_last_block_height_before(
+        &self,
+        time: OffsetDateTime,
+    ) -> Result<Option<i64>, sqlx::Error> {
+        trace!("get_last_block_height_before");
+        let start = Instant::now();
+
+        let maybe_record = sqlx::query!(
+            r#"
+                SELECT height
+                FROM block
+                WHERE timestamp < $1
+                ORDER BY timestamp DESC
+                LIMIT 1
+            "#,
+            time
+        )
+        .fetch_optional(&self.connection_pool)
+        .await?;
+        log_db_operation_time("get_last_block_height_before", start);
+
+        if let Some(row) = maybe_record {
+            Ok(Some(row.height))
+        } else {
+            Ok(None)
+        }
+    }
+
+    pub(crate) async fn get_signed_between(
+        &self,
+        consensus_address: &str,
+        start_height: i64,
+        end_height: i64,
+    ) -> Result<i64, sqlx::Error> {
+        trace!("get_signed_between");
+        let start = Instant::now();
+
+        let count = sqlx::query!(
+            r#"
+                SELECT COUNT(*) as count FROM pre_commit
+                WHERE 
+                    validator_address = $1
+                    AND height >= $2
+                    AND height <= $3
+            "#,
+            consensus_address,
+            start_height,
+            end_height
+        )
+        .fetch_one(&self.connection_pool)
+        .await?
+        .count;
+        log_db_operation_time("get_signed_between", start);
+
+        Ok(count.expect("Could not find the count"))
+    }
+
+    pub(crate) async fn get_precommit(
+        &self,
+        consensus_address: &str,
+        height: i64,
+    ) -> Result<Option<CommitSignature>, sqlx::Error> {
+        trace!("get_precommit");
+        let start = Instant::now();
+
+        let res = sqlx::query_as(
+            r#"
+                SELECT * FROM pre_commit
+                WHERE validator_address = $1
+                AND height = $2
+            "#,
+        )
+        .bind(consensus_address)
+        .bind(height)
+        .fetch_optional(&self.connection_pool)
+        .await?;
+        log_db_operation_time("get_precommit", start);
+
+        Ok(res)
+    }
+
+    pub(crate) async fn get_block_validators(
+        &self,
+        height: i64,
+    ) -> Result<Vec<Validator>, sqlx::Error> {
+        trace!("get_block_validators");
+        let start = Instant::now();
+
+        let res = sqlx::query_as!(
+            Validator,
+            r#"
+                SELECT * FROM validator 
+                WHERE EXISTS (
+                    SELECT 1 FROM pre_commit
+                    WHERE height = $1
+                    AND pre_commit.validator_address = validator.consensus_address
+                )
+            "#,
+            height
+        )
+        .fetch_all(&self.connection_pool)
+        .await?;
+        log_db_operation_time("get_block_validators", start);
+
+        Ok(res)
+    }
+
+    pub(crate) async fn get_validators(&self) -> Result<Vec<Validator>, sqlx::Error> {
+        trace!("get_validators");
+        let start = Instant::now();
+
+        let res = sqlx::query_as("SELECT * FROM validator")
+            .fetch_all(&self.connection_pool)
+            .await?;
+        log_db_operation_time("get_validators", start);
+
+        Ok(res)
+    }
+
+    pub(crate) async fn get_last_processed_height(&self) -> Result<i64, sqlx::Error> {
+        trace!("get_last_processed_height");
+        let start = Instant::now();
+
+        let maybe_record = sqlx::query!(
+            r#"
+                SELECT last_processed_height FROM metadata
+            "#
+        )
+        .fetch_optional(&self.connection_pool)
+        .await?;
+        log_db_operation_time("get_last_processed_height", start);
+
+        if let Some(row) = maybe_record {
+            Ok(row.last_processed_height)
+        } else {
+            Ok(-1)
+        }
+    }
+
+    pub(crate) async fn get_pruned_height(&self) -> Result<i64, sqlx::Error> {
+        trace!("get_pruned_height");
+        let start = Instant::now();
+
+        let maybe_record = sqlx::query!(
+            r#"
+                SELECT last_pruned_height FROM pruning
+            "#
+        )
+        .fetch_optional(&self.connection_pool)
+        .await?;
+
+        log_db_operation_time("get_pruned_height", start);
+
+        if let Some(row) = maybe_record {
+            Ok(row.last_pruned_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 = Postgres>,
+{
+    trace!("insert_validator");
+    let start = Instant::now();
+
+    sqlx::query!(
+        r#"
+            INSERT INTO validator (consensus_address, consensus_pubkey)
+            VALUES ($1, $2)
+            ON CONFLICT (consensus_address) DO NOTHING
+        "#,
+        consensus_address,
+        consensus_pubkey
+    )
+    .execute(executor)
+    .await?;
+    log_db_operation_time("insert_validator", start);
+
+    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 = Postgres>,
+{
+    trace!("insert_block");
+    let start = Instant::now();
+
+    sqlx::query!(
+        r#"
+            INSERT INTO block (height, hash, num_txs, total_gas, proposer_address, timestamp)
+            VALUES ($1, $2, $3, $4, $5, $6)
+            ON CONFLICT DO NOTHING
+        "#,
+        height,
+        hash,
+        num_txs as i32,
+        total_gas,
+        proposer_address,
+        timestamp
+    )
+    .execute(executor)
+    .await?;
+    log_db_operation_time("insert_block", start);
+
+    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 = Postgres>,
+{
+    trace!("insert_precommit");
+    let start = Instant::now();
+
+    sqlx::query!(
+        r#"
+            INSERT INTO pre_commit (validator_address, height, timestamp, voting_power, proposer_priority)
+            VALUES ($1, $2, $3, $4, $5)
+            ON CONFLICT (validator_address, timestamp) DO NOTHING
+        "#,
+        validator_address,
+        height,
+        timestamp,
+        voting_power,
+        proposer_priority
+    )
+    .execute(executor)
+    .await?;
+    log_db_operation_time("insert_precommit", start);
+
+    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,
+    num_messages: i64,
+    messages: Vec<cosmrs::Any>,
+    memo: String,
+    signatures: Vec<String>,
+    signer_infos: Vec<SignerInfo>,
+    fee: Fee,
+    gas_wanted: i64,
+    gas_used: i64,
+    raw_log: String,
+    executor: E,
+) -> Result<(), sqlx::Error>
+where
+    E: Executor<'a, Database = Postgres>,
+{
+    trace!("insert_transaction");
+    let start = Instant::now();
+
+    sqlx::query!(
+        r#"
+            INSERT INTO transaction (hash, height, "index", success, num_messages, messages, memo, signatures, signer_infos, fee, gas_wanted, gas_used, raw_log)
+            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
+            ON CONFLICT (hash) DO UPDATE
+                SET height = excluded.height,
+                    "index" = excluded."index",
+                    success = excluded.success,
+                    num_messages = excluded.num_messages,
+                    messages = excluded.messages,
+                    memo = excluded.memo,
+                    signatures = excluded.signatures,
+                    signer_infos = excluded.signer_infos,
+                    fee = excluded.fee,
+                    gas_wanted = excluded.gas_wanted,
+                    gas_used = excluded.gas_used,
+                    raw_log = excluded.raw_log
+        "#,
+        hash,
+        height,
+        index as i32,
+        success,
+        num_messages as i32,
+        serde_json::json!(messages.iter().map(|msg| {
+            serde_json::json!({
+                "type_url": msg.type_url,
+                "value": base64::encode(&msg.value)
+            })
+        }).collect::<Vec<_>>()),
+        memo,
+        &signatures,
+        serde_json::json!(signer_infos.iter().map(|info| {
+            serde_json::json!({
+                "public_key": {
+                    "type_url": info.public_key.type_url,
+                    "value": base64::encode(&info.public_key.value)
+                },
+                "mode_info": info.mode_info,
+                "sequence": info.sequence
+            })
+        }).collect::<Vec<_>>()),
+        serde_json::json!({
+            "amount": fee.amount,
+            "gas_limit": fee.gas_limit,
+            "payer": fee.payer,
+            "granter": fee.granter
+        }),
+        gas_wanted,
+        gas_used,
+        raw_log
+    )
+    .execute(executor)
+    .await?;
+    log_db_operation_time("insert_transaction", start);
+
+    Ok(())
+}
+
+#[instrument(skip(executor))]
+pub(crate) async fn insert_message<'a, E>(
+    transaction_hash: String,
+    index: i64,
+    typ: String,
+    value: cosmrs::Any,
+    involved_accounts: Vec<String>,
+    height: i64,
+    executor: E,
+) -> Result<(), sqlx::Error>
+where
+    E: Executor<'a, Database = Postgres>,
+{
+    trace!("insert_message");
+    let start = Instant::now();
+
+    sqlx::query!(
+        r#"
+            INSERT INTO message (transaction_hash, "index", type, value, involved_accounts_addresses, height)
+            VALUES ($1, $2, $3, $4, $5, $6)
+            ON CONFLICT (transaction_hash, "index") DO UPDATE
+                SET height = excluded.height,
+                type = excluded.type,
+                value = excluded.value,
+                involved_accounts_addresses = excluded.involved_accounts_addresses
+        "#,
+            transaction_hash,
+            index,
+            typ,
+            serde_json::json!({
+                "type_url": value.type_url,
+                "value": base64::encode(&value.value)
+            }),
+            &involved_accounts,
+            height
+    )
+    .execute(executor)
+    .await?;
+    log_db_operation_time("insert_message", start);
+
+    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 = Postgres>,
+{
+    trace!("update_last_processed");
+    let start = Instant::now();
+
+    sqlx::query!(
+        "UPDATE metadata SET last_processed_height = GREATEST(last_processed_height, $1)",
+        height
+    )
+    .execute(executor)
+    .await?;
+    log_db_operation_time("update_last_processed", start);
+
+    Ok(())
+}
+
+#[instrument(skip(executor))]
+pub(crate) async fn update_last_pruned<'a, E>(height: i64, executor: E) -> Result<(), sqlx::Error>
+where
+    E: Executor<'a, Database = Postgres>,
+{
+    trace!("update_last_pruned");
+    let start = Instant::now();
+
+    sqlx::query!("UPDATE pruning SET last_pruned_height = $1", height)
+        .execute(executor)
+        .await?;
+    log_db_operation_time("update_last_pruned", start);
+
+    Ok(())
+}
+
+pub(crate) async fn prune_blocks<'a, E>(oldest_to_keep: i64, executor: E) -> Result<(), sqlx::Error>
+where
+    E: Executor<'a, Database = Postgres>,
+{
+    trace!("prune_blocks");
+    let start = Instant::now();
+
+    sqlx::query!("DELETE FROM block WHERE height < $1", oldest_to_keep)
+        .execute(executor)
+        .await?;
+    log_db_operation_time("prune_blocks", start);
+
+    Ok(())
+}
+
+pub(crate) async fn prune_pre_commits<'a, E>(
+    oldest_to_keep: i64,
+    executor: E,
+) -> Result<(), sqlx::Error>
+where
+    E: Executor<'a, Database = Postgres>,
+{
+    trace!("prune_pre_commits");
+    let start = Instant::now();
+
+    sqlx::query!("DELETE FROM pre_commit WHERE height < $1", oldest_to_keep)
+        .execute(executor)
+        .await?;
+    log_db_operation_time("prune_pre_commits", start);
+
+    Ok(())
+}
+
+pub(crate) async fn prune_transactions<'a, E>(
+    oldest_to_keep: i64,
+    executor: E,
+) -> Result<(), sqlx::Error>
+where
+    E: Executor<'a, Database = Postgres>,
+{
+    trace!("prune_transactions");
+    let start = Instant::now();
+
+    sqlx::query!(
+        "DELETE FROM transaction WHERE height < $1",
+        oldest_to_keep
+    )
+    .execute(executor)
+    .await?;
+    log_db_operation_time("prune_transactions", start);
+
+    Ok(())
+}
+
+pub(crate) async fn prune_messages<'a, E>(
+    oldest_to_keep: i64,
+    executor: E,
+) -> Result<(), sqlx::Error>
+where
+    E: Executor<'a, Database = Postgres>,
+{
+    trace!("prune_messages");
+    let start = Instant::now();
+
+    sqlx::query!("DELETE FROM message WHERE height < $1", oldest_to_keep)
+        .execute(executor)
+        .await?;
+    log_db_operation_time("prune_messages", start);
+
+    Ok(())
+}
+
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct Coin {
+    pub denom: String,
+    pub amount: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct SignerInfo {
+    pub public_key: cosmrs::Any,
+    pub mode_info: String,
+    pub sequence: i64,
+}
+
+#[derive(Debug, Clone)]
+pub struct Fee {
+    pub amount: Vec<Coin>,
+    pub gas_limit: i64,
+    pub payer: String,
+    pub granter: String,
+}

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

@@ -0,0 +1,420 @@
+// 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, prune_blocks, prune_messages, prune_pre_commits, prune_transactions,
+    update_last_processed, update_last_pruned, StorageManager, Coin, Fee, SignerInfo,
+};
+use crate::storage::models::{CommitSignature, Validator};
+use sqlx::types::time::OffsetDateTime;
+use sqlx::{ConnectOptions, Postgres, Transaction};
+use std::str::FromStr;
+use tendermint::block::{Commit, CommitSig};
+use tendermint::Block;
+use tendermint_rpc::endpoint::validators;
+use tokio::time::Instant;
+use tracing::{debug, error, info, instrument, trace, warn};
+
+mod helpers;
+mod manager;
+pub mod models;
+
+pub type StorageTransaction = Transaction<'static, Postgres>;
+
+#[derive(Clone)]
+pub struct ScraperStorage {
+    pub(crate) manager: StorageManager,
+}
+
+pub(crate) fn log_db_operation_time(op_name: &str, start_time: Instant) {
+    let elapsed = start_time.elapsed();
+    let formatted = humantime::format_duration(elapsed);
+
+    match elapsed.as_millis() {
+        v if v > 10000 => error!("{op_name} took {formatted} to execute"),
+        v if v > 1000 => warn!("{op_name} took {formatted} to execute"),
+        v if v > 100 => info!("{op_name} took {formatted} to execute"),
+        v if v > 10 => debug!("{op_name} took {formatted} to execute"),
+        _ => trace!("{op_name} took {formatted} to execute"),
+    }
+}
+
+impl ScraperStorage {
+    #[instrument]
+    pub async fn init(database_url: &str) -> Result<Self, ScraperError> {
+        let mut opts = sqlx::postgres::PgConnectOptions::from_str(database_url)
+            .map_err(|err| ScraperError::InternalDatabaseError(err))?;
+
+        // TODO: do we want auto_vacuum ?
+
+        opts.disable_statement_logging();
+
+        let connection_pool = match sqlx::PgPool::connect_with(opts).await {
+            Ok(db) => db,
+            Err(err) => {
+                error!("Failed to connect to PostgreSQL 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(self))]
+    pub async fn prune_storage(
+        &self,
+        oldest_to_keep: u32,
+        current_height: u32,
+    ) -> Result<(), ScraperError> {
+        let start = Instant::now();
+
+        let mut tx = self.begin_processing_tx().await?;
+
+        prune_messages(oldest_to_keep.into(), &mut tx).await?;
+        prune_transactions(oldest_to_keep.into(), &mut tx).await?;
+        prune_pre_commits(oldest_to_keep.into(), &mut tx).await?;
+        prune_blocks(oldest_to_keep.into(), &mut tx).await?;
+        update_last_pruned(current_height.into(), &mut tx).await?;
+
+        let commit_start = Instant::now();
+        tx.commit()
+            .await
+            .map_err(|source| ScraperError::StorageTxCommitFailure { source })?;
+        log_db_operation_time("committing pruning tx", commit_start);
+
+        log_db_operation_time("pruning storage", start);
+        Ok(())
+    }
+
+    #[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 lowest_block_height(&self) -> Result<Option<i64>, ScraperError> {
+        Ok(self.manager.get_lowest_block().await?)
+    }
+
+    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_blocks_between(
+        &self,
+        start_time: OffsetDateTime,
+        end_time: OffsetDateTime,
+    ) -> Result<i64, 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);
+        };
+
+        Ok(block_end - block_start)
+    }
+
+    pub async fn get_signed_between(
+        &self,
+        consensus_address: &str,
+        start_height: i64,
+        end_height: i64,
+    ) -> Result<i64, 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<i64, 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_precommit(
+        &self,
+        consensus_address: &str,
+        height: i64,
+    ) -> Result<Option<CommitSignature>, ScraperError> {
+        Ok(self
+            .manager
+            .get_precommit(consensus_address, height)
+            .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_all_known_validators(&self) -> Result<Vec<Validator>, ScraperError> {
+        Ok(self.manager.get_validators().await?)
+    }
+
+    pub async fn get_last_processed_height(&self) -> Result<i64, ScraperError> {
+        Ok(self.manager.get_last_processed_height().await?)
+    }
+
+    pub async fn get_pruned_height(&self) -> Result<i64, ScraperError> {
+        Ok(self.manager.get_pruned_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
+    for chain_tx in &block.transactions {
+        persist_transaction(chain_tx, 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_transaction(
+    chain_tx: &ParsedTransactionResponse,
+    tx: &mut Transaction<'_, Postgres>,
+) -> Result<(), ScraperError> {
+    let signer_infos = chain_tx
+        .tx
+        .auth_info
+        .signer_infos
+        .iter()
+        .map(|info| SignerInfo {
+            public_key: info.public_key.clone()
+                .map(|pk| pk.into())
+                .unwrap_or_default(),
+            mode_info: serde_json::to_string(&serde_json::json!({
+                "single": {
+                    "mode": match &info.mode_info {
+                        cosmrs::tx::ModeInfo::Single(s) => s.mode as u32,
+                        _ => 0,
+                    }
+                }
+            })).unwrap_or_default(),
+            sequence: info.sequence as i64,
+        })
+        .collect();
+
+    let fee = Fee {
+        amount: chain_tx
+            .tx
+            .auth_info
+            .fee
+            .amount
+            .iter()
+            .map(|coin| Coin {
+                denom: coin.denom.to_string(),
+                amount: coin.amount.to_string(),
+            })
+            .collect(),
+        gas_limit: chain_tx.tx.auth_info.fee.gas_limit as i64,
+        payer: chain_tx.tx.auth_info.fee.payer
+            .clone()
+            .map(|id| id.to_string())
+            .unwrap_or_default(),
+        granter: chain_tx.tx.auth_info.fee.granter
+            .clone()
+            .map(|id| id.to_string())
+            .unwrap_or_default(),
+    };
+
+    let signatures = chain_tx.tx.signatures
+        .iter()
+        .map(|sig| base64::encode(sig))
+        .collect();
+
+    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.messages.clone(),
+        chain_tx.tx.body.memo.clone(),
+        signatures,
+        signer_infos,
+        fee,
+        chain_tx.tx_result.gas_wanted,
+        chain_tx.tx_result.gas_used,
+        chain_tx.tx_result.log.clone(),
+        &mut *tx,
+    )
+    .await?;
+
+    for (index, msg) in chain_tx.tx.body.messages.iter().enumerate() {
+        let involved_accounts = extract_involved_accounts(msg);
+        insert_message(
+            chain_tx.hash.to_string(),
+            index as i64,
+            msg.type_url.clone(),
+            msg.clone(),
+            involved_accounts,
+            chain_tx.height.into(),
+            &mut *tx,
+        )
+        .await?;
+    }
+
+    Ok(())
+}
+
+fn extract_involved_accounts(_msg: &cosmrs::Any) -> Vec<String> {
+    // This is a placeholder implementation
+    // TODO: Implement proper account extraction based on message type
+    vec![]
+}

common/nyxd-scraper-psql/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, Eq, PartialEq, Hash, 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,
+}