integration of frigate ephemeral scanning

Author: sdmg15

Cargo.lock

@@ -14,7 +14,7 @@ use bdk_sp::{
         hex::{DisplayHex, FromHex},
         key::Secp256k1,
         script::PushBytesBuf,
-        secp256k1::{PublicKey, Scalar},
+        secp256k1::{PublicKey, Scalar, SecretKey},
         Address, Amount, Block, BlockHash, FeeRate, Network, OutPoint, PrivateKey, ScriptBuf,
         Sequence, Transaction, TxOut, Txid,
     },
@@ -34,6 +34,7 @@ use bdk_sp_oracles::{
         TrustedPeer, UnboundedReceiver, Warning,
     },
     filters::kyoto::{FilterEvent, FilterSubscriber},
+    frigate::{self, FrigateClient, FrigateListener, History, SubscribeRequest},
     tweaks::blindbit::{BlindbitSubscriber, TweakEvent},
 };
 use bdk_sp_wallet::{
@@ -161,6 +162,24 @@ pub enum Commands {
         #[clap(long)]
         hash: Option<BlockHash>,
     },
+
+    ScanFrigrate {
+        #[clap(flatten)]
+        rpc_args: RpcArgs,
+        /// The scan private key for which outputs will be scanned for.
+        #[clap(long)]
+        scan_priv_key: SecretKey,
+        /// The spend public key for which outputs will be scanned for.
+        #[clap(long)]
+        spend_pub_key: PublicKey,
+        /// An optional start parameter from where the scanning will start. When not specified it starts from Taproot activation height.
+        #[clap(long)]
+        start: Option<u64>,
+        /// Optional list of labels to scan for. It always scan for change index with label 0.
+        #[clap(long)]
+        labels: Option<Vec<u32>>,
+    },
+
     Create {
         /// Network
         #[clap(long, short, default_value = "signet")]
@@ -567,6 +586,41 @@ async fn main() -> anyhow::Result<()> {
                 );
             }
         }
