Exp 191 refactor inscription manager

crates/btcio/src/broadcaster/error.rs

@@ -9,6 +9,9 @@ pub enum BroadcasterError {
     #[error("client: {0}")]
     Client(#[from] anyhow::Error),
 
+    #[error("expected tx not found in db. Idx {0}")]
+    TxNotFound(u64),
+
     #[error("{0}")]
     Other(String),
 }

crates/btcio/src/broadcaster/handle.rs

@@ -41,6 +41,10 @@ impl L1BroadcastHandle {
 
         Ok(idx)
     }
+
+    pub async fn get_tx_entry_by_id_async(&self, txid: Buf32) -> DbResult<Option<L1TxEntry>> {
+        self.ops.get_tx_entry_by_id_async(txid).await
+    }
 }
 
 pub fn spawn_broadcaster_task(

crates/btcio/src/broadcaster/state.rs

@@ -1,7 +1,9 @@
 use std::{collections::BTreeMap, sync::Arc};
 
 use alpen_express_db::types::{L1TxEntry, L1TxStatus};
+use bitcoin::Txid;
 use express_storage::BroadcastDbOps;
+use tracing::*;
 
 use super::error::{BroadcasterError, BroadcasterResult};
 
@@ -72,6 +74,9 @@ async fn filter_unfinalized_from_db(
             break;
         };
 
+        let status = &txentry.status;
+        let txid = ops.get_txid_async(idx).await?.map(Txid::from);
+        debug!(?idx, ?txid, ?status, "TxEntry");
         match txentry.status {
             L1TxStatus::Finalized { height: _ } | L1TxStatus::Excluded { reason: _ } => {}
             _ => {

crates/btcio/src/broadcaster/task.rs

@@ -1,7 +1,6 @@
 use std::{collections::BTreeMap, sync::Arc, time::Duration};
 
 use alpen_express_db::types::{ExcludeReason, L1TxEntry, L1TxStatus};
-use alpen_express_primitives::buf::Buf32;
 use bitcoin::{hashes::Hash, Txid};
 use express_storage::{ops::l1tx_broadcast, BroadcastDbOps};
 use tokio::sync::mpsc::Receiver;
@@ -38,8 +37,8 @@
             _ = interval.tick() => {}
 
             Some((idx, txentry)) = entry_receiver.recv() => {
-                let txid = get_txid_str(idx, ops.as_ref()).await?;
-                info!(%idx, %txid, "Received txentry");
+                let txid: Option<Txid> = ops.get_txid_async(idx).await?.map(Into::into);
+                info!(%idx, ?txid, "Received txentry");
 
                 // Insert into state's unfinalized entries. Need not update next_idx because that
                 // will be handled in state.next() call
@@ -66,22 +65,6 @@
     }
 }
 
-async fn get_txid(idx: u64, ops: &BroadcastDbOps) -> BroadcasterResult<Buf32> {
-    ops.get_txid_async(idx)
-        .await?
-        .ok_or(BroadcasterError::Other(format!(
-            "No txid entry found for idx {}",
-            idx
-        )))
-}
-
-async fn get_txid_str(idx: u64, ops: &BroadcastDbOps) -> BroadcasterResult<String> {
-    let txid: Buf32 = get_txid(idx, ops).await?;
-    let mut id = txid.0;
-    id.reverse();
-    Ok(hex::encode(id))
-}
-
 /// Processes unfinalized entries and returns entries idxs that are finalized
 async fn process_unfinalized_entries(
     unfinalized_entries: &BTreeMap<u64, L1TxEntry>,
@@ -92,7 +75,7 @@
     let mut updated_entries = BTreeMap::new();
 
     for (idx, txentry) in unfinalized_entries.iter() {
-        info!(%idx, "processing txentry");
+        debug!(%idx, "processing txentry");
         let updated_status = handle_entry(rpc_client, txentry, *idx, ops.as_ref()).await?;
 
         if let Some(status) = updated_status {
@@ -125,7 +108,10 @@
     idx: u64,
     ops: &BroadcastDbOps,
 ) -> BroadcasterResult<Option<L1TxStatus>> {
-    let txid = get_txid(idx, ops).await?;
+    let txid = ops
+        .get_txid_async(idx)
+        .await?
+        .ok_or(BroadcasterError::TxNotFound(idx))?;
     match txentry.status {
         L1TxStatus::Unpublished => {
             // Try to publish

crates/btcio/src/writer/builder.rs

@@ -12,12 +12,11 @@ use bitcoin::{
         script,
     },
     consensus::deserialize,
-    hashes::{sha256d, Hash},
+    hashes::Hash,
     key::{TapTweak, TweakedPublicKey, UntweakedKeypair},
     script::PushBytesBuf,
     secp256k1::{
-        self, constants::SCHNORR_SIGNATURE_SIZE, schnorr::Signature, Secp256k1, SecretKey,
-        XOnlyPublicKey,
+        self, constants::SCHNORR_SIGNATURE_SIZE, schnorr::Signature, Secp256k1, XOnlyPublicKey,
     },
     sighash::{Prevouts, SighashCache},
     taproot::{
@@ -103,7 +102,6 @@ pub async fn build_inscription_txs(
     rpc_client: &Arc<impl SeqL1Client + L1Client>,
     config: &WriterConfig,
 ) -> anyhow::Result<(Transaction, Transaction)> {
-    // let (signature, pub_key) = sign_blob_with_private_key(&payload, &config.private_key)?;
     let utxos = rpc_client.get_utxos().await?;
     let utxos = utxos
         .into_iter()
@@ -213,22 +211,6 @@ pub fn create_inscription_transactions(
     Ok((unsigned_commit_tx, reveal_tx))
 }
 
-// Signs a message with a private key
-pub fn sign_blob_with_private_key(
-    blob: &[u8],
-    private_key: &SecretKey,
-) -> anyhow::Result<(Vec<u8>, Vec<u8>)> {
-    let message = sha256d::Hash::hash(blob).to_byte_array();
-    let secp = Secp256k1::new();
-    let public_key = secp256k1::PublicKey::from_secret_key(&secp, private_key);
-    let msg = secp256k1::Message::from_digest_slice(&message).unwrap();
-    let sig = secp.sign_ecdsa(&msg, private_key);
-    Ok((
-        sig.serialize_compact().to_vec(),
-        public_key.serialize().to_vec(),
-    ))
-}
-
 fn get_size(
     inputs: &[TxIn],
     outputs: &[TxOut],

crates/btcio/src/writer/config.rs

@@ -31,7 +31,7 @@
             sequencer_address: addr,
             rollup_name,
             // TODO: get these from config as well
-            inscription_fee_policy: InscriptionFeePolicy::Fixed(100),
+            inscription_fee_policy: InscriptionFeePolicy::Smart,
             poll_duration_ms: 1000,
             amount_for_reveal_txn: 1000,
         })

crates/btcio/src/writer/mod.rs

@@ -1,8 +1,9 @@
-mod broadcast;
 mod builder;
 pub mod config;
-pub mod utils;
-mod watcher;
-mod writer_handler;
+mod signer;
+mod task;
 
-pub use writer_handler::*;
+#[cfg(test)]
+mod test_utils;
+
+pub use task::{start_inscription_task, InscriptionHandle};

crates/btcio/src/writer/signer.rs

@@ -0,0 +1,88 @@
+use std::sync::Arc;
+
+use alpen_express_db::types::{BlobEntry, L1TxEntry};
+use alpen_express_primitives::buf::Buf32;
+use bitcoin::Transaction;
+use tracing::debug;
+
+use super::{builder::build_inscription_txs, config::WriterConfig};
+use crate::{
+    broadcaster::L1BroadcastHandle,
+    rpc::traits::{L1Client, SeqL1Client},
+};
+
+type BlobIdx = u64;
+
+/// Create inscription transactions corresponding to a [`BlobEntry`].
+///
+/// This is useful when receiving a new intent as well as when
+/// broadcasting fails because the input UTXOs have been spent
+/// by something else already.
+pub async fn create_and_sign_blob_inscriptions(
+    blobentry: &BlobEntry,
+    bhandle: &L1BroadcastHandle,
+    client: Arc<impl L1Client + SeqL1Client>,
+    config: &WriterConfig,
+) -> anyhow::Result<(Buf32, Buf32)> {
+    let (commit, reveal) = build_inscription_txs(&blobentry.blob, &client, config).await?;
+
+    debug!("Signing commit transaction {}", commit.compute_txid());
+    let signed_commit: Transaction = client.sign_raw_transaction_with_wallet(commit).await?;
+
+    let cid: Buf32 = signed_commit.compute_txid().into();
+    let rid: Buf32 = reveal.compute_txid().into();
+
+    let centry = L1TxEntry::from_tx(&signed_commit);
+    let rentry = L1TxEntry::from_tx(&reveal);
+
+    // These don't need to be atomic. It will be handled by writer task if it does not find both
+    // commit-reveal txs in db by triggering re-signing.
+    let _ = bhandle.insert_new_tx_entry(cid, centry).await?;
+    let _ = bhandle.insert_new_tx_entry(rid, rentry).await?;
+    Ok((cid, rid))
+}
+
+#[cfg(test)]
+mod test {
+    use std::sync::Arc;
+
+    use alpen_express_db::types::{BlobEntry, BlobL1Status};
+    use alpen_express_primitives::hash;
+
+    use super::*;
+    use crate::{
+        test_utils::TestBitcoinClient,
+        writer::test_utils::{get_broadcast_handle, get_config, get_inscription_ops},
+    };
+
+    #[tokio::test]
+    async fn test_create_and_sign_blob_inscriptions() {
+        let iops = get_inscription_ops();
+        let bcast_handle = get_broadcast_handle();
+        let client = Arc::new(TestBitcoinClient::new(1));
+        let config = get_config();
+
+        // First insert an unsigned blob
+        let entry = BlobEntry::new_unsigned([1; 100].to_vec());
+
+        assert_eq!(entry.status, BlobL1Status::Unsigned);
+        assert_eq!(entry.commit_txid, Buf32::zero());
+        assert_eq!(entry.reveal_txid, Buf32::zero());
+
+        let intent_hash = hash::raw(&entry.blob);
+        iops.put_blob_entry_async(intent_hash, entry.clone())
+            .await
+            .unwrap();
+
+        let (cid, rid) =
+            create_and_sign_blob_inscriptions(&entry, bcast_handle.as_ref(), client, &config)
+                .await
+                .unwrap();
+
+        // Check if corresponding txs exist in db
+        let ctx = bcast_handle.get_tx_entry_by_id_async(cid).await.unwrap();
+        let rtx = bcast_handle.get_tx_entry_by_id_async(rid).await.unwrap();
+        assert!(ctx.is_some());
+        assert!(rtx.is_some());
+    }
+}