More apis (#32)

* added total voting power, added sse chain sync

* fmt, clippy

Cargo.toml

@@ -55,6 +55,7 @@ async-stream = "0.3.5"
 futures-core = "0.3.30"
 futures-util = "0.3.30"
 futures = "0.3.30"
+tokio-stream = "0.1.15"
 tokio-retry = "0.3"
 thiserror = "1.0.56"
 diesel_migrations = { version = "2.1.0", default-features = false, features = [

swagger.yml

@@ -125,6 +125,18 @@ paths:
                 type: array
                 items:
                   $ref: '#/components/schemas/Withdraw'  
+  /api/v1/pos/voting-power:
+    get:
+      summary: Get the total voting power
+      responses:
+        '200':
+          description: The total voting power.
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: '#/components/schemas/VotingPower'  
   /api/v1/gov/proposal:
     get:
       summary: Get a list of governance proposals
@@ -421,6 +433,11 @@ components:
           minimum: 0
         withdrawEpoch: 
           type: integer
+    VotingPower:
+      type: object
+      properties:
+        votingPower:
+          type: integer
     Balance:
       type: object
       properties:

webserver/Cargo.toml

@@ -38,6 +38,8 @@ axum-trace-id.workspace = true
 axum-macros.workspace = true
 orm.workspace = true
 diesel.workspace = true
+futures.workspace = true
+tokio-stream.workspace = true
 # shared.workspace = true
 deadpool-redis = "0.13.0"
 

webserver/src/app.rs

@@ -19,8 +19,8 @@ use tower_http::trace::TraceLayer;
 use crate::appstate::AppState;
 use crate::config::AppConfig;
 use crate::handler::{
-    balance as balance_handlers, governance as gov_handlers,
-    pos as pos_handlers,
+    balance as balance_handlers, chain as chain_handlers,
+    governance as gov_handlers, pos as pos_handlers,
 };
 use crate::state::common::CommonState;
 
@@ -50,6 +50,10 @@ impl ApplicationServer {
                     get(pos_handlers::get_withdraws),
                 )
                 .route("/pos/reward/:address", get(pos_handlers::get_rewards))
+                .route(
+                    "/pos/voting-power",
+                    get(pos_handlers::get_total_voting_power),
+                )
                 .route(
                     "/gov/proposal",
                     get(gov_handlers::get_governance_proposals),
@@ -74,6 +78,7 @@ impl ApplicationServer {
                     "/account/:address",
                     get(balance_handlers::get_address_balance),
                 )
+                .route("/chain/sync", get(chain_handlers::sync_height))
                 .with_state(common_state)
         };
 

webserver/src/handler/chain.rs

@@ -0,0 +1,28 @@
+use std::convert::Infallible;
+use std::time::Duration;
+
+use axum::extract::State;
+use axum::response::sse::{Event, KeepAlive};
+use axum::response::Sse;
+use futures::Stream;
+use tokio_stream::StreamExt;
+
+use crate::state::common::CommonState;
+
+pub async fn sync_height(
+    State(state): State<CommonState>,
+) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
+    let stream = tokio_stream::wrappers::IntervalStream::new(
+        tokio::time::interval(Duration::from_secs(3)),
+    )
+    .then(move |_| {
+        let state = state.clone();
+
+        async move {
+            let height = state.chain_service.find_latest_height().await;
+            Ok(Event::default().data(height.to_string()))
+        }
+    });
+
+    Sse::new(stream).keep_alive(KeepAlive::default())
+}

webserver/src/handler/mod.rs

@@ -1,3 +1,4 @@
 pub mod balance;
+pub mod chain;
 pub mod governance;
 pub mod pos;

webserver/src/handler/pos.rs

@@ -6,7 +6,9 @@ use axum_trace_id::TraceId;
 
 use crate::dto::pos::PoSQueryParams;
 use crate::error::api::ApiError;
-use crate::response::pos::{Bond, Reward, Unbond, ValidatorWithId, Withdraw};
+use crate::response::pos::{
+    Bond, Reward, TotalVotingPower, Unbond, ValidatorWithId, Withdraw,
+};
 use crate::response::utils::PaginatedResponse;
 use crate::state::common::CommonState;
 
@@ -71,3 +73,13 @@ pub async fn get_rewards(
     let rewards = state.pos_service.get_rewards_by_address(address).await?;
     Ok(Json(rewards))
 }
+
+#[debug_handler]
+pub async fn get_total_voting_power(
+    _trace_id: TraceId<String>,
+    _headers: HeaderMap,
+    State(state): State<CommonState>,
+) -> Result<Json<TotalVotingPower>, ApiError> {
+    let total_voting_power = state.pos_service.get_total_voting_power().await?;
+    Ok(Json(TotalVotingPower { total_voting_power }))
+}

webserver/src/repository/balance.rs

@@ -37,9 +37,9 @@ impl BalanceRepoTrait for BalanceRepo {
                 .filter(balances::dsl::owner.eq(address))
                 .select(BalanceDb::as_select())
                 .get_results(conn)
-                .unwrap()
         })
         .await
+        .map_err(|e| e.to_string())?
         .map_err(|e| e.to_string())
     }
 }

