add put_value, get_values, add some tests

src/impls.rs

@@ -15,7 +15,7 @@
  */
 
 use crate::{KEYS_TABLE_NAME, VALUES_TABLE_NAME, DB_PATH, TRUSTED_TIMESTAMP_SERVICE_ID, TRUSTED_TIMESTAMP_FUNCTION_NAME};
-use crate::results::{Key};
+use crate::results::{Key, Record};
 use marine_sqlite_connector::{Connection, Result as SqliteResult, Error as SqliteError, State};
 use fluence::{CallParameters};
 use eyre;
@@ -54,13 +54,14 @@ pub(crate) fn create_values_table() -> bool {
     connection
         .execute(f!("
             CREATE TABLE IF NOT EXISTS {VALUES_TABLE_NAME} (
-                key TEXT PRIMARY KEY,
+                key TEXT,
                 value TEXT,
                 peer_id TEXT,
                 relay_id TEXT,
                 service_id TEXT,
                 timestamp_created INTEGER,
-                timestamp_accessed INTEGER
+                timestamp_accessed INTEGER,
+                PRIMARY KEY (key, peer_id)
                 );
         "),
         ).is_ok()
@@ -114,3 +115,50 @@ pub fn republish_key_impl(key: Key, current_timestamp: u64) -> SqliteResult<()>
 
     update_key(&get_connection()?, key.key, key.peer_id, key.timestamp_created, current_timestamp)
 }
+
+pub fn put_value_impl(key: String, value: String, current_timestamp: u64, relay_id: Vec<String>) -> SqliteResult<()> {
+    let call_parameters = fluence::get_call_parameters();
+    check_timestamp_tetraplets(&call_parameters, 2)
+        .map_err(|e| SqliteError { code: None, message: Some(e.to_string()) })?;
+
+    let connection = get_connection()?;
+
+    let _key = get_key_metadata_helper(&connection, key.clone())?;
+    let relay_id = if relay_id.len() == 0 { "".to_string() } else { relay_id[0].clone() };
+    let peer_id = call_parameters.init_peer_id;
+    let service_id = call_parameters.service_id;
+
+    connection.execute(
+        f!("INSERT OR REPLACE INTO {VALUES_TABLE_NAME} \
+                    VALUES ('{key}', '{value}', '{peer_id}', '{relay_id}',\
+                    '{service_id}', '{current_timestamp}', '{current_timestamp}')")
+    )
+}
+
+pub fn get_values_impl(key: String, current_timestamp: u64) -> SqliteResult<Vec<Record>> {
+    let call_parameters = fluence::get_call_parameters();
+    check_timestamp_tetraplets(&call_parameters, 1)
+        .map_err(|e| SqliteError { code: None, message: Some(e.to_string()) })?;
+
+    let connection = get_connection()?;
+
+    let mut statement = connection.prepare(
+        f!("UPDATE {VALUES_TABLE_NAME} \
+                     SET timestamp_accessed = '{current_timestamp}' \
+                     WHERE key = '{key}' \
+                     RETURNING value, peer_id, relay_id, service_id, timestamp_created"))?;
+
+    let mut result: Vec<Record> = vec![];
+
+    while let State::Row = statement.next()? {
+        result.push(Record{
+            value: statement.read::<String>(0)?,
+            peer_id:statement.read::<String>(1)?,
+            relay_id: statement.read::<String>(2)?,
+            service_id: statement.read::<String>(3)?,
+            timestamp_created: statement.read::<i64>(4)? as u64,
+        })
+    }
+
+    Ok(result)
+}

src/main.rs

@@ -19,8 +19,8 @@ mod results;
 mod tests;
 mod impls;
 
-use crate::results::{Key, GetKeyMetadataResult, RegisterKeyResult, RepublishKeyResult, PutValueResult};
+use crate::results::{Key, GetKeyMetadataResult, RegisterKeyResult, RepublishKeyResult, PutValueResult, GetValuesResult};
-use crate::impls::{create_keys_table, create_values_table, register_key_impl, get_key_metadata_impl, republish_key_impl};
+use crate::impls::{create_keys_table, create_values_table, register_key_impl, get_key_metadata_impl, republish_key_impl, put_value_impl, get_values_impl};
 
 use fluence::marine;
 use fluence::module_manifest;
@@ -61,16 +61,16 @@ pub fn republish_key(key: Key, current_timestamp: u64) -> RepublishKeyResult {
 }
 
 // VALUES
-// #[marine]
+#[marine]
-// pub fn put_value(key: String, value: String, current_timestamp: u64, relay_id: Vec<String>) -> PutValueResult {
+pub fn put_value(key: String, value: String, current_timestamp: u64, relay_id: Vec<String>) -> PutValueResult {
-//     put_value_impl(key, value, current_timestamp, relay_id).into()
+    put_value_impl(key, value, current_timestamp, relay_id).into()
-// }
+}
 
-//
+
-// #[marine]
+#[marine]
-// pub fn get_values(key: String, current_timestamp: u64) -> GetValuesResult {
+pub fn get_values(key: String, current_timestamp: u64) -> GetValuesResult {
-//     get_values_impl(key, current_timestamp).into()
+    get_values_impl(key, current_timestamp).into()
-// }
+}
 
 // #[marine]
 // pub fn republish_values(key: String, records: Vec<Record>, current_timestamp: u64) -> RepublishValuesResult  {

src/results.rs

@@ -91,7 +91,6 @@ pub struct Record {
     pub relay_id: String,
     pub service_id: String,
     pub timestamp_created: u64,
-    pub timestamp_accessed: u64,
 }
 
 #[marine]

src/tests.rs

@@ -107,10 +107,7 @@ mod tests {
     #[marine_test(config_path = "../Config.toml", modules_dir = "../artifacts/")]
     fn register_key_empty_cp() {
         clear_db();
-        let key = "some_key".to_string();
+        let result = aqua_dht.register_key("some_key".to_string(), 123u64);
-        let timestamp = 123u64;
-
-        let result = aqua_dht.register_key(key.clone(), timestamp);
         assert!(!result.success);
         assert_eq!(result.error, "you should use peer.timestamp_ms to pass timestamp");
     }
@@ -118,9 +115,6 @@ mod tests {
     #[marine_test(config_path = "../Config.toml", modules_dir = "../artifacts/")]
     fn register_key_invalid_cp() {
         clear_db();
-        let key = "some_key".to_string();
-        let timestamp = 123u64;
-
         let mut invalid_cp = CallParameters::default();
         invalid_cp.tetraplets.push(vec![]);
         invalid_cp.tetraplets.push(vec![SecurityTetraplet {
@@ -130,7 +124,7 @@ mod tests {
             json_path: "some json path".to_string(),
         }]);
 
-        let result = aqua_dht.register_key_cp(key.clone(), timestamp, invalid_cp);
+        let result = aqua_dht.register_key_cp("some_key".to_string(), 123u64, invalid_cp);
         assert!(!result.success);
         assert_eq!(result.error, "you should use peer.timestamp_ms to pass timestamp");
     }
@@ -219,4 +213,55 @@ mod tests {
         assert!(!result.success);
         assert_eq!(result.error, "key already exists with different peer_id");
     }
+
+    #[marine_test(config_path = "../Config.toml", modules_dir = "../artifacts/")]
+    fn put_value_empty_cp() {
+        clear_db();
+        let result = aqua_dht.put_value("some_key".to_string(), "value".to_string(), 123u64, vec![]);
+        assert!(!result.success);
+        assert_eq!(result.error, "you should use peer.timestamp_ms to pass timestamp");
+    }
+
+    #[marine_test(config_path = "../Config.toml", modules_dir = "../artifacts/")]
+    fn put_value_invalid_cp() {
+        clear_db();
+
+        let mut invalid_cp = CallParameters::default();
+        invalid_cp.tetraplets.push(vec![]);
+        invalid_cp.tetraplets.push(vec![SecurityTetraplet {
+            peer_pk: "some peer_pk".to_string(),
+            service_id: "INVALID SERVICE ID".to_string(),
+            function_name: "INVALID FUNCTION NAME".to_string(),
+            json_path: "some json path".to_string(),
+        }]);
+
+        let result = aqua_dht.put_value_cp("some_key".to_string(), "value".to_string(), 123u64, vec![], invalid_cp);
+        assert!(!result.success);
+        assert_eq!(result.error, "you should use peer.timestamp_ms to pass timestamp");
+    }
+
+    #[marine_test(config_path = "../Config.toml", modules_dir = "../artifacts/")]
+    fn get_values_empty_cp() {
+        clear_db();
+        let result = aqua_dht.get_values("some_key".to_string(), 123u64);
+        assert!(!result.success);
+        assert_eq!(result.error, "you should use peer.timestamp_ms to pass timestamp");
+    }
+
+    #[marine_test(config_path = "../Config.toml", modules_dir = "../artifacts/")]
+    fn get_values_invalid_cp() {
+        clear_db();
+        let mut invalid_cp = CallParameters::default();
+        invalid_cp.tetraplets.push(vec![]);
+        invalid_cp.tetraplets.push(vec![SecurityTetraplet {
+            peer_pk: "some peer_pk".to_string(),
+            service_id: "INVALID SERVICE ID".to_string(),
+            function_name: "INVALID FUNCTION NAME".to_string(),
+            json_path: "some json path".to_string(),
+        }]);
+
+        let result = aqua_dht.get_values_cp("some_key".to_string(), 123u64, invalid_cp);
+        assert!(!result.success);
+        assert_eq!(result.error, "you should use peer.timestamp_ms to pass timestamp");
+    }
 }