f Add PayjoinScheduler and PayjoinExecuter

Author: jbesraa

Cargo.toml

@@ -57,7 +57,7 @@ lightning-liquidity = { version = "0.1.0-alpha.1", features = ["std"] }
 
 bdk = { version = "0.29.0", default-features = false, features = ["std", "async-interface", "use-esplora-async", "sqlite-bundled", "keys-bip39"]}
 
-reqwest = { version = "0.11", default-features = false, features = ["json", "rustls-tls"] }
+reqwest = { version = "0.11", default-features = false, features = ["json", "rustls-tls", "blocking"] }
 rusqlite = { version = "0.28.0", features = ["bundled"] }
 bitcoin = "0.30.2"
 bip39 = "2.0.0"
@@ -69,6 +69,11 @@ tokio = { version = "1", default-features = false, features = [ "rt-multi-thread
 esplora-client = { version = "0.6", default-features = false }
 libc = "0.2"
 uniffi = { version = "0.26.0", features = ["build"], optional = true }
+payjoin = { version = "0.13.0", features = ["receive", "send"] }
+http-body-util = "0.1.0"
+hyper = {version = "1.2.0", features = ["http1", "server"]}
+bytes = "1.5.0"
+hyper-util = {version = "0.1.3", features = ["tokio"] }
 
 [target.'cfg(vss)'.dependencies]
 vss-client = "0.2"

bindings/kotlin/ldk-node-jvm/lib/src/test/kotlin/org/lightningdevkit/ldknode/LibraryTest.kt

@@ -119,6 +119,7 @@ class LibraryTest {
         config1.listeningAddresses = listOf(listenAddress1)
         config1.network = Network.REGTEST
         config1.logLevel = LogLevel.TRACE
+        config1.payjoin_server_port = 1345
 
         println("Config 1: $config1")
 
@@ -127,6 +128,7 @@ class LibraryTest {
         config2.listeningAddresses = listOf(listenAddress2)
         config2.network = Network.REGTEST
         config2.logLevel = LogLevel.TRACE
+        config2.payjoin_server_port = 1346
         println("Config 2: $config2")
 
         val builder1 = Builder.fromConfig(config1)

bindings/ldk_node.udl

@@ -15,6 +15,7 @@ dictionary Config {
 	sequence<PublicKey> trusted_peers_0conf;
 	u64 probing_liquidity_limit_multiplier;
 	LogLevel log_level;
+	u16 payjoin_server_port;
 };
 
 interface Builder {

bindings/python/src/ldk_node/test_ldk_node.py

@@ -1,4 +1,5 @@
 import unittest
+import random
 import tempfile
 import time
 import subprocess
@@ -82,6 +83,7 @@ def send_to_address(address, amount_sats):
 
 def setup_node(tmp_dir, esplora_endpoint, listening_addresses):
     config = default_config()
+    config.payjoin_server_port = random.randint(3000, 6000)
     builder = Builder.from_config(config)
     builder.set_storage_dir_path(tmp_dir)
     builder.set_esplora_server(esplora_endpoint)

src/builder.rs

@@ -12,13 +12,15 @@ use crate::logger::{log_error, FilesystemLogger, Logger};
 use crate::message_handler::NodeCustomMessageHandler;
 use crate::payment_store::PaymentStore;
 use crate::peer_store::PeerStore;
+use crate::pjoin::LDKPayjoinExecuter;
 use crate::sweep::OutputSweeper;
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
 	ChainMonitor, ChannelManager, FakeMessageRouter, GossipSync, KeysManager, NetworkGraph,
 	OnionMessenger, PeerManager,
 };
 use crate::wallet::Wallet;
+use crate::LDKPayjoin;
 use crate::{LogLevel, Node};
 
 use lightning::chain::{chainmonitor, BestBlock, Watch};
@@ -944,6 +946,13 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		};
 
 	let (stop_sender, _) = tokio::sync::watch::channel(());
+	let payjoin_executer = LDKPayjoinExecuter::new(
+		Arc::clone(&wallet),
+		Arc::clone(&logger),
+		Arc::clone(&peer_manager),
+		Arc::clone(&channel_manager),
+	);
+	let payjoin = Arc::new(LDKPayjoin::new(payjoin_executer));
 
 	Ok(Node {
 		runtime,
@@ -957,6 +966,7 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		channel_manager,
 		chain_monitor,
 		output_sweeper,
+		payjoin,
 		peer_manager,
 		keys_manager,
 		network_graph,

src/config.rs

@@ -44,6 +44,9 @@ pub(crate) const WALLET_SYNC_INTERVAL_MINIMUM_SECS: u64 = 10;
 // The length in bytes of our wallets' keys seed.
 pub(crate) const WALLET_KEYS_SEED_LEN: usize = 64;
 
+// Port used by the Payjoin HTTP server.
+pub(crate) const DEFAULT_PAYJOIN_HTTP_SERVER_PORT: u16 = 3227;
+
 #[derive(Debug, Clone)]
 /// Represents the configuration of an [`Node`] instance.
 ///
@@ -104,6 +107,8 @@ pub struct Config {
 	///
 	/// Any messages below this level will be excluded from the logs.
 	pub log_level: LogLevel,
+	/// Payjoin server port
+	pub payjoin_server_port: u16,
 }
 
 impl Default for Config {
@@ -120,6 +125,7 @@ impl Default for Config {
 			trusted_peers_0conf: Vec::new(),
 			probing_liquidity_limit_multiplier: DEFAULT_PROBING_LIQUIDITY_LIMIT_MULTIPLIER,
 			log_level: DEFAULT_LOG_LEVEL,
+			payjoin_server_port: DEFAULT_PAYJOIN_HTTP_SERVER_PORT,
 		}
 	}
 }

src/lib.rs

@@ -89,14 +89,16 @@ mod logger;
 mod message_handler;
 mod payment_store;
 mod peer_store;
+mod pj_new_crate;
 mod sweep;
 mod tx_broadcaster;
 mod types;
 #[cfg(feature = "uniffi")]
 mod uniffi_types;
 mod wallet;
 
-pub use bip39;
+use crate::pj_new_crate::ScheduledChannel;
+use crate::pjoin::LDKPayjoin;
 pub use bitcoin;
 pub use lightning;
 pub use lightning_invoice;
@@ -107,6 +109,8 @@ pub use error::Error as NodeError;
 use error::Error;
 
 pub use event::Event;
+use payjoin::Uri;
+mod pjoin;
 pub use types::ChannelConfig;
 
 pub use io::utils::generate_entropy_mnemonic;
@@ -157,18 +161,18 @@ use lightning_transaction_sync::EsploraSyncClient;
 use lightning::routing::router::{PaymentParameters, RouteParameters};
 use lightning_invoice::{payment, Bolt11Invoice, Currency};
 
-use bitcoin::hashes::sha256::Hash as Sha256;
 use bitcoin::hashes::Hash;
 use bitcoin::secp256k1::PublicKey;
+use bitcoin::{hashes::sha256::Hash as Sha256, Amount};
 
 use bitcoin::{Address, Txid};
 
 use rand::Rng;
 
-use std::default::Default;
-use std::net::ToSocketAddrs;
+use std::net::{SocketAddr, ToSocketAddrs};
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::{Duration, Instant, SystemTime};
+use std::{default::Default, str::FromStr};
 
 #[cfg(feature = "uniffi")]
 uniffi::include_scaffolding!("ldk_node");
@@ -188,6 +192,7 @@ pub struct Node<K: KVStore + Sync + Send + 'static> {
 	channel_manager: Arc<ChannelManager<K>>,
 	chain_monitor: Arc<ChainMonitor<K>>,
 	output_sweeper: Arc<Sweeper<K>>,
+	payjoin: Arc<LDKPayjoin<K>>,
 	peer_manager: Arc<PeerManager<K>>,
 	keys_manager: Arc<KeysManager>,
 	network_graph: Arc<NetworkGraph>,
@@ -461,7 +466,30 @@ impl<K: KVStore + Sync + Send + 'static> Node<K> {
 			});
 		}
 
-		// Regularly reconnect to persisted peers.
+		let payjoin_handler = Arc::clone(&self.payjoin);
+		let mut stop_payjoin_server = self.stop_sender.subscribe();
+		let pj_port = self.config.payjoin_server_port;
+		runtime.spawn(async move {
+			let addr = SocketAddr::from(([127, 0, 0, 1], pj_port));
+			let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
+			loop {
+				let (stream, _) = match listener.accept().await {
+					Ok(res) => res,
+					Err(e) => {
+						println!("Failed to accept incoming payjoin connection: {}", e);
+						continue;
+					},
+				};
+				tokio::select! {
+					_ = stop_payjoin_server.changed() => {
+						return;
+					}
+					_ = payjoin_handler.serve(stream) => {}
+				}
+			}
+		});
+
+		// Regularly reconnect to channel peers.
 		let connect_pm = Arc::clone(&self.peer_manager);
 		let connect_logger = Arc::clone(&self.logger);
 		let connect_peer_store = Arc::clone(&self.peer_store);
@@ -667,6 +695,33 @@ impl<K: KVStore + Sync + Send + 'static> Node<K> {
 		self.runtime.read().unwrap().is_some()
 	}
 