webserver/src/repository/chain.rs

@@ -0,0 +1,38 @@
+use axum::async_trait;
+use diesel::dsl::max;
+use diesel::{QueryDsl, RunQueryDsl};
+use orm::schema::block_crawler_state;
+
+use crate::appstate::AppState;
+
+#[derive(Clone)]
+pub struct ChainRepository {
+    pub(crate) app_state: AppState,
+}
+
+#[async_trait]
+pub trait ChainRepositoryTrait {
+    fn new(app_state: AppState) -> Self;
+
+    async fn find_latest_height(&self) -> Result<Option<i32>, String>;
+}
+
+#[async_trait]
+impl ChainRepositoryTrait for ChainRepository {
+    fn new(app_state: AppState) -> Self {
+        Self { app_state }
+    }
+
+    async fn find_latest_height(&self) -> Result<Option<i32>, String> {
+        let conn = self.app_state.get_db_connection().await;
+
+        conn.interact(move |conn| {
+            block_crawler_state::dsl::block_crawler_state
+                .select(max(block_crawler_state::dsl::height))
+                .first::<Option<i32>>(conn)
+        })
+        .await
+        .map_err(|e| e.to_string())?
+        .map_err(|e| e.to_string())
+    }
+}

webserver/src/repository/governance.rs

@@ -70,9 +70,9 @@ impl GovernanceRepoTrait for GovernanceRepo {
                 .select(GovernanceProposalDb::as_select())
                 .paginate(page)
                 .load_and_count_pages(conn)
-                .unwrap()
         })
         .await
+        .map_err(|e| e.to_string())?
         .map_err(|e| e.to_string())
     }
 
@@ -109,9 +109,9 @@ impl GovernanceRepoTrait for GovernanceRepo {
                 .select(GovernanceProposalDb::as_select())
                 .paginate(page)
                 .load_and_count_pages(conn)
-                .unwrap()
         })
         .await
+        .map_err(|e| e.to_string())?
         .map_err(|e| e.to_string())
     }
 
@@ -128,9 +128,9 @@ impl GovernanceRepoTrait for GovernanceRepo {
                 .select(GovernanceProposalVoteDb::as_select())
                 .paginate(page)
                 .load_and_count_pages(conn)
-                .unwrap()
         })
         .await
+        .map_err(|e| e.to_string())?
         .map_err(|e| e.to_string())
     }
 
@@ -148,9 +148,9 @@ impl GovernanceRepoTrait for GovernanceRepo {
                 ))
                 .select(GovernanceProposalVoteDb::as_select())
                 .get_results(conn)
-                .unwrap()
         })
         .await
+        .map_err(|e| e.to_string())?
         .map_err(|e| e.to_string())
     }
 }

webserver/src/repository/mod.rs

@@ -1,4 +1,5 @@
 pub mod balance;
+pub mod chain;
 pub mod governance;
 pub mod pos;
 pub mod utils;

webserver/src/repository/pos.rs

@@ -1,4 +1,5 @@
 use axum::async_trait;
