use postgres in chain scraper

Author: jstuczyn

Cargo.lock

@@ -11,11 +11,7 @@ use crate::storage::transaction::PostgresStorageTransaction;
 use async_trait::async_trait;
 use nyxd_scraper_shared::storage::helpers::log_db_operation_time;
 use nyxd_scraper_shared::storage::{NyxdScraperStorage, NyxdScraperStorageError};
-use sqlx::postgres::PgConnectOptions;
 use sqlx::types::time::{OffsetDateTime, PrimitiveDateTime};
-use sqlx::ConnectOptions;
-use std::fmt::Debug;
-use std::path::Path;
 use tokio::time::Instant;
 use tracing::{debug, error, info, instrument};
 

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

@@ -4,7 +4,7 @@
 use crate::storage::models::{CommitSignature, Validator};
 use nyxd_scraper_shared::storage::helpers::log_db_operation_time;
 use sqlx::types::time::PrimitiveDateTime;
-use sqlx::types::{Json, JsonValue};
+use sqlx::types::JsonValue;
 use sqlx::{Executor, Postgres};
 use tokio::time::Instant;
 use tracing::{instrument, trace};

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

@@ -24,7 +24,7 @@ nym-bin-common = { path = "../common/bin-common", features = ["output_format"] }
 nym-network-defaults = { path = "../common/network-defaults" }
 nym-task = { path = "../common/task" }
 nym-validator-client = { path = "../common/client-libs/validator-client" }
-nyxd-scraper-sqlite = { path = "../common/nyxd-scraper-sqlite" }
+nyxd-scraper-psql = { path = "../common/nyxd-scraper-psql" }
 reqwest = { workspace = true, features = ["rustls-tls"] }
 schemars = { workspace = true }
 serde = { workspace = true, features = ["derive"] }

nyx-chain-watcher/Cargo.toml

