mirror of https://github.com/lukechilds/sensei.git
John Cantrell
3 years ago
22 changed files with 1653 additions and 724 deletions
File diff suppressed because it is too large
@ -0,0 +1,300 @@ |
|||
use base64; |
|||
use bitcoin::blockdata::block::Block; |
|||
use bitcoin::blockdata::transaction::Transaction; |
|||
use bitcoin::consensus::encode; |
|||
use bitcoin::hash_types::{BlockHash, Txid}; |
|||
use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator}; |
|||
use lightning_block_sync::http::HttpEndpoint; |
|||
use lightning_block_sync::rpc::RpcClient; |
|||
use lightning_block_sync::{AsyncBlockSourceResult, BlockHeaderData, BlockSource}; |
|||
use serde_json; |
|||
use std::collections::HashMap; |
|||
use std::sync::atomic::{AtomicU32, Ordering}; |
|||
use std::sync::Arc; |
|||
use std::time::Duration; |
|||
use tokio::sync::Mutex; |
|||
|
|||
use bitcoin::hashes::hex::FromHex; |
|||
use lightning_block_sync::http::JsonResponse; |
|||
use std::convert::TryInto; |
|||
pub struct FeeResponse { |
|||
pub feerate_sat_per_kw: Option<u32>, |
|||
pub errored: bool, |
|||
} |
|||
|
|||
impl TryInto<FeeResponse> for JsonResponse { |
|||
type Error = std::io::Error; |
|||
fn try_into(self) -> std::io::Result<FeeResponse> { |
|||
let errored = !self.0["errors"].is_null(); |
|||
Ok(FeeResponse { |
|||
errored, |
|||
feerate_sat_per_kw: match self.0["feerate"].as_f64() { |
|||
// Bitcoin Core gives us a feerate in BTC/KvB, which we need to convert to
|
|||
// satoshis/KW. Thus, we first multiply by 10^8 to get satoshis, then divide by 4
|
|||
// to convert virtual-bytes into weight units.
|
|||
Some(feerate_btc_per_kvbyte) => { |
|||
Some((feerate_btc_per_kvbyte * 100_000_000.0 / 4.0).round() as u32) |
|||
} |
|||
None => None, |
|||
}, |
|||
}) |
|||
} |
|||
} |
|||
|
|||
pub struct BlockchainInfo { |
|||
pub latest_height: usize, |
|||
pub latest_blockhash: BlockHash, |
|||
pub chain: String, |
|||
} |
|||
|
|||
impl TryInto<BlockchainInfo> for JsonResponse { |
|||
type Error = std::io::Error; |
|||
fn try_into(self) -> std::io::Result<BlockchainInfo> { |
|||
Ok(BlockchainInfo { |
|||
latest_height: self.0["blocks"].as_u64().unwrap() as usize, |
|||
latest_blockhash: BlockHash::from_hex(self.0["bestblockhash"].as_str().unwrap()) |
|||
.unwrap(), |
|||
chain: self.0["chain"].as_str().unwrap().to_string(), |
|||
}) |
|||
} |
|||
} |
|||
pub struct BitcoindClient { |
|||
bitcoind_rpc_client: Arc<Mutex<RpcClient>>, |
|||
host: String, |
|||
port: u16, |
|||
rpc_user: String, |
|||
rpc_password: String, |
|||
fees: Arc<HashMap<Target, AtomicU32>>, |
|||
handle: tokio::runtime::Handle, |
|||
} |
|||
|
|||
#[derive(Clone, Eq, Hash, PartialEq)] |
|||
pub enum Target { |
|||
Background, |
|||
Normal, |
|||
HighPriority, |
|||
} |
|||
|
|||
impl BlockSource for &BitcoindClient { |
|||
fn get_header<'a>( |
|||
&'a mut self, |
|||
header_hash: &'a BlockHash, |
|||
height_hint: Option<u32>, |
|||
) -> AsyncBlockSourceResult<'a, BlockHeaderData> { |
|||
Box::pin(async move { |
|||
let mut rpc = self.bitcoind_rpc_client.lock().await; |
|||
rpc.get_header(header_hash, height_hint).await |
|||
}) |
|||
} |
|||
|
|||
fn get_block<'a>( |
|||
&'a mut self, |
|||
header_hash: &'a BlockHash, |
|||
) -> AsyncBlockSourceResult<'a, Block> { |
|||
Box::pin(async move { |
|||
let mut rpc = self.bitcoind_rpc_client.lock().await; |
|||
rpc.get_block(header_hash).await |
|||
}) |
|||
} |
|||
|
|||
fn get_best_block<'a>(&'a mut self) -> AsyncBlockSourceResult<(BlockHash, Option<u32>)> { |
|||
Box::pin(async move { |
|||
let mut rpc = self.bitcoind_rpc_client.lock().await; |
|||
rpc.get_best_block().await |
|||
}) |
|||
} |
|||
} |
|||
|
|||
/// The minimum feerate we are allowed to send, as specify by LDK.
|
|||
const MIN_FEERATE: u32 = 253; |
|||
|
|||
impl BitcoindClient { |
|||
pub async fn new( |
|||
host: String, |
|||
port: u16, |
|||
rpc_user: String, |
|||
rpc_password: String, |
|||
handle: tokio::runtime::Handle, |
|||
) -> std::io::Result<Self> { |
|||
let http_endpoint = HttpEndpoint::for_host(host.clone()).with_port(port); |
|||
let rpc_credentials = |
|||
base64::encode(format!("{}:{}", rpc_user.clone(), rpc_password.clone())); |
|||
let mut bitcoind_rpc_client = RpcClient::new(&rpc_credentials, http_endpoint)?; |
|||
let _dummy = bitcoind_rpc_client |
|||
.call_method::<BlockchainInfo>("getblockchaininfo", &vec![]) |
|||
.await |
|||
.map_err(|_| { |
|||
std::io::Error::new(std::io::ErrorKind::PermissionDenied, |
|||
"Failed to make initial call to bitcoind - please check your RPC user/password and access settings") |
|||
})?; |
|||
let mut fees: HashMap<Target, AtomicU32> = HashMap::new(); |
|||
fees.insert(Target::Background, AtomicU32::new(MIN_FEERATE)); |
|||
fees.insert(Target::Normal, AtomicU32::new(2000)); |
|||
fees.insert(Target::HighPriority, AtomicU32::new(5000)); |
|||
let client = Self { |
|||
bitcoind_rpc_client: Arc::new(Mutex::new(bitcoind_rpc_client)), |
|||
host, |
|||
port, |
|||
rpc_user, |
|||
rpc_password, |
|||
fees: Arc::new(fees), |
|||
handle: handle.clone(), |
|||
}; |
|||
BitcoindClient::poll_for_fee_estimates( |
|||
client.fees.clone(), |
|||
client.bitcoind_rpc_client.clone(), |
|||
handle, |
|||
); |
|||
Ok(client) |
|||
} |
|||
|
|||
fn poll_for_fee_estimates( |
|||
fees: Arc<HashMap<Target, AtomicU32>>, |
|||
rpc_client: Arc<Mutex<RpcClient>>, |
|||
handle: tokio::runtime::Handle, |
|||
) { |
|||
handle.spawn(async move { |
|||
loop { |
|||
let background_estimate = { |
|||
let mut rpc = rpc_client.lock().await; |
|||
let background_conf_target = serde_json::json!(144); |
|||
let background_estimate_mode = serde_json::json!("ECONOMICAL"); |
|||
let resp = rpc |
|||
.call_method::<FeeResponse>( |
|||
"estimatesmartfee", |
|||
&vec![background_conf_target, background_estimate_mode], |
|||
) |
|||
.await |
|||
.unwrap(); |
|||
match resp.feerate_sat_per_kw { |
|||
Some(feerate) => std::cmp::max(feerate, MIN_FEERATE), |
|||
None => MIN_FEERATE, |
|||
} |
|||
}; |
|||
|
|||
let normal_estimate = { |
|||
let mut rpc = rpc_client.lock().await; |
|||
let normal_conf_target = serde_json::json!(18); |
|||
let normal_estimate_mode = serde_json::json!("ECONOMICAL"); |
|||
let resp = rpc |
|||
.call_method::<FeeResponse>( |
|||
"estimatesmartfee", |
|||
&vec![normal_conf_target, normal_estimate_mode], |
|||
) |
|||
.await |
|||
.unwrap(); |
|||
match resp.feerate_sat_per_kw { |
|||
Some(feerate) => std::cmp::max(feerate, MIN_FEERATE), |
|||
None => 2000, |
|||
} |
|||
}; |
|||
|
|||
let high_prio_estimate = { |
|||
let mut rpc = rpc_client.lock().await; |
|||
let high_prio_conf_target = serde_json::json!(6); |
|||
let high_prio_estimate_mode = serde_json::json!("CONSERVATIVE"); |
|||
let resp = rpc |
|||
.call_method::<FeeResponse>( |
|||
"estimatesmartfee", |
|||
&vec![high_prio_conf_target, high_prio_estimate_mode], |
|||
) |
|||
.await |
|||
.unwrap(); |
|||
|
|||
match resp.feerate_sat_per_kw { |
|||
Some(feerate) => std::cmp::max(feerate, MIN_FEERATE), |
|||
None => 5000, |
|||
} |
|||
}; |
|||
|
|||
fees.get(&Target::Background) |
|||
.unwrap() |
|||
.store(background_estimate, Ordering::Release); |
|||
fees.get(&Target::Normal) |
|||
.unwrap() |
|||
.store(normal_estimate, Ordering::Release); |
|||
fees.get(&Target::HighPriority) |
|||
.unwrap() |
|||
.store(high_prio_estimate, Ordering::Release); |
|||
tokio::time::sleep(Duration::from_secs(60)).await; |
|||
} |
|||
}); |
|||
} |
|||
|
|||
pub fn get_new_rpc_client(&self) -> std::io::Result<RpcClient> { |
|||
let http_endpoint = HttpEndpoint::for_host(self.host.clone()).with_port(self.port); |
|||
let rpc_credentials = base64::encode(format!( |
|||
"{}:{}", |
|||
self.rpc_user.clone(), |
|||
self.rpc_password.clone() |
|||
)); |
|||
RpcClient::new(&rpc_credentials, http_endpoint) |
|||
} |
|||
|
|||
pub async fn send_raw_transaction(&self, raw_tx: String) { |
|||
let mut rpc = self.bitcoind_rpc_client.lock().await; |
|||
|
|||
let raw_tx_json = serde_json::json!(raw_tx); |
|||
rpc.call_method::<Txid>("sendrawtransaction", &[raw_tx_json]) |
|||
.await |
|||
.unwrap(); |
|||
} |
|||
|
|||
pub async fn get_blockchain_info(&self) -> BlockchainInfo { |
|||
let mut rpc = self.bitcoind_rpc_client.lock().await; |
|||
rpc.call_method::<BlockchainInfo>("getblockchaininfo", &vec![]) |
|||
.await |
|||
.unwrap() |
|||
} |
|||
} |
|||
|
|||
impl FeeEstimator for BitcoindClient { |
|||
fn get_est_sat_per_1000_weight(&self, confirmation_target: ConfirmationTarget) -> u32 { |
|||
match confirmation_target { |
|||
ConfirmationTarget::Background => self |
|||
.fees |
|||
.get(&Target::Background) |
|||
.unwrap() |
|||
.load(Ordering::Acquire), |
|||
ConfirmationTarget::Normal => self |
|||
.fees |
|||
.get(&Target::Normal) |
|||
.unwrap() |
|||
.load(Ordering::Acquire), |
|||
ConfirmationTarget::HighPriority => self |
|||
.fees |
|||
.get(&Target::HighPriority) |
|||
.unwrap() |
|||
.load(Ordering::Acquire), |
|||
} |
|||
} |
|||
} |
|||
|
|||
impl BroadcasterInterface for BitcoindClient { |
|||
fn broadcast_transaction(&self, tx: &Transaction) { |
|||
let bitcoind_rpc_client = self.bitcoind_rpc_client.clone(); |
|||
let tx_serialized = serde_json::json!(encode::serialize_hex(tx)); |
|||
self.handle.spawn(async move { |
|||
let mut rpc = bitcoind_rpc_client.lock().await; |
|||
// This may error due to RL calling `broadcast_transaction` with the same transaction
|
|||
// multiple times, but the error is safe to ignore.
|
|||
match rpc |
|||
.call_method::<Txid>("sendrawtransaction", &vec![tx_serialized]) |
|||
.await |
|||
{ |
|||
Ok(_) => {} |
|||
Err(e) => { |
|||
let err_str = e.get_ref().unwrap().to_string(); |
|||
if !err_str.contains("Transaction already in block chain") |
|||
&& !err_str.contains("Inputs missing or spent") |
|||
&& !err_str.contains("bad-txns-inputs-missingorspent") |
|||
&& !err_str.contains("non-BIP68-final") |
|||
&& !err_str.contains("insufficient fee, rejecting replacement ") |
|||
{ |
|||
panic!("{}", e); |
|||
} |
|||
} |
|||
} |
|||
}); |
|||
} |
|||
} |
|||
@ -0,0 +1,21 @@ |
|||
use std::sync::Arc; |
|||
|
|||
use bitcoin::Transaction; |
|||
use lightning::chain::chaininterface::BroadcasterInterface; |
|||
|
|||
use super::{bitcoind_client::BitcoindClient, listener_database::ListenerDatabase}; |
|||
|
|||
pub struct SenseiBroadcaster { |
|||
pub bitcoind_client: Arc<BitcoindClient>, |
|||
pub listener_database: ListenerDatabase, |
|||
} |
|||
|
|||
impl BroadcasterInterface for SenseiBroadcaster { |
|||
fn broadcast_transaction(&self, tx: &Transaction) { |
|||
self.bitcoind_client.broadcast_transaction(tx); |
|||
|
|||
// TODO: there's a bug here if the broadcast fails
|
|||
// best solution is to probably setup a zmq listener
|
|||
self.listener_database.process_mempool_tx(tx); |
|||
} |
|||
} |
|||
@ -0,0 +1,65 @@ |
|||
use std::{ |
|||
collections::HashMap, |
|||
sync::{Arc, Mutex}, |
|||
}; |
|||
|
|||
use crate::node::{ChainMonitor, ChannelManager}; |
|||
use bitcoin::{Block, BlockHeader}; |
|||
use lightning::chain::Listen; |
|||
|
|||
use super::listener_database::ListenerDatabase; |
|||
|
|||
pub struct SenseiChainListener { |
|||
listeners: Mutex<HashMap<String, (Arc<ChainMonitor>, Arc<ChannelManager>, ListenerDatabase)>>, |
|||
} |
|||
|
|||
impl SenseiChainListener { |
|||
pub fn new() -> Self { |
|||
Self { |
|||
listeners: Mutex::new(HashMap::new()), |
|||
} |
|||
} |
|||
|
|||
fn get_key( |
|||
&self, |
|||
listener: &(Arc<ChainMonitor>, Arc<ChannelManager>, ListenerDatabase), |
|||
) -> String { |
|||
listener.1.get_our_node_id().to_string() |
|||
} |
|||
|
|||
pub fn add_listener( |
|||
&self, |
|||
listener: (Arc<ChainMonitor>, Arc<ChannelManager>, ListenerDatabase), |
|||
) { |
|||
let mut listeners = self.listeners.lock().unwrap(); |
|||
listeners.insert(self.get_key(&listener), listener); |
|||
} |
|||
|
|||
pub fn remove_listener( |
|||
&self, |
|||
listener: (Arc<ChainMonitor>, Arc<ChannelManager>, ListenerDatabase), |
|||
) { |
|||
let mut listeners = self.listeners.lock().unwrap(); |
|||
listeners.remove(&self.get_key(&listener)); |
|||
} |
|||
} |
|||
|
|||
impl Listen for SenseiChainListener { |
|||
fn block_connected(&self, block: &Block, height: u32) { |
|||
let listeners = self.listeners.lock().unwrap(); |
|||
for (chain_monitor, channel_manager, listener_database) in listeners.values() { |
|||
channel_manager.block_connected(block, height); |
|||
chain_monitor.block_connected(block, height); |
|||
listener_database.block_connected(block, height); |
|||
} |
|||
} |
|||
|
|||
fn block_disconnected(&self, header: &BlockHeader, height: u32) { |
|||
let listeners = self.listeners.lock().unwrap(); |
|||
for (chain_monitor, channel_manager, listener_database) in listeners.values() { |
|||
channel_manager.block_disconnected(header, height); |
|||
chain_monitor.block_disconnected(header, height); |
|||
listener_database.block_disconnected(header, height); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,251 @@ |
|||
use bdk::{ |
|||
database::{BatchOperations, Database, SqliteDatabase}, |
|||
BlockTime, KeychainKind, LocalUtxo, TransactionDetails, |
|||
}; |
|||
use bitcoin::{Block, BlockHeader, OutPoint, Script, Transaction, TxOut, Txid}; |
|||
use lightning::chain::Listen; |
|||
|
|||
use crate::database::node::NodeDatabase; |
|||
|
|||
#[derive(Clone)] |
|||
pub struct ListenerDatabase { |
|||
bdk_db_path: String, |
|||
node_db_path: String, |
|||
} |
|||
|
|||
impl ListenerDatabase { |
|||
pub fn new(bdk_db_path: String, node_db_path: String) -> Self { |
|||
Self { |
|||
bdk_db_path, |
|||
node_db_path, |
|||
} |
|||
} |
|||
|
|||
pub fn process_mempool_tx(&self, tx: &Transaction) { |
|||
let mut database = SqliteDatabase::new(self.bdk_db_path.clone()); |
|||
let mut internal_max_deriv = None; |
|||
let mut external_max_deriv = None; |
|||
|
|||
self.process_tx( |
|||
tx, |
|||
&mut database, |
|||
None, |
|||
None, |
|||
&mut internal_max_deriv, |
|||
&mut external_max_deriv, |
|||
); |
|||
|
|||
let current_ext = database |
|||
.get_last_index(KeychainKind::External) |
|||
.unwrap() |
|||
.unwrap_or(0); |
|||
let first_ext_new = external_max_deriv.map(|x| x + 1).unwrap_or(0); |
|||
if first_ext_new > current_ext { |
|||
database |
|||
.set_last_index(KeychainKind::External, first_ext_new) |
|||
.unwrap(); |
|||
} |
|||
|
|||
let current_int = database |
|||
.get_last_index(KeychainKind::Internal) |
|||
.unwrap() |
|||
.unwrap_or(0); |
|||
let first_int_new = internal_max_deriv.map(|x| x + 1).unwrap_or(0); |
|||
if first_int_new > current_int { |
|||
database |
|||
.set_last_index(KeychainKind::Internal, first_int_new) |
|||
.unwrap(); |
|||
} |
|||
} |
|||
|
|||
pub fn process_tx( |
|||
&self, |
|||
tx: &Transaction, |
|||
database: &mut SqliteDatabase, |
|||
confirmation_height: Option<u32>, |
|||
confirmation_time: Option<u64>, |
|||
internal_max_deriv: &mut Option<u32>, |
|||
external_max_deriv: &mut Option<u32>, |
|||
) { |
|||
let mut incoming: u64 = 0; |
|||
let mut outgoing: u64 = 0; |
|||
|
|||
let mut inputs_sum: u64 = 0; |
|||
let mut outputs_sum: u64 = 0; |
|||
|
|||
// look for our own inputs
|
|||
for (i, input) in tx.input.iter().enumerate() { |
|||
if let Some(previous_output) = database |
|||
.get_previous_output(&input.previous_output) |
|||
.unwrap() |
|||
{ |
|||
inputs_sum += previous_output.value; |
|||
|
|||
if database.is_mine(&previous_output.script_pubkey).unwrap() { |
|||
outgoing += previous_output.value; |
|||
|
|||
database.del_utxo(&input.previous_output).unwrap(); |
|||
} |
|||
} |
|||
} |
|||
|
|||
for (i, output) in tx.output.iter().enumerate() { |
|||
// to compute the fees later
|
|||
outputs_sum += output.value; |
|||
|
|||
// this output is ours, we have a path to derive it
|
|||
if let Some((keychain, child)) = database |
|||
.get_path_from_script_pubkey(&output.script_pubkey) |
|||
.unwrap() |
|||
{ |
|||
database |
|||
.set_utxo(&LocalUtxo { |
|||
outpoint: OutPoint::new(tx.txid(), i as u32), |
|||
txout: output.clone(), |
|||
keychain, |
|||
}) |
|||
.unwrap(); |
|||
incoming += output.value; |
|||
|
|||
// TODO: implement this
|
|||
|
|||
if keychain == KeychainKind::Internal |
|||
&& (internal_max_deriv.is_none() || child > internal_max_deriv.unwrap_or(0)) |
|||
{ |
|||
*internal_max_deriv = Some(child); |
|||
} else if keychain == KeychainKind::External |
|||
&& (external_max_deriv.is_none() || child > external_max_deriv.unwrap_or(0)) |
|||
{ |
|||
*external_max_deriv = Some(child); |
|||
} |
|||
} |
|||
} |
|||
|
|||
if incoming > 0 || outgoing > 0 { |
|||
let tx = TransactionDetails { |
|||
txid: tx.txid(), |
|||
transaction: Some(tx.clone()), |
|||
received: incoming, |
|||
sent: outgoing, |
|||
confirmation_time: BlockTime::new(confirmation_height, confirmation_time), |
|||
verified: true, |
|||
fee: Some(inputs_sum.saturating_sub(outputs_sum)), |
|||
}; |
|||
|
|||
database.set_tx(&tx).unwrap(); |
|||
} |
|||
} |
|||
} |
|||
|
|||
impl Listen for ListenerDatabase { |
|||
fn block_connected(&self, block: &Block, height: u32) { |
|||
let mut database = SqliteDatabase::new(self.bdk_db_path.clone()); |
|||
let mut internal_max_deriv = None; |
|||
let mut external_max_deriv = None; |
|||
|
|||
// iterate all transactions in the block, looking for ones we care about
|
|||
for tx in &block.txdata { |
|||
self.process_tx( |
|||
tx, |
|||
&mut database, |
|||
Some(height), |
|||
Some(block.header.time.into()), |
|||
&mut internal_max_deriv, |
|||
&mut external_max_deriv, |
|||
) |
|||
} |
|||
|
|||
let current_ext = database |
|||
.get_last_index(KeychainKind::External) |
|||
.unwrap() |
|||
.unwrap_or(0); |
|||
let first_ext_new = external_max_deriv.map(|x| x + 1).unwrap_or(0); |
|||
if first_ext_new > current_ext { |
|||
database |
|||
.set_last_index(KeychainKind::External, first_ext_new) |
|||
.unwrap(); |
|||
} |
|||
|
|||
let current_int = database |
|||
.get_last_index(KeychainKind::Internal) |
|||
.unwrap() |
|||
.unwrap_or(0); |
|||
let first_int_new = internal_max_deriv.map(|x| x + 1).unwrap_or(0); |
|||
if first_int_new > current_int { |
|||
database |
|||
.set_last_index(KeychainKind::Internal, first_int_new) |
|||
.unwrap(); |
|||
} |
|||
|
|||
// TODO: there's probably a bug here.
|
|||
// need to atomicly update bdk database and this last_sync
|
|||
let mut node_database = NodeDatabase::new(self.node_db_path.clone()); |
|||
node_database.update_last_sync(block.block_hash()).unwrap(); |
|||
} |
|||
|
|||
fn block_disconnected(&self, header: &BlockHeader, height: u32) { |
|||
let mut database = SqliteDatabase::new(self.bdk_db_path.clone()); |
|||
let mut deleted_txids = vec![]; |
|||
|
|||
// delete all transactions with this height
|
|||
for details in database.iter_txs(false).unwrap() { |
|||
match details.confirmation_time { |
|||
Some(c) if c.height < height => continue, |
|||
_ => { |
|||
database.del_tx(&details.txid, false).unwrap(); |
|||
deleted_txids.push(details.txid) |
|||
} |
|||
}; |
|||
} |
|||
|
|||
// delete all utxos from the deleted txs
|
|||
if deleted_txids.len() > 0 { |
|||
for utxo in database.iter_utxos().unwrap() { |
|||
if deleted_txids.contains(&utxo.outpoint.txid) { |
|||
database.del_utxo(&utxo.outpoint).unwrap(); |
|||
} |
|||
} |
|||
} |
|||
|
|||
// TODO: update the keychain indexes?
|
|||
//
|
|||
|
|||
// TODO: there's probably a bug here.
|
|||
// need to atomicly update bdk database and this last_sync
|
|||
let mut node_database = NodeDatabase::new(self.node_db_path.clone()); |
|||
node_database |
|||
.update_last_sync(header.prev_blockhash) |
|||
.unwrap(); |
|||
} |
|||
} |
|||
|
|||
pub(crate) trait DatabaseUtils: Database { |
|||
fn is_mine(&self, script: &Script) -> Result<bool, bdk::Error> { |
|||
self.get_path_from_script_pubkey(script) |
|||
.map(|o| o.is_some()) |
|||
} |
|||
|
|||
fn get_raw_tx_or<D>(&self, txid: &Txid, default: D) -> Result<Option<Transaction>, bdk::Error> |
|||
where |
|||
D: FnOnce() -> Result<Option<Transaction>, bdk::Error>, |
|||
{ |
|||
self.get_tx(txid, true)? |
|||
.map(|t| t.transaction) |
|||
.flatten() |
|||
.map_or_else(default, |t| Ok(Some(t))) |
|||
} |
|||
|
|||
fn get_previous_output(&self, outpoint: &OutPoint) -> Result<Option<TxOut>, bdk::Error> { |
|||
self.get_raw_tx(&outpoint.txid)? |
|||
.map(|previous_tx| { |
|||
if outpoint.vout as usize >= previous_tx.output.len() { |
|||
Err(bdk::Error::InvalidOutpoint(*outpoint)) |
|||
} else { |
|||
Ok(previous_tx.output[outpoint.vout as usize].clone()) |
|||
} |
|||
}) |
|||
.transpose() |
|||
} |
|||
} |
|||
|
|||
impl<T: Database> DatabaseUtils for T {} |
|||
@ -0,0 +1,97 @@ |
|||
use std::{sync::Arc, time::Duration}; |
|||
|
|||
use crate::{ |
|||
config::SenseiConfig, |
|||
node::{ChainMonitor, ChannelManager}, |
|||
}; |
|||
use bitcoin::BlockHash; |
|||
use lightning::chain::{BestBlock, Listen}; |
|||
use lightning_block_sync::poll::ValidatedBlockHeader; |
|||
use lightning_block_sync::SpvClient; |
|||
use lightning_block_sync::{init, poll, UnboundedCache}; |
|||
use std::ops::Deref; |
|||
|
|||
use super::{ |
|||
bitcoind_client::BitcoindClient, listener::SenseiChainListener, |
|||
listener_database::ListenerDatabase, |
|||
}; |
|||
|
|||
pub struct SenseiChainManager { |
|||
config: SenseiConfig, |
|||
pub listener: Arc<SenseiChainListener>, |
|||
pub bitcoind_client: Arc<BitcoindClient>, |
|||
} |
|||
|
|||
impl SenseiChainManager { |
|||
pub async fn new(config: SenseiConfig) -> Result<Self, crate::error::Error> { |
|||
let listener = Arc::new(SenseiChainListener::new()); |
|||
|
|||
let bitcoind_client = Arc::new( |
|||
BitcoindClient::new( |
|||
config.bitcoind_rpc_host.clone(), |
|||
config.bitcoind_rpc_port, |
|||
config.bitcoind_rpc_username.clone(), |
|||
config.bitcoind_rpc_password.clone(), |
|||
tokio::runtime::Handle::current(), |
|||
) |
|||
.await |
|||
.expect("invalid bitcoind rpc config"), |
|||
); |
|||
|
|||
let block_source_poller = bitcoind_client.clone(); |
|||
let listener_poller = listener.clone(); |
|||
tokio::spawn(async move { |
|||
let derefed = &mut block_source_poller.deref(); |
|||
let mut cache = UnboundedCache::new(); |
|||
let chain_tip = init::validate_best_block_header(derefed).await.unwrap(); |
|||
let chain_poller = poll::ChainPoller::new(derefed, config.network); |
|||
let mut spv_client = |
|||
SpvClient::new(chain_tip, chain_poller, &mut cache, listener_poller); |
|||
loop { |
|||
spv_client.poll_best_tip().await.unwrap(); |
|||
tokio::time::sleep(Duration::from_secs(1)).await; |
|||
} |
|||
}); |
|||
|
|||
Ok(Self { |
|||
config, |
|||
listener, |
|||
bitcoind_client, |
|||
}) |
|||
} |
|||
|
|||
pub async fn synchronize_to_tip( |
|||
&self, |
|||
chain_listeners: Vec<(BlockHash, &(dyn Listen + Send + Sync))>, |
|||
) -> Result<ValidatedBlockHeader, crate::error::Error> { |
|||
let chain_tip = init::synchronize_listeners( |
|||
&mut self.bitcoind_client.deref(), |
|||
self.config.network, |
|||
&mut UnboundedCache::new(), |
|||
chain_listeners, |
|||
) |
|||
.await |
|||
.unwrap(); |
|||
|
|||
Ok(chain_tip) |
|||
} |
|||
|
|||
pub async fn keep_in_sync( |
|||
&self, |
|||
channel_manager: Arc<ChannelManager>, |
|||
chain_monitor: Arc<ChainMonitor>, |
|||
listener_database: ListenerDatabase, |
|||
) -> Result<(), crate::error::Error> { |
|||
let chain_listener = (chain_monitor, channel_manager, listener_database); |
|||
self.listener.add_listener(chain_listener); |
|||
Ok(()) |
|||
} |
|||
|
|||
pub async fn get_best_block(&self) -> Result<BestBlock, crate::error::Error> { |
|||
let blockchain_info = self.bitcoind_client.get_blockchain_info().await; |
|||
Ok(BestBlock::new( |
|||
blockchain_info.latest_blockhash, |
|||
blockchain_info.latest_height as u32, |
|||
)) |
|||
} |
|||
} |
|||
@ -0,0 +1,6 @@ |
|||
pub mod bitcoind_client; |
|||
pub mod broadcaster; |
|||
pub mod listener; |
|||
pub mod listener_database; |
|||
pub mod manager; |
|||
pub mod wallet; |
|||
@ -0,0 +1,128 @@ |
|||
use bdk::bitcoin::{Address, Script, Transaction}; |
|||
use bdk::blockchain::{noop_progress, Blockchain}; |
|||
use bdk::database::BatchDatabase; |
|||
use bdk::wallet::{AddressIndex, Wallet}; |
|||
use bdk::SignOptions; |
|||
|
|||
use lightning::chain::chaininterface::BroadcasterInterface; |
|||
use lightning::chain::chaininterface::{ConfirmationTarget, FeeEstimator}; |
|||
use std::sync::{Mutex, MutexGuard}; |
|||
use crate::error::Error; |
|||
|
|||
/// Lightning Wallet
|
|||
///
|
|||
/// A wrapper around a bdk::Wallet to fulfill many of the requirements
|
|||
/// needed to use lightning with LDK.
|
|||
pub struct LightningWallet<B, D> { |
|||
inner: Mutex<Wallet<B, D>>, |
|||
} |
|||
|
|||
impl<B, D> LightningWallet<B, D> |
|||
where |
|||
B: Blockchain, |
|||
D: BatchDatabase, |
|||
{ |
|||
/// create a new lightning wallet from your bdk wallet
|
|||
pub fn new(wallet: Wallet<B, D>) -> Self { |
|||
LightningWallet { |
|||
inner: Mutex::new(wallet), |
|||
} |
|||
} |
|||
|
|||
pub fn get_unused_address(&self) -> Result<Address, Error> { |
|||
let wallet = self.inner.lock().unwrap(); |
|||
let address_info = wallet.get_address(AddressIndex::LastUnused)?; |
|||
Ok(address_info.address) |
|||
} |
|||
|
|||
pub fn construct_funding_transaction( |
|||
&self, |
|||
output_script: &Script, |
|||
value: u64, |
|||
target_blocks: usize, |
|||
) -> Result<Transaction, Error> { |
|||
let wallet = self.inner.lock().unwrap(); |
|||
|
|||
let mut tx_builder = wallet.build_tx(); |
|||
let fee_rate = wallet.client().estimate_fee(target_blocks)?; |
|||
|
|||
tx_builder |
|||
.add_recipient(output_script.clone(), value) |
|||
.fee_rate(fee_rate) |
|||
.enable_rbf(); |
|||
|
|||
let (mut psbt, _tx_details) = tx_builder.finish()?; |
|||
|
|||
let _finalized = wallet.sign(&mut psbt, SignOptions::default())?; |
|||
|
|||
Ok(psbt.extract_tx()) |
|||
} |
|||
|
|||
pub fn get_balance(&self) -> Result<u64, Error> { |
|||
let wallet = self.inner.lock().unwrap(); |
|||
wallet.get_balance().map_err(Error::Bdk) |
|||
} |
|||
|
|||
pub fn get_wallet(&self) -> MutexGuard<Wallet<B, D>> { |
|||
self.inner.lock().unwrap() |
|||
} |
|||
|
|||
fn sync(&self) -> Result<(), Error> { |
|||
let wallet = self.inner.lock().unwrap(); |
|||
wallet.sync(noop_progress(), None)?; |
|||
Ok(()) |
|||
} |
|||
} |
|||
|
|||
impl<B, D> From<Wallet<B, D>> for LightningWallet<B, D> |
|||
where |
|||
B: Blockchain, |
|||
D: BatchDatabase, |
|||
{ |
|||
fn from(wallet: Wallet<B, D>) -> Self { |
|||
Self::new(wallet) |
|||
} |
|||
} |
|||
|
|||
impl<B, D> FeeEstimator for LightningWallet<B, D> |
|||
where |
|||
B: Blockchain, |
|||
D: BatchDatabase, |
|||
{ |
|||
fn get_est_sat_per_1000_weight(&self, confirmation_target: ConfirmationTarget) -> u32 { |
|||
let wallet = self.inner.lock().unwrap(); |
|||
|
|||
let target_blocks = match confirmation_target { |
|||
ConfirmationTarget::Background => 6, |
|||
ConfirmationTarget::Normal => 3, |
|||
ConfirmationTarget::HighPriority => 1, |
|||
}; |
|||
|
|||
let estimate = wallet |
|||
.client() |
|||
.estimate_fee(target_blocks) |
|||
.unwrap_or_default(); |
|||
let sats_per_vbyte = estimate.as_sat_vb() as u32; |
|||
sats_per_vbyte * 253 |
|||
} |
|||
} |
|||
|
|||
impl<B, D> BroadcasterInterface for LightningWallet<B, D> |
|||
where |
|||
B: Blockchain, |
|||
D: BatchDatabase, |
|||
{ |
|||
fn broadcast_transaction(&self, tx: &Transaction) { |
|||
let wallet = self.inner.lock().unwrap(); |
|||
let _result = wallet.client().broadcast(tx); |
|||
} |
|||
} |
|||
|
|||
#[cfg(test)] |
|||
mod tests { |
|||
#[test] |
|||
fn it_works() { |
|||
let result = 2 + 2; |
|||
assert_eq!(result, 4); |
|||
} |
|||
} |
|||
Loading…
Reference in new issue