feat: add get mutation header (#474)

* feat: add get mutation header

* fix: improve the var name

* feat: add scan and get mutation

* feat: add scan mutation headers

Author: imotai

bridge/contracts/DB3Rollup.sol

@@ -8,6 +8,7 @@ contract DB3Rollup {
     // the locked balance of an address
     mapping(address => uint256) private _balances;
     event Deposit(address _from, uint256 amount);
+
     constructor(Db3Token tokenContract) {
         _tokenContract = tokenContract;
     }

format.sh

@@ -1,6 +1,7 @@
 #! /bin/bash
 
 cargo fmt
+npx buf format -w src/proto/proto
 cd bridge && npx prettier --write 'contracts/**/*.sol'
 
 

src/base/src/lib.rs

@@ -17,3 +17,4 @@
 
 pub mod bson_util;
 pub mod strings;
+pub mod times;

src/base/src/times.rs

@@ -0,0 +1,25 @@
+//
+// times.rs
+// Copyright (C) 2023 db3.network Author imotai <[email protected]>
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+use std::time::SystemTime;
+
+pub fn get_current_time_in_secs() -> u64 {
+    match SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) {
+        Ok(n) => n.as_secs(),
+        Err(_) => 0,
+    }
+}

src/crypto/src/id.rs

@@ -85,6 +85,19 @@ impl TxId {
         Self::try_from_bytes(base64ct::Base64::decode_vec(input).unwrap().as_slice())
     }
 
+    pub fn try_from_hex(input: &str) -> std::result::Result<Self, DB3Error> {
+        if input.starts_with("0x") {
+            let new_input = &input[2..];
+            let data = hex::decode(new_input)
+                .map_err(|e| DB3Error::KeyCodecError(format!("fail to decode tx id for {e}")))?;
+            Self::try_from_bytes(data.as_slice())
+        } else {
+            let data = hex::decode(input)
+                .map_err(|e| DB3Error::KeyCodecError(format!("fail to decode tx id for {e}")))?;
+            Self::try_from_bytes(data.as_slice())
+        }
+    }
+
     pub fn try_from_bytes(data: &[u8]) -> std::result::Result<Self, DB3Error> {
         let arr: [u8; TX_ID_LENGTH] = data.try_into().map_err(|_| DB3Error::InvalidAddress)?;
         Ok(Self { data: arr })

src/node/src/command.rs

@@ -413,6 +413,7 @@ impl DB3Command {
             block_store_cf_name: "block_store_cf".to_string(),
             tx_store_cf_name: "tx_store_cf".to_string(),
             message_max_buffer: 4 * 1024,
+            scan_max_limit: 50,
         };
         let state_config = StateStoreConfig {
             db_path: state_db_path.to_string(),

src/node/src/storage_node_light_impl.rs

@@ -17,14 +17,15 @@
 
 use crate::mutation_utils::MutationUtil;
 use db3_crypto::db3_address::DB3Address;
-use db3_crypto::id::DbId;
+use db3_crypto::id::{DbId, TxId};
 use db3_error::Result;
+use db3_proto::db3_mutation_v2_proto::{MutationAction, MutationRollupStatus};
 use db3_proto::db3_storage_proto::{
-    storage_node_server::StorageNode, ExtraItem, GetNonceRequest, GetNonceResponse,
-    SendMutationRequest, SendMutationResponse,
+    storage_node_server::StorageNode, ExtraItem, GetMutationBodyRequest, GetMutationBodyResponse,
+    GetMutationHeaderRequest, GetMutationHeaderResponse, GetNonceRequest, GetNonceResponse,
+    ScanMutationHeaderRequest, ScanMutationHeaderResponse, SendMutationRequest,
+    SendMutationResponse,
 };
-
-use db3_proto::db3_mutation_v2_proto::MutationAction;
 use db3_storage::mutation_store::{MutationStore, MutationStoreConfig};
 use db3_storage::state_store::{StateStore, StateStoreConfig};
 use tonic::{Request, Response, Status};
@@ -56,6 +57,47 @@ impl StorageNodeV2Impl {
 
 #[tonic::async_trait]
 impl StorageNode for StorageNodeV2Impl {
+    async fn get_mutation_body(
+        &self,
+        request: Request<GetMutationBodyRequest>,
+    ) -> std::result::Result<Response<GetMutationBodyResponse>, Status> {
+        let r = request.into_inner();
+        let tx_id =
+            TxId::try_from_hex(r.id.as_str()).map_err(|e| Status::internal(format!("{e}")))?;
+        let body = self
+            .storage
+            .get_mutation(&tx_id)
+            .map_err(|e| Status::internal(format!("{e}")))?;
+        Ok(Response::new(GetMutationBodyResponse { body }))
+    }
+
+    async fn scan_mutation_header(
+        &self,
+        request: Request<ScanMutationHeaderRequest>,
+    ) -> std::result::Result<Response<ScanMutationHeaderResponse>, Status> {
+        let r = request.into_inner();
+        let headers = self
+            .storage
+            .scan_mutation_headers(r.start, r.limit)
+            .map_err(|e| Status::internal(format!("{e}")))?;
+        Ok(Response::new(ScanMutationHeaderResponse { headers }))
+    }
+
+    async fn get_mutation_header(
+        &self,
+        request: Request<GetMutationHeaderRequest>,
+    ) -> std::result::Result<Response<GetMutationHeaderResponse>, Status> {
+        let r = request.into_inner();
+        let header = self
+            .storage
+            .get_mutation_header(r.block_id, r.order_id)
+            .map_err(|e| Status::internal(format!("{e}")))?;
+        Ok(Response::new(GetMutationHeaderResponse {
+            header,
+            status: MutationRollupStatus::Pending.into(),
+            rollup_tx: vec![],
+        }))
+    }
     async fn get_nonce(
         &self,
         request: Request<GetNonceRequest>,
@@ -88,9 +130,9 @@ impl StorageNode for StorageNodeV2Impl {
         match self.state_store.incr_nonce(&address, nonce) {
             Ok(_) => {
                 // mutation id
-                let id = self
+                let (id, block, order) = self
                     .storage
-                    .add_mutation(&r.payload, r.signature.as_bytes(), &address)
+                    .add_mutation(&r.payload, r.signature.as_str(), &address)
                     .map_err(|e| Status::internal(format!("{e}")))?;
                 match action {
                     MutationAction::CreateDocumentDb => {
@@ -100,18 +142,23 @@ impl StorageNode for StorageNodeV2Impl {
                             key: "db_addr".to_string(),
                             value: db_addr,
                         };
+
                         Ok(Response::new(SendMutationResponse {
                             id,
                             code: 0,
                             msg: "ok".to_string(),
                             items: vec![item],
+                            block,
+                            order,
                         }))
                     }
                     _ => Ok(Response::new(SendMutationResponse {
                         id,
                         code: 0,
                         msg: "ok".to_string(),
                         items: vec![],
+                        block,
+                        order,
                     })),
                 }
             }
@@ -120,6 +167,8 @@ impl StorageNode for StorageNodeV2Impl {
                 code: 1,
                 msg: "bad nonce".to_string(),
                 items: vec![],
+                block: 0,
+                order: 0,
             })),
         }
     }

src/proto/proto/db3_account.proto

@@ -22,12 +22,10 @@ package db3_account_proto;
 //
 //
 message Account {
-    uint64 bills = 1;
-    uint64 credits = 2;
-    uint64 total_storage_in_bytes = 3;
-    uint64 total_mutation_count = 4;
-    uint64 total_session_count = 5;
-    uint64 nonce = 6;
+  uint64 bills = 1;
+  uint64 credits = 2;
+  uint64 total_storage_in_bytes = 3;
+  uint64 total_mutation_count = 4;
+  uint64 total_session_count = 5;
+  uint64 nonce = 6;
 }
-
-

src/proto/proto/db3_account_v2.proto

@@ -15,12 +15,11 @@
 // limitations under the License.
 //
 
-
 syntax = "proto3";
 
 package db3_account_v2_proto;
 
 message Account {
-    uint64 nonce = 1;
-    uint64 total_storage_in_bytes = 2;
+  uint64 nonce = 1;
+  uint64 total_storage_in_bytes = 2;
 }

src/proto/proto/db3_base.proto

@@ -16,26 +16,25 @@
 // limitations under the License.
 //
 
-
 syntax = "proto3";
 package db3_base_proto;
 
 enum ChainRole {
-    SettlementChain = 0;
-    StorageShardChain = 10;
-    DvmComputingChain = 20;
+  SettlementChain = 0;
+  StorageShardChain = 10;
+  DvmComputingChain = 20;
 }
 
 enum ChainId {
-    MainNet = 0;
-    TestNet = 10;
-    DevNet = 20;
+  MainNet = 0;
+  TestNet = 10;
+  DevNet = 20;
 }
 
 message BroadcastMeta {
-    uint64 nonce = 1;
-    // the chain id of db3
-    ChainId chain_id = 2;
-    // the chain role of db3
-    ChainRole chain_role = 3;
+  uint64 nonce = 1;
+  // the chain id of db3
+  ChainId chain_id = 2;
+  // the chain role of db3
+  ChainRole chain_role = 3;
 }

src/proto/proto/db3_bill.proto

@@ -20,28 +20,28 @@ syntax = "proto3";
 package db3_bill_proto;
 
 enum BillType {
-    BillForMutation = 0;
-    BillForQuery = 1;
-    BillForMint = 2;
+  BillForMutation = 0;
+  BillForQuery = 1;
+  BillForMint = 2;
 }
 
 message Bill {
-    // uint256
-    uint64 gas_fee = 1;
-    // the block id 
-    uint64 block_id = 2;
-    // query session or mutation
-    BillType bill_type = 4;
-    // time on block created
-    uint64 time = 5;
-    // mutation id or query session id
-    bytes tx_id = 6;
-    // owner address
-    bytes owner = 8;
-    // to address
-    bytes to = 9;
+  // uint256
+  uint64 gas_fee = 1;
+  // the block id
+  uint64 block_id = 2;
+  // query session or mutation
+  BillType bill_type = 4;
+  // time on block created
+  uint64 time = 5;
+  // mutation id or query session id
+  bytes tx_id = 6;
+  // owner address
+  bytes owner = 8;
+  // to address
+  bytes to = 9;
 }
 
 message BillQueryRequest {
-    uint64 block_height = 1;
+  uint64 block_height = 1;
 }

src/proto/proto/db3_database.proto

@@ -23,19 +23,19 @@ package db3_database_proto;
 // the definition of database
 //
 message Database {
-    bytes address = 1;
-    // the owner of the Database
-    bytes sender = 2;
-    // the history of database modification
-    repeated bytes tx = 3;
-    repeated Collection collections = 4;
-    string desc =5;
+  bytes address = 1;
+  // the owner of the Database
+  bytes sender = 2;
+  // the history of database modification
+  repeated bytes tx = 3;
+  repeated Collection collections = 4;
+  string desc = 5;
 }
 
 message Collection {
-    bytes id = 1;
-    string name = 2;
-    repeated Index index_list = 3;
+  bytes id = 1;
+  string name = 2;
+  repeated Index index_list = 3;
 }
 
 message Document {
@@ -46,7 +46,6 @@ message Document {
 }
 
 message Index {
-
   // A field in an index.
   // The field_path describes which field is indexed, the value_mode describes
   // how the field value is indexed.
@@ -432,7 +431,4 @@ message StructuredQuery {
   // * The number of values cannot be greater than the number of fields
   //   specified in the `ORDER BY` clause.
   // Cursor end_at = 8;
-
-
 }
-