Repository: dbpunk-labs/db3 (language: rust)

Issue:
Light Store: support document ownership management
## Motivations

<!--
If your feature request is related to a problem, please describe it.
-->

- Would you like to implement this feature? [y/n]

## Solution

<!-- Describe the solution you'd like. -->

## Alternatives

<!-- Describe any alternative solutions or features you've considered. -->

## Additional context

<!-- Add any other context or screenshots about the feature request here. -->



Discussion / hints:

Fix patch:
diff --git a/src/crypto/src/db3_address.rs b/src/crypto/src/db3_address.rs
index d28ce025..1f44c6db 100644
--- a/src/crypto/src/db3_address.rs
+++ b/src/crypto/src/db3_address.rs
@@ -37,6 +37,7 @@ pub struct DB3Address(
 
 impl DB3Address {
     pub const ZERO: Self = Self([0u8; DB3_ADDRESS_LENGTH]);
+    pub const ONE: Self = Self([1u8; DB3_ADDRESS_LENGTH]);
 
     pub fn to_vec(&self) -> Vec<u8> {
         self.0.to_vec()
diff --git a/src/error/src/lib.rs b/src/error/src/lib.rs
index aa58b391..95b808d6 100644
--- a/src/error/src/lib.rs
+++ b/src/error/src/lib.rs
@@ -33,6 +33,8 @@ pub enum DB3Error {
     InvalidSignature(String),
     #[error("fail to codec key with error {0}")]
     KeyCodecError(String),
+    #[error("fail to verify the owner with error {0}")]
+    OwnerVerifyFailed(String),
     #[error("fail to apply mutation with error {0}")]
     ApplyMutationError(String),
     #[error("fail to submit mutation session with error {0}")]
@@ -71,8 +73,10 @@ pub enum DB3Error {
     QueryDatabaseError(String),
     #[error("database with addr {0} was not found")]
     DatabaseNotFound(String),
-    #[error("collection with name {0} was not found")]
-    CollectionNotFound(String),
+    #[error("collection with name {0} was not found in db {1}")]
+    CollectionNotFound(String, String),
+    #[error("collection {0} already exist in db {1}")]
+    CollectionAlreadyExist(String, String),
     #[error("the address does not match the public key")]
     InvalidSigner,
     #[error("fail to generate key for {0}")]
@@ -127,6 +131,8 @@ pub enum DB3Error {
     RollupError(String),
     #[error("invalid collection name for error {0}")]
     InvalidCollectionNameError(String),
+    #[error("invalid mutation for error {0}")]
+    InvalidMutationError(String),
 }
 
 pub type Result<T> = std::result::Result<T, DB3Error>;
diff --git a/src/node/src/indexer_impl.rs b/src/node/src/indexer_impl.rs
index 637b412c..e2b25771 100644
--- a/src/node/src/indexer_impl.rs
+++ b/src/node/src/indexer_impl.rs
@@ -74,7 +74,7 @@ impl IndexerNodeImpl {
     async fn handle_event(&self, event: EventMessageV2, store_sdk: &StoreSDKV2) -> Result<()> {
         match event.event {
             Some(event_message::Event::BlockEvent(be)) => {
-                info!(
+                debug!(
                     "Receive BlockEvent: Block\t{}\tMutationCount\t{}",
                     be.block_id, be.mutation_count,
                 );
@@ -164,17 +164,22 @@ impl IndexerNodeImpl {
                         let db_addr = DB3Address::try_from(db_address_ref)
                             .map_err(|e| DB3Error::WriteStoreError(format!("{e}")))?;
                         if let Some(Body::DocumentMutation(ref doc_mutation)) = &body.body {
-                            let mut docs = Vec::<(String, i64)>::new();
-                            for (j, buf) in doc_mutation.documents.iter().enumerate() {
+                            if doc_mutation.documents.len() != doc_mutation.ids.len() {
+                                let msg = format!(
+                                    "doc ids size {} not equal to documents size {}",
+                                    doc_mutation.ids.len(),
+                                    doc_mutation.documents.len()
+                                );
+                                warn!("{}", msg.as_str());
+                                return Err(DB3Error::InvalidMutationError(msg));
+                            }
+                            let mut docs = Vec::<String>::new();
+                            for buf in doc_mutation.documents.iter() {
                                 let document = bytes_to_bson_document(buf.clone())
                                     .map_err(|e| DB3Error::WriteStoreError(format!("{e}")))?;
                                 let doc_str = document.to_string();
                                 debug!("add document: {}", doc_str);
-                                if doc_mutation.ids.len() <= j {
-                                    warn!("no doc id for document {}", doc_str);
-                                    break;
-                                }
-                                docs.push((doc_str, doc_mutation.ids[j]));
+                                docs.push(doc_str);
                             }
                             self.db_store
                                 .update_docs(
@@ -182,6 +187,7 @@ impl IndexerNodeImpl {
                                     &address,
                                     doc_mutation.collection_name.as_str(),
                                     &docs,
+                                    &doc_mutation.ids,
                                 )
                                 .map_err(|e| DB3Error::WriteStoreError(format!("{e}")))?;
                             info!(
@@ -244,7 +250,8 @@ impl IndexerNodeImpl {
                                     "add documents with db_addr {}, collection_name: {}, from owner {}, ids: {:?}",
                                     db_addr.to_hex().as_str(),
                                     doc_mutation.collection_name.as_str(),
-                                    address.to_hex().as_str(), ids,
+                                    address.to_hex().as_str(),
+                                    ids
                                 );
                         }
                     }
diff --git a/src/node/src/storage_node_light_impl.rs b/src/node/src/storage_node_light_impl.rs
index 4a50b5b7..b0e7341f 100644
--- a/src/node/src/storage_node_light_impl.rs
+++ b/src/node/src/storage_node_light_impl.rs
@@ -41,6 +41,7 @@ use db3_proto::db3_storage_proto::{
     Subscription as SubscriptionV2,
 };
 
+use db3_base::bson_util::bytes_to_bson_document;
 use db3_storage::db_store_v2::{DBStoreV2, DBStoreV2Config};
 use db3_storage::mutation_store::{MutationStore, MutationStoreConfig};
 use db3_storage::state_store::{StateStore, StateStoreConfig};
@@ -56,7 +57,8 @@ use tokio::task;
 use tokio::time::{sleep, Duration as TokioDuration};
 use tokio_stream::wrappers::ReceiverStream;
 use tonic::{Request, Response, Status};
-use tracing::{info, warn};
+use tracing::{debug, info, warn};
+
 pub struct StorageNodeV2Config {
     pub store_config: MutationStoreConfig,
     pub state_config: StateStoreConfig,
@@ -114,7 +116,7 @@ impl StorageNodeV2Impl {
             info!("start the block producer thread");
             while local_running.load(Ordering::Relaxed) {
                 sleep(TokioDuration::from_millis(local_block_interval)).await;
-                info!(
+                debug!(
                     "produce block {}",
                     local_storage.get_current_block().unwrap_or(0)
                 );
@@ -131,7 +133,7 @@ impl StorageNodeV2Impl {
                         };
                         match local_event_sender.send(msg) {
                             Ok(_) => {
-                                info!("broadcast block event {}, {}", block_id, mutation_count);
+                                debug!("broadcast block event {}, {}", block_id, mutation_count);
                             }
                             Err(e) => {
                                 warn!("the broadcast channel error for {:?}", e);
@@ -203,7 +205,7 @@ impl StorageNodeV2Impl {
                             info!("subscribers len : {}", subscribers.len());
                         }
                         Ok(event) = event_sub.recv() => {
-                            info!("receive event {:?}", event);
+                            debug!("receive event {:?}", event);
                             for (key , (sender, sub)) in subscribers.iter() {
                                 if sender.is_closed() {
                                     to_be_removed.insert(key.clone());
@@ -216,7 +218,7 @@ impl StorageNodeV2Impl {
                                     }
                                     match sender.try_send(Ok(event.clone())) {
                                         Ok(_) => {
-                                            info!("send event to addr 0x{}", hex::encode(key.as_ref()));
+                                            debug!("send event to addr 0x{}", hex::encode(key.as_ref()));
                                             break;
                                         }
                                         Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
@@ -449,9 +451,9 @@ impl StorageNode for StorageNodeV2Impl {
                 // mutation id
                 let (id, block, order) = self
                     .storage
-                    .add_mutation(&r.payload, r.signature.as_str(), &address, nonce)
+                    .generate_mutation_block_and_order(&r.payload, r.signature.as_str())
                     .map_err(|e| Status::internal(format!("{e}")))?;
-                match action {
+                let response = match action {
                     MutationAction::CreateDocumentDb => {
                         let mut items: Vec<ExtraItem> = Vec::new();
                         for body in dm.bodies {
@@ -482,14 +484,14 @@ impl StorageNode for StorageNodeV2Impl {
                                 break;
                             }
                         }
-                        Ok(Response::new(SendMutationResponse {
+                        Response::new(SendMutationResponse {
                             id,
                             code: 0,
                             msg: "ok".to_string(),
                             items,
                             block,
                             order,
-                        }))
+                        })
                     }
                     MutationAction::AddCollection => {
                         let mut items: Vec<ExtraItem> = Vec::new();
@@ -521,24 +523,163 @@ impl StorageNode for StorageNodeV2Impl {
                                 items.push(item);
                             }
                         }
-                        Ok(Response::new(SendMutationResponse {
+                        Response::new(SendMutationResponse {
                             id,
                             code: 0,
                             msg: "ok".to_string(),
                             items,
                             block,
                             order,
-                        }))
+                        })
+                    }
+                    MutationAction::AddDocument => {
+                        let mut items: Vec<ExtraItem> = Vec::new();
+                        for (_i, body) in dm.bodies.iter().enumerate() {
+                            let db_address_ref: &[u8] = body.db_address.as_ref();
+                            let db_addr = DB3Address::try_from(db_address_ref)
+                                .map_err(|e| Status::internal(format!("{e}")))?;
+                            if let Some(Body::DocumentMutation(ref doc_mutation)) = &body.body {
+                                let mut docs = Vec::<String>::new();
+                                for buf in doc_mutation.documents.iter() {
+                                    let document = bytes_to_bson_document(buf.clone())
+                                        .map_err(|e| Status::internal(format!("{e}")))?;
+                                    docs.push(document.to_string());
+                                }
+                                let ids = self
+                                    .db_store
+                                    .add_docs(
+                                        &db_addr,
+                                        &address,
+                                        doc_mutation.collection_name.as_str(),
+                                        &docs,
+                                    )
+                                    .map_err(|e| Status::internal(format!("{e}")))?;
+                                info!(
+                                    "add documents with db_addr {}, collection_name: {}, from owner {}, document size: {}",
+                                    db_addr.to_hex().as_str(),
+                                    doc_mutation.collection_name.as_str(),
+                                    address.to_hex().as_str(),
+                                    ids.len()
+                                );
+                                // return document keys
+                                for id in ids {
+                                    let item = ExtraItem {
+                                        key: "document".to_string(),
+                                        value: id.to_string(),
+                                    };
+                                    items.push(item);
+                                }
+                            }
+                        }
+                        Response::new(SendMutationResponse {
+                            id,
+                            code: 0,
+                            msg: "ok".to_string(),
+                            items,
+                            block,
+                            order,
+                        })
+                    }
+                    MutationAction::UpdateDocument => {
+                        for (_i, body) in dm.bodies.iter().enumerate() {
+                            let db_address_ref: &[u8] = body.db_address.as_ref();
+                            let db_addr = DB3Address::try_from(db_address_ref)
+                                .map_err(|e| Status::internal(format!("{e}")))?;
+                            if let Some(Body::DocumentMutation(ref doc_mutation)) = &body.body {
+                                if doc_mutation.documents.len() != doc_mutation.ids.len() {
+                                    let msg = format!(
+                                        "doc ids size {} not equal to documents size {}",
+                                        doc_mutation.ids.len(),
+                                        doc_mutation.documents.len()
+                                    );
+                                    warn!("{}", msg.as_str());
+                                    return Err(Status::internal(msg));
+                                }
+                                let mut docs = Vec::<String>::new();
+                                for buf in doc_mutation.documents.iter() {
+                                    let document = bytes_to_bson_document(buf.clone())
+                                        .map_err(|e| Status::internal(format!("{e}")))?;
+                                    let doc_str = document.to_string();
+                                    debug!("update document: {}", doc_str);
+                                    docs.push(doc_str);
+                                }
+                                self.db_store
+                                    .update_docs(
+                                        &db_addr,
+                                        &address,
+                                        doc_mutation.collection_name.as_str(),
+                                        &docs,
+                                        &doc_mutation.ids,
+                                    )
+                                    .map_err(|e| Status::internal(format!("{e}")))?;
+                                info!(
+                                    "update documents with db_addr {}, collection_name: {}, from owner {}",
+                                    db_addr.to_hex().as_str(),
+                                    doc_mutation.collection_name.as_str(),
+                                    address.to_hex().as_str()
+                                );
+                            }
+                        }
+                        Response::new(SendMutationResponse {
+                            id,
+                            code: 0,
+                            msg: "ok".to_string(),
+                            items: vec![],
+                            block,
+                            order,
+                        })
                     }
-                    _ => Ok(Response::new(SendMutationResponse {
+                    MutationAction::DeleteDocument => {
+                        for (_i, body) in dm.bodies.iter().enumerate() {
+                            let db_address_ref: &[u8] = body.db_address.as_ref();
+                            let db_addr = DB3Address::try_from(db_address_ref)
+                                .map_err(|e| Status::internal(format!("{e}")))?;
+                            if let Some(Body::DocumentMutation(ref doc_mutation)) = &body.body {
+                                self.db_store
+                                    .delete_docs(
+                                        &db_addr,
+                                        &address,
+                                        doc_mutation.collection_name.as_str(),
+                                        &doc_mutation.ids,
+                                    )
+                                    .map_err(|e| Status::internal(format!("{e}")))?;
+                                info!(
+                                    "delete documents with db_addr {}, collection_name: {}, from owner {}",
+                                    db_addr.to_hex().as_str(),
+                                    doc_mutation.collection_name.as_str(),
+                                    address.to_hex().as_str()
+                                );
+                            }
+                        }
+                        Response::new(SendMutationResponse {
+                            id,
+                            code: 0,
+                            msg: "ok".to_string(),
+                            items: vec![],
+                            block,
+                            order,
+                        })
+                    }
+                    _ => Response::new(SendMutationResponse {
                         id,
                         code: 0,
                         msg: "ok".to_string(),
                         items: vec![],
                         block,
                         order,
-                    })),
-                }
+                    }),
+                };
+                self.storage
+                    .add_mutation(
+                        &r.payload,
+                        r.signature.as_str(),
+                        &address,
+                        nonce,
+                        block,
+                        order,
+                    )
+                    .map_err(|e| Status::internal(format!("{e}")))?;
+                Ok(response)
             }
             Err(_e) => Ok(Response::new(SendMutationResponse {
                 id: "".to_string(),
diff --git a/src/proto/proto/db3_database_v2.proto b/src/proto/proto/db3_database_v2.proto
index ad89574f..4e512dbd 100644
--- a/src/proto/proto/db3_database_v2.proto
+++ b/src/proto/proto/db3_database_v2.proto
@@ -73,7 +73,6 @@ message Index {
 message Document {
   int64 id = 1;
   string doc = 2;
-  string owner = 3;
 }
 
 message QueryParameter {
diff --git a/src/storage/src/db_doc_key_v2.rs b/src/storage/src/db_doc_key_v2.rs
new file mode 100644
index 00000000..2044371e
--- /dev/null
+++ b/src/storage/src/db_doc_key_v2.rs
@@ -0,0 +1,59 @@
+//
+// collection_key.rs
+// Copyright (C) 2023 db3.network Author imotai <codego.me@gmail.com>
+//
+// 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 db3_crypto::db3_address::DB3Address;
+use db3_error::Result;
+use std::fmt;
+const DOC_PREFIX: &str = "/doc/";
+/// DBDocKey with db address, doc id
+pub struct DbDocKeyV2<'a>(pub &'a DB3Address, pub i64);
+impl<'a> DbDocKeyV2<'a> {
+    ///
+    /// encode the database key
+    ///
+    pub fn encode(&self) -> Result<Vec<u8>> {
+        let mut encoded_key = DOC_PREFIX.as_bytes().to_vec();
+        encoded_key.extend_from_slice(self.0.as_ref());
+        encoded_key.extend_from_slice(self.1.to_be_bytes().as_ref());
+        Ok(encoded_key)
+    }
+}
+
+impl fmt::Display for DbDocKeyV2<'_> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "{}/{}/{}", DOC_PREFIX, self.0.to_hex(), self.1,)
+    }
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use db3_crypto::db3_address::DB3_ADDRESS_LENGTH;
+    #[test]
+    fn it_works() {}
+
+    #[test]
+    fn test_db_doc_key_v2() {
+        let db_addr = DB3Address::ZERO;
+        let doc_id = 1;
+        let db_doc_key = DbDocKeyV2(&db_addr, doc_id);
+        let encoded_key = db_doc_key.encode().unwrap();
+        assert_eq!(encoded_key.len(), DOC_PREFIX.len() + DB3_ADDRESS_LENGTH + 8);
+        assert_eq!(
+            "/doc//0x0000000000000000000000000000000000000000/1",
+            db_doc_key.to_string()
+        );
+    }
+}
diff --git a/src/storage/src/db_store.rs b/src/storage/src/db_store.rs
index c2e80ce9..2092bdc1 100644
--- a/src/storage/src/db_store.rs
+++ b/src/storage/src/db_store.rs
@@ -352,6 +352,7 @@ impl DbStore {
                     } else {
                         return Err(DB3Error::CollectionNotFound(
                             document_mutation.collection_name.to_string(),
+                            db_id.to_hex(),
                         ));
                     }
                 }
diff --git a/src/storage/src/db_store_v2.rs b/src/storage/src/db_store_v2.rs
index 25b6f946..d914357d 100644
--- a/src/storage/src/db_store_v2.rs
+++ b/src/storage/src/db_store_v2.rs
@@ -1,3 +1,4 @@
+use std::collections::BTreeMap;
 //
 // ns_store.rs
 // Copyright (C) 2022 db3.network Author imotai <codego.me@gmail.com>
@@ -21,6 +22,7 @@ use db3_crypto::id::DbId;
 use db3_crypto::id_v2::OpEntryId;
 
 use crate::collection_key;
+use crate::db_doc_key_v2::DbDocKeyV2;
 use crate::doc_store::{DocStore, DocStoreConfig};
 use db3_error::{DB3Error, Result};
 use db3_proto::db3_database_v2_proto::{
@@ -30,7 +32,7 @@ use db3_proto::db3_mutation_v2_proto::{CollectionMutation, DocumentDatabaseMutat
 use prost::Message;
 use rocksdb::{DBRawIteratorWithThreadMode, DBWithThreadMode, MultiThreaded, Options, WriteBatch};
 use std::path::Path;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use tracing::info;
 
 type StorageEngine = DBWithThreadMode<MultiThreaded>;
@@ -49,12 +51,15 @@ pub struct DBStoreV2Config {
     pub enable_doc_store: bool,
     pub doc_store_conf: DocStoreConfig,
 }
-
+struct DBState {
+    pub db_doc_order: BTreeMap<String, i64>,
+}
 #[derive(Clone)]
 pub struct DBStoreV2 {
     config: DBStoreV2Config,
     se: Arc<StorageEngine>,
     doc_store: Arc<DocStore>,
+    db_state: Arc<Mutex<DBState>>,
 }
 
 impl DBStoreV2 {
@@ -90,9 +95,28 @@ impl DBStoreV2 {
             config,
             se,
             doc_store,
+            db_state: Arc::new(Mutex::new(DBState {
+                db_doc_order: BTreeMap::new(),
+            })),
         })
     }
 
+    /// increase db doc order
+    fn increase_db_doc_order(&self, db_addr_hex: &String) -> Result<i64> {
+        match self.db_state.lock() {
+            Ok(mut state) => match state.db_doc_order.get_mut(db_addr_hex) {
+                Some(order) => {
+                    *order += 1;
+                    Ok(*order)
+                }
+                None => {
+                    state.db_doc_order.insert(db_addr_hex.clone(), 1);
+                    Ok(1)
+                }
+            },
+            Err(e) => Err(DB3Error::WriteStoreError(format!("{e}"))),
+        }
+    }
     pub fn get_collection_of_database(&self, db_addr: &DB3Address) -> Result<Vec<Collection>> {
         self.get_entries_with_prefix::<Collection>(
             db_addr.as_ref(),
@@ -205,7 +229,12 @@ impl DBStoreV2 {
             ));
         }
 
-        //TODO add mutex control
+        if self.is_db_collection_exist(db_addr, collection.collection_name.as_str())? {
+            return Err(DB3Error::ReadStoreError(
+                "collection with name exist".to_string(),
+            ));
+        }
+
         let ck = collection_key::build_collection_key(db_addr, collection.collection_name.as_str())
             .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
 
@@ -262,30 +291,21 @@ impl DBStoreV2 {
     pub fn update_docs(
         &self,
         db_addr: &DB3Address,
-        _sender: &DB3Address,
+        sender: &DB3Address,
         col_name: &str,
-        docs: &[(String, i64)],
+        docs: &Vec<String>,
+        doc_ids: &Vec<i64>,
     ) -> Result<()> {
-        let ck = collection_key::build_collection_key(db_addr, col_name)
-            .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
-        let collection_store_cf_handle = self
-            .se
-            .cf_handle(self.config.collection_store_cf_name.as_str())
-            .ok_or(DB3Error::ReadStoreError("cf is not found".to_string()))?;
-        let ck_ref: &[u8] = ck.as_ref();
-        let value = self
-            .se
-            .get_cf(&collection_store_cf_handle, ck_ref)
-            .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
-        if let None = value {
-            return Err(DB3Error::ReadStoreError(format!(
-                "collection with name {} does not exist",
-                col_name
-            )));
+        if !self.is_db_collection_exist(db_addr, col_name)? {
+            return Err(DB3Error::CollectionNotFound(
+                col_name.to_string(),
+                db_addr.to_hex(),
+            ));
         }
+        self.verify_docs_ownership(sender, db_addr, doc_ids)?;
         if self.config.enable_doc_store {
             //TODO add id-> owner mapping to control the permissions
-            self.doc_store.patch_docs(db_addr, col_name, docs)
+            self.doc_store.patch_docs(db_addr, col_name, docs, &doc_ids)
         } else {
             Ok(())
         }
@@ -315,16 +335,16 @@ impl DBStoreV2 {
             )));
         }
         if self.config.enable_doc_store {
-            let docs = self.doc_store.execute_query(db_addr, col_name, query)?;
-            //TODO add owner
-            Ok(docs
-                .iter()
-                .map(|doc| Document {
-                    id: doc.0,
-                    doc: doc.1.to_string(),
-                    owner: "".to_string(),
+            let result = self.doc_store.execute_query(db_addr, col_name, query)?;
+            let mut documents = vec![];
+
+            for (id, doc) in result {
+                documents.push(Document {
+                    id,
+                    doc: doc.to_string(),
                 })
-                .collect())
+            }
+            Ok(documents)
         } else {
             Ok(vec![])
         }
@@ -333,30 +353,20 @@ impl DBStoreV2 {
     pub fn delete_docs(
         &self,
         db_addr: &DB3Address,
-        _sender: &DB3Address,
+        sender: &DB3Address,
         col_name: &str,
-        ids: &[i64],
+        doc_ids: &Vec<i64>,
     ) -> Result<()> {
-        let ck = collection_key::build_collection_key(db_addr, col_name)
-            .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
-        let collection_store_cf_handle = self
-            .se
-            .cf_handle(self.config.collection_store_cf_name.as_str())
-            .ok_or(DB3Error::ReadStoreError("cf is not found".to_string()))?;
-        let ck_ref: &[u8] = ck.as_ref();
-        let value = self
-            .se
-            .get_cf(&collection_store_cf_handle, ck_ref)
-            .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
-        if let None = value {
-            return Err(DB3Error::ReadStoreError(format!(
-                "collection with name {} does not exist",
-                col_name
-            )));
+        if !self.is_db_collection_exist(db_addr, col_name)? {
+            return Err(DB3Error::CollectionNotFound(
+                col_name.to_string(),
+                db_addr.to_hex(),
+            ));
         }
+        self.verify_docs_ownership(sender, db_addr, doc_ids)?;
         if self.config.enable_doc_store {
             //TODO add id-> owner mapping to control the permissions
-            self.doc_store.delete_docs(db_addr, col_name, ids)
+            self.doc_store.delete_docs(db_addr, col_name, doc_ids)
         } else {
             Ok(())
         }
@@ -365,7 +375,7 @@ impl DBStoreV2 {
     pub fn add_docs(
         &self,
         db_addr: &DB3Address,
-        _sender: &DB3Address,
+        sender: &DB3Address,
         col_name: &str,
         docs: &Vec<String>,
     ) -> Result<Vec<i64>> {
@@ -386,14 +396,115 @@ impl DBStoreV2 {
                 col_name
             )));
         }
+        let db_addr_hex = db_addr.to_hex();
+        let mut doc_ids = vec![];
+        for _ in 0..docs.len() {
+            let doc_id = self.increase_db_doc_order(&db_addr_hex)?;
+            doc_ids.push(doc_id);
+        }
+
+        // add db+id-> owner mapping to control the permissions
+        self.create_doc_ownership(sender, db_addr, &doc_ids)?;
         if self.config.enable_doc_store {
-            //TODO add id-> owner mapping to control the permissions
-            self.doc_store.add_str_docs(db_addr, col_name, docs)
-        } else {
-            Ok(vec![])
+            self.doc_store
+                .add_str_docs(db_addr, col_name, docs, &doc_ids)?;
+        }
+        Ok(doc_ids)
+    }
+
+    /// verify if the collection exists in the given db
+    pub fn is_db_collection_exist(&self, db_addr: &DB3Address, col_name: &str) -> Result<bool> {
+        let ck = collection_key::build_collection_key(db_addr, col_name)
+            .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
+        let collection_store_cf_handle = self
+            .se
+            .cf_handle(self.config.collection_store_cf_name.as_str())
+            .ok_or(DB3Error::ReadStoreError("cf is not found".to_string()))?;
+        let ck_ref: &[u8] = ck.as_ref();
+        let value = self
+            .se
+            .get_cf(&collection_store_cf_handle, ck_ref)
+            .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
+        Ok(value.is_some())
+    }
+
+    /// verify the ownership of the doc ids
+    pub fn verify_docs_ownership(
+        &self,
+        sender: &DB3Address,
+        db_addr: &DB3Address,
+        doc_ids: &Vec<i64>,
+    ) -> Result<()> {
+        let doc_owner_store_cf_handle = self
+            .se
+            .cf_handle(self.config.doc_owner_store_cf_name.as_str())
+            .ok_or(DB3Error::ReadStoreError("cf is not found".to_string()))?;
+        for id in doc_ids {
+            let db_doc_key = DbDocKeyV2(db_addr, *id).encode().unwrap();
+            let value = self
+                .se
+                .get_cf(&doc_owner_store_cf_handle, db_doc_key)
+                .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
+            if let Some(owner) = value {
+                if owner != sender.as_ref() {
+                    return Err(DB3Error::OwnerVerifyFailed(format!(
+                        "doc owner is not the sender"
+                    )));
+                }
+            } else {
+                return Err(DB3Error::OwnerVerifyFailed(format!(
+                    "doc owner key not found"
+                )));
+            }
         }
+        Ok(())
     }
 
+    pub fn get_doc_key_from_doc_id(&self, doc_id: i64) -> Result<Vec<u8>> {
+        let doc_owner_store_cf_handle = self
+            .se
+            .cf_handle(self.config.doc_owner_store_cf_name.as_str())
+            .ok_or(DB3Error::ReadStoreError("cf is not found".to_string()))?;
+        let value = self
+            .se
+            .get_cf(&doc_owner_store_cf_handle, doc_id.to_be_bytes().as_ref())
+            .map_err(|e| DB3Error::ReadStoreError(format!("{e}")))?;
+        match value {
+            Some(v) => Ok(v),
+            None => {
+                return Err(DB3Error::ReadStoreError(format!(
+                    "doc owner key not found for doc id {}",
+                    doc_id
+                )))
+            }
+        }
+    }
+    /// create db+id-> owner mapping to control the permissions
+    pub fn create_doc_ownership(
+        &self,
+        sender: &DB3Address,
+        db_addr: &DB3Address,
+        doc_ids: &Vec<i64>,
+    ) -> Result<()> {
+        let doc_owner_store_cf_handle = self
+            .se
+            .cf_handle(self.config.doc_owner_store_cf_name.as_str())
+            .ok_or(DB3Error::ReadStoreError("cf is not found".to_string()))?;
+        let mut batch = WriteBatch::default();
+        for id in doc_ids {
+            let db_doc_key = DbDocKeyV2(db_addr, *id);
+            let encoded_db_doc_key = db_doc_key.encode()?;
+            batch.put_cf(
+                &doc_owner_store_cf_handle,
+                &encoded_db_doc_key,
+                sender.as_ref(),
+            );
+        }
+        self.se
+            .write(batch)
+            .map_err(|e| DB3Error::WriteStoreError(format!("{e}")))?;
+        Ok(())
+    }
     pub fn create_doc_database(
         &self,
         sender: &DB3Address,
@@ -561,4 +672,47 @@ mod tests {
             assert!(false);
         }
     }
+
+    #[test]
+    fn test_increase_db_doc_order_ut() {
+        let tmp_dir_path = TempDir::new("new_database").expect("create temp dir");
+        let real_path = tmp_dir_path.path().to_str().unwrap().to_string();
+        let config = DBStoreV2Config {
+            db_path: real_path,
+            db_store_cf_name: "db".to_string(),
+            doc_store_cf_name: "doc".to_string(),
+            collection_store_cf_name: "cf2".to_string(),
+            index_store_cf_name: "index".to_string(),
+            doc_owner_store_cf_name: "doc_owner".to_string(),
+            db_owner_store_cf_name: "db_owner".to_string(),
+            scan_max_limit: 50,
+            enable_doc_store: false,
+            doc_store_conf: DocStoreConfig::default(),
+        };
+        let result = DBStoreV2::new(config);
+        assert_eq!(result.is_ok(), true);
+        let db_m = DocumentDatabaseMutation {
+            db_desc: "test_desc".to_string(),
+        };
+        let db3_store = result.unwrap();
+        let result = db3_store.create_doc_database(&DB3Address::ZERO, &db_m, 1, 1, 1, 1);
+        assert!(result.is_ok());
+        let db_id_1 = result.unwrap();
+        let result = db3_store.create_doc_database(&DB3Address::ZERO, &db_m, 2, 1, 2, 1);
+        assert!(result.is_ok());
+        let db_id_2 = result.unwrap();
+
+        let result = db3_store
+            .increase_db_doc_order(&db_id_1.address().to_hex())
+            .unwrap();
+        assert_eq!(result, 1);
+        let result = db3_store
+            .increase_db_doc_order(&db_id_1.address().to_hex())
+            .unwrap();
+        assert_eq!(result, 2);
+        let result = db3_store
+            .increase_db_doc_order(&db_id_2.address().to_hex())
+            .unwrap();
+        assert_eq!(result, 1);
+    }
 }
diff --git a/src/storage/src/doc_key_v2.rs b/src/storage/src/doc_key_v2.rs
new file mode 100644
index 00000000..65657ceb
--- /dev/null
+++ b/src/storage/src/doc_key_v2.rs
@@ -0,0 +1,165 @@
+//
+// collection_key.rs
+// Copyright (C) 2023 db3.network Author imotai <codego.me@gmail.com>
+//
+// 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 db3_crypto::db3_address::{DB3Address, DB3_ADDRESS_LENGTH};
+use db3_error::{DB3Error, Result};
+use std::fmt;
+const DOC: &str = "/doc";
+/// DocOwnerKey with owner address, block id, order id, entry id
+pub struct DocOwnerKeyV2(pub DB3Address, pub u64, pub u32, pub u32);
+
+impl DocOwnerKeyV2 {
+    ///
+    /// encode the document owner key
+    ///
+    pub fn encode(&self) -> Vec<u8> {
+        let mut encoded_key = DOC.as_bytes().to_vec();
+        encoded_key.extend_from_slice(self.0.as_ref());
+        encoded_key.extend_from_slice(self.1.to_be_bytes().as_ref());
+        encoded_key.extend_from_slice(self.2.to_be_bytes().as_ref());
+        encoded_key.extend_from_slice(self.3.to_be_bytes().as_ref());
+        encoded_key
+    }
+
+    ///
+    /// decode the document owner key
+    ///
+    #[allow(dead_code)]
+    pub fn decode(data: &[u8]) -> Result<Self> {
+        const MIN_KEY_TOTAL_LEN: usize = DB3_ADDRESS_LENGTH + DOC.len();
+        if data.len() < MIN_KEY_TOTAL_LEN {
+            return Err(DB3Error::KeyCodecError(
+                "the length of data is invalid".to_string(),
+            ));
+        }
+        let address_offset = DOC.len();
+        let data_slice: &[u8; DB3_ADDRESS_LENGTH] = &data
+            [address_offset..address_offset + DB3_ADDRESS_LENGTH]
+            .try_into()
+            .map_err(|e| DB3Error::KeyCodecError(format!("{e}")))?;
+        let owner = DB3Address::from(data_slice);
+        let block_id_offset = address_offset + DB3_ADDRESS_LENGTH;
+        let block_id_slice: &[u8; 8] = &data[block_id_offset..block_id_offset + 8]
+            .try_into()
+            .map_err(|e| DB3Error::KeyCodecError(format!("{e}")))?;
+        let block_id = u64::from_be_bytes(*block_id_slice);
+
+        let order_id_offset = block_id_offset + 8;
+        let order_id_slice: &[u8; 4] = &data[order_id_offset..order_id_offset + 4]
+            .try_into()
+            .map_err(|e| DB3Error::KeyCodecError(format!("{e}")))?;
+        let order_id = u32::from_be_bytes(*order_id_slice);
+
+        let entry_id_offset = order_id_offset + 4;
+        let entry_id_slice: &[u8; 4] = &data[entry_id_offset..entry_id_offset + 4]
+            .try_into()
+            .map_err(|e| DB3Error::KeyCodecError(format!("{e}")))?;
+        let entry_id = u32::from_be_bytes(*entry_id_slice);
+        Ok(Self(owner, block_id, order_id, entry_id))
+    }
+    /// decode the document key from string
+    pub fn from_str(doc_key_str: &str) -> Result<Self> {
+        let tokens: Vec<_> = doc_key_str.split("/").collect();
+        if tokens.len() != 6 {
+            return Err(DB3Error::KeyCodecError(
+                "the length of key is invalid".to_string(),
+            ));
+        }
+        let (_, prefix, owner_hex, block_id_str, order_id_str, entry_id_str) = (
+            tokens[0], tokens[1], tokens[2], tokens[3], tokens[4], tokens[5],
+        );
+        if prefix != "doc" {
+            return Err(DB3Error::KeyCodecError(
+                "the prefix of key is invalid".to_string(),
+            ));
+        }
+        let owner = DB3Address::from_hex(owner_hex)?;
+        let block_id: u64 = block_id_str
+            .parse()
+            .map_err(|e| DB3Error::KeyCodecError(format!("{e}")))?;
+        let order_id: u32 = order_id_str
+            .parse()
+            .map_err(|e| DB3Error::KeyCodecError(format!("{e}")))?;
+        let entry_id: u32 = entry_id_str
+            .parse()
+            .map_err(|e| DB3Error::KeyCodecError(format!("{e}")))?;
+        Ok(Self(owner, block_id, order_id, entry_id))
+    }
+
+    pub fn verify_owner(&self, owner: &DB3Address) -> Result<()> {
+        if owner.as_ref() == self.0.as_ref() {
+            Ok(())
+        } else {
+            Err(DB3Error::OwnerVerifyFailed(
+                "the owner is invalid".to_string(),
+            ))
+        }
+    }
+}
+
+impl fmt::Display for DocOwnerKeyV2 {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(
+            f,
+            "{}/{}/{}/{}/{}",
+            DOC,
+            self.0.to_hex(),
+            self.1,
+            self.2,
+            self.3
+        )
+    }
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn it_works() {}
+
+    #[test]
+    fn encode_and_decode_ut() {
+        let key = DocOwnerKeyV2(DB3Address::default(), 1, 2, 3);
+        let encoded = key.encode();
+        let decoded = DocOwnerKeyV2::decode(&encoded).unwrap();
+        assert_eq!(key.0.to_hex(), decoded.0.to_hex());
+        assert_eq!(key.1, decoded.1);
+        assert_eq!(key.2, decoded.2);
+        assert_eq!(key.3, decoded.3);
+    }
+
+    #[test]
+    fn verify_owner_ut() {
+        let key = DocOwnerKeyV2(DB3Address::default(), 1, 2, 3);
+        let owner = DB3Address::default();
+        let result = key.verify_owner(&owner);
+        assert!(result.is_ok());
+    }
+
+    #[test]
+    fn from_str_ut() {
+        let key = DocOwnerKeyV2(DB3Address::default(), 1, 2, 3);
+        let key_str = key.to_string();
+        assert_eq!(
+            "/doc/0x0000000000000000000000000000000000000000/1/2/3",
+            key_str
+        );
+        let decoded = DocOwnerKeyV2::from_str(&key_str).unwrap();
+        assert_eq!(key.0.to_hex(), decoded.0.to_hex());
+        assert_eq!(key.1, decoded.1);
+        assert_eq!(key.2, decoded.2);
+        assert_eq!(key.3, decoded.3);
+    }
+}
diff --git a/src/storage/src/doc_store.rs b/src/storage/src/doc_store.rs
index 4c4ae40b..d3fca041 100644
--- a/src/storage/src/doc_store.rs
+++ b/src/storage/src/doc_store.rs
@@ -150,7 +150,7 @@ impl DocStore {
         }
     }
 
-    pub fn delete_docs(&self, db_addr: &DB3Address, col_name: &str, ids: &[i64]) -> Result<()> {
+    pub fn delete_docs(&self, db_addr: &DB3Address, col_name: &str, ids: &Vec<i64>) -> Result<()> {
         let db_opt = self.get_db_ref(db_addr);
         if let Some(db) = db_opt {
             for id in ids {
@@ -266,12 +266,13 @@ impl DocStore {
         &self,
         db_addr: &DB3Address,
         col_name: &str,
-        pairs: &[(String, i64)],
+        docs: &[String],
+        doc_ids: &[i64],
     ) -> Result<()> {
         let db_opt = self.get_db_ref(db_addr);
         if let Some(db) = db_opt {
-            for pair in pairs {
-                db.patch(col_name, &pair.0.as_str(), pair.1)
+            for idx in 0..docs.len() {
+                db.patch(col_name, &docs[idx].as_str(), doc_ids[idx])
                     .map_err(|e| DB3Error::WriteStoreError(format!("{e}")))?;
             }
             Ok(())
@@ -306,17 +307,15 @@ impl DocStore {
         db_addr: &DB3Address,
         col_name: &str,
         docs: &Vec<String>,
-    ) -> Result<Vec<i64>> {
+        ids: &Vec<i64>,
+    ) -> Result<()> {
         let db_opt = self.get_db_ref(db_addr);
         if let Some(db) = db_opt {
-            let mut ids = Vec::new();
-            for doc in docs {
-                let id = db
-                    .put_new(col_name, doc)
+            for (i, doc) in docs.iter().enumerate() {
+                db.put(col_name, doc, ids[i])
                     .map_err(|e| DB3Error::WriteStoreError(format!("{e}")))?;
-                ids.push(id);
             }
-            Ok(ids)
+            Ok(())
         } else {
             Err(DB3Error::WriteStoreError(format!(
                 "no database found with addr {}",
diff --git a/src/storage/src/lib.rs b/src/storage/src/lib.rs
index 80c6512e..b4b5508b 100644
--- a/src/storage/src/lib.rs
+++ b/src/storage/src/lib.rs
@@ -21,11 +21,13 @@ pub mod bill_store;
 pub mod collection_key;
 pub mod commit_store;
 mod db3_document;
+pub mod db_doc_key_v2;
 mod db_key;
 pub mod db_owner_key;
 pub mod db_owner_key_v2;
 pub mod db_store;
 pub mod db_store_v2;
+pub mod doc_key_v2;
 pub mod doc_store;
 pub mod key;
 pub mod mutation_store;
diff --git a/src/storage/src/mutation_store.rs b/src/storage/src/mutation_store.rs
index ce47170d..af2f4f08 100644
--- a/src/storage/src/mutation_store.rs
+++ b/src/storage/src/mutation_store.rs
@@ -26,7 +26,7 @@ use prost::Message;
 use rocksdb::{DBWithThreadMode, MultiThreaded, Options, WriteBatch};
 use std::path::Path;
 use std::sync::{Arc, Mutex};
-use tracing::{debug, info};
+use tracing::info;
 
 type StorageEngine = DBWithThreadMode<MultiThreaded>;
 
@@ -457,17 +457,28 @@ impl MutationStore {
         self.get_record::<MutationBody>(self.config.tx_store_cf_name.as_str(), tx_id.as_ref())
     }
 
+    pub fn generate_mutation_block_and_order(
+        &self,
+        payload: &[u8],
+        signature: &str,
+    ) -> Result<(String, u64, u32)> {
+        let tx_id = TxId::from((payload, signature.as_bytes()));
+        let hex_id = tx_id.to_hex();
+        let (block, order) = self.increase_order()?;
+        Ok((hex_id, block, order))
+    }
+
     pub fn add_mutation(
         &self,
         payload: &[u8],
         signature: &str,
         sender: &DB3Address,
         nonce: u64,
+        block: u64,
+        order: u32,
     ) -> Result<(String, u64, u32)> {
         let tx_id = TxId::from((payload, signature.as_bytes()));
         let hex_id = tx_id.to_hex();
-        debug!("the tx id is {}", hex_id);
-        let (block, order) = self.increase_order()?;
         let mut encoded_id: Vec<u8> = Vec::new();
         encoded_id.extend_from_slice(&block.to_be_bytes());
         encoded_id.extend_from_slice(&order.to_be_bytes());
@@ -556,14 +567,34 @@ mod tests {
         if let Ok(store) = result {
             let payload: Vec<u8> = vec![1];
             let signature: &str = "0xasdasdsad";
-            let result = store.add_mutation(payload.as_ref(), signature, &DB3Address::ZERO, 1);
+            let (_id, block, order) = store
+                .generate_mutation_block_and_order(payload.as_ref(), signature)
+                .unwrap();
+            let result = store.add_mutation(
+                payload.as_ref(),
+                signature,
+                &DB3Address::ZERO,
+                1,
+                block,
+                order,
+            );
             assert!(result.is_ok());
             if let Ok(headers) = store.scan_mutation_headers(0, 1) {
                 assert_eq!(1, headers.len());
             } else {
                 assert!(false);
             }
-            let result = store.add_mutation(payload.as_ref(), signature, &DB3Address::ZERO, 1);
+            let (_id, block, order) = store
+                .generate_mutation_block_and_order(payload.as_ref(), signature)
+                .unwrap();
+            let result = store.add_mutation(
+                payload.as_ref(),
+                signature,
+                &DB3Address::ZERO,
+                1,
+                block,
+                order,
+            );
             assert!(result.is_ok());
             if let Ok(headers) = store.scan_mutation_headers(0, 1) {
                 assert_eq!(1, headers.len());
@@ -645,7 +676,17 @@ mod tests {
             }
             let payload: Vec<u8> = vec![1];
             let signature: &str = "0xasdasdsad";
-            let result = store.add_mutation(payload.as_ref(), signature, &DB3Address::ZERO, 1);
+            let (_id, block, order) = store
+                .generate_mutation_block_and_order(payload.as_ref(), signature)
+                .unwrap();
+            let result = store.add_mutation(
+                payload.as_ref(),
+                signature,
+                &DB3Address::ZERO,
+                1,
+                block,
+                order,
+            );
             assert!(result.is_ok());
             let result = store.get_range_mutations(0, 1);
             if let Ok(r) = result {
@@ -677,7 +718,17 @@ mod tests {
         if let Ok(store) = result {
             let payload: Vec<u8> = vec![1];
             let signature: &str = "0xasdasdsad";
-            let result = store.add_mutation(payload.as_ref(), signature, &DB3Address::ZERO, 1);
+            let (_id, block, order) = store
+                .generate_mutation_block_and_order(payload.as_ref(), signature)
+                .unwrap();
+            let result = store.add_mutation(
+                payload.as_ref(),
+                signature,
+                &DB3Address::ZERO,
+                1,
+                block,
+                order,
+            );
             assert!(result.is_ok());
             if let Ok((id, block, order)) = result {
                 if let Ok(Some(v)) = store.get_mutation_header(block, order) {


Test patch: