cashierd: resume watch after restart

sql/cashier.sql

@@ -1,4 +1,3 @@
-
 CREATE TABLE IF NOT EXISTS main_keypairs(
     keypair_id INTEGER PRIMARY KEY NOT NULL,
    	token_key_private BLOB NOT NULL,
@@ -12,7 +11,8 @@ CREATE TABLE IF NOT EXISTS deposit_keypairs(
    	token_key_private BLOB NOT NULL,
     token_key_public BLOB NOT NULL,
 	network BLOB NOT NULL,
-	asset_id BLOB NOT NULL
+	asset_id BLOB NOT NULL,
+	confirm BLOB NOT NULL
 );
 
 CREATE TABLE IF NOT EXISTS withdraw_keypairs(

src/bin/cashierd.rs

@@ -130,7 +130,7 @@ impl Cashierd {
                     } else {
                         main_keypair = deserialize(&main_keypairs[0].0)?;
                     }
-                    
+
                     let sol_client = SolClient::new(serialize(&main_keypair)).await?;
 
                     bridge2.add_clients("sol".into(), sol_client).await?;
@@ -146,8 +146,8 @@ impl Cashierd {
                     let _btc_client = BtcClient::new(btc_endpoint)?;
                     // NOTE bitcoin is not implemented yet
                     //
-                    // TODO check if there is main_keypair inside 
-                    // cashierdb before generating new one 
+                    // TODO check if there is main_keypair inside
+                    // cashierdb before generating new one
                     //
                     //bridge2.add_clients("btc".into(), btc_client).await?;
                 }
@@ -157,6 +157,12 @@ impl Cashierd {
             }
         }
 
+        let resume_watch_deposit_keys_task = executor.spawn(Self::resume_watch_deposit_keys(
+            self.bridge.clone(),
+            self.cashier_wallet.clone(),
+            self.features.clone(),
+        ));
+
         self.client.lock().await.start().await?;
 
         let (notify, recv_coin) = async_channel::unbounded::<(jubjub::SubgroupPoint, u64)>();
@@ -191,11 +197,35 @@ impl Cashierd {
 
         listen_and_serve(cfg, self.clone()).await?;
 
+        resume_watch_deposit_keys_task.cancel().await;
         listen_for_receiving_coins_task.cancel().await;
         cashier_client_subscriber_task.cancel().await;
         Ok(())
     }
 
+    async fn resume_watch_deposit_keys(
+        bridge: Arc<Bridge>,
+        cashier_wallet: Arc<CashierDb>,
+        features: HashMap<String, String>,
+    ) -> Result<()> {
+        for (network, _) in features.iter() {
+            let keypairs_to_watch = cashier_wallet.get_deposit_token_keys_by_network(&network)?;
+
+            for keypair in keypairs_to_watch {
+                let bridge = bridge.clone();
+                let bridge_subscribtion = bridge.subscribe().await;
+                bridge_subscribtion
+                    .sender
+                    .send(bridge::BridgeRequests {
+                        network: network.to_owned(),
+                        payload: bridge::BridgeRequestsPayload::Watch(Some((keypair.0, keypair.1))),
+                    })
+                    .await?;
+            }
+        }
+        Ok(())
+    }
+
     async fn listen_for_receiving_coins(
         bridge: Arc<Bridge>,
         cashier_wallet: Arc<CashierDb>,
@@ -214,13 +244,12 @@ impl Cashierd {
 
         // send a request to bridge to send equivalent amount of
         // received drk coin to token publickey
-        if let Some((addr, network, asset_id)) = token {
+        if let Some((addr, network, _asset_id)) = token {
             bridge_subscribtion
                 .sender
                 .send(bridge::BridgeRequests {
                     network: network.to_string(),
-                    asset_id,
-                    payload: bridge::BridgeRequestsPayload::SendRequest(addr.clone(), amount),
+                    payload: bridge::BridgeRequestsPayload::Send(addr.clone(), amount),
                 })
                 .await?;
 
@@ -230,7 +259,7 @@ impl Cashierd {
             let error_code = res.error as u32;
             if error_code == 0 {
                 match res.payload {
-                    bridge::BridgeResponsePayload::SendResponse => {
+                    bridge::BridgeResponsePayload::Send => {
                         // TODO Send the received coins to the main address
                         cashier_wallet.confirm_withdraw_key_record(&addr, &network)?;
                     }
@@ -272,22 +301,32 @@ impl Cashierd {
             let drk_pub_key = bs58::decode(&drk_pub_key).into_vec()?;
             let drk_pub_key: jubjub::SubgroupPoint = deserialize(&drk_pub_key)?;
 
-            // TODO check if the drk public key is already exist
-            let _check = self
+            // check if the drk public key is already exist
+            let check = self
                 .cashier_wallet
                 .get_deposit_token_keys_by_dkey_public(&drk_pub_key, &network)?;
 
             let bridge = self.bridge.clone();
             let bridge_subscribtion = bridge.subscribe().await;
 
-            bridge_subscribtion
-                .sender
-                .send(bridge::BridgeRequests {
-                    network: network.clone(),
-                    asset_id,
-                    payload: bridge::BridgeRequestsPayload::WatchRequest,
-                })
-                .await?;
+            if check.is_empty() {
+                bridge_subscribtion
+                    .sender
+                    .send(bridge::BridgeRequests {
+                        network: network.clone(),
+                        payload: bridge::BridgeRequestsPayload::Watch(None),
+                    })
+                    .await?;
+            } else {
+                let keypair = check[0].to_owned();
+                bridge_subscribtion
+                    .sender
+                    .send(bridge::BridgeRequests {
+                        network: network.clone(),
+                        payload: bridge::BridgeRequestsPayload::Watch(Some((keypair.0, keypair.1))),
+                    })
+                    .await?;
+            }
 
             let bridge_res = bridge_subscribtion.receiver.recv().await?;
 
@@ -298,9 +337,9 @@ impl Cashierd {
             }
 
             match bridge_res.payload {
-                bridge::BridgeResponsePayload::WatchResponse(token_priv, token_pub) => {
+                bridge::BridgeResponsePayload::Watch(token_priv, token_pub) => {
                     // add pairings to db
-                    self.cashier_wallet.put_exchange_keys(
+                    self.cashier_wallet.put_deposit_keys(
                         &drk_pub_key,
                         &token_priv,
                         &serialize(&token_pub),
@@ -310,6 +349,9 @@ impl Cashierd {
 
                     return Ok(token_pub);
                 }
+                bridge::BridgeResponsePayload::Address(token_pub) => {
+                    return Ok(token_pub);
+                }
                 _ => Err(Error::BridgeError(
                     "Receive unknown value from Subscription".into(),
                 )),

src/service/bridge.rs

@@ -7,7 +7,6 @@ use std::collections::HashMap;
 
 pub struct BridgeRequests {
     pub network: String,
-    pub asset_id: jubjub::Fr,
     pub payload: BridgeRequestsPayload,
 }
 
@@ -17,13 +16,14 @@ pub struct BridgeResponse {
 }
 
 pub enum BridgeRequestsPayload {
-    SendRequest(Vec<u8>, u64), // send (address, amount)
-    WatchRequest,
+    Send(Vec<u8>, u64),                // send (address, amount)
+    Watch(Option<(Vec<u8>, Vec<u8>)>), // if already has a keypair
 }
 
 pub enum BridgeResponsePayload {
-    WatchResponse(Vec<u8>, String),
-    SendResponse,
+    Watch(Vec<u8>, String),
+    Address(String),
+    Send,
     Empty,
 }
 
@@ -111,19 +111,31 @@ impl Bridge {
         let client = &self.clients.lock().await[&network];
 
         match req.payload {
-            BridgeRequestsPayload::WatchRequest => {
-                let sub = client.subscribe().await?;
-                let res = BridgeResponse {
-                    error: BridgeResponseError::NoError,
-                    payload: BridgeResponsePayload::WatchResponse(sub.secret_key, sub.public_key),
-                };
-                rep.send(res).await?;
-            }
-            BridgeRequestsPayload::SendRequest(addr, amount) => {
+            BridgeRequestsPayload::Watch(val) => match val {
+                Some((private_key, public_key)) => {
+                    let pub_key = client
+                        .subscribe_with_keypair(private_key, public_key)
+                        .await?;
+                    let res = BridgeResponse {
+                        error: BridgeResponseError::NoError,
+                        payload: BridgeResponsePayload::Address(pub_key),
+                    };
+                    rep.send(res).await?;
+                }
+                None => {
+                    let sub = client.subscribe().await?;
+                    let res = BridgeResponse {
+                        error: BridgeResponseError::NoError,
+                        payload: BridgeResponsePayload::Watch(sub.secret_key, sub.public_key),
+                    };
+                    rep.send(res).await?;
+                }
+            },
+            BridgeRequestsPayload::Send(addr, amount) => {
                 client.send(addr, amount).await?;
                 let res = BridgeResponse {
                     error: BridgeResponseError::NoError,
-                    payload: BridgeResponsePayload::SendResponse,
+                    payload: BridgeResponsePayload::Send,
                 };
                 rep.send(res).await?;
             }
@@ -136,6 +148,15 @@ impl Bridge {
 #[async_trait]
 pub trait TokenClient {
     async fn subscribe(&self) -> Result<TokenSubscribtion>;
+
+    // should check if the keypair in not already subscribed
+    async fn subscribe_with_keypair(
+        &self,
+        private_key: Vec<u8>,
+        public_key: Vec<u8>,
+    ) -> Result<String>;
+
     async fn get_notifier(&self) -> Result<async_channel::Receiver<TokenNotification>>;
+
     async fn send(&self, address: Vec<u8>, amount: u64) -> Result<()>;
 }

src/service/btc.rs

@@ -177,14 +177,22 @@ impl TokenClient for BtcClient {
         })
     }
 
+    async fn subscribe_with_keypair(
+        &self,
+        _private_key: Vec<u8>,
+        _public_key: Vec<u8>,
+    ) -> Result<String> {
+        // TODO this not implemented yet
+        Ok(String::new())
+    }
+
     async fn get_notifier(&self) -> Result<async_channel::Receiver<TokenNotification>> {
         // TODO this not implemented yet
         let (_, notifier) = async_channel::unbounded();
         Ok(notifier)
     }
     async fn send(&self, _address: Vec<u8>, _amount: u64) -> Result<()> {
-        // TODO
-
+        // TODO this not implemented yet
         Ok(())
     }
 }

src/service/sol.rs

@@ -122,8 +122,8 @@ impl SolClient {
             JsonResult::Err(e) => {
                 // receive an error
                 debug!(
-                        target: "SOL BRIDGE",
-                        "Error on subscription: {:?}", e.error.message.to_string());
+                    target: "SOL BRIDGE",
+                    "Error on subscription: {:?}", e.error.message.to_string());
             }
 
             JsonResult::Notif(n) => {
@@ -197,18 +197,17 @@ impl SolClient {
         Ok(())
     }
 
-    async fn unsubscribe(&self, sub_id: u64, pubkey: &Pubkey) -> Result<()> {
-        let sub_msg = jsonrpc::request(json!("accountUnsubscribe"), json!([json!(sub_id)]));
-        self.subscribe_channel.0.send(sub_msg).await?;
-        self.subscriptions.lock().await.remove(pubkey);
-        Ok(())
-    }
-}
+    async fn handle_subscribe_request(&self, keypair: Keypair) -> Result<()> {
 
-#[async_trait]
-impl TokenClient for SolClient {
-    async fn subscribe(&self) -> Result<TokenSubscribtion> {
-        let keypair = Keypair::generate(&mut OsRng);
+        // check first if it's not already subscribed
+        if self
+            .subscriptions
+            .lock()
+            .await
+            .contains_key(&keypair.pubkey())
+        {
+            return Ok(());
+        }
 
         // Parameters for subscription to events related to `pubkey`.
         let sub_params = SubscribeParams {
@@ -227,10 +226,6 @@ impl TokenClient for SolClient {
             .get_balance(&keypair.pubkey())
             .map_err(|err| SolFailed::from(err))?;
 
-        let public_key = keypair.pubkey().to_string();
-        // NOTE we send keypair for sol as secret_key
-        let secret_key = serialize(&keypair);
-
         // add to subscriptions list
         self.subscriptions
             .lock()
@@ -240,12 +235,48 @@ impl TokenClient for SolClient {
         //  send
         self.subscribe_channel.0.send(sub_msg).await?;
 
+        Ok(())
+    }
+
+    async fn unsubscribe(&self, sub_id: u64, pubkey: &Pubkey) -> Result<()> {
+        let sub_msg = jsonrpc::request(json!("accountUnsubscribe"), json!([json!(sub_id)]));
+        self.subscribe_channel.0.send(sub_msg).await?;
+        self.subscriptions.lock().await.remove(pubkey);
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl TokenClient for SolClient {
+    async fn subscribe(&self) -> Result<TokenSubscribtion> {
+        let keypair = Keypair::generate(&mut OsRng);
+
+        let public_key = keypair.pubkey().to_string();
+        let secret_key = serialize(&keypair);
+
+        self.handle_subscribe_request(keypair).await?;
+
         Ok(TokenSubscribtion {
             secret_key,
             public_key,
         })
     }
 
+    // in solana case private key it's the same as keypair
+    async fn subscribe_with_keypair(
+        &self,
+        private_key: Vec<u8>,
+        _public_key: Vec<u8>,
+    ) -> Result<String> {
+        let keypair: Keypair = deserialize(&private_key)?;
+
+        let public_key = keypair.pubkey().to_string();
+
+        self.handle_subscribe_request(keypair).await?;
+
+        Ok(public_key)
+    }
+
     async fn get_notifier(&self) -> Result<async_channel::Receiver<TokenNotification>> {
         Ok(self.notify_channel.1.clone())
     }
@@ -272,7 +303,6 @@ impl TokenClient for SolClient {
 
 /// Derive an associated token address from given owner and mint
 fn get_associated_token_account(owner: &Pubkey, mint: &Pubkey) -> (Pubkey, u8) {
-
     let associated_token =
         Pubkey::from_str("ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL").unwrap();
 
@@ -401,5 +431,3 @@ impl From<crate::error::Error> for SolFailed {
 }
 
 pub type SolResult<T> = std::result::Result<T, SolFailed>;
-
-

src/wallet/cashierdb.rs

@@ -154,7 +154,7 @@ impl CashierDb {
         Ok(())
     }
 
-    pub fn put_exchange_keys(
+    pub fn put_deposit_keys(
         &self,
         d_key_public: &jubjub::SubgroupPoint,
         token_key_private: &Vec<u8>,
@@ -173,17 +173,20 @@ impl CashierDb {
         let asset_id = self.get_value_serialized(asset_id)?;
         let network = self.get_value_serialized(network)?;
 
+        let confirm = self.get_value_serialized(&false)?;
+
         conn.execute(
             "INSERT INTO deposit_keypairs
-            (d_key_public, token_key_private, token_key_public, network, asset_id)
+            (d_key_public, token_key_private, token_key_public, network, asset_id, confirm)
             VALUES 
-            (:d_key_public, :token_key_private, :token_key_public, :network, :asset_id)",
+            (:d_key_public, :token_key_private, :token_key_public, :network, :asset_id, :confirm)",
             named_params! {
                 ":d_key_public": &d_key_public,
                 ":token_key_private": token_key_private,
                 ":token_key_public": token_key_public,
                 ":network": &network,
                 ":asset_id": &asset_id,
+                ":confirm": &confirm,
             },
         )?;
         Ok(())
@@ -278,15 +281,58 @@ impl CashierDb {
         conn.pragma_update(None, "key", &self.password)?;
 
         let network = self.get_value_serialized(network)?;
+        let confirm = self.get_value_serialized(&false)?;
 
         let mut stmt = conn.prepare(
             "SELECT token_key_private, token_key_public 
             FROM deposit_keypairs 
             WHERE d_key_public = :d_key_public 
-            AND network = :network ;",
+            AND network = :network 
+            AND confirm = :confirm ;",
+        )?;
+        let keys_iter = stmt.query_map::<(Vec<u8>, Vec<u8>), _, _>(
+            &[
+                (":d_key_public", &d_key_public),
+                (":network", &network),
+                (":confirm", &confirm),
+            ],
+            |row| Ok((row.get(0)?, row.get(1)?)),
+        )?;
+
+        let mut keys = vec![];
+
+        for k in keys_iter {
+            keys.push(k?);
+        }
+
+        Ok(keys)
+    }
+
+    // return private and public keys as a tuple
+    pub fn get_deposit_token_keys_by_network(
+        &self,
+        network: &String,
+    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
+        debug!(target: "CASHIERDB", "Check for existing dkey");
+        // open connection
+        let conn = Connection::open(&self.path)?;
+        // unlock database
+        conn.pragma_update(None, "key", &self.password)?;
+
+        let network = self.get_value_serialized(network)?;
+        let confirm = self.get_value_serialized(&false)?;
+
+        let mut stmt = conn.prepare(
+            "SELECT token_key_private, token_key_public 
+            FROM deposit_keypairs 
+            WHERE network = :network 
+            AND confirm = :confirm ;",
         )?;
         let keys_iter = stmt.query_map::<(Vec<u8>, Vec<u8>), _, _>(
-            &[(":d_key_public", &d_key_public), (":network", &network)],
+            &[
+                (":network", &network),
+                (":confirm", &confirm),
+            ],
             |row| Ok((row.get(0)?, row.get(1)?)),
         )?;
 
@@ -374,6 +420,33 @@ impl CashierDb {
         Ok(())
     }
 
+    pub fn confirm_deposit_key_record(
+        &self,
+        d_key_public: &jubjub::SubgroupPoint,
+        network: &String,
+    ) -> Result<()> {
+        debug!(target: "CASHIERDB", "Confirm withdraw keys");
+
+        // open connection
+        let conn = Connection::open(&self.path)?;
+        // unlock database
+        conn.pragma_update(None, "key", &self.password)?;
+
+        let network = self.get_value_serialized(network)?;
+        let confirm = self.get_value_serialized(&true)?;
+        let d_key_public = self.get_value_serialized(d_key_public)?;
+
+        conn.execute(
+            "UPDATE deposit_keypairs 
+            SET confirm = ?1  
+            WHERE d_key_public = ?2 
+            AND network = ?3;",
+            params![confirm, d_key_public, network],
+        )?;
+
+        Ok(())
+    }
+
     // TODO convert this to generic function work with different tokens
     //pub fn put_btc_utxo(
     //    &self,
@@ -473,7 +546,7 @@ mod tests {
         let public2 = zcash_primitives::constants::SPENDING_KEY_GENERATOR * secret2;
         let asset_id: jubjub::Fr = jubjub::Fr::random(&mut OsRng);
 
-        wallet.put_exchange_keys(
+        wallet.put_deposit_keys(
             &public2,
             &token_addr_private,
             &token_addr,
@@ -488,6 +561,12 @@ mod tests {
         assert_eq!(keys[0].0, token_addr_private);
         assert_eq!(keys[0].1, token_addr);
 
+        wallet.confirm_deposit_key_record(&public2, &network)?;
+
+        let keys = wallet.get_deposit_token_keys_by_dkey_public(&public2, &network)?;
+
+        assert_eq!(keys.len(), 0);
+
         std::fs::remove_file(walletdb_path)?;
 
         Ok(())