feat: implement ns-inscriber cli tool

.gitignore

@@ -2,6 +2,7 @@
 # will have compiled files and executables
 debug/
 target/
+keys/
 
 # Remove Cargo.lock from gitignore if creating an executable, leave it for libraries
 # More information here https://doc.rust-lang.org/cargo/guide/cargo-toml-vs-cargo-lock.html

crates/ns-indexer/Cargo.toml

@@ -40,7 +40,7 @@ reqwest = { version = "0.11", features = [
 ], default-features = false }
 bitcoin = { version = "0.31", features = ["serde", "base64", "rand"] }
 dotenvy = "0.15"
-faster-hex = "0.9"
+hex = "0.4"
 bitcoincore-rpc-json = "0.18.0"
 scylla = "0.10"
 tower = "0.4"

crates/ns-indexer/README.md

@@ -6,4 +6,8 @@ More information about the protocol can be found in the [protocol documentation]
 
 ```sh
 cargo run --package ns-indexer --bin ns-indexer
+```
+
+```sh
+cargo build --release --target x86_64-unknown-linux-musl --package ns-indexer --bin ns-indexer
 ```

crates/ns-indexer/cql/schema.cql

@@ -123,3 +123,18 @@ CREATE TABLE IF NOT EXISTS checkpoint (
     AND compaction = {'class': 'SizeTieredCompactionStrategy'}
     AND compression = {'sstable_compression': 'LZ4Compressor'}
     AND default_time_to_live = 0;
+
+CREATE TABLE IF NOT EXISTS utxo (
+    txid          BLOB,    -- transaction id that contains this inscription
+    vout          INT,     -- output index in transaction
+    amount        BIGINT,  -- unspend amount in satoshi
+    address       BLOB,    -- p2tr address
+    PRIMARY KEY (txid, vout)
+) WITH CLUSTERING ORDER BY (vout ASC)
+    AND caching = {'enabled': 'true'}
+    AND comment = 'unspent TX outputs'
+    AND compaction = {'class': 'SizeTieredCompactionStrategy'}
+    AND compression = {'sstable_compression': 'LZ4Compressor'}
+    AND default_time_to_live = 0;
+
+CREATE INDEX utxo_address ON utxo (address);

.env.sample → crates/ns-indexer/sample.env

@@ -1,17 +1,16 @@
-BITCOIN_RPC_URL=http://127.0.0.1:18443
-BITCOIN_RPC_USER=test
-BITCOIN_RPC_PASSWORD=123456
-
-# ----- ns-indexer env -----
 LOG_LEVEL=info
+
 INDEXER_SERVER_WORKER_THREADS=0 # defaults to the number of cpus on the system
 INDEXER_SERVER_ADDR=0.0.0.0:8080
 INDEXER_SERVER_NOSCAN=false # run as API server
+INDEXER_UTXO=false # index UTXO when scanning
 INDEXER_START_HEIGHT=0
-# more nodes split by comma
-SCYLLA_NODES=127.0.0.1:9042
+
+SCYLLA_NODES=127.0.0.1:9042 # more nodes split by comma
 SCYLLA_USERNAME=""
 SCYLLA_PASSWORD=""
 SCYLLA_KEYSPACE=ns_indexer
 
-# ----- ns-inscriber env -----
+BITCOIN_RPC_URL=http://127.0.0.1:18443
+BITCOIN_RPC_USER=test
+BITCOIN_RPC_PASSWORD=123456

crates/ns-indexer/src/api/inscription.rs

@@ -42,7 +42,7 @@ impl InscriptionAPI {
         ctx.set("action", "get_best_inscription".into()).await;
 
         let best_inscriptions_state = api.state.best_inscriptions.read().await;
-        let mut inscription = best_inscriptions_state.last().cloned();
+        let mut inscription = best_inscriptions_state.back().cloned();
         if inscription.is_none() {
             let last_accepted_state = api.state.last_accepted.read().await;
             inscription = last_accepted_state.clone();
@@ -107,7 +107,9 @@ impl InscriptionAPI {
     ) -> Result<PackObject<SuccessResponse<Vec<Inscription>>>, HTTPError> {
         ctx.set("action", "list_best_inscriptions".into()).await;
         let best_inscriptions_state = api.state.best_inscriptions.read().await;
-        Ok(to.with(SuccessResponse::new(best_inscriptions_state.clone())))
+        Ok(to.with(SuccessResponse::new(
+            best_inscriptions_state.iter().cloned().collect(),
+        )))
     }
 
     pub async fn list_by_block_height(

crates/ns-indexer/src/api/mod.rs

@@ -13,10 +13,12 @@ use crate::indexer::{Indexer, IndexerState};
 mod inscription;
 mod name;
 mod service;
+mod utxo;
 
 pub use inscription::InscriptionAPI;
 pub use name::NameAPI;
 pub use service::ServiceAPI;
+pub use utxo::UtxoAPI;
 
 #[derive(Serialize, Deserialize)]
 pub struct AppVersion {
@@ -99,6 +101,11 @@ pub struct QueryPubkey {
     pub pubkey: String,
 }
 
+#[derive(Debug, Deserialize, Validate)]
+pub struct QueryAddress {
+    pub address: String,
+}
+
 fn validate_name(name: &str) -> Result<(), ValidationError> {
     if !ns::valid_name(name) {
         return Err(ValidationError::new("invalid name"));

crates/ns-indexer/src/api/name.rs

@@ -67,8 +67,12 @@ impl NameAPI {
     ) -> Result<PackObject<SuccessResponse<Vec<String>>>, HTTPError> {
         input.validate()?;
 
-        let mut pubkey = [0u8; 32];
-        faster_hex::hex_decode(input.pubkey.as_bytes(), &mut pubkey)
+        let key = if input.pubkey.starts_with("0x") {
+            &input.pubkey[2..]
+        } else {
+            input.pubkey.as_str()
+        };
+        let pubkey = hex::decode(key)
             .map_err(|_| HTTPError::new(400, format!("Invalid pubkey: {}", input.pubkey)))?;
         ctx.set_kvs(vec![("action", "list_names_by_pubkey".into())])
             .await;

crates/ns-indexer/src/api/utxo.rs

@@ -0,0 +1,68 @@
+use axum::{
+    extract::{Query, State},
+    Extension,
+};
+use bitcoin::{Address, AddressType};
+use std::{collections::BTreeMap, str::FromStr, sync::Arc};
+use validator::Validate;
+
+use axum_web::{
+    context::ReqContext,
+    erring::{HTTPError, SuccessResponse},
+    object::PackObject,
+};
+
+use crate::api::{IndexerAPI, QueryAddress};
+use crate::db;
+use crate::utxo::UTXO;
+
+pub struct UtxoAPI;
+
+impl UtxoAPI {
+    pub async fn list(
+        State(app): State<Arc<IndexerAPI>>,
+        Extension(ctx): Extension<Arc<ReqContext>>,
+        to: PackObject<()>,
+        input: Query<QueryAddress>,
+    ) -> Result<PackObject<SuccessResponse<Vec<UTXO>>>, HTTPError> {
+        input.validate()?;
+
+        let address = Address::from_str(input.address.as_str())
+            .map_err(|_| HTTPError::new(400, format!("invalid address: {}", input.address)))?
+            .assume_checked();
+
+        match address.address_type() {
+            Some(AddressType::P2tr) | Some(AddressType::P2wpkh) => {}
+            other => {
+                return Err(HTTPError::new(
+                    400,
+                    format!("only support p2tr address, got: {:?}", other),
+                ));
+            }
+        }
+
+        ctx.set_kvs(vec![("action", "list_utxos_by_address".into())])
+            .await;
+
+        let address = address.script_pubkey().as_bytes().to_vec();
+        let utxos = db::Utxo::list(&app.scylla, &address).await?;
+        let mut utxos: BTreeMap<(&Vec<u8>, u32), UTXO> = utxos
+            .iter()
+            .map(|utxo| ((&utxo.txid, utxo.vout as u32), utxo.to_utxo()))
+            .collect();
+
+        let confirming_utxos = app.state.confirming_utxos.read().await;
+        for utxo in confirming_utxos.iter() {
+            for spent in &utxo.1 {
+                utxos.remove(&(&spent.txid, spent.vout));
+            }
+            for (_, unspent) in &utxo.2 {
+                utxos.insert((&unspent.txid, unspent.vout), unspent.clone());
+            }
+        }
+
+        let mut utxos = utxos.into_values().collect::<Vec<_>>();
+        utxos.sort_by(|a, b| a.amount.partial_cmp(&b.amount).unwrap());
+        Ok(to.with(SuccessResponse::new(utxos)))
+    }
+}

crates/ns-indexer/src/bin/main.rs

@@ -57,7 +57,16 @@ fn main() -> anyhow::Result<()> {
             rpcpassword,
         })
         .await?;
-        let indexer = Indexer::new(&IndexerOptions { scylla }).await?;
+
+        let indexer = Indexer::new(&IndexerOptions {
+            scylla,
+            index_utxo: std::env::var("INDEXER_UTXO")
+                .unwrap_or("false".to_string())
+                .parse::<bool>()
+                .unwrap(),
+        })
+        .await?;
+
         let last_accepted_height = indexer.initialize().await?;
         let start_height = if last_accepted_height > 0 {
             last_accepted_height + 1

crates/ns-indexer/src/db/mod.rs

@@ -2,9 +2,11 @@ mod model_inscription;
 mod model_name_state;
 mod model_service_protocol;
 mod model_service_state;
+mod model_utxo;
 pub mod scylladb;
 
 pub use model_inscription::{Checkpoint, Inscription, InvalidInscription};
 pub use model_name_state::NameState;
 pub use model_service_protocol::ServiceProtocol;
 pub use model_service_state::ServiceState;
+pub use model_utxo::Utxo;

crates/ns-indexer/src/db/model_inscription.rs

@@ -386,6 +386,16 @@ impl Inscription {
         statements.push(checkpoint_query.as_str());
         values.push(checkpoint_params);
 
+        if statements.len() > 500 {
+            log::info!(target: "ns-indexer",
+                action = "save_checkpoint",
+                statements = statements.len(),
+                block_height = checkpoint.block_height,
+                height = checkpoint.height;
+                "",
+            );
+        }
+
         let _ = db.batch(statements, values).await?;
         Ok(())
     }

crates/ns-indexer/src/db/model_utxo.rs

@@ -0,0 +1,131 @@
+use std::vec;
+
+use scylla_orm::{ColumnsMap, CqlValue, ToCqlVal};
+use scylla_orm_macros::CqlOrm;
+
+use crate::db::scylladb;
+use crate::utxo;
+
+#[derive(Debug, Default, Clone, CqlOrm, PartialEq)]
+pub struct Utxo {
+    pub txid: Vec<u8>,
+    pub vout: i32,
+    pub amount: i64,
+    pub address: Vec<u8>,
+
+    pub _fields: Vec<String>, // selected fields，field with `_` will be ignored by CqlOrm
+}
+
+impl Utxo {
+    pub fn from_utxo(address: Vec<u8>, value: &utxo::UTXO) -> Self {
+        Self {
+            txid: value.txid.clone(),
+            vout: value.vout as i32,
+            amount: value.amount as i64,
+            address,
+            _fields: vec![],
+        }
+    }
+
+    pub fn to_utxo(&self) -> utxo::UTXO {
+        utxo::UTXO {
+            txid: self.txid.clone(),
+            vout: self.vout as u32,
+            amount: self.amount as u64,
+        }
+    }
+
+    pub async fn handle_utxo(
+        db: &scylladb::ScyllaDB,
+        spent: &Vec<utxo::UTXO>,
+        unspent: &Vec<(Vec<u8>, utxo::UTXO)>,
+    ) -> anyhow::Result<()> {
+        // delete spent utxos
+        let mut start = 0;
+        while start < spent.len() {
+            let end = if start + 1000 > spent.len() {
+                spent.len()
+            } else {
+                start + 1000
+            };
+            let mut statements: Vec<&str> = Vec::with_capacity(end - start);
+            let mut values: Vec<Vec<CqlValue>> = Vec::with_capacity(end - start);
+            let query = "DELETE FROM utxo WHERE txid=? AND vout=?";
+
+            for tx in &spent[start..end] {
+                statements.push(query);
+                values.push(vec![tx.txid.to_cql(), (tx.vout as i32).to_cql()]);
+            }
+
+            if statements.len() > 500 {
+                log::info!(target: "ns-indexer",
+                    action = "handle_spent_utxos",
+                    statements = statements.len();
+                    "",
+                );
+            }
+
+            let _ = db.batch(statements, values).await?;
+            start = end;
+        }
+
+        let mut start = 0;
+        while start < unspent.len() {
+            let end = if start + 1000 > unspent.len() {
+                unspent.len()
+            } else {
+                start + 1000
+            };
+            let mut statements: Vec<&str> = Vec::with_capacity(unspent.len());
+            let mut values: Vec<Vec<CqlValue>> = Vec::with_capacity(unspent.len());
+            let query = "INSERT INTO utxo (txid,vout,amount,address) VALUES (?,?,?,?)";
+
+            for tx in &unspent[start..end] {
+                statements.push(query);
+                let tx = Self::from_utxo(tx.0.clone(), &tx.1);
+                values.push(vec![
+                    tx.txid.to_cql(),
+                    tx.vout.to_cql(),
+                    tx.amount.to_cql(),
+                    tx.address.to_cql(),
+                ]);
+            }
+
+            if statements.len() > 500 {
+                log::info!(target: "ns-indexer",
+                    action = "handle_unspent_utxos",
+                    statements = statements.len();
+                    "",
+                );
+            }
+
+            let _ = db.batch(statements, values).await?;
+            start = end;
+        }
+
+        Ok(())
+    }
+
+    pub async fn list(db: &scylladb::ScyllaDB, address: &Vec<u8>) -> anyhow::Result<Vec<Self>> {
+        let fields = Self::fields();
+
+        let query = format!(
+            "SELECT {} FROM utxo WHERE address=? USING TIMEOUT 3s",
+            fields.clone().join(",")
+        );
+        let params = (address.to_cql(),);
+        let rows = db.execute_iter(query, params).await?;
+
+        let mut res: Vec<Self> = Vec::with_capacity(rows.len());
+        for row in rows {
+            let mut doc = Self::default();
+            let mut cols = ColumnsMap::with_capacity(fields.len());
+            cols.fill(row, &fields)?;
+            doc.fill(&cols);
+            doc._fields = fields.clone();
+            res.push(doc);
+        }
+
+        Ok(res)
+    }
+}

crates/ns-indexer/src/db/scylladb.rs

@@ -186,7 +186,7 @@ mod tests {
 
     #[tokio::test(flavor = "current_thread")]
     async fn exec_cqls_works() {
-        dotenvy::from_filename(".env.sample").expect(".env file not found");
+        dotenvy::from_filename("sample.env").expect(".env file not found");
         let db = get_db().await;
 
         let schema = std::include_str!("../../cql/schema.cql");