Feat: API get_l2_block_by_id (#146)

* Feat: Adds API `get_l2_block_by_id`.

* Adds `get_l2_block_by_id.ts` to integration test.

* Updates dependency `common-rs`.

* Fixes the failed asserts.

Author: silathdiir

.env

@@ -1,6 +1,6 @@
 RUN_MODE=development # development or staging
 RUST_BACKTRACE=1
-RUST_LOG=rollup_state_manager=debug
+RUST_LOG=rollup_state_manager=debug,info
 
 NTXS=2
 BALANCELEVELS=3

.github/workflows/integration-test.yml

@@ -87,4 +87,5 @@ jobs:
       - name: Run tests
         run: |
           cd ./examples/js/
+          npx ts-node get_l2_block_by_id.ts
           npx ts-node get_token_balance.ts

Cargo.lock

@@ -0,0 +1,81 @@
+import { grpcClient } from "./grpc_client";
+import { kafkaProducer } from "./kafka_producer";
+import { sleep } from "./util";
+import { strict as assert } from "assert";
+
+const blockId = 0;
+const userId = 0;
+
+const kafkaUserValue = {
+  user_id: userId,
+  l1_address: "0x6286d0A2FC1d4C12a4ACc274018b401c68157Fdb",
+  l2_pubkey:
+    "0x5d182c51bcfe99583d7075a7a0c10d96bef82b8a059c4bf8c5f6e7124cf2bba3"
+};
+
+const kafkaBalanceValue = {
+  timestamp: 16264463600,
+  user_id: userId,
+  asset: "ETH",
+  business: "deposit",
+  change: "3",
+  balance: "3",
+  balance_available: "3",
+  balance_frozen: "0",
+  detail: JSON.stringify({ id: 0 })
+};
+
+async function main() {
+  try {
+    await mainTest();
+  } catch (error) {
+    console.error("Caught error:", error);
+    process.exit(1);
+  }
+}
+
+async function mainTest() {
+  await kafkaProducer.Init();
+
+  await registerUser();
+  await depositBalance();
+  await sleep(1000);
+  await kafkaProducer.Stop();
+
+  await getL2BlockByIdTest();
+}
+
+async function registerUser() {
+  await kafkaProducer.send([
+    {
+      key: "registeruser",
+      value: JSON.stringify(kafkaUserValue)
+    }
+  ]);
+}
+
+async function depositBalance() {
+  await kafkaProducer.send([
+    {
+      key: "balances",
+      value: JSON.stringify(kafkaBalanceValue)
+    }
+  ]);
+}
+
+async function getL2BlockByIdTest() {
+  const res = await grpcClient.l2BlockQuery(blockId);
+  assert(res["created_time"]);
+  assert.equal(res["txs"].length, 2);
+  assert.equal(res["tx_num"], "2");
+  assert.equal(res["real_tx_num"], "2");
+  assert.equal(res["status"], "INITED");
+  assert.equal(
+    res["new_root"],
+    "Fr(0x1cd19866300cc822c6ead46a420ecf2b4dacd26b49d3a5cbf737761f3fa2dd01)"
+  );
+
+  console.log("getL2BlockByIdTest passed");
+}
+
+main();

examples/js/get_l2_block_by_id.ts

@@ -4,7 +4,7 @@ import { sleep } from "./util";
 import { strict as assert } from "assert";
 
 const tokenId = 0;
-const userId = 0;
+const userId = 1;
 
 const kafkaUserValue = {
   user_id: userId,

examples/js/get_token_balance.ts

@@ -17,6 +17,9 @@ class Client {
     this.client = caller(`${server}`, { file, load }, "RollupState");
   }
 
+  async l2BlockQuery(block_id): Promise<Map<string, any>> {
+    return (await this.client.l2BlockQuery({ block_id }));
+  }
   async tokenBalanceQuery(account_id, token_id, token_address, token_name): Promise<Map<string, any>> {
     return (await this.client.tokenBalanceQuery({
       account_id,

examples/js/grpc_client.ts

@@ -5,6 +5,13 @@ package rollup_state;
 import "google/api/annotations.proto";
 
 service RollupState {
+  rpc L2BlockQuery(L2BlockQueryRequest) returns (L2BlockQueryResponse) {
+    option (google.api.http) = {
+      post : "/api/l2_block/{block_id}"
+      body : "*"
+    };
+  }
+
   rpc TokenBalanceQuery(TokenBalanceQueryRequest) returns (TokenBalanceQueryResponse) {
     option (google.api.http) = {
       post : "/api/token_balance/{account_id}/{token_id}"
@@ -13,6 +20,23 @@ service RollupState {
   }
 }
 
+message L2BlockQueryRequest {
+  int64 block_id = 1;
+}
+
+message L2BlockQueryResponse {
+  string new_root = 1;
+  double created_time = 2;
+  uint64 tx_num = 3;
+  uint64 real_tx_num = 4;
+  TaskStatus status = 5;
+  repeated Tx txs = 6;
+  // TODO: Adds `l1_tx_hash: string`.
+  message Tx {
+      repeated string detail = 1; // TODO: Fixes to decoding TX in issue #132.
+  }
+}
+
 message TokenBalanceQueryRequest {
   uint32 account_id = 1;
   uint32 token_id = 2;
@@ -21,7 +45,15 @@ message TokenBalanceQueryRequest {
 }
 
 message TokenBalanceQueryResponse {
-    string balance = 1;
-    string balance_raw = 2;
-    uint32 precision = 3;
+  string balance = 1;
+  string balance_raw = 2;
+  uint32 precision = 3;
+}
+
+enum TaskStatus {
+  INITED = 0;
+  WITGENING = 1;
+  READY = 2;
+  ASSIGNED = 3;
+  PROVED = 4;
 }

proto/rollup_state.proto

@@ -1,17 +1,56 @@
+use crate::config::Settings;
 use crate::grpc::rpc::*;
 use crate::state::global::GlobalState;
 use crate::test_utils::types::{get_token_id_by_name, prec_token_id};
+use crate::types::l2::L2BlockSerde;
+use fluidex_common::db::models::{l2_block, tablenames, task};
+use fluidex_common::db::DbType;
 use fluidex_common::types::FrExt;
+use fluidex_common::utils::timeutil::FTimestamp;
 use std::sync::{Arc, RwLock};
-use tonic::Status;
+use tonic::{Code, Status};
 
 pub struct Controller {
+    db_pool: sqlx::Pool<DbType>,
     state: Arc<RwLock<GlobalState>>,
 }
 
 impl Controller {
-    pub fn new(state: Arc<RwLock<GlobalState>>) -> Self {
-        Self { state }
+    pub async fn new(state: Arc<RwLock<GlobalState>>) -> Self {
+        let db_pool = sqlx::postgres::PgPool::connect(Settings::db()).await.unwrap();
+        Self { db_pool, state }
+    }
+
+    pub async fn l2_block_query(&self, request: L2BlockQueryRequest) -> Result<L2BlockQueryResponse, Status> {
+        let block_id = request.block_id;
+        let l2_block = get_l2_block_by_id(&self.db_pool, block_id).await?;
+
+        let status = match get_task_status_by_block_id(&self.db_pool, block_id).await? {
+            task::TaskStatus::Inited => TaskStatus::Inited,
+            task::TaskStatus::Witgening => TaskStatus::Witgening,
+            task::TaskStatus::Ready => TaskStatus::Ready,
+            task::TaskStatus::Assigned => TaskStatus::Assigned,
+            task::TaskStatus::Proved => TaskStatus::Proved,
+        };
+
+        let witness: L2BlockSerde = serde_json::from_value(l2_block.witness).unwrap();
+        let tx_num = witness.encoded_txs.len() as u64;
+        let txs = witness
+            .encoded_txs
+            .iter()
+            .map(|tx| l2_block_query_response::Tx {
+                detail: tx.iter().map(|fr_str| fr_str.0.to_decimal_string()).collect(),
+            })
+            .collect();
+
+        Ok(L2BlockQueryResponse {
+            new_root: l2_block.new_root,
+            created_time: FTimestamp::from(&l2_block.created_time).0,
+            tx_num,
+            real_tx_num: tx_num, // TODO: Needs to decode and filter out NOP txs.
+            status: status as i32,
+            txs,
+        })
     }
 
     pub fn token_balance_query(&self, request: TokenBalanceQueryRequest) -> Result<TokenBalanceQueryResponse, Status> {
@@ -33,3 +72,40 @@ impl Controller {
         })
     }
 }
+
+async fn get_l2_block_by_id(db_pool: &sqlx::Pool<DbType>, block_id: i64) -> Result<l2_block::L2Block, Status> {
+    let stmt = format!(
+        "select block_id, new_root, witness, created_time
+        from {}
+        where block_id = $1
+        order by created_time desc limit 1",
+        tablenames::L2_BLOCK,
+    );
+    match sqlx::query_as::<_, l2_block::L2Block>(&stmt)
+        .bind(block_id)
+        .fetch_one(db_pool)
+        .await
+    {
+        Ok(l2_block) => Ok(l2_block),
+        Err(sqlx::Error::RowNotFound) => Err(Status::new(Code::NotFound, "db l2_block record not found")),
+        Err(err) => {
+            println!("{:?}", err);
+            Err(Status::new(Code::Internal, "db table l2_block fetch error"))
+        }
+    }
+}
+
+async fn get_task_status_by_block_id(db_pool: &sqlx::Pool<DbType>, block_id: i64) -> Result<task::TaskStatus, Status> {
+    let stmt = format!(
+        "select status
+        from {}
+        where block_id = $1
+        order by created_time desc limit 1",
+        tablenames::TASK,
+    );
+    match sqlx::query_as(&stmt).bind(block_id).fetch_one(db_pool).await {
+        Ok((task_status,)) => Ok(task_status),
+        Err(sqlx::Error::RowNotFound) => Err(Status::new(Code::NotFound, "db task record not found")),
+        Err(_) => Err(Status::new(Code::Internal, "db table task fetch error")),
+    }
+}

src/grpc/controller.rs

@@ -9,15 +9,19 @@ pub struct Handler {
 }
 
 impl Handler {
-    pub fn new(state: Arc<RwLock<GlobalState>>) -> Self {
+    pub async fn new(state: Arc<RwLock<GlobalState>>) -> Self {
         Self {
-            controller: Controller::new(state),
+            controller: Controller::new(state).await,
         }
     }
 }
 
 #[tonic::async_trait]
 impl rollup_state_server::RollupState for Handler {
+    async fn l2_block_query(&self, request: Request<L2BlockQueryRequest>) -> Result<Response<L2BlockQueryResponse>, Status> {
+        Ok(Response::new(self.controller.l2_block_query(request.into_inner()).await?))
+    }
+
     async fn token_balance_query(&self, request: Request<TokenBalanceQueryRequest>) -> Result<Response<TokenBalanceQueryResponse>, Status> {
         Ok(Response::new(self.controller.token_balance_query(request.into_inner())?))
     }

src/grpc/handler.rs

@@ -22,7 +22,7 @@ pub fn run_grpc_server(addr: SocketAddr, state: Arc<RwLock<GlobalState>>) -> any
             tx.send(()).ok();
         });
 
-        let handler = Handler::new(state);
+        let handler = Handler::new(state).await;
 
         tonic::transport::Server::builder()
             .add_service(RollupStateServer::new(handler))