Add ChannelScheduler and PayjoinHandler to Node

  `ChannelSchedulder` is used by `PayjoinHandler` in order to manage
  scheduled channels for opening a channel through incoming payjoin
  requests.

  `PayjoinHandler` starts an http server in the background and allows
  the node to accept payjoin transactions on the new config arguemnt
  `payjoin_server_port` utilising `lightning_payjoin` crate.

Author: jbesraa

Cargo.toml

@@ -86,12 +86,14 @@ winapi = { version = "0.3", features = ["winbase"] }
 
 [dev-dependencies]
 lightning = { git = "https://github.com/jbesraa/rust-lightning.git", rev = "d3e2d5a", features = ["std", "_test_utils"] }
+reqwest = { version = "0.11", default-features = false, features = ["blocking"] }
 # lightning = { version = "0.0.121", features = ["std", "_test_utils"] }
 #lightning = { git = "https://github.com/lightningdevkit/rust-lightning", branch="main", features = ["std", "_test_utils"] }
 electrum-client = { version = "0.15.1", default-features = true }
 bitcoincore-rpc = { version = "0.17.0", default-features = false }
 proptest = "1.0.0"
 regex = "1.5.6"
+payjoin = { git = "https://github.com/jbesraa/rust-payjoin.git", rev = "9e4f454", features = ["send"] }
 
 [target.'cfg(not(no_download))'.dev-dependencies]
 electrsd = { version = "0.26.0", features = ["legacy", "esplora_a33e97e1", "bitcoind_25_0"] }

src/builder.rs

@@ -10,6 +10,7 @@ use crate::io::sqlite_store::SqliteStore;
 use crate::liquidity::LiquiditySource;
 use crate::logger::{log_error, FilesystemLogger, Logger};
 use crate::message_handler::NodeCustomMessageHandler;
+use crate::payjoin_handler::PayjoinHandler;
 use crate::payment_store::PaymentStore;
 use crate::peer_store::PeerStore;
 use crate::sweep::OutputSweeper;
@@ -38,6 +39,7 @@ use lightning::util::persist::{
 };
 use lightning::util::ser::ReadableArgs;
 
+use lightning_payjoin::scheduler::ChannelScheduler;
 use lightning_persister::fs_store::FilesystemStore;
 
 use lightning_transaction_sync::EsploraSyncClient;
@@ -93,6 +95,12 @@ struct LiquiditySourceConfig {
 	lsps2_service: Option<(SocketAddress, PublicKey, Option<String>)>,
 }
 