+        Commands::ScanFrigrate {
+            rpc_args,
+            scan_priv_key,
+            spend_pub_key,
+            start,
+            labels,
+        } => {
+            let client = FrigateClient::new(
+                &rpc_args.url,
+                rpc_args.rpc_user.as_deref(),
+                rpc_args.rpc_password.as_deref(),
+            )?;
+            // send a subscribe request
+            let subscribe_params = SubscribeRequest {
+                scan_priv_key,
+                spend_pub_key: spend_pub_key.into(),
+                start_height: start,
+                labels,
+            };
+
+            let (history_tx, mut history_rx) = tokio::sync::mpsc::unbounded_channel::<History>();
+            let frigate_listener = FrigateListener::new(client, history_tx);
+            tokio::task::spawn(async move { frigate_listener.run(subscribe_params).await });
+
+            loop {
+                if let Some(history) = history_rx.recv().await {
+                    if history.progress >= 1.0 {
+                        break;
+                    }
+                    // Make another request to .get the txdetails
+                    // Check that the outputs in the txs really belongs to the wallet
+                    // Update the wallet state with them
+                }
+            }
+        }
         Commands::Balance => {
             fn print_balances<'a>(
                 title_str: &'a str,

cli/v2/src/main.rs

@@ -13,9 +13,10 @@ redb = "2.4.0"
 rayon = "1.11.0"
 reqwest = { version = "0.12.23", features = ["json", "rustls-tls", "http2", "charset"], default-features = false }
 serde = { version = "1.0.219", features = ["serde_derive"] }
-serde_json = "1.0.142"
+serde_json = { version = "1.0.142", features = ["raw_value"]}
 url = "2.5.4"
 tracing = "0.1.41"
+jsonrpc = "=0.18.0"
 
 [lints]
 workspace = true

oracles/Cargo.toml

@@ -0,0 +1,184 @@
+use bip157::tokio;
+use bip157::tokio::io::{AsyncReadExt, AsyncWriteExt};
+use bip157::tokio::sync::mpsc::UnboundedSender;
+use bitcoin::{key::TweakedPublicKey, secp256k1::SecretKey, PublicKey};
+use jsonrpc::simple_http::{self, SimpleHttpTransport};
+use jsonrpc::Client;
+use serde::{Deserialize, Serialize};
+use serde_json::value::to_raw_value;
+use serde_json::Value;
+
+#[derive(Debug)]
+pub enum FrigateError {
+    JsonRpc(jsonrpc::Error),
+    ParseUrl(url::ParseError),
+    Serde(serde_json::Error),
+}
+
+impl From<serde_json::Error> for FrigateError {
+    fn from(value: serde_json::Error) -> Self {
+        FrigateError::Serde(value)
+    }
+}
+
+impl From<url::ParseError> for FrigateError {
+    fn from(value: url::ParseError) -> Self {
+        Self::ParseUrl(value)
+    }
+}
+
+impl From<jsonrpc::Error> for FrigateError {
+    fn from(value: jsonrpc::Error) -> Self {
+        Self::JsonRpc(value)
+    }
+}
+
+pub struct FrigateClient {
+    pub host_url: String,
+    client: Client,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct History {
+    pub height: u64,
+    pub tx_hash: bitcoin::BlockHash,
+    pub tweak_key: TweakedPublicKey,
+    pub progress: f32,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct NotifPayload {
+    scan_private_key: SecretKey,
+    spend_public_key: PublicKey,
+    address: String,
+    labels: Option<Vec<u32>>,
+    start_height: u64,
+    progress: f32,
+    history: Vec<History>,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct SubscribeRequest {
+    pub scan_priv_key: SecretKey,
+    pub spend_pub_key: PublicKey,
+    pub start_height: Option<u64>,
+    pub labels: Option<Vec<u32>>,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct UnsubscribeRequest {
+    pub scan_privkey: SecretKey,
+    pub spend_pubkey: PublicKey,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct GetRequest {
+    tx_hash: bitcoin::BlockHash,
+}
+
+const SUBSCRIBE_RPC_METHOD: &str = "blockchain.silentpayments.subscribe";
+const UNSUBSCRIBE_RPC_METHOD: &str = "blockchain.silentpayments.unsubscribe";
+const GET_RPC_METHOD: &str = "blockchain.transaction.get";
+const SUBSCRIBE_OWNED_OUTPUTS: &str = "blockchain.scripthash.subscribe";
+
+impl FrigateClient {
+    pub fn new(
+        host_url: &str,
+        user: Option<&str>,
+        password: Option<&str>,
+    ) -> Result<Self, simple_http::Error> {
+        let transport = SimpleHttpTransport::builder()
+            .url(host_url)?
+            .auth(user.unwrap_or(""), password)
+            .build();
+
+        Ok(Self {
+            host_url: host_url.to_string(),
+            client: Client::with_transport(transport),
+        })
+    }
+
+    pub fn subscribe(&self, req: &SubscribeRequest) -> Result<String, FrigateError> {
+        let params = to_raw_value(&serde_json::json!(req))?;
+        let req = self
+            .client
+            .build_request(SUBSCRIBE_RPC_METHOD, Some(&params));
+        let res = self.client.send_request(req)?;
+        Ok(res.result()?)
+    }
+
+    pub fn unsubscribe(&self, req: &UnsubscribeRequest) -> Result<String, FrigateError> {
+        let params = to_raw_value(&serde_json::json!(req))?;
+        let req = self
+            .client
+            .build_request(UNSUBSCRIBE_RPC_METHOD, Some(&params));
+        let res = self.client.send_request(req)?;
+        Ok(res.result()?)
+    }
+
+    pub fn get(&self, req: &GetRequest) -> Result<String, FrigateError> {
+        let params = to_raw_value(req)?;
+        let req = self.client.build_request(GET_RPC_METHOD, Some(&params));
+        Ok("".to_string())
+    }
+}
+
+pub struct FrigateListener {
+    frigate_client: FrigateClient,
+    sender: UnboundedSender<History>,
+}
+
+impl FrigateListener {
+    pub fn new(frigate_client: FrigateClient, sender: UnboundedSender<History>) -> Self {
+        Self {
+            frigate_client,
+            sender,
+        }
+    }
+
+    pub async fn run(&self, params: SubscribeRequest) -> Result<(), FrigateError> {
+        let raw_subscribe_params = serde_json::json!(params);
+        let mut tcp_stream = tokio::net::TcpStream::connect(self.frigate_client.host_url.clone())
+            .await
+            .map_err(|e| FrigateError::JsonRpc(jsonrpc::Error::Transport(Box::new(e))))?;
+        let (mut reader, mut writer) = tcp_stream.split();
+
+        writer
+            .write_all(raw_subscribe_params.to_string().as_bytes())
+            .await
+            .map_err(|e| FrigateError::JsonRpc(jsonrpc::Error::Transport(Box::new(e))))?;
+
+        loop {
+            let mut buffer = vec![0; 1024];
+            let n = reader
+                .read(&mut buffer)
+                .await
+                .map_err(|e| FrigateError::JsonRpc(jsonrpc::Error::Transport(Box::new(e))))?;
+            if n == 0 {
+                break;
+            }
+            let response_str = String::from_utf8_lossy(&buffer[..n]);
+            let result: Value = serde_json::from_str(&response_str).map_err(FrigateError::Serde)?;
+
+            if result["result"].is_string() {
+                tracing::info!("Subscribed to silent payment address: {:?}", result);
+            } else if result["params"].is_object() {
+                let mut history: History =
+                    serde_json::from_value(result["params"]["history"].clone())
+                        .map_err(FrigateError::Serde)?;
+                history.progress = result["params"]["progress"].as_f64().unwrap_or(0.0) as f32;
+                let progress = history.progress;
+
+                self.sender
+                    .send(history)
+                    .map_err(|e| FrigateError::JsonRpc(jsonrpc::Error::Transport(Box::new(e))))?;
+
+                if progress >= 1.0 {
+                    tracing::info!("Scanning completed");
+                    break;
+                }
+            }
+        }
+        Ok(())
+    }
+}

oracles/src/frigate/mod.rs

@@ -1,3 +1,4 @@
 pub mod filters;
+pub mod frigate;
 pub mod tweaks;
 pub use bip157;