+use diesel::dsl::sum;
 use diesel::{
     BoolExpressionMethods, ExpressionMethods, QueryDsl, RunQueryDsl,
     SelectableHelper,
@@ -51,6 +52,8 @@ pub trait PosRepositoryTrait {
         &self,
         address: String,
     ) -> Result<Vec<PoSRewardDb>, String>;
+
+    async fn get_total_voting_power(&self) -> Result<Option<i64>, String>;
 }
 
 #[async_trait]
@@ -70,9 +73,9 @@ impl PosRepositoryTrait for PosRepository {
                 .select(ValidatorDb::as_select())
                 .paginate(page)
                 .load_and_count_pages(conn)
-                .unwrap()
         })
         .await
+        .map_err(|e| e.to_string())?
         .map_err(|e| e.to_string())
     }
 
@@ -104,9 +107,9 @@ impl PosRepositoryTrait for PosRepository {
                 .filter(bonds::dsl::address.eq(address))
                 .select(BondDb::as_select())
                 .get_results(conn)
-                .unwrap()
         })
         .await
+        .map_err(|e| e.to_string())?
         .map_err(|e| e.to_string())
     }
 
@@ -121,9 +124,9 @@ impl PosRepositoryTrait for PosRepository {
                 .filter(unbonds::dsl::address.eq(address))
                 .select(UnbondDb::as_select())
                 .get_results(conn)
-                .unwrap()
         })
         .await
+        .map_err(|e| e.to_string())?
         .map_err(|e| e.to_string())
     }
 
@@ -143,9 +146,9 @@ impl PosRepositoryTrait for PosRepository {
                 )
                 .select(UnbondDb::as_select())
                 .get_results(conn)
-                .unwrap()
         })
         .await
+        .map_err(|e| e.to_string())?
         .map_err(|e| e.to_string())
     }
 
@@ -160,9 +163,22 @@ impl PosRepositoryTrait for PosRepository {
                 .filter(pos_rewards::dsl::owner.eq(address))
                 .select(PoSRewardDb::as_select())
                 .get_results(conn)
-                .unwrap()
         })
         .await
+        .map_err(|e| e.to_string())?
+        .map_err(|e| e.to_string())
+    }
+
+    async fn get_total_voting_power(&self) -> Result<Option<i64>, String> {
+        let conn = self.app_state.get_db_connection().await;
+
+        conn.interact(move |conn| {
+            validators::table
+                .select(sum(validators::dsl::voting_power))
+                .first(conn)
+        })
+        .await
+        .map_err(|e| e.to_string())?
         .map_err(|e| e.to_string())
     }
 }

webserver/src/response/pos.rs

@@ -48,6 +48,12 @@ pub struct Reward {
     pub validator: ValidatorWithId,
 }
 
+#[derive(Clone, Debug, Deserialize, Serialize, Default)]
+#[serde(rename_all = "camelCase")]
+pub struct TotalVotingPower {
+    pub total_voting_power: u64,
+}
+
 impl From<ValidatorDb> for Validator {
     fn from(value: ValidatorDb) -> Self {
         Self {

webserver/src/service/chain.rs

@@ -0,0 +1,24 @@
+use crate::appstate::AppState;
+use crate::repository::chain::{ChainRepository, ChainRepositoryTrait};
+
+#[derive(Clone)]
+pub struct ChainService {
+    chain_repo: ChainRepository,
+}
+
+impl ChainService {
+    pub fn new(app_state: AppState) -> Self {
+        Self {
+            chain_repo: ChainRepository::new(app_state),
+        }
+    }
+
+    // return Result with ChainError
+    pub async fn find_latest_height(&self) -> u64 {
+        self.chain_repo
+            .find_latest_height()
+            .await
+            .unwrap()
+            .unwrap_or_default() as u64
+    }
+}

webserver/src/service/mod.rs

@@ -1,3 +1,4 @@
 pub mod balance;
+pub mod chain;
 pub mod governance;
 pub mod pos;

webserver/src/service/pos.rs

@@ -146,4 +146,14 @@ impl PosService {
         }
         Ok(rewards)
     }
+
+    pub async fn get_total_voting_power(&self) -> Result<u64, PoSError> {
+        let total_voting_power_db = self
+            .pos_repo
+            .get_total_voting_power()
+            .await
+            .map_err(PoSError::Database)?;
+
+        Ok(total_voting_power_db.unwrap_or_default() as u64)
+    }
 }