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

src/builder.rs

@@ -11,6 +11,7 @@ use crate::io::sqlite_store::SqliteStore;
 use crate::liquidity::LiquiditySource;
 use crate::logger::{log_error, log_info, FilesystemLogger, Logger};
 use crate::message_handler::NodeCustomMessageHandler;
+use crate::payjoin_handler::PayjoinHandler;
 use crate::payment_store::PaymentStore;
 use crate::peer_store::PeerStore;
 use crate::tx_broadcaster::TransactionBroadcaster;
@@ -39,6 +40,7 @@ use lightning::util::persist::{
 use lightning::util::ser::ReadableArgs;
 use lightning::util::sweep::OutputSweeper;
 
+use lightning_payjoin::scheduler::ChannelScheduler;
 use lightning_persister::fs_store::FilesystemStore;
 
 use lightning_transaction_sync::EsploraSyncClient;
@@ -94,6 +96,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 }
@@ -173,6 +181,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 {
@@ -189,12 +198,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,
 		}
 	}
 
@@ -249,6 +260,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.
@@ -367,6 +387,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,
@@ -391,6 +412,7 @@ impl NodeBuilder {
 			seed_bytes,
 			logger,
 			kv_store,
+			self.payjoin_config.as_ref(),
 		)
 	}
 }
@@ -461,6 +483,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.
@@ -524,7 +553,7 @@ fn build_with_store_internal(
 	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<DynStore>,
+	logger: Arc<FilesystemLogger>, kv_store: Arc<DynStore>, payjoin_config: Option<&PayjoinConfig>,
 ) -> Result<Node, BuildError> {
 	// Initialize the on-chain wallet and chain access
 	let xprv = bitcoin::bip32::ExtendedPrivKey::new_master(config.network.into(), &seed_bytes)
@@ -557,6 +586,7 @@ fn build_with_store_internal(
 		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)) => {
@@ -567,6 +597,7 @@ fn build_with_store_internal(
 			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(),
@@ -585,6 +616,7 @@ fn build_with_store_internal(
 			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(),
@@ -974,6 +1006,17 @@ fn build_with_store_internal(
 	};
 
 	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));
@@ -994,6 +1037,8 @@ fn build_with_store_internal(
 		channel_manager,
 		chain_monitor,
 		output_sweeper,
+		payjoin_handler,
+		channel_scheduler,
 		peer_manager,
 		connection_manager,
 		keys_manager,

src/event.rs

@@ -28,6 +28,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;
@@ -305,6 +306,7 @@ where
 	runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>,
 	logger: L,
 	config: Arc<Config>,
+	channel_scheduler: Arc<tokio::sync::Mutex<ChannelScheduler>>,
 }
 
 impl<L: Deref + Clone + Sync + Send + 'static> EventHandler<L>
@@ -316,6 +318,7 @@ where
 		output_sweeper: Arc<Sweeper>, network_graph: Arc<NetworkGraph>,
 		payment_store: Arc<PaymentStore<L>>, peer_store: Arc<PeerStore<L>>,
 		runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>, logger: L, config: Arc<Config>,
+		channel_scheduler: Arc<tokio::sync::Mutex<ChannelScheduler>>,
 	) -> Self {
 		Self {
 			event_queue,
@@ -324,6 +327,7 @@ where
 			output_sweeper,
 			network_graph,
 			payment_store,
+			channel_scheduler,
 			peer_store,
 			logger,
 			runtime,
@@ -338,6 +342,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
@@ -348,6 +353,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.0,
+					);
+					return {};
+				}
+
 				// Sign the final funding transaction and broadcast it.
 				match self.wallet.create_funding_transaction(
 					output_script,

src/lib.rs

@@ -108,6 +108,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::ChannelConfig;
 mod payjoin_handler;
 
@@ -191,6 +193,8 @@ pub struct Node {
 	output_sweeper: Arc<Sweeper>,
 	peer_manager: Arc<PeerManager>,
 	connection_manager: Arc<ConnectionManager<Arc<FilesystemLogger>>>,
+	payjoin_handler: Option<Arc<PayjoinHandler>>,
+	channel_scheduler: Arc<tokio::sync::Mutex<ChannelScheduler>>,
 	keys_manager: Arc<KeysManager>,
 	network_graph: Arc<NetworkGraph>,
 	gossip_source: Arc<GossipSource>,
@@ -501,6 +505,28 @@ impl Node {
 			});
 		}
 
+		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_cm = Arc::clone(&self.connection_manager);
 		let connect_pm = Arc::clone(&self.peer_manager);
@@ -638,6 +664,7 @@ impl Node {
 			Arc::clone(&self.runtime),
 			Arc::clone(&self.logger),
 			Arc::clone(&self.config),
+			Arc::clone(&self.channel_scheduler),
 		));
 
 		// Setup background processing
@@ -707,6 +734,67 @@ impl Node {
 		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>();
+		let payjoin_handler = self.payjoin_handler.as_ref().ok_or(Error::WalletOperationFailed)?; // fix error
+		payjoin_handler
+			.schedule_channel(
+				bitcoin::Amount::from_sat(channel_amount_sats),
+				node_id,
+				user_channel_id,
+			)
+			.await;
+		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> {
+		if let Some(payjoin_handler) = self.payjoin_handler.clone() {
+			Ok(payjoin_handler.list_scheduled_channels().await)
+		} else {
+			Ok(Vec::new())
+		}
+	}
+
 	/// Disconnects all peers, stops all running background tasks, and shuts down [`Node`].
 	///
 	/// After this returns most API methods will return [`Error::NotRunning`].