drk: broadcasted transactions history handling added

bin/darkfid/src/rpc_wallet.rs

@@ -166,6 +166,24 @@ impl Darkfid {
                     continue
                 }
 
+                QueryType::Text => {
+                    let Some(ref row) = row else {
+                        error!("[RPC] wallet.query_row_single: Got None for QueryType::Text");
+                        return server_error(RpcError::NoRowsFoundInWallet, id, None)
+                    };
+
+                    let value: String = match row.try_get(col) {
+                        Ok(v) => v,
+                        Err(e) => {
+                            error!("[RPC] wallet.query_row_single: {}", e);
+                            return JsonError::new(ParseError, None, id).into()
+                        }
+                    };
+
+                    ret.push(json!(value));
+                    continue
+                }
+
                 _ => unreachable!(),
             }
         }
@@ -275,6 +293,18 @@ impl Darkfid {
                         row_ret.push(json!(value));
                     }
 
+                    QueryType::Text => {
+                        let value: String = match row.try_get(col) {
+                            Ok(v) => v,
+                            Err(e) => {
+                                error!("[RPC] wallet.query_row_multi: {}", e);
+                                return JsonError::new(ParseError, None, id).into()
+                            }
+                        };
+
+                        row_ret.push(json!(value));
+                    }
+
                     _ => unreachable!(),
                 }
             }
@@ -322,6 +352,7 @@ impl Darkfid {
 
                     query = query.bind(val);
                 }
+
                 QueryType::Blob => {
                     let val: Vec<u8> = match serde_json::from_value(pair[1].clone()) {
                         Ok(v) => v,
@@ -361,6 +392,18 @@ impl Darkfid {
                     query = query.bind(val);
                 }
 
+                QueryType::Text => {
+                    let val: String = match serde_json::from_value(pair[1].clone()) {
+                        Ok(v) => v,
+                        Err(e) => {
+                            error!("[RPC] wallet.exec_sql: Failed casting value to String: {}", e);
+                            return JsonError::new(ParseError, None, id).into()
+                        }
+                    };
+
+                    query = query.bind(val);
+                }
+
                 _ => return JsonError::new(InvalidParams, None, id).into(),
             }
         }

bin/drk/src/main.rs

@@ -70,6 +70,9 @@ mod rpc_blockchain;
 mod cli_util;
 use cli_util::{parse_token_pair, parse_value_pair};
 
+/// Wallet functionality related to drk operations
+mod wallet;
+
 /// Wallet functionality related to DAO
 mod wallet_dao;
 use wallet_dao::DaoParams;
@@ -80,6 +83,9 @@ mod wallet_money;
 /// Wallet functionality related to arbitrary tokens
 mod wallet_token;
 
+/// Wallet functionality related to transactions history
+mod wallet_txs_history;
+
 #[derive(Parser)]
 #[command(about = cli_desc!())]
 struct Args {
@@ -362,6 +368,17 @@ enum ExplorerSubcmd {
 
     /// Read a transaction from stdin and simulate it
     SimulateTx,
+
+    /// Fetch broadcasted transactions history
+    TxsHistory {
+        /// Fetch specific history record (optional)
+        tx_hash: Option<String>,
+
+        #[arg(long)]
+        /// Encode specific history record transaction
+        /// to base58.
+        encode: bool,
+    },
 }
 
 #[derive(Subcommand)]
@@ -509,6 +526,7 @@ async fn main() -> Result<()> {
             let drk = Drk::new(args.endpoint).await?;
 
             if initialize {
+                drk.initialize_wallet().await?;
                 drk.initialize_money().await?;
                 drk.initialize_dao().await?;
                 return Ok(())
@@ -1076,7 +1094,7 @@ async fn main() -> Result<()> {
                 {
                     tx
                 } else {
-                    eprintln!("Transaction was not found!");
+                    eprintln!("Transaction was not found");
                     exit(1);
                 };
 
@@ -1113,6 +1131,43 @@ async fn main() -> Result<()> {
 
                 Ok(())
             }
+
+            ExplorerSubcmd::TxsHistory { tx_hash, encode } => {
+                let drk = Drk::new(args.endpoint).await?;
+
+                if let Some(c) = tx_hash {
+                    let (tx_hash, status, tx) = drk.get_tx_history_record(&c).await?;
+
+                    if encode {
+                        println!("{}", bs58::encode(&serialize(&tx)).into_string());
+                        exit(1)
+                    }
+
+                    println!("Transaction ID: {}", tx_hash);
+                    println!("Status: {}", status);
+                    println!("{:?}", tx);
+
+                    return Ok(())
+                }
+
+                let map = drk.get_txs_history().await?;
+
+                // Create a prettytable with the new data:
+                let mut table = Table::new();
+                table.set_format(*format::consts::FORMAT_NO_BORDER_LINE_SEPARATOR);
+                table.set_titles(row!["Transaction Hash", "Status"]);
+                for (txs_hash, status) in map.iter() {
+                    table.add_row(row![txs_hash, status]);
+                }
+
+                if table.is_empty() {
+                    println!("No transactions found");
+                } else {
+                    println!("{}", table);
+                }
+
+                Ok(())
+            }
         },
 
         Subcmd::Alias(cmd) => match cmd {

bin/drk/src/rpc_blockchain.rs

@@ -94,6 +94,7 @@ impl Drk {
                     eprintln!("Deserialized successfully. Scanning block...");
                     self.scan_block_money(&block_data).await?;
                     self.scan_block_dao(&block_data).await?;
+                    self.update_tx_history_records_status(&block_data.txs, "Finalized").await?;
                 }
 
                 JsonResult::Error(e) => {
@@ -169,13 +170,8 @@ impl Drk {
 
         let txid = serde_json::from_value(rep)?;
 
-        // At this point the tx is successfully broadcasted. We can add the
-        // temp data into the wallet. Once scanned, it should mean that the
-        // transaction was finalized, so at that point we actually add the
-        // missing data. For now it'll be in an "unconfirmed" state.
-        // TODO: Do the same for Money::*
-        //self.wallet_apply_unconfirmed_dao_data(tx).await?;
-        //self.wallet_apply_unconfirmed_money_data(tx).await?;
+        // Store transactions history record
+        self.insert_tx_history_record(tx).await?;
 
         Ok(txid)
     }
@@ -235,6 +231,7 @@ impl Drk {
             self.reset_daos().await?;
             self.reset_dao_proposals().await?;
             self.reset_dao_votes().await?;
+            self.update_all_tx_history_records_status("Rejected").await?;
             0
         } else {
             self.last_scanned_slot().await?
@@ -280,6 +277,7 @@ impl Drk {
                 eprintln!("Found");
                 self.scan_block_money(&block).await?;
                 self.scan_block_dao(&block).await?;
+                self.update_tx_history_records_status(&block.txs, "Finalized").await?;
             } else {
                 eprintln!("Not found");
                 // Write down the slot number into back to the wallet
@@ -333,10 +331,11 @@ impl Drk {
                     let params = n.params.as_array().unwrap()[0].as_str().unwrap();
                     let bytes = bs58::decode(params).into_vec()?;
 
-                    let txs_hash: String = deserialize(&bytes)?;
+                    let tx_hash: String = deserialize(&bytes)?;
                     eprintln!("===================================");
-                    eprintln!("Erroneous transaction: {}", txs_hash);
+                    eprintln!("Erroneous transaction: {}", tx_hash);
                     eprintln!("===================================");
+                    self.update_tx_history_record_status(&tx_hash, "Rejected").await?;
                 }
 
                 JsonResult::Error(e) => {

bin/drk/src/wallet.rs

@@ -0,0 +1,43 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use anyhow::Result;
+use darkfi::rpc::jsonrpc::JsonRequest;
+use serde_json::json;
+
+use super::Drk;
+
+impl Drk {
+    /// Initialize wallet with tables for drk
+    pub async fn initialize_wallet(&self) -> Result<()> {
+        let wallet_schema = include_str!("../wallet.sql");
+
+        // We perform a request to darkfid with the schema to initialize
+        // the necessary tables in the wallet.
+        let req = JsonRequest::new("wallet.exec_sql", json!([wallet_schema]));
+        let rep = self.rpc_client.request(req).await?;
+
+        if rep == true {
+            eprintln!("Successfully initialized wallet schema for drk");
+        } else {
+            eprintln!("[initialize_wallet] Got unexpected reply from darkfid: {}", rep);
+        }
+
+        Ok(())
+    }
+}

bin/drk/src/wallet_txs_history.rs

@@ -0,0 +1,191 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2023 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use anyhow::{anyhow, Result};
+use darkfi::{rpc::jsonrpc::JsonRequest, tx::Transaction, wallet::walletdb::QueryType};
+use darkfi_serial::{deserialize, serialize};
+use serde_json::json;
+
+use super::Drk;
+
+// Wallet SQL table constant names. These have to represent the `wallet.sql`
+// SQL schema.
+const WALLET_TXS_HISTORY_TABLE: &str = "transactions_history";
+const WALLET_TXS_HISTORY_COL_TX_HASH: &str = "transaction_hash";
+const WALLET_TXS_HISTORY_COL_STATUS: &str = "status";
+const WALLET_TXS_HISTORY_COL_TX: &str = "tx";
+
+impl Drk {
+    /// Fetch all transactions history records, excluding bytes column.
+    pub async fn get_txs_history(&self) -> Result<Vec<(String, String)>> {
+        let mut ret = vec![];
+
+        let query = format!(
+            "SELECT {}, {} FROM {};",
+            WALLET_TXS_HISTORY_COL_TX_HASH, WALLET_TXS_HISTORY_COL_STATUS, WALLET_TXS_HISTORY_TABLE
+        );
+
+        let params = json!([
+            query,
+            QueryType::Text as u8,
+            WALLET_TXS_HISTORY_COL_TX_HASH,
+            QueryType::Text as u8,
+            WALLET_TXS_HISTORY_COL_STATUS,
+        ]);
+
+        let req = JsonRequest::new("wallet.query_row_multi", params);
+        let rep = self.rpc_client.request(req).await?;
+
+        let Some(rows) = rep.as_array() else {
+            return Err(anyhow!("[txs_history] Unexpected response from darkfid: {}", rep));
+        };
+
+        for row in rows {
+            let tx_hash: String = serde_json::from_value(row[0].clone())?;
+            let status: String = serde_json::from_value(row[1].clone())?;
+            ret.push((tx_hash, status));
+        }
+
+        Ok(ret)
+    }
+
+    /// Get a transaction history record.
+    pub async fn get_tx_history_record(
+        &self,
+        tx_hash: &str,
+    ) -> Result<(String, String, Transaction)> {
+        let query = format!(
+            "SELECT * FROM {} WHERE {} = {};",
+            WALLET_TXS_HISTORY_TABLE, WALLET_TXS_HISTORY_COL_TX_HASH, tx_hash
+        );
+
+        let params = json!([
+            query,
+            QueryType::Text as u8,
+            WALLET_TXS_HISTORY_COL_TX_HASH,
+            QueryType::Text as u8,
+            WALLET_TXS_HISTORY_COL_STATUS,
+            QueryType::Blob as u8,
+            WALLET_TXS_HISTORY_COL_TX,
+        ]);
+
+        let req = JsonRequest::new("wallet.query_row_single", params);
+        let rep = self.rpc_client.request(req).await?;
+
+        let Some(arr) = rep.as_array() else {
+            return Err(anyhow!("[get_tx_history_record] Unexpected response from darkfid: {}", rep));
+        };
+
+        if arr.len() != 3 {
+            return Err(anyhow!("Did not find transaction record with hash {}", tx_hash))
+        }
+
+        let tx_hash: String = serde_json::from_value(arr[0].clone())?;
+
+        let status: String = serde_json::from_value(arr[1].clone())?;
+
+        let tx_bytes: Vec<u8> = serde_json::from_value(arr[2].clone())?;
+        let tx: Transaction = deserialize(&tx_bytes)?;
+
+        Ok((tx_hash, status, tx))
+    }
+
+    /// Insert a [`Transaction`] history record into the wallet.
+    pub async fn insert_tx_history_record(&self, tx: &Transaction) -> Result<()> {
+        let query = format!(
+            "INSERT INTO {} ({}, {}, {}) VALUES (?1, ?2, ?3);",
+            WALLET_TXS_HISTORY_TABLE,
+            WALLET_TXS_HISTORY_COL_TX_HASH,
+            WALLET_TXS_HISTORY_COL_STATUS,
+            WALLET_TXS_HISTORY_COL_TX,
+        );
+
+        let params = json!([
+            query,
+            QueryType::Text as u8,
+            tx.hash().to_string(),
+            QueryType::Text as u8,
+            "Broadcasted",
+            QueryType::Blob as u8,
+            serialize(tx),
+        ]);
+
+        let req = JsonRequest::new("wallet.exec_sql", params);
+        let _ = self.rpc_client.request(req).await?;
+
+        Ok(())
+    }
+
+    /// Update a transactions history record status to the given one.
+    pub async fn update_tx_history_record_status(&self, tx_hash: &str, status: &str) -> Result<()> {
+        let query = format!(
+            "UPDATE {} SET {} = ?1 WHERE {} = ?2;",
+            WALLET_TXS_HISTORY_TABLE, WALLET_TXS_HISTORY_COL_STATUS, WALLET_TXS_HISTORY_COL_TX_HASH,
+        );
+
+        let params = json!([query, QueryType::Text as u8, status, QueryType::Text as u8, tx_hash,]);
+
+        let req = JsonRequest::new("wallet.exec_sql", params);
+        let _ = self.rpc_client.request(req).await?;
+
+        Ok(())
+    }
+
+    /// Update given transactions history record statuses to the given one.
+    pub async fn update_tx_history_records_status(
+        &self,
+        txs: &Vec<Transaction>,
+        status: &str,
+    ) -> Result<()> {
+        if txs.is_empty() {
+            return Ok(())
+        }
+
+        let txs_hashes: Vec<String> = txs.into_iter().map(|tx| tx.hash().to_string()).collect();
+        let txs_hashes_string = format!("{:?}", txs_hashes).replace("[", "(").replace("]", ")");
+        let query = format!(
+            "UPDATE {} SET {} = ?1 WHERE {} IN {};",
+            WALLET_TXS_HISTORY_TABLE,
+            WALLET_TXS_HISTORY_COL_STATUS,
+            WALLET_TXS_HISTORY_COL_TX_HASH,
+            txs_hashes_string
+        );
+
+        let params = json!([query, QueryType::Text as u8, status,]);
+
+        let req = JsonRequest::new("wallet.exec_sql", params);
+        let _ = self.rpc_client.request(req).await?;
+
+        Ok(())
+    }
+
+    /// Update all transaction history records statuses to the given one.
+    pub async fn update_all_tx_history_records_status(&self, status: &str) -> Result<()> {
+        let query = format!(
+            "UPDATE {} SET {} = ?1",
+            WALLET_TXS_HISTORY_TABLE, WALLET_TXS_HISTORY_COL_STATUS,
+        );
+
+        let params = json!([query, QueryType::Text as u8, status,]);
+
+        let req = JsonRequest::new("wallet.exec_sql", params);
+        let _ = self.rpc_client.request(req).await?;
+
+        Ok(())
+    }
+}

bin/drk/wallet.sql

@@ -0,0 +1,9 @@
+-- Wallet definitions for drk.
+-- We store data that is needed for wallet operations.
+
+-- Broadcasted transactions history
+CREATE TABLE IF NOT EXISTS transactions_history (
+    transaction_hash TEXT PRIMARY KEY NOT NULL,
+    status TEXT NOT NULL,
+	tx BLOB NOT NULL
+);

src/wallet/walletdb.rs

@@ -48,8 +48,10 @@ pub enum QueryType {
     OptionInteger = 0x02,
     /// OptionBlob gets decoded into `Option<Vec<u8>>`
     OptionBlob = 0x03,
+    /// Text gets decoded into `String`
+    Text = 0x04,
     /// Last type, increment this when you add new types.
-    Last = 0x04,
+    Last = 0x05,
 }
 
 impl From<u8> for QueryType {
@@ -59,6 +61,7 @@ impl From<u8> for QueryType {
             0x01 => Self::Blob,
             0x02 => Self::OptionInteger,
             0x03 => Self::OptionBlob,
+            0x04 => Self::Text,
             _ => unimplemented!(),
         }
     }