feat(rpc): add get_total_served and get_commodity_served endpoints

core/rpc/src/handlers.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use axum::{Extension, Json};
 use draco_interfaces::{
-    types::{EpochInfo, NodeInfo, ProtocolParams},
+    types::{CommodityServed, EpochInfo, NodeInfo, ProtocolParams, TotalServed},
     RpcInterface, SyncQueryRunnerInterface,
 };
 use fleek_crypto::{EthAddress, NodePublicKey};
@@ -70,6 +70,11 @@ impl RpcServer {
                 "flk_get_protocol_params",
                 get_protocol_params_handler::<Q, I>,
             )
+            .with_method("flk_get_total_served", get_total_served_handler::<Q, I>)
+            .with_method(
+                "flk_get_commodity_served",
+                get_commodity_served_handler::<Q, I>,
+            )
             .with_method("flk_get_reputation", get_reputation_handler::<Q, I>);
 
         RpcServer(server.finish())
@@ -200,3 +205,20 @@ pub async fn get_protocol_params_handler<Q: SyncQueryRunnerInterface, I: RpcInte
 ) -> Result<u128> {
     Ok(data.0.query_runner().get_protocol_params(params))
 }
+
+pub async fn get_total_served_handler<Q: SyncQueryRunnerInterface, I: RpcInterface<Q>>(
+    data: Data<Arc<I>>,
+    Params(params): Params<u64>,
+) -> Result<TotalServed> {
+    Ok(data.0.query_runner().get_total_served(params))
+}
+
+pub async fn get_commodity_served_handler<Q: SyncQueryRunnerInterface, I: RpcInterface<Q>>(
+    data: Data<Arc<I>>,
+    Params(params): Params<NodeKeyParam>,
+) -> Result<CommodityServed> {
+    Ok(data
+        .0
+        .query_runner()
+        .get_commodity_served(&params.public_key))
+}

core/rpc/src/tests.rs

@@ -9,7 +9,10 @@ use draco_application::{
     query_runner::QueryRunner,
 };
 use draco_interfaces::{
-    types::{EpochInfo, NodeInfo, ProtocolParams, Staking, UpdateRequest, Worker as NodeWorker},
+    types::{
+        CommodityServed, EpochInfo, NodeInfo, ProtocolParams, Staking, TotalServed, UpdateRequest,
+        Worker as NodeWorker,
+    },
     ApplicationInterface, MempoolSocket, RpcInterface, SyncQueryRunnerInterface,
     WithStartAndShutdown,
 };
@@ -1200,3 +1203,122 @@ async fn test_rpc_get_protocol_params() -> Result<()> {
 
     Ok(())
 }
+
+#[test]
+async fn test_rpc_get_total_served() -> Result<()> {
+    // Init application service and store total served in application state.
+    let mut genesis = Genesis::load().unwrap();
+
+    let total_served = TotalServed {
+        served: vec![1000],
+        reward_pool: 1_000_u32.into(),
+    };
+    genesis.total_served.insert(0, total_served.clone());
+
+    let app = Application::init(AppConfig {
+        genesis: Some(genesis),
+        mode: Mode::Test,
+    })
+    .await
+    .unwrap();
+    let query_runner = app.sync_query();
+    app.start().await;
+
+    // Init rpc service
+    let port = 30016;
+    let mut rpc = Rpc::init(
+        RpcConfig::default(),
+        MockWorker::mempool_socket(),
+        query_runner,
+    )
+    .await?;
+    rpc.config.port = port;
+
+    task::spawn(async move {
+        rpc.start().await;
+    });
+    wait_for_server_start(port).await?;
+
+    let req = json!({
+        "jsonrpc": "2.0",
+        "method":"flk_get_total_served",
+        "params": 0,
+        "id":1,
+    });
+
+    let response = make_request(port, req.to_string()).await?;
+
+    if response.status().is_success() {
+        let value: Value = response.json().await?;
+        if value.get("result").is_some() {
+            // Parse the response as a successful response
+            let success_response: RpcSuccessResponse<TotalServed> = serde_json::from_value(value)?;
+            assert_eq!(total_served, success_response.result);
+        } else {
+            panic!("Rpc Error: {value}")
+        }
+    } else {
+        panic!("Request failed with status: {}", response.status());
+    }
+    Ok(())
+}
+
+#[test]
+async fn test_rpc_get_commodity_served() -> Result<()> {
+    // Init application service and store total served in application state.
+    let mut genesis = Genesis::load().unwrap();
+
+    let node_secret_key = NodeSecretKey::generate();
+    let node_public_key = node_secret_key.to_pk();
+    genesis
+        .current_epoch_served
+        .insert(node_public_key.to_base64(), vec![1000]);
+
+    let app = Application::init(AppConfig {
+        genesis: Some(genesis),
+        mode: Mode::Test,
+    })
+    .await
+    .unwrap();
+    let query_runner = app.sync_query();
+    app.start().await;
+
+    // Init rpc service
+    let port = 30017;
+    let mut rpc = Rpc::init(
+        RpcConfig::default(),
+        MockWorker::mempool_socket(),
+        query_runner,
+    )
+    .await?;
+    rpc.config.port = port;
+
+    task::spawn(async move {
+        rpc.start().await;
+    });
+    wait_for_server_start(port).await?;
+
+    let req = json!({
+        "jsonrpc": "2.0",
+        "method":"flk_get_commodity_served",
+        "params": {"public_key": node_public_key},
+        "id":1,
+    });
+
+    let response = make_request(port, req.to_string()).await?;
+
+    if response.status().is_success() {
+        let value: Value = response.json().await?;
+        if value.get("result").is_some() {
+            // Parse the response as a successful response
+            let success_response: RpcSuccessResponse<CommodityServed> =
+                serde_json::from_value(value)?;
+            assert_eq!(vec![1000], success_response.result);
+        } else {
+            panic!("Rpc Error: {value}")
+        }
+    } else {
+        panic!("Request failed with status: {}", response.status());
+    }
+    Ok(())
+}