+	/// Request a new channel to be opened with a remote peer.
+	pub async fn schedule_payjoin_channel(
+		&self, channel_amount_sats: u64, push_msat: Option<u64>, announce_channel: bool,
+		node_id: PublicKey,
+	) -> Result<String, Error> {
+		let channel =
+			ScheduledChannel::new(channel_amount_sats, push_msat, announce_channel, node_id);
+		self.payjoin.schedule(channel).await;
+		let bip21 = self.payjoin_bip21(channel_amount_sats);
+		bip21
+	}
+
+	/// Generate a BIP21 URI for a payjoin request.
+	pub fn payjoin_bip21(&self, amount_sats: u64) -> Result<String, Error> {
+		let address = self.wallet.get_new_address()?;
+		let amount = Amount::from_sat(amount_sats);
+		let pj = format!("https://0.0.0.0:{}/payjoin", self.config.payjoin_server_port);
+		let pj_uri_string = format!("{}?amount={}&pj={}", address.to_qr_uri(), amount.to_btc(), pj);
+		assert!(Uri::from_str(&pj_uri_string).is_ok());
+		Ok(pj_uri_string)
+	}
+
+	/// List all scheduled payjoin channels.
+	pub async fn list_scheduled_channels(&self) -> Result<Vec<ScheduledChannel>, Error> {
+		Ok(self.payjoin.list_scheduled_channels().await)
+	}
+
 	/// Disconnects all peers, stops all running background tasks, and shuts down [`Node`].
 	///
 	/// After this returns most API methods will return [`Error::NotRunning`].