@@ -6,9 +6,9 @@ use crate::env::vars::{
 use crate::http::state::BankScraperModuleState;
 use async_trait::async_trait;
 use nym_validator_client::nyxd::{Any, Coin, CosmosCoin, Hash, Msg, MsgSend, Name};
-use nyxd_scraper_sqlite::{
-    MsgModule, NyxdScraperTransaction, ParsedTransactionResponse, PruningOptions, ScraperError,
-    SqliteNyxdScraper,
+use nyxd_scraper_psql::{
+    MsgModule, NyxdScraperTransaction, ParsedTransactionResponse, PostgresNyxdScraper,
+    PruningOptions, ScraperError,
 };
 use sqlx::SqlitePool;
 use std::fs;
@@ -18,7 +18,7 @@ pub(crate) async fn run_chain_scraper(
     config: &crate::config::Config,
     db_pool: SqlitePool,
     shared_state: BankScraperModuleState,
-) -> anyhow::Result<SqliteNyxdScraper> {
+) -> anyhow::Result<PostgresNyxdScraper> {
     let websocket_url = std::env::var("NYXD_WS").expect("NYXD_WS not defined");
 
     let rpc_url = std::env::var("NYXD").expect("NYXD not defined");
@@ -47,16 +47,16 @@ pub(crate) async fn run_chain_scraper(
 
     if nuke_db {
         warn!("☢️☢️☢️ NUKING THE SCRAPER DATABASE");
-        fs::remove_file(config.chain_scraper_database_path())?;
+        fs::remove_file(config.chain_scraper_connection_string())?;
     }
 
-    let scraper = SqliteNyxdScraper::builder(nyxd_scraper_sqlite::Config {
+    let scraper = PostgresNyxdScraper::builder(nyxd_scraper_psql::Config {
         websocket_url,
         rpc_url,
-        database_storage: config.chain_scraper_database_path().into(),
+        database_storage: config.chain_scraper_connection_string.clone(),
         pruning_options: PruningOptions::nothing(),
         store_precommits: false,
-        start_block: nyxd_scraper_sqlite::StartingBlockOpts {
+        start_block: nyxd_scraper_psql::StartingBlockOpts {
             start_block_height,
             use_best_effort_start_height,
         },

nyx-chain-watcher/src/chain_scraper/mod.rs

@@ -5,38 +5,46 @@ use crate::cli::DEFAULT_NYX_CHAIN_WATCHER_ID;
 use crate::config::payments_watcher::HttpAuthenticationOptions::AuthorizationBearerToken;
 use crate::config::payments_watcher::PaymentWatcherConfig;
 use crate::config::{default_config_filepath, Config, ConfigBuilder, PaymentWatchersConfig};
+use crate::env::vars::*;
 use crate::error::NyxChainWatcherError;
 use nym_config::save_unformatted_config_to_file;
 use nym_validator_client::nyxd::AccountId;
 use std::str::FromStr;
 
 #[derive(clap::Args, Debug)]
-pub(crate) struct Args {}
+pub(crate) struct Args {
+    /// (Override) Postgres connection string for chain scraper history
+    #[arg(long, env = NYX_CHAIN_WATCHER_HISTORY_DATABASE_PATH, alias = "chain_history_db_path")]
+    pub(crate) chain_history_db_connection_string: String,
+}
 
-pub(crate) async fn execute(_args: Args) -> Result<(), NyxChainWatcherError> {
+pub(crate) async fn execute(args: Args) -> Result<(), NyxChainWatcherError> {
     let config_path = default_config_filepath();
     let data_dir = Config::default_data_directory(&config_path)?;
 
-    let builder = ConfigBuilder::new(config_path.clone(), data_dir).with_payment_watcher_config(
-        PaymentWatchersConfig {
-            watchers: vec![PaymentWatcherConfig {
-                id: DEFAULT_NYX_CHAIN_WATCHER_ID.to_string(),
-                webhook_url: "https://webhook.site".to_string(),
-                watch_for_transfer_recipient_accounts: vec![AccountId::from_str(
-                    "n17g9a2pwwkg8m60wf59pq6mv0c2wusg9ukparkz",
-                )
-                .unwrap()],
-                authentication: Some(AuthorizationBearerToken {
-                    token: "1234".to_string(),
-                }),
-                description: None,
-                watch_for_chain_message_types: vec![
-                    "/cosmos.bank.v1beta1.MsgSend".to_string(),
-                    "/ibc.applications.transfer.v1.MsgTransfer".to_string(),
-                ],
-            }],
-        },
-    );
+    let builder = ConfigBuilder::new(
+        config_path.clone(),
+        data_dir,
+        args.chain_history_db_connection_string,
+    )
+    .with_payment_watcher_config(PaymentWatchersConfig {
+        watchers: vec![PaymentWatcherConfig {
+            id: DEFAULT_NYX_CHAIN_WATCHER_ID.to_string(),
+            webhook_url: "https://webhook.site".to_string(),
+            watch_for_transfer_recipient_accounts: vec![AccountId::from_str(
+                "n17g9a2pwwkg8m60wf59pq6mv0c2wusg9ukparkz",
+            )
+            .unwrap()],
+            authentication: Some(AuthorizationBearerToken {
+                token: "1234".to_string(),
+            }),
+            description: None,
+            watch_for_chain_message_types: vec![
+                "/cosmos.bank.v1beta1.MsgSend".to_string(),
+                "/ibc.applications.transfer.v1.MsgTransfer".to_string(),
+            ],
+        }],
+    });
 
     let config = builder.build();
 

nyx-chain-watcher/src/cli/commands/init.rs

@@ -10,9 +10,9 @@ pub(crate) struct Args {
     #[arg(long, env = NYX_CHAIN_WATCHER_DATABASE_PATH)]
     pub(crate) chain_watcher_db_path: Option<String>,
 
-    /// (Override) SQLite database file path for chain scraper history
-    #[arg(long, env = NYX_CHAIN_WATCHER_HISTORY_DATABASE_PATH)]
-    pub(crate) chain_history_db_path: Option<String>,
+    /// (Override) Postgres connection string for chain scraper history
+    #[arg(long, env = NYX_CHAIN_WATCHER_HISTORY_DATABASE_PATH, alias = "chain_history_db_path")]
+    pub(crate) chain_history_db_connection_string: String,
 
     /// (Override) Watch for transfers to these recipient accounts
     #[clap(

nyx-chain-watcher/src/cli/commands/run/args.rs

@@ -13,8 +13,8 @@ pub(crate) fn get_run_config(args: Args) -> Result<Config, NyxChainWatcherError>
         mut watch_for_chain_message_types,
         webhook_auth,
         ref chain_watcher_db_path,
-        ref chain_history_db_path,
         webhook_url,
+        ..
     } = args;
 
     // if there are no args set, then try load the config
@@ -42,18 +42,17 @@ pub(crate) fn get_run_config(args: Args) -> Result<Config, NyxChainWatcherError>
     let config_path = default_config_filepath();
     let data_dir = Config::default_data_directory(&config_path)?;
 
-    let mut builder = ConfigBuilder::new(config_path, data_dir);
+    let mut builder = ConfigBuilder::new(
+        config_path,
+        data_dir,
+        args.chain_history_db_connection_string,
+    );
 
     if let Some(db_path) = chain_watcher_db_path {
         info!("Overriding database url with '{db_path}'");
         builder = builder.with_db_path(db_path.clone());
     }
 
-    if let Some(db_path) = chain_history_db_path {
-        info!("Overriding chain history database url with '{db_path}'");
-        builder = builder.with_chain_scraper_db_path(db_path.clone());
-    }
-
     if let Some(webhook_url) = webhook_url {
         let authentication =
             webhook_auth.map(|token| HttpAuthenticationOptions::AuthorizationBearerToken { token });

nyx-chain-watcher/src/cli/commands/run/config.rs

@@ -125,7 +125,7 @@ pub(crate) async fn execute(args: Args, http_port: u16) -> Result<(), NyxChainWa
     );
     info!(
         "Chain History Database path is {:?}",
-        std::path::Path::new(&config.chain_scraper_database_path()).canonicalize()
+        std::path::Path::new(&config.chain_scraper_connection_string()).canonicalize()
     );
 
     // Ensure parent directory exists

nyx-chain-watcher/src/cli/commands/run/mod.rs

@@ -20,7 +20,6 @@ use crate::error::NyxChainWatcherError;
 const DEFAULT_NYM_CHAIN_WATCHER_DIR: &str = "nym-chain-watcher";
 
 pub(crate) const DEFAULT_NYM_CHAIN_WATCHER_DB_FILENAME: &str = "nyx_chain_watcher.sqlite";
-pub(crate) const DEFAULT_NYM_CHAIN_SCRAPER_HISTORY_DB_FILENAME: &str = "chain_history.sqlite";
 
 /// Derive default path to nym-chain-watcher's config directory.
 /// It should get resolved to `$HOME/.nym/nym-chain-watcher/config`
@@ -44,22 +43,25 @@ pub struct ConfigBuilder {
 
     pub db_path: Option<String>,
 
-    pub chain_scraper_db_path: Option<String>,
-
+    pub chain_scraper_connection_string: String,
     pub payment_watcher_config: Option<PaymentWatchersConfig>,
 
     pub logging: Option<LoggingSettings>,
 }
 
 impl ConfigBuilder {
-    pub fn new(config_path: PathBuf, data_dir: PathBuf) -> Self {
+    pub fn new(
+        config_path: PathBuf,
+        data_dir: PathBuf,
+        chain_scraper_connection_string: String,
+    ) -> Self {
         ConfigBuilder {
             config_path,
             data_dir,
             payment_watcher_config: None,
             logging: None,
             db_path: None,
-            chain_scraper_db_path: None,
+            chain_scraper_connection_string,
         }
     }
 
@@ -68,11 +70,6 @@ impl ConfigBuilder {
         self
     }
 
-    pub fn with_chain_scraper_db_path(mut self, chain_scraper_db_path: String) -> Self {
-        self.chain_scraper_db_path = Some(chain_scraper_db_path);
-        self
-    }
-
     #[allow(dead_code)]
     pub fn with_payment_watcher_config(
         mut self,
@@ -95,7 +92,7 @@ impl ConfigBuilder {
             payment_watcher_config: self.payment_watcher_config.unwrap_or_default(),
             data_dir: self.data_dir,
             db_path: self.db_path,
-            chain_scraper_db_path: self.chain_scraper_db_path,
+            chain_scraper_connection_string: self.chain_scraper_connection_string,
         }
     }
 }
@@ -113,8 +110,7 @@ pub struct Config {
     #[serde(skip)]
     db_path: Option<String>,
 
-    #[serde(skip)]
-    chain_scraper_db_path: Option<String>,
+    pub chain_scraper_connection_string: String,
 
     #[serde(default)]
     pub payment_watcher_config: PaymentWatchersConfig,
@@ -208,14 +204,8 @@ impl Config {
         })
     }
 
-    pub fn chain_scraper_database_path(&self) -> String {
-        self.chain_scraper_db_path.clone().unwrap_or_else(|| {
-            let mut path = self.data_dir.clone().to_path_buf();
-            path.push(DEFAULT_NYM_CHAIN_SCRAPER_HISTORY_DB_FILENAME);
-            path.to_str()
-                .unwrap_or(DEFAULT_NYM_CHAIN_SCRAPER_HISTORY_DB_FILENAME)
-                .to_string()
-        })
+    pub fn chain_scraper_connection_string(&self) -> String {
+        self.chain_scraper_connection_string.clone()
     }
 
     // simple wrapper that reads config file and assigns path location