run the migrations from the data observatory and not the base chain scraper

Author: mmsinclair

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

@@ -13,7 +13,7 @@ use nyxd_scraper_shared::storage::helpers::log_db_operation_time;
 use nyxd_scraper_shared::storage::{NyxdScraperStorage, NyxdScraperStorageError};
 use sqlx::types::time::{OffsetDateTime, PrimitiveDateTime};
 use tokio::time::Instant;
-use tracing::{debug, error, info, instrument, warn};
+use tracing::{debug, error, info, instrument};
 
 #[derive(Clone)]
 pub struct PostgresScraperStorage {
@@ -22,7 +22,10 @@ pub struct PostgresScraperStorage {
 
 impl PostgresScraperStorage {
     #[instrument]
-    pub async fn init(connection_string: &str) -> Result<Self, PostgresScraperError> {
+    pub async fn init(
+        connection_string: &str,
+        run_migrations: &bool,
+    ) -> Result<Self, PostgresScraperError> {
         debug!("initialising scraper database with '{connection_string}'",);
 
         let connection_pool = match sqlx::PgPool::connect(connection_string).await {
@@ -33,12 +36,13 @@ impl PostgresScraperStorage {
             }
         };
 
-        if let Err(err) = sqlx::migrate!("./sql_migrations")
-            .run(&connection_pool)
-            .await
-        {
-            warn!("Failed to initialize SQLx database: {err}");
-            // return Err(err.into());
+        if *run_migrations {
+            if let Err(err) = sqlx::migrate!("./sql_migrations")
+                .run(&connection_pool)
+                .await
+            {
+                return Err(err.into());
+            }
         }
 
         info!("Database migration finished!");
@@ -192,8 +196,11 @@ impl PostgresScraperStorage {
 impl NyxdScraperStorage for PostgresScraperStorage {
     type StorageTransaction = PostgresStorageTransaction;
 
-    async fn initialise(storage: &str) -> Result<Self, NyxdScraperStorageError> {
-        PostgresScraperStorage::init(storage)
+    async fn initialise(
+        storage: &str,
+        run_migrations: &bool,
+    ) -> Result<Self, NyxdScraperStorageError> {
+        PostgresScraperStorage::init(storage, run_migrations)
             .await
             .map_err(NyxdScraperStorageError::from)
     }

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

@@ -48,6 +48,8 @@ pub struct Config {
     pub store_precommits: bool,
 
     pub start_block: StartingBlockOpts,
+
+    pub run_migrations: bool,
 }
 
 pub struct NyxdScraperBuilder<S> {
@@ -161,7 +163,7 @@ where
 
     pub async fn new(config: Config) -> Result<Self, ScraperError> {
         config.pruning_options.validate()?;
-        let storage = S::initialise(&config.database_storage).await?;
+        let storage = S::initialise(&config.database_storage, &config.run_migrations).await?;
         let rpc_client = RpcClient::new(&config.rpc_url)?;
 
         Ok(NyxdScraper {

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

@@ -33,7 +33,10 @@ pub trait NyxdScraperStorage: Clone + Sized {
     type StorageTransaction: NyxdScraperTransaction;
 
     /// Either connection string (postgres) or storage path (sqlite)
-    async fn initialise(storage: &str) -> Result<Self, NyxdScraperStorageError>;
+    async fn initialise(
+        storage: &str,
+        run_migrations: &bool,
+    ) -> Result<Self, NyxdScraperStorageError>;
 
     async fn begin_processing_tx(
         &self,

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

@@ -207,7 +207,10 @@ impl SqliteScraperStorage {
 impl NyxdScraperStorage for SqliteScraperStorage {
     type StorageTransaction = SqliteStorageTransaction;
 
-    async fn initialise(storage: &str) -> Result<Self, NyxdScraperStorageError> {
+    async fn initialise(
+        storage: &str,
+        _run_migrations: &bool,
+    ) -> Result<Self, NyxdScraperStorageError> {
         SqliteScraperStorage::init(storage)
             .await
             .map_err(NyxdScraperStorageError::from)

nym-data-observatory/src/chain_scraper/mod.rs

@@ -1,8 +1,7 @@
 use crate::cli::commands::run::Args;
 use crate::db::DbPool;
 use nyxd_scraper_psql::{PostgresNyxdScraper, PruningOptions};
-use std::fs;
-use tracing::{info, warn};
+use tracing::info;
 
 pub(crate) mod webhook;
 
@@ -29,6 +28,7 @@ pub(crate) async fn run_chain_scraper(
             start_block_height: args.start_block_height,
             use_best_effort_start_height,
         },
+        run_migrations: false, // ignore the base migrations
     })
     .with_msg_module(crate::modules::wasm::WasmModule::new(connection_pool))
     .with_tx_module(webhook::WebhookModule::new(config.clone())?);

nym-data-observatory/src/db/mod.rs

@@ -1,13 +1,16 @@
 use anyhow::{Result, anyhow};
-use sqlx::{Postgres, postgres::PgConnectOptions};
+use sqlx::{Postgres, migrate::Migrator, postgres::PgConnectOptions};
 use std::str::FromStr;
+use tracing::info;
 
 pub(crate) mod models;
 pub(crate) mod queries {
     pub mod price;
     pub mod wasm;
 }
 
+static MIGRATOR: Migrator = sqlx::migrate!("./migrations");
+
 pub(crate) type DbPool = sqlx::Pool<Postgres>;
 
 pub(crate) struct Storage {
@@ -22,6 +25,13 @@ impl Storage {
             .await
             .map_err(|err| anyhow!("Failed to connect to {}: {}", &connection_url, err))?;
 
+        MIGRATOR
+            .run(&pool)
+            .await
+            .map_err(|err| anyhow!("Failed to run migrations: {}", err))?;
+
+        info!("✅ Successfully migrated the database");
+
         Ok(Storage { pool })
     }
 

nym-validator-rewarder/src/config/mod.rs

@@ -135,6 +135,7 @@ impl Config {
                 start_block_height: None,
                 use_best_effort_start_height: true,
             },
+            run_migrations: true,
         })
     }
 

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

@@ -60,6 +60,7 @@ pub(crate) async fn run_chain_scraper(
             start_block_height,
             use_best_effort_start_height,
         },
+        run_migrations: true,
     })
     .with_msg_module(BankScraperModule::new(
         db_pool,