Handle reorgs

.gitignore

@@ -3,3 +3,4 @@ target/
 .DS_Store
 *.db
 *.age
+App.toml

src/cli.rs

@@ -668,7 +668,7 @@ async fn process_command(command: Command, zec: &mut CoinDef, txbytes: &mut Vec<
             )?;
             let txb = serde_cbor::to_vec(&tx)?;
             println!("{}", hex::encode(&txb));
-            store_tx_details(&connection, id, &tx.txid, &txb)?;
+            store_tx_details(&connection, id, height, &tx.txid, &txb)?;
         }
         Command::GenDiversifiedAddress { account, pools } => {
             let connection = zec.connection()?;

src/db.rs

@@ -124,6 +124,7 @@ pub fn reset_tables(connection: &Connection) -> Result<()> {
     connection.execute(
         "CREATE TABLE IF NOT EXISTS txdetails(
         id_tx INTEGER PRIMARY KEY,
+        height INTEGER NOT NULL,
         txid BLOB NOT NULL UNIQUE,
         data BLOB NOT NULL)",
         [],

src/db/notes.rs

@@ -390,6 +390,7 @@ pub fn reset_scan(network: &Network, connection: &Connection, height: Option<u32
     connection.execute("DELETE FROM notes WHERE height >= ?1", [height])?;
     connection.execute("DELETE FROM witnesses WHERE height >= ?1", [height])?;
     connection.execute("DELETE FROM txdetails", [])?;
+    connection.execute("DELETE FROM msgs", [])?;
     connection.execute("UPDATE notes SET spent = NULL WHERE spent >= ?1", [height])?;
 
     Ok(height)
@@ -403,11 +404,16 @@ pub fn rewind_checkpoint(connection: &Connection) -> Result<()> {
 }
 
 pub fn rewind(connection: &Connection, height: u32) -> Result<()> {
-    connection.execute("DELETE FROM blcks WHERE height >= ?1", [height])?;
-    connection.execute("DELETE FROM txs WHERE height >= ?1", [height])?;
-    connection.execute("DELETE FROM notes WHERE height >= ?1", [height])?;
-    connection.execute("DELETE FROM witnesses WHERE height >= ?1", [height])?;
-    connection.execute("UPDATE notes SET spent = NULL WHERE spent >= ?1", [height])?;
+    let height = connection.query_row(
+        "SELECT height FROM blcks WHERE height <= ?1 ORDER BY height DESC LIMIT 1", [height], |r| r.get::<_, u32>(0))?;
+    tracing::info!("Dropping sync data after @{height}");
+    connection.execute("DELETE FROM blcks WHERE height > ?1", [height])?;
+    connection.execute("DELETE FROM txs WHERE height > ?1", [height])?;
+    connection.execute("DELETE FROM notes WHERE height > ?1", [height])?;
+    connection.execute("DELETE FROM witnesses WHERE height > ?1", [height])?;
+    connection.execute("DELETE FROM txdetails WHERE height > ?1", [height])?;
+    connection.execute("DELETE FROM msgs WHERE height > ?1", [height])?;
+    connection.execute("UPDATE notes SET spent = NULL WHERE spent > ?1", [height])?;
     Ok(())
 }
 
@@ -420,11 +426,11 @@ pub fn get_txid(connection: &Connection, id: u32) -> Result<(Vec<u8>, u32)> {
     Ok((txid, timestamp))
 }
 
-pub fn store_tx_details(connection: &Connection, id: u32, txid: &Hash, data: &[u8]) -> Result<()> {
+pub fn store_tx_details(connection: &Connection, id: u32, height: u32, txid: &Hash, data: &[u8]) -> Result<()> {
     connection.execute(
-        "INSERT INTO txdetails(id_tx, txid, data)
-        VALUES (?1, ?2, ?3) ON CONFLICT DO NOTHING",
-        params![id, txid, data],
+        "INSERT INTO txdetails(id_tx, height, txid, data)
+        VALUES (?1, ?2, ?3, ?4) ON CONFLICT DO NOTHING",
+        params![id, height, txid, data],
     )?;
     Ok(())
 }

src/txdetails.rs

@@ -295,7 +295,7 @@ pub async fn retrieve_tx_details(
             tx,
         )?;
         let tx_bin = bincode::serialize(&txd)?;
-        store_tx_details(&connection.lock(), id_tx, &txid, &tx_bin)?;
+        store_tx_details(&connection.lock(), id_tx, height, &txid, &tx_bin)?;
         let (tx_address, tx_memo) =
             get_tx_primary_address_memo(network, &account_addrs, &rtx, &txd)?;
         update_tx_primary_address_memo(&connection.lock(), id_tx, tx_address, tx_memo)?;