feat: update inscription transfer logic

Author: Ludo Galabru

components/chainhook-event-observer/src/indexer/ordinals/db/mod.rs

@@ -16,6 +16,8 @@ use crate::{
     utils::Context,
 };
 
+use super::ord::height::Height;
+
 fn get_default_ordinals_db_file_path(base_dir: &PathBuf) -> PathBuf {
     let mut destination_path = base_dir.clone();
     destination_path.push("bitcoin_block_traversal.sqlite");
@@ -54,9 +56,9 @@ pub fn initialize_ordinal_state_storage(path: &PathBuf, ctx: &Context) -> Connec
         "CREATE TABLE IF NOT EXISTS inscriptions (
             inscription_id TEXT NOT NULL PRIMARY KEY,
             outpoint_to_watch TEXT NOT NULL,
-            satoshi_id TEXT NOT NULL,
+            ordinal_number INTEGER NOT NULL,
             inscription_number INTEGER NOT NULL,
-            offset NOT NULL
+            offset INTEGER NOT NULL
         )",
         [],
     ) {
@@ -69,7 +71,7 @@ pub fn initialize_ordinal_state_storage(path: &PathBuf, ctx: &Context) -> Connec
         ctx.try_log(|logger| slog::error!(logger, "{}", e.to_string()));
     }
     if let Err(e) = conn.execute(
-        "CREATE INDEX IF NOT EXISTS index_inscriptions_on_satoshi_id ON inscriptions(satoshi_id);",
+        "CREATE INDEX IF NOT EXISTS index_inscriptions_on_ordinal_number ON inscriptions(ordinal_number);",
         [],
     ) {
         ctx.try_log(|logger| slog::error!(logger, "{}", e.to_string()));
@@ -240,7 +242,7 @@ pub fn retrieve_compacted_block_from_index(
 ) -> Option<CompactedBlock> {
     let args: &[&dyn ToSql] = &[&block_id.to_sql().unwrap()];
     let mut stmt = storage_conn
-        .prepare("SELECT compacted_bytes FROM blocks WHERE id = ?1")
+        .prepare("SELECT compacted_bytes FROM blocks WHERE id = ?")
         .unwrap();
     let result_iter = stmt
         .query_map(args, |row| {
@@ -255,35 +257,14 @@ pub fn retrieve_compacted_block_from_index(
     return None;
 }
 
-pub fn scan_existing_inscriptions_id(
-    inscription_id: &str,
-    storage_conn: &Connection,
-) -> Option<String> {
-    let args: &[&dyn ToSql] = &[&inscription_id.to_sql().unwrap()];
-    let mut stmt = storage_conn
-        .prepare("SELECT inscription_id FROM inscriptions WHERE inscription_id = ?1")
-        .unwrap();
-    let result_iter = stmt
-        .query_map(args, |row| {
-            let inscription_id: String = row.get(0).unwrap();
-            Ok(inscription_id)
-        })
-        .unwrap();
-
-    for result in result_iter {
-        return Some(result.unwrap());
-    }
-    return None;
-}
-
 pub fn store_new_inscription(
     inscription_data: &OrdinalInscriptionRevealData,
     storage_conn: &Connection,
     ctx: &Context,
 ) {
     if let Err(e) = storage_conn.execute(
-        "INSERT INTO inscriptions (inscription_id, outpoint_to_watch, satoshi_id, inscription_number) VALUES (?1, ?2)",
-        rusqlite::params![&inscription_data.inscription_id, &inscription_data.outpoint_post_inscription, &inscription_data.inscription_id, &inscription_data.inscription_id],
+        "INSERT INTO inscriptions (inscription_id, outpoint_to_watch, ordinal_number, inscription_number, offset) VALUES (?1, ?2, ?3, ?4, ?5)",
+        rusqlite::params![&inscription_data.inscription_id, &inscription_data.satpoint_post_inscription[0..inscription_data.satpoint_post_inscription.len()-2], &inscription_data.ordinal_number, &inscription_data.inscription_number, 0],
     ) {
         ctx.try_log(|logger| slog::error!(logger, "{}", e.to_string()));
     }
@@ -297,7 +278,7 @@ pub fn update_transfered_inscription(
     ctx: &Context,
 ) {
     if let Err(e) = storage_conn.execute(
-        "UPDATE inscriptions SET outpoint_to_watch = ?1, offset = ?2 WHERE inscription_id = ?3",
+        "UPDATE inscriptions SET outpoint_to_watch = ?, offset = ? WHERE inscription_id = ?",
         rusqlite::params![&outpoint_post_transfer, &offset, &inscription_id],
     ) {
         ctx.try_log(|logger| slog::error!(logger, "{}", e.to_string()));
@@ -314,61 +295,48 @@ pub fn find_last_inscription_number(
             "SELECT inscription_number FROM inscriptions ORDER BY inscription_number DESC LIMIT 1",
         )
         .unwrap();
-    let result_iter = stmt
-        .query_map(args, |row| {
-            let inscription_number: u64 = row.get(0).unwrap();
-            Ok(inscription_number)
-        })
-        .unwrap();
-
-    for result in result_iter {
-        return Ok(result.unwrap());
+    let mut rows = stmt.query(args).unwrap();
+    while let Ok(Some(row)) = rows.next() {
+        let inscription_number: u64 = row.get(0).unwrap();
+        return Ok(inscription_number);
     }
-    return Ok(0);
+    Ok(0)
 }
 
-pub fn find_inscription_with_satoshi_id(
-    satoshi_id: &str,
+pub fn find_inscription_with_ordinal_number(
+    ordinal_number: &u64,
     storage_conn: &Connection,
     ctx: &Context,
 ) -> Option<String> {
-    let args: &[&dyn ToSql] = &[&satoshi_id.to_sql().unwrap()];
+    let args: &[&dyn ToSql] = &[&ordinal_number.to_sql().unwrap()];
     let mut stmt = storage_conn
-        .prepare("SELECT inscription_id FROM inscriptions WHERE satoshi_id = ?1")
+        .prepare("SELECT inscription_id FROM inscriptions WHERE ordinal_number = ?")
         .unwrap();
-    let result_iter = stmt
-        .query_map(args, |row| {
-            let inscription_id: String = row.get(0).unwrap();
-            Ok(inscription_id)
-        })
-        .unwrap();
-
-    for result in result_iter {
-        return Some(result.unwrap());
+    let mut rows = stmt.query(args).unwrap();
+    while let Ok(Some(row)) = rows.next() {
+        let inscription_id: String = row.get(0).unwrap();
+        return Some(inscription_id);
     }
     return None;
 }
 
 pub fn find_inscriptions_at_wached_outpoint(
-    txin: &str,
+    outpoint: &str,
     storage_conn: &Connection,
-) -> Vec<(String, u64, String, u64)> {
-    let args: &[&dyn ToSql] = &[&txin.to_sql().unwrap()];
+) -> Vec<(String, u64, u64, u64)> {
+    let args: &[&dyn ToSql] = &[&outpoint.to_sql().unwrap()];
     let mut stmt = storage_conn
-        .prepare("SELECT inscription_id, inscription_number, satoshi_id, offset FROM inscriptions WHERE outpoint_to_watch = ?1 ORDER BY offset ASC")
+        .prepare("SELECT inscription_id, inscription_number, ordinal_number, offset FROM inscriptions WHERE outpoint_to_watch = ? ORDER BY offset ASC")
         .unwrap();
     let mut results = vec![];
-    let result_iter = stmt
-        .query_map(args, |row| {
-            let inscription_id: String = row.get(0).unwrap();
-            let inscription_number: u64 = row.get(1).unwrap();
-            let satoshi_id: String = row.get(2).unwrap();
-            let offset: u64 = row.get(1).unwrap();
-            results.push((inscription_id, inscription_number, satoshi_id, offset));
-            Ok(())
-        })
-        .unwrap();
-
+    let mut rows = stmt.query(args).unwrap();
+    while let Ok(Some(row)) = rows.next() {
+        let inscription_id: String = row.get(0).unwrap();
+        let inscription_number: u64 = row.get(1).unwrap();
+        let ordinal_number: u64 = row.get(2).unwrap();
+        let offset: u64 = row.get(3).unwrap();
+        results.push((inscription_id, inscription_number, ordinal_number, offset));
+    }
     return results;
 }
 
@@ -430,51 +398,61 @@ pub async fn build_bitcoin_traversal_local_storage(
     let bitcoin_config = bitcoin_config.clone();
     let moved_ctx = ctx.clone();
     let block_data_tx_moved = block_data_tx.clone();
-    let handle_1 = hiro_system_kit::thread_named("Block data retrieval").spawn(move || {
-        while let Ok(Some((block_height, block_hash))) = block_hash_rx.recv() {
-            let moved_bitcoin_config = bitcoin_config.clone();
-            let block_data_tx = block_data_tx_moved.clone();
-            let moved_ctx = moved_ctx.clone();
-            retrieve_block_data_pool.execute(move || {
-                moved_ctx.try_log(|logger| slog::info!(logger, "Fetching block #{block_height}"));
-                let future = retrieve_full_block_breakdown_with_retry(
-                    &moved_bitcoin_config,
-                    &block_hash,
-                    &moved_ctx,
-                );
-                let block_data = hiro_system_kit::nestable_block_on(future).unwrap();
-                let _ = block_data_tx.send(Some(block_data));
-            });
-            let res = retrieve_block_data_pool.join();
-            res
-        }
-    }).expect("unable to spawn thread");
-
-    let handle_2 = hiro_system_kit::thread_named("Block data compression").spawn(move || {
-        while let Ok(Some(block_data)) = block_data_rx.recv() {
-            let block_compressed_tx_moved = block_compressed_tx.clone();
-            compress_block_data_pool.execute(move || {
-                let compressed_block = CompactedBlock::from_full_block(&block_data);
-                let _ = block_compressed_tx_moved
-                    .send(Some((block_data.height as u32, compressed_block)));
-            });
+    let handle_1 = hiro_system_kit::thread_named("Block data retrieval")
+        .spawn(move || {
+            while let Ok(Some((block_height, block_hash))) = block_hash_rx.recv() {
+                let moved_bitcoin_config = bitcoin_config.clone();
+                let block_data_tx = block_data_tx_moved.clone();
+                let moved_ctx = moved_ctx.clone();
+                retrieve_block_data_pool.execute(move || {
+                    moved_ctx
+                        .try_log(|logger| slog::info!(logger, "Fetching block #{block_height}"));
+                    let future = retrieve_full_block_breakdown_with_retry(
+                        &moved_bitcoin_config,
+                        &block_hash,
+                        &moved_ctx,
+                    );
+                    let block_data = hiro_system_kit::nestable_block_on(future).unwrap();
+                    let _ = block_data_tx.send(Some(block_data));
+                });
+                let res = retrieve_block_data_pool.join();
+                res
+            }
+        })
+        .expect("unable to spawn thread");
+
+    let handle_2 = hiro_system_kit::thread_named("Block data compression")
+        .spawn(move || {
+            while let Ok(Some(block_data)) = block_data_rx.recv() {
+                let block_compressed_tx_moved = block_compressed_tx.clone();
+                compress_block_data_pool.execute(move || {
+                    let compressed_block = CompactedBlock::from_full_block(&block_data);
+                    let _ = block_compressed_tx_moved
+                        .send(Some((block_data.height as u32, compressed_block)));
+                });
 
-            let res = compress_block_data_pool.join();
-            // let _ = block_compressed_tx.send(None);
-            res
-        }
-    }).expect("unable to spawn thread");
+                let res = compress_block_data_pool.join();
+                // let _ = block_compressed_tx.send(None);
+                res
+            }
+        })
+        .expect("unable to spawn thread");
 
     let mut blocks_stored = 0;
     while let Ok(Some((block_height, compacted_block))) = block_compressed_rx.recv() {
         ctx.try_log(|logger| slog::info!(logger, "Storing block #{block_height}"));
         write_compacted_block_to_index(block_height, &compacted_block, &storage_conn, &ctx);
-        blocks_stored+= 1;
+        blocks_stored += 1;
         if blocks_stored == end_block - start_block {
             let _ = block_data_tx.send(None);
             let _ = block_hash_tx.send(None);
-            ctx.try_log(|logger| slog::info!(logger, "Local ordinals storage successfully seeded with #{blocks_stored} blocks"));
-            return Ok(())
+            ctx.try_log(|logger| {
+                slog::info!(
+                    logger,
+                    "Local ordinals storage successfully seeded with #{blocks_stored} blocks"
+                )
+            });
+            return Ok(());
         }
     }
 
@@ -488,7 +466,7 @@ pub fn retrieve_satoshi_point_using_local_storage(
     block_identifier: &BlockIdentifier,
     transaction_identifier: &TransactionIdentifier,
     ctx: &Context,
-) -> Result<(u64, u64), String> {
+) -> Result<(u64, u64, u64), String> {
     let mut ordinal_offset = 0;
     let mut ordinal_block_number = block_identifier.index as u32;
     let txid = {
@@ -505,20 +483,23 @@ pub fn retrieve_satoshi_point_using_local_storage(
             }
         };
 
+        let coinbase_txid = &res.0 .0 .0;
+        let txid = tx_cursor.0;
+
         ctx.try_log(|logger| {
-            slog::debug!(
+            slog::info!(
                 logger,
                 "{ordinal_block_number}:{:?}:{:?}",
-                hex::encode(&res.0 .0 .0),
-                hex::encode(txid)
+                hex::encode(&coinbase_txid),
+                hex::encode(&txid)
             )
         });
 
+        // to remove
         std::thread::sleep(std::time::Duration::from_millis(300));
 
-        // evaluate exit condition: did we reach a coinbase transaction?
-        let coinbase_txid = &res.0 .0 .0;
-        if coinbase_txid.eq(&tx_cursor.0) {
+        // evaluate exit condition: did we reach the **final** coinbase transaction
+        if coinbase_txid.eq(&txid) {
             let coinbase_value = &res.0 .0 .1;
             if ordinal_offset.lt(coinbase_value) {
                 break;
@@ -527,7 +508,7 @@ pub fn retrieve_satoshi_point_using_local_storage(
             // loop over the transaction fees to detect the right range
             let cut_off = ordinal_offset - coinbase_value;
             let mut accumulated_fees = 0;
-            for (txid, inputs, outputs) in res.0 .1 {
+            for (_, inputs, outputs) in res.0 .1 {
                 let mut total_in = 0;
                 for (_, _, _, input_value) in inputs.iter() {
                     total_in += input_value;
@@ -559,14 +540,14 @@ pub fn retrieve_satoshi_point_using_local_storage(
             }
         } else {
             // isolate the target transaction
-            for (txid, inputs, outputs) in res.0 .1 {
+            for (txid_n, inputs, outputs) in res.0 .1 {
                 // we iterate over the transactions, looking for the transaction target
-                if !txid.eq(&tx_cursor.0) {
+                if !txid_n.eq(&txid) {
                     continue;
                 }
 
                 ctx.try_log(|logger| {
-                    slog::debug!(logger, "Evaluating {}: {:?}", hex::encode(&txid), outputs)
+                    slog::info!(logger, "Evaluating {}: {:?}", hex::encode(&txid_n), outputs)
                 });
 
                 let mut sats_out = 0;
@@ -575,21 +556,35 @@ pub fn retrieve_satoshi_point_using_local_storage(
                         break;
                     }
                     ctx.try_log(|logger| {
-                        slog::debug!(logger, "Adding {} from output #{}", output_value, index)
+                        slog::info!(logger, "Adding {} from output #{}", output_value, index)
                     });
                     sats_out += output_value;
                 }
                 sats_out += ordinal_offset;
+                ctx.try_log(|logger| {
+                    slog::info!(
+                        logger,
+                        "Adding offset {ordinal_offset} to sats_out {sats_out}"
+                    )
+                });
 
                 let mut sats_in = 0;
                 for (txin, block_height, vout, txin_value) in inputs.into_iter() {
                     sats_in += txin_value;
+                    ctx.try_log(|logger| {
+                        slog::info!(
+                            logger,
+                            "Adding txin_value {txin_value} to sats_in {sats_in} (txin: {})",
+                            hex::encode(&txin)
+                        )
+                    });
+
                     if sats_in >= sats_out {
                         ordinal_offset = sats_out - (sats_in - txin_value);
                         ordinal_block_number = block_height;
 
                         ctx.try_log(|logger| slog::debug!(logger, "Block {ordinal_block_number} / Tx {} / [in:{sats_in}, out:{sats_out}]: {block_height} -> {ordinal_block_number}:{ordinal_offset} -> {}:{vout}",
-                        hex::encode(&txid),
+                        hex::encode(&txid_n),
                         hex::encode(&txin)));
                         tx_cursor = (txin, vout as usize);
                         break;
@@ -598,6 +593,9 @@ pub fn retrieve_satoshi_point_using_local_storage(
             }
         }
     }
-    Ok((ordinal_block_number.into(), ordinal_offset))
-}
 
+    let height = Height(ordinal_block_number.into());
+    let ordinal_number = height.starting_sat().0 + ordinal_offset;
+
+    Ok((ordinal_block_number.into(), ordinal_offset, ordinal_number))
+}