gov: search proposal by text (#29)

* search proprosals by text

* minor swagger fix

* wip

* upsert rewards

* return 404 if proposal is not found

* wip

* wip

* wip

chain/Cargo.toml

@@ -13,11 +13,11 @@ name = "chain"
 path = "src/main.rs"
 
 [dependencies]
-tokio.workspace = true 
+tokio.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true
 serde_json.workspace = true
-clap.workspace = true 
+clap.workspace = true
 anyhow.workspace = true
 namada_sdk.workspace = true
 namada_core.workspace = true
@@ -30,6 +30,7 @@ deadpool-diesel.workspace = true
 diesel.workspace = true
 orm.workspace = true
 clap-verbosity-flag.workspace = true
+futures.workspace = true
 
 [build-dependencies]
-vergen = { version = "8.0.0", features = ["build", "git", "gitcl"] }
+vergen = { version = "8.0.0", features = ["build", "git", "gitcl"] }

chain/run.sh

@@ -1,2 +1,2 @@
 
-cargo run -- --tendermint-url http://127.0.0.1:27657 --checksums-filepath ../artifacts/checksums.json --chain-id localnet.a38cf62f63db8c1a1f3c9 --database-url postgres://postgres:[email protected]:5435/indexer_local
+cargo run -- --tendermint-url http://127.0.0.1:27657 --chain-id local.988b6b47aadd23b12b8a9a03 --database-url postgres://postgres:[email protected]:5435/namada-indexer

chain/src/main.rs

@@ -18,6 +18,7 @@ use clap::Parser;
 use clap_verbosity_flag::LevelFilter;
 use deadpool_diesel::postgres::Object;
 use diesel::{ExpressionMethods, QueryDsl, RunQueryDsl};
+use orm::migrations::run_migrations;
 use orm::schema::pos_rewards;
 use shared::block::Block;
 use shared::block_result::BlockResult;
@@ -44,7 +45,7 @@ async fn main() -> Result<(), MainError> {
             .unwrap_or_else(|| {
                 panic!("{} must be defined in namada storage.", code_path)
             });
-        checksums.add(code_path, code);
+        checksums.add(code_path, code.to_lowercase());
     }
 
     let log_level = match config.verbosity.log_level_filter() {
@@ -66,6 +67,12 @@ async fn main() -> Result<(), MainError> {
     let app_state = AppState::new(config.database_url).into_db_error()?;
     let conn = Arc::new(app_state.get_db_connection().await.into_db_error()?);
 
+    // Run migrations
+    run_migrations(&conn)
+        .await
+        .context_db_interact_error()
+        .into_db_error()?;
+
     initial_query(&client, &conn).await?;
 
     let last_block_height = db_service::get_last_synched_block(&conn)
@@ -227,6 +234,8 @@ async fn initial_query(
         let pos_crawler_epoch =
             get_pos_crawler_state(conn).await.into_db_error();
 
+        println!("{:?}, {:?}", epoch, pos_crawler_epoch);
+
         match pos_crawler_epoch {
             Ok(pos_crawler_epoch) if pos_crawler_epoch.epoch == epoch => {
                 break;
@@ -236,7 +245,7 @@ async fn initial_query(
 
         tracing::info!("Waiting for PoS service update...");
 
-        sleep(Duration::from_secs(5)).await;
+        sleep(Duration::from_secs(1)).await;
     }
 
     let balances = query_all_balances(client).await.into_rpc_error()?;

chain/src/services/namada.rs

@@ -2,6 +2,7 @@ use std::collections::HashSet;
 use std::str::FromStr;
 
 use anyhow::{anyhow, Context};
+use futures::StreamExt;
 use namada_core::storage::{
     BlockHeight as NamadaSdkBlockHeight, Epoch as NamadaSdkEpoch,
 };
@@ -67,33 +68,40 @@ pub async fn get_epoch_at_block_height(
     Ok(epoch.0 as Epoch)
 }
 
-// TODO: remove unwraps
 pub async fn query_balance(
     client: &HttpClient,
     balance_changes: &HashSet<BalanceChange>,
 ) -> anyhow::Result<Balances> {
-    let mut res: Balances = vec![];
-
-    for balance_change in balance_changes {
-        let owner =
-            NamadaSdkAddress::from_str(&balance_change.address.to_string())
-                .context("Failed to parse owner address")?;
-        let token =
-            NamadaSdkAddress::from_str(&balance_change.token.to_string())
-                .context("Failed to parse token address")?;
-
-        let amount = rpc::get_token_balance(client, &token, &owner)
-            .await
-            .unwrap_or_default();
-
-        res.push(Balance {
-            owner: Id::from(owner),
-            token: Id::from(token),
-            amount: Amount::from(amount),
-        });
-    }
-
-    anyhow::Ok(res)
+    Ok(futures::stream::iter(balance_changes)
+        .filter_map(|balance_change| async move {
+            tracing::info!(
+                "Fetching balance change for {} ...",
+                balance_change.address
+            );
+
+            let owner =
+                NamadaSdkAddress::from_str(&balance_change.address.to_string())
+                    .context("Failed to parse owner address")
+                    .ok()?;
+            let token =
+                NamadaSdkAddress::from_str(&balance_change.token.to_string())
+                    .context("Failed to parse token address")
+                    .ok()?;
+
+            let amount = rpc::get_token_balance(client, &token, &owner)
+                .await
+                .unwrap_or_default();
+
+            Some(Balance {
+                owner: Id::from(owner),
+                token: Id::from(token),
+                amount: Amount::from(amount),
+            })
+        })
+        .map(futures::future::ready)
+        .buffer_unordered(20)
+        .collect::<Vec<_>>()
+        .await)
 }
 
 pub async fn query_all_balances(

docker-compose.yml

@@ -13,8 +13,7 @@ services:
     environment:
       DATABASE_URL: postgres://postgres:password@postgres:5435/namada-indexer
       TENDERMINT_URL: http://localhost:27657
-      CHAIN_ID: local.1321312131
-      CHECKSUMS_FILE: /app/checksums.json
+      CHAIN_ID: local.988b6b47aadd23b12b8a9a03
 
   governance:
     restart: on-failure

governance/run.sh

@@ -1,2 +1,2 @@
 
-cargo run -- --cargo-env development --tendermint-url http://127.0.0.1:27657 --database-url postgres://postgres:[email protected]:5435/indexer_local
+cargo run -- --tendermint-url http://127.0.0.1:27657 --database-url postgres://postgres:[email protected]:5435/namada-indexer --sleep-for 15

governance/src/main.rs

@@ -9,8 +9,9 @@ use governance::config::AppConfig;
 use governance::services::{db as db_service, namada as namada_service};
 use governance::state::AppState;
 use orm::governance_proposal::GovernanceProposalUpdateStatusDb;
+use orm::migrations::run_migrations;
 use orm::schema::governance_proposals;
-use shared::error::{AsDbError, AsRpcError, MainError};
+use shared::error::{AsDbError, AsRpcError, ContextDbInteractError, MainError};
 use tendermint_rpc::HttpClient;
 use tokio::signal;
 use tokio::time::sleep;
@@ -45,6 +46,13 @@ async fn main() -> anyhow::Result<()> {
 
     let app_state = AppState::new(config.database_url).into_db_error()?;
 
+    let conn = Arc::new(app_state.get_db_connection().await.into_db_error()?);
+    // Run migrations
+    run_migrations(&conn)
+        .await
+        .context_db_interact_error()
+        .into_db_error()?;
+
     let retry_strategy = FixedInterval::from_millis(5000).map(jitter);
     let exit_handle = must_exit_handle();
 

orm/migrations/2024-05-09-042930_init_unbonds/up.sql

@@ -9,6 +9,6 @@ CREATE TABLE unbonds (
     CONSTRAINT fk_validator_id FOREIGN KEY(validator_id) REFERENCES validators(id) ON DELETE CASCADE
 );
 
-ALTER TABLE unbonds ADD UNIQUE (address, validator_id, withdraw_epoch, epoch);
+ALTER TABLE unbonds ADD UNIQUE (address, validator_id, withdraw_epoch);
 
 CREATE INDEX index_unbonds_owner_withdraw_epoch ON unbonds (address, withdraw_epoch);

orm/migrations/2024-05-09-131859_pos_rewards/up.sql

@@ -8,4 +8,6 @@ CREATE TABLE pos_rewards (
   CONSTRAINT fk_validator_id FOREIGN KEY(validator_id) REFERENCES validators(id) ON DELETE CASCADE
 );
 
+ALTER TABLE pos_rewards ADD UNIQUE (owner, validator_id);
+
 CREATE INDEX index_pos_rewards_owner ON pos_rewards USING HASH  (owner);

pos/run.sh

@@ -1,2 +1,2 @@
 
-cargo run -- --tendermint-url http://127.0.0.1:27657 --database-url postgres://postgres:[email protected]:5435/indexer_local
+cargo run -- --tendermint-url http://127.0.0.1:27657 --database-url postgres://postgres:[email protected]:5435/namada-indexer

rewards/run.sh

@@ -0,0 +1,2 @@
+
+cargo run -- --tendermint-url http://127.0.0.1:27657 --database-url postgres://postgres:[email protected]:5435/namada-indexer --sleep-for 15

rewards/src/main.rs

@@ -4,13 +4,15 @@ use std::time::Duration;
 
 use clap::Parser;
 use clap_verbosity_flag::LevelFilter;
+use diesel::upsert::excluded;
 use diesel::{ExpressionMethods, QueryDsl, RunQueryDsl, SelectableHelper};
+use orm::migrations::run_migrations;
 use orm::pos_rewards::PosRewardInsertDb;
 use orm::validators::ValidatorDb;
 use rewards::config::AppConfig;
 use rewards::services::namada as namada_service;
 use rewards::state::AppState;
-use shared::error::{AsDbError, AsRpcError, MainError};
+use shared::error::{AsDbError, AsRpcError, ContextDbInteractError, MainError};
 use tendermint_rpc::HttpClient;
 use tokio::signal;
 use tokio::time::sleep;
@@ -45,6 +47,13 @@ async fn main() -> anyhow::Result<()> {
 
     let app_state = AppState::new(config.database_url).into_db_error()?;
 
+    let conn = Arc::new(app_state.get_db_connection().await.into_db_error()?);
+    // Run migrations
+    run_migrations(&conn)
+        .await
+        .context_db_interact_error()
+        .into_db_error()?;
+
     let retry_strategy = FixedInterval::from_millis(5000).map(jitter);
     let exit_handle = must_exit_handle();
 
@@ -100,7 +109,11 @@ async fn main() -> anyhow::Result<()> {
                                         })
                                         .collect::<Vec<_>>(),
                                 )
-                                .on_conflict_do_nothing()
+                                .on_conflict((orm::schema::pos_rewards::columns::owner, orm::schema::pos_rewards::columns::validator_id))
+                                .do_update()
+                                .set(
+                                    orm::schema::pos_rewards::columns::raw_amount.eq(excluded(orm::schema::pos_rewards::columns::raw_amount))
+                                )
                                 .execute(transaction_conn)
                         }
                     )

shared/src/transaction.rs

@@ -26,6 +26,7 @@ pub enum TransactionKind {
 
 impl TransactionKind {
     pub fn from(tx_kind_name: &str, data: &[u8]) -> Self {
+        println!("tx kind: {}", tx_kind_name);
         match tx_kind_name {
             "tx_transfer" => {
                 let transfer_data =
@@ -159,6 +160,7 @@ impl Transaction {
                         let tx_data = transaction.data().unwrap_or_default();
                         TransactionKind::from(&tx_kind_name, &tx_data)
                     } else {
+                        println!("UNKNOWN: {}", id);
                         TransactionKind::Unknown
                     }
                 } else {