+#[derive(Debug, Clone)]
+struct PayjoinConfig {
+	payjoin_directory: lightning_payjoin::Url,
+	payjoin_relay: lightning_payjoin::Url,
+}
+
 impl Default for LiquiditySourceConfig {
 	fn default() -> Self {
 		Self { lsps2_service: None }
@@ -166,6 +174,7 @@ pub struct NodeBuilder {
 	chain_data_source_config: Option<ChainDataSourceConfig>,
 	gossip_source_config: Option<GossipSourceConfig>,
 	liquidity_source_config: Option<LiquiditySourceConfig>,
+	payjoin_config: Option<PayjoinConfig>,
 }
 
 impl NodeBuilder {
@@ -182,12 +191,14 @@ impl NodeBuilder {
 		let chain_data_source_config = None;
 		let gossip_source_config = None;
 		let liquidity_source_config = None;
+		let payjoin_config = None;
 		Self {
 			config,
 			entropy_source_config,
 			chain_data_source_config,
 			gossip_source_config,
 			liquidity_source_config,
+			payjoin_config,
 		}
 	}
 
@@ -242,6 +253,15 @@ impl NodeBuilder {
 		self
 	}
 
+	/// Configures the [`Node`] instance to source its gossip data from the given RapidGossipSync
+	/// server.
+	pub fn set_payjoin_config(
+		&mut self, payjoin_directory: lightning_payjoin::Url, payjoin_relay: lightning_payjoin::Url,
+	) -> &mut Self {
+		self.payjoin_config = Some(PayjoinConfig { payjoin_directory, payjoin_relay });
+		self
+	}
+
 	/// Configures the [`Node`] instance to source its inbound liquidity from the given
 	/// [LSPS2](https://github.com/BitcoinAndLightningLayerSpecs/lsp/blob/main/LSPS2/README.md)
 	/// service.
@@ -362,6 +382,7 @@ impl NodeBuilder {
 			self.chain_data_source_config.as_ref(),
 			self.gossip_source_config.as_ref(),
 			self.liquidity_source_config.as_ref(),
+			self.payjoin_config.as_ref(),
 			seed_bytes,
 			logger,
 			vss_store,
@@ -385,6 +406,7 @@ impl NodeBuilder {
 			self.chain_data_source_config.as_ref(),
 			self.gossip_source_config.as_ref(),
 			self.liquidity_source_config.as_ref(),
+			self.payjoin_config.as_ref(),
 			seed_bytes,
 			logger,
 			kv_store,
@@ -458,6 +480,13 @@ impl ArcedNodeBuilder {
 		self.inner.write().unwrap().set_gossip_source_rgs(rgs_server_url);
 	}
 
+	/// Configures the [`Node`] instance to use payjoin.
+	pub fn set_payjoin_config(
+		&self, payjoin_directory: lightning_payjoin::Url, payjoin_relay: lightning_payjoin::Url,
+	) {
+		self.inner.write().unwrap().set_payjoin_config(payjoin_directory, payjoin_relay);
+	}
+
 	/// Configures the [`Node`] instance to source its inbound liquidity from the given
 	/// [LSPS2](https://github.com/BitcoinAndLightningLayerSpecs/lsp/blob/main/LSPS2/README.md)
 	/// service.
@@ -522,8 +551,9 @@ impl ArcedNodeBuilder {
 fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 	config: Arc<Config>, chain_data_source_config: Option<&ChainDataSourceConfig>,
 	gossip_source_config: Option<&GossipSourceConfig>,
-	liquidity_source_config: Option<&LiquiditySourceConfig>, seed_bytes: [u8; 64],
-	logger: Arc<FilesystemLogger>, kv_store: Arc<K>,
+	liquidity_source_config: Option<&LiquiditySourceConfig>,
+	payjoin_config: Option<&PayjoinConfig>, seed_bytes: [u8; 64], logger: Arc<FilesystemLogger>,
+	kv_store: Arc<K>,
 ) -> Result<Node<K>, BuildError> {
 	// Initialize the on-chain wallet and chain access
 	let xprv = bitcoin::bip32::ExtendedPrivKey::new_master(config.network.into(), &seed_bytes)
@@ -556,6 +586,7 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		log_error!(logger, "Failed to set up wallet: {}", e);
 		BuildError::WalletSetupFailed
 	})?;
+	let channel_scheduler = Arc::new(tokio::sync::Mutex::new(ChannelScheduler::new()));
 
 	let (blockchain, tx_sync, tx_broadcaster, fee_estimator) = match chain_data_source_config {
 		Some(ChainDataSourceConfig::Esplora(server_url)) => {
@@ -566,6 +597,7 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 			let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
 				tx_sync.client().clone(),
 				Arc::clone(&logger),
+				Arc::clone(&channel_scheduler),
 			));
 			let fee_estimator = Arc::new(OnchainFeeEstimator::new(
 				tx_sync.client().clone(),
@@ -584,6 +616,7 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 			let tx_broadcaster = Arc::new(TransactionBroadcaster::new(
 				tx_sync.client().clone(),
 				Arc::clone(&logger),
+				Arc::clone(&channel_scheduler),
 			));
 			let fee_estimator = Arc::new(OnchainFeeEstimator::new(
 				tx_sync.client().clone(),
@@ -945,6 +978,17 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		};
 
 	let (stop_sender, _) = tokio::sync::watch::channel(());
+	let payjoin_handler = if let Some(payjoin_config) = payjoin_config {
+		Some(Arc::new(PayjoinHandler::new(
+			Arc::clone(&wallet),
+			Arc::clone(&channel_manager),
+			Arc::clone(&channel_scheduler),
+			payjoin_config.payjoin_directory.clone(),
+			payjoin_config.payjoin_relay.clone(),
+		)))
+	} else {
+		None
+	};
 
 	let is_listening = Arc::new(AtomicBool::new(false));
 	let latest_wallet_sync_timestamp = Arc::new(RwLock::new(None));
@@ -965,6 +1009,8 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		channel_manager,
 		chain_monitor,
 		output_sweeper,
+		payjoin_handler,
+		channel_scheduler,
 		peer_manager,
 		keys_manager,
 		network_graph,

src/event.rs

@@ -29,6 +29,7 @@ use bitcoin::blockdata::locktime::absolute::LockTime;
 use bitcoin::secp256k1::PublicKey;
 use bitcoin::OutPoint;
 
+use lightning_payjoin::scheduler::ChannelScheduler;
 use rand::{thread_rng, Rng};
 
 use core::future::Future;
@@ -302,6 +303,7 @@ where
 	output_sweeper: Arc<Sweeper<K>>,
 	network_graph: Arc<NetworkGraph>,
 	payment_store: Arc<PaymentStore<K, L>>,
+	channel_scheduler: Arc<tokio::sync::Mutex<ChannelScheduler>>,
 	peer_store: Arc<PeerStore<K, L>>,
 	runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>,
 	logger: L,
@@ -318,6 +320,7 @@ where
 		network_graph: Arc<NetworkGraph>, payment_store: Arc<PaymentStore<K, L>>,
 		peer_store: Arc<PeerStore<K, L>>, runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>,
 		logger: L, config: Arc<Config>,
+		channel_scheduler: Arc<tokio::sync::Mutex<ChannelScheduler>>,
 	) -> Self {
 		Self {
 			event_queue,
@@ -326,6 +329,7 @@ where
 			output_sweeper,
 			network_graph,
 			payment_store,
+			channel_scheduler,
 			peer_store,
 			logger,
 			runtime,
@@ -340,6 +344,7 @@ where
 				counterparty_node_id,
 				channel_value_satoshis,
 				output_script,
+				user_channel_id,
 				..
 			} => {
 				// Construct the raw transaction with the output that is paid the amount of the
@@ -350,6 +355,16 @@ where
 				let cur_height = self.channel_manager.current_best_block().height();
 				let locktime = LockTime::from_height(cur_height).unwrap_or(LockTime::ZERO);
 
+				let mut channel_scheduler = self.channel_scheduler.lock().await;
+				if channel_scheduler.is_channel_created(user_channel_id) {
+					channel_scheduler.set_channel_accepted(
+						user_channel_id,
+						output_script,
+						temporary_channel_id,
+					);
+					return {};
+				}
+
 				// Sign the final funding transaction and broadcast it.
 				match self.wallet.create_funding_transaction(
 					output_script,

src/lib.rs

@@ -107,6 +107,8 @@ pub use error::Error as NodeError;
 use error::Error;
 
 pub use event::Event;
+use lightning_payjoin::scheduler::{ChannelScheduler, ScheduledChannel};
+use payjoin_handler::PayjoinHandler;
 pub use types::{BestBlock, ChannelConfig};
 mod payjoin_handler;
 
@@ -190,6 +192,8 @@ pub struct Node<K: KVStore + Sync + Send + 'static> {
 	channel_manager: Arc<ChannelManager<K>>,
 	chain_monitor: Arc<ChainMonitor<K>>,
 	output_sweeper: Arc<Sweeper<K>>,
+	payjoin_handler: Option<Arc<PayjoinHandler<K>>>,
+	channel_scheduler: Arc<tokio::sync::Mutex<ChannelScheduler>>,
 	peer_manager: Arc<PeerManager<K>>,
 	keys_manager: Arc<KeysManager>,
 	network_graph: Arc<NetworkGraph>,
@@ -501,6 +505,28 @@ impl<K: KVStore + Sync + Send + 'static> Node<K> {
 			});
 		}
 
+		if let Some(payjoin_handler) = &self.payjoin_handler {
+			let mut stop_payjoin_server = self.stop_sender.subscribe();
+			let payjoin_handler = Arc::clone(&payjoin_handler);
+			let payjoin_check_interval = 2;
+			runtime.spawn(async move {
+				let mut payjoin_interval =
+					tokio::time::interval(Duration::from_secs(payjoin_check_interval));
+				payjoin_interval.reset();
+				payjoin_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+				loop {
+					tokio::select! {
+						_ = stop_payjoin_server.changed() => {
+							return;
+						}
+						_ = payjoin_interval.tick() => {
+							let _ = payjoin_handler.process_request().await;
+						}
+					}
+				}
+			});
+		}
+
 		// Regularly reconnect to persisted peers.
 		let connect_pm = Arc::clone(&self.peer_manager);
 		let connect_logger = Arc::clone(&self.logger);
@@ -639,6 +665,7 @@ impl<K: KVStore + Sync + Send + 'static> Node<K> {
 			Arc::clone(&self.runtime),
 			Arc::clone(&self.logger),
 			Arc::clone(&self.config),
+			Arc::clone(&self.channel_scheduler),
 		));
 
 		// Setup background processing
@@ -708,6 +735,60 @@ impl<K: KVStore + Sync + Send + 'static> Node<K> {
 		Ok(())
 	}
 
+	/// Open a channel with funds coming from a payjoin transaction.
+	///
+	/// This method will add a new channel to the scheduler and start the process of opening a
+	/// channel with the give node until it reaches the `ChannelAccepted` state.
+	///
+	/// In order to finalise the channel opening process, the user must make a payjoin payment to
+	/// the returned BIP21 URL.
+	pub async fn open_payjoin_channel(
+		&self, channel_amount_sats: u64, push_msat: Option<u64>, announce_channel: bool,
+		node_id: PublicKey, address: SocketAddress,
+	) -> Result<String, Error> {
+		let rt_lock = self.runtime.read().unwrap();
+		if rt_lock.is_none() {
+			return Err(Error::NotRunning);
+		}
+		let runtime = rt_lock.as_ref().unwrap();
+		let user_channel_id: u128 = rand::thread_rng().gen::<u128>();
+		self.channel_scheduler.lock().await.schedule(
+			bitcoin::Amount::from_sat(channel_amount_sats),
+			node_id,
+			user_channel_id,
+		);
+		let user_config = UserConfig {
+			channel_handshake_limits: Default::default(),
+			channel_handshake_config: ChannelHandshakeConfig {
+				announced_channel: announce_channel,
+				..Default::default()
+			},
+			..Default::default()
+		};
+		let push_msat = push_msat.unwrap_or(0);
+		self.internal_connect_open_channel(
+			node_id,
+			channel_amount_sats,
+			push_msat,
+			user_channel_id,
+			address,
+			Some(user_config),
+			runtime,
+		)?;
+		let payjoin_uri = self
+			.payjoin_handler
+			.as_ref()
+			.ok_or(Error::WalletOperationFailed)?
+			.payjoin_uri(bitcoin::Amount::from_sat(channel_amount_sats))
+			.await;
+		Ok(payjoin_uri)
+	}
+
+	/// List all scheduled payjoin channels.
+	pub async fn list_scheduled_channels(&self) -> Result<Vec<ScheduledChannel>, Error> {
+		Ok(self.channel_scheduler.lock().await.list_channels().clone())
+	}
+
 	/// Disconnects all peers, stops all running background tasks, and shuts down [`Node`].
 	///
 	/// After this returns most API methods will return [`Error::NotRunning`].