Merge pull request 'Remove processes cache' (#1) from remove_process_cache into dev

Reviewed-on: #1

src/api.rs

@@ -23,7 +23,7 @@ use sdk_common::aes_gcm::aead::{Aead, Payload};
 use sdk_common::crypto::{
     decrypt_with_key, encrypt_with_key, generate_key, verify_merkle_proof, AeadCore, Aes256Gcm, AnkSharedSecretHash, KeyInit, AAD
 };
-use sdk_common::process::{check_tx_for_process_updates, lock_processes, Process, ProcessState};
+use sdk_common::process::{Process, ProcessState};
 use sdk_common::serialization::{OutPointMemberMap, OutPointProcessMap};
 use sdk_common::signature::{AnkHash, AnkMessageHash, AnkValidationNoHash, AnkValidationYesHash, Proof};
 use sdk_common::sp_client::bitcoin::blockdata::fee_rate;
@@ -346,65 +346,6 @@ pub fn dump_wallet() -> ApiResult<String> {
     Ok(serde_json::to_string(device.get_sp_client()).unwrap())
 }
 
-#[wasm_bindgen]
-pub fn reset_process_cache() -> ApiResult<()> {
-    let mut cached_processes = lock_processes()?;
-
-    *cached_processes = HashMap::new();
-
-    debug_assert!(cached_processes.is_empty());
-
-    Ok(())
-}
-
-#[wasm_bindgen]
-pub fn dump_process_cache() -> ApiResult<String> {
-    let cached_processes = lock_processes()?;
-
-    let serializable_cache = cached_processes
-        .iter()
-        .map(|(outpoint, process)| {
-            (
-                outpoint.to_string(),
-                serde_json::to_value(&process).unwrap(),
-            )
-        })
-        .collect::<serde_json::Map<String, Value>>();
-
-    let json_string = serde_json::to_string(&serializable_cache)
-        .map_err(|e| ApiError::new(format!("Failed to serialize process cache: {}", e)))?;
-
-    Ok(json_string)
-}
-
-#[wasm_bindgen]
-pub fn set_process_cache(processes: JsValue) -> ApiResult<()> {
-    let parsed_processes: HashMap<OutPoint, Process> = serde_wasm_bindgen::from_value(processes)?;
-
-    let mut cached_processes = lock_processes()?;
-
-    if !cached_processes.is_empty() {
-        // Don't overwrite processes in memory
-        return Err(ApiError::new("Processes cache is not empty".to_owned()));
-    }
-
-    *cached_processes = parsed_processes;
-
-    Ok(())
-}
-
-#[wasm_bindgen]
-pub fn add_to_process_cache(process_id: String, process: String) -> ApiResult<()> {
-    let process_id = OutPoint::from_str(&process_id)?;
-    let process: Process = serde_json::from_str(&process)?;
-
-    let mut cached_processes = lock_processes()?;
-
-    cached_processes.insert(process_id, process);
-
-    Ok(())
-}
-
 #[wasm_bindgen]
 pub fn reset_shared_secrets() -> ApiResult<()> {
     let mut shared_secrets = lock_shared_secrets()?;
@@ -467,7 +408,6 @@ pub fn reset_device() -> ApiResult<()> {
     *device = Device::default();
 
     reset_shared_secrets()?;
-    reset_process_cache()?;
 
     Ok(())
 }
@@ -479,6 +419,60 @@ pub fn get_txid(transaction: String) -> ApiResult<String> {
     Ok(tx.txid().to_string())
 }
 
+#[wasm_bindgen]
+pub fn get_prevouts(transaction: String) -> ApiResult<Vec<String>> {
+    let tx: Transaction = deserialize(&Vec::from_hex(&transaction)?)?;
+
+    Ok(tx.input.iter().map(|input| input.previous_output.to_string()).collect())
+}
+
+#[wasm_bindgen]
+pub fn get_opreturn(transaction: String) -> ApiResult<String> {
+    let tx: Transaction = deserialize(&Vec::from_hex(&transaction)?)?;
+
+    let op_return_outputs: Vec<_> = tx
+        .output
+        .iter()
+        .filter(|o| o.script_pubkey.is_op_return())
+        .collect();
+    if op_return_outputs.len() != 1 {
+        return Err(ApiError::new(
+            "Transaction must contain exactly one op_return output".to_owned(),
+        ));
+    }
+    let data = &op_return_outputs
+        .into_iter()
+        .next()
+        .unwrap()
+        .script_pubkey
+        .as_bytes()[2..];
+    if data.len() != 32 {
+        return Err(ApiError::new("commited data is not 32B long".to_owned()));
+    }
+
+    Ok(data.to_lower_hex_string())
+}
+
+#[wasm_bindgen]
+pub fn process_commit_new_state(mut process: Process, state_id: String, new_tip: String) -> ApiResult<Process> {
+    let state_id_array: [u8; 32] = Vec::from_hex(&state_id)?.try_into().unwrap();
+    let new_state: ProcessState;
+    if let Ok(commited_state) = process.get_state_for_id(&state_id_array) {
+        new_state = commited_state.clone();
+    } else {
+        new_state = ProcessState {
+            commited_in: process.get_process_tip()?,
+            state_id: state_id_array,
+            ..Default::default()
+        };
+    }
+
+    process.remove_all_concurrent_states()?;
+    process.insert_concurrent_state(new_state)?;
+    process.update_states_tip(OutPoint::from_str(&new_tip)?)?;
+    Ok(process)
+}
+
 fn handle_transaction(
     updated: HashMap<OutPoint, OwnedOutput>,
     tx: &Transaction,
@@ -494,27 +488,12 @@ fn handle_transaction(
         local_member = local_device.to_member();
     }
 
-    let op_return: Vec<&sdk_common::sp_client::bitcoin::TxOut> = tx
-        .output
-        .iter()
-        .filter(|o| o.script_pubkey.is_op_return())
-        .collect();
-    if op_return.len() != 1 {
-        return Err(AnyhowError::msg(
-            "Transaction must have exactly one op_return output",
-        ));
-    }
-    let commitment =
-        AnkPrdHash::from_slice(&op_return.first().unwrap().script_pubkey.as_bytes()[2..])?;
-
     // Basically a transaction that destroyed utxo is a transaction we sent.
     let utxo_destroyed: HashMap<&OutPoint, &OwnedOutput> = updated
         .iter()
         .filter(|(outpoint, output)| output.spend_status != OutputSpendStatus::Unspent)
         .collect();
 
-    let mut shared_secrets = lock_shared_secrets()?;
-
     // empty utxo_destroyed means we received this transaction
     if utxo_destroyed.is_empty() {
         let shared_point = sp_utils::receiving::calculate_ecdh_shared_secret(
@@ -523,10 +502,12 @@ fn handle_transaction(
         );
         let shared_secret = AnkSharedSecretHash::from_shared_point(shared_point);
 
+        let mut shared_secrets = lock_shared_secrets()?;
+
         // We keep the shared_secret as unconfirmed
         shared_secrets.add_unconfirmed_secret(shared_secret);
 
-        // We also return it
+        // We also return it for permanent storage
         let mut new_secret = SecretsStore::new();
         new_secret.add_unconfirmed_secret(shared_secret);
 
@@ -564,29 +545,6 @@ fn process_transaction(
 ) -> anyhow::Result<ApiReturn> {
     let tx = deserialize::<Transaction>(&Vec::from_hex(&tx_hex)?)?;
 
-    // Before anything, check if this transaction spends the tip of a process we know about
-    match check_tx_for_process_updates(&tx) {
-        Ok(outpoint) => {
-            let processes = lock_processes()?;
-            let process = processes.get(&outpoint).unwrap();
-            let new_state = process.get_latest_commited_state().unwrap();
-            let diffs = if let Ok(diffs) = create_diffs(&lock_local_device()?, process, new_state, members_list) { diffs } else { vec![] };
-            let updated_process = UpdatedProcess {
-                process_id: outpoint,
-                current_process: process.clone(),
-                diffs,
-                ..Default::default()
-            };
-            let api_return = ApiReturn {
-                updated_process: Some(updated_process),
-                ..Default::default()
-            };
-            debug!("Found an update for process {:?}", api_return.updated_process.as_ref().unwrap().process_id);
-            return Ok(api_return);
-        }
-        Err(e) => debug!("Failed to find process update: {}", e)
-    }
-
     let tweak_data = PublicKey::from_str(&tweak_data_hex)?;
 
     let updated: HashMap<OutPoint, OwnedOutput>;
@@ -597,9 +555,12 @@ fn process_transaction(
 
     if updated.len() > 0 {
         return handle_transaction(updated, &tx, tweak_data);
+    } else {
+        log::debug!("Transaction is not ours");
+        return Ok(ApiReturn {
+            ..Default::default()
+        });
     }
-
-    Err(anyhow::Error::msg("Transaction is not our"))
 }
 
 #[wasm_bindgen]
@@ -608,7 +569,7 @@ pub fn parse_new_tx(new_tx_msg: String, block_height: u32, members_list: OutPoin
 
     if let Some(error) = new_tx.error {
         return Err(ApiError::new(format!(
-            "NewTx returned with an error: {}",
+            "NewTx message returned with an error: {}",
             error
         )));
     }
@@ -742,6 +703,7 @@ fn handle_prd(
     prd: Prd,
     secret: AnkSharedSecretHash,
     members_list: &OutPointMemberMap,
+    processes: &OutPointProcessMap
 ) -> AnyhowResult<ApiReturn> {
     debug!("handle_prd: {:#?}", prd);
     // Connect is a bit different here because there's no associated process
@@ -752,13 +714,10 @@ fn handle_prd(
 
     let outpoint = prd.process_id;
 
-    let mut processes = lock_processes()?;
+    let mut relevant_process: Process = if let Some(relevant_process) = processes.0.get(&outpoint) {
-    let relevant_process = match processes.entry(outpoint) {
+        relevant_process.clone()
-        std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(),
+    } else {
-        std::collections::hash_map::Entry::Vacant(entry) => {
+        Process::new(outpoint)
-            debug!("Creating new process for outpoint: {}", outpoint);
-            entry.insert(Process::new(outpoint))
-        }
     };
 
     match prd.prd_type {
@@ -972,18 +931,19 @@ fn handle_prd(
 fn handle_decrypted_message(
     secret: AnkSharedSecretHash,
     plain: Vec<u8>,
-    members_list: &OutPointMemberMap
+    members_list: &OutPointMemberMap,
+    processes: &OutPointProcessMap
 ) -> anyhow::Result<ApiReturn> {
     let local_address: SilentPaymentAddress = lock_local_device()?.get_address();
     if let Ok(prd) = Prd::extract_from_message(&plain, local_address) {
-        handle_prd(prd, secret, members_list)
+        handle_prd(prd, secret, members_list, processes)
     } else {
         Err(anyhow::Error::msg("Failed to handle decrypted message"))
     }
 }
 
 #[wasm_bindgen]
-pub fn parse_cipher(cipher_msg: String, members_list: OutPointMemberMap) -> ApiResult<ApiReturn> {
+pub fn parse_cipher(cipher_msg: String, members_list: OutPointMemberMap, processes: OutPointProcessMap) -> ApiResult<ApiReturn> {
     // Check that the cipher is not empty or too long
     if cipher_msg.is_empty() || cipher_msg.len() > MAX_PRD_PAYLOAD_SIZE {
         return Err(ApiError::new(
@@ -995,7 +955,7 @@ pub fn parse_cipher(cipher_msg: String, members_list: OutPointMemberMap) -> ApiR
 
     let decrypt_res = lock_shared_secrets()?.try_decrypt(&cipher);
     if let Ok((secret, plain)) = decrypt_res {
-        return handle_decrypted_message(secret, plain, &members_list)
+        return handle_decrypted_message(secret, plain, &members_list, &processes)
             .map_err(|e| ApiError::new(format!("Failed to handle decrypted message: {}", e)));
     }
 
@@ -1197,15 +1157,6 @@ pub fn create_new_process(
 
     process.insert_concurrent_state(new_state.clone())?;
 
-    {
-        let mut processes = lock_processes()?;
-        // If we already have an entry with this outpoint, something's wrong
-        if processes.contains_key(&process_id) {
-            return Err(ApiError::new("There's already a process for this outpoint".to_owned()));
-        }
-        processes.insert(process_id, process.clone());
-    }
-
     let commit_msg = CommitMessage::new(
         process_id, 
         pcd_commitment, 
@@ -1287,13 +1238,6 @@ pub fn update_process(
     // Add the new state to the process
     process.insert_concurrent_state(new_state.clone())?;
 
-    {
-        // save the process to wasm memory
-        let mut processes = lock_processes()?;
-        // To keep it simple, assume that db is true and replace what we have in memory
-        processes.insert(process_id, process.clone());
-    }
-
     let updated_process = UpdatedProcess {
         process_id,
         current_process: process,