working POC

Cargo.toml

@@ -28,14 +28,15 @@ panic = 'abort'     # Abort on panic
 default = []
 
 [dependencies]
-lightning = { version = "0.0.121", features = ["std"] }
-lightning-invoice = { version = "0.29.0" }
-lightning-net-tokio = { version = "0.0.121" }
-lightning-persister = { version = "0.0.121" }
-lightning-background-processor = { version = "0.0.121", features = ["futures"] }
-lightning-rapid-gossip-sync = { version = "0.0.121" }
-lightning-transaction-sync = { version = "0.0.121", features = ["esplora-async-https", "time"] }
-lightning-liquidity = { version = "0.1.0-alpha.1", features = ["std"] }
+# lightning = { version = "0.0.121", features = ["std"] }
+lightning = { path = "../rust-lightning/lightning", features = ["std"] }
+# lightning-invoice = { version = "0.29.0" }
+# lightning-net-tokio = { version = "0.0.121" }
+# lightning-persister = { version = "0.0.121" }
+# lightning-background-processor = { version = "0.0.121", features = ["futures"] }
+# lightning-rapid-gossip-sync = { version = "0.0.121" }
+# lightning-transaction-sync = { version = "0.0.121", features = ["esplora-async-https", "time"] }
+# lightning-liquidity = { version = "0.1.0-alpha.1", features = ["std"] }
 
 #lightning = { git = "https://github.com/lightningdevkit/rust-lightning", branch="main", features = ["std"] }
 #lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", branch="main" }
@@ -46,14 +47,13 @@ lightning-liquidity = { version = "0.1.0-alpha.1", features = ["std"] }
 #lightning-transaction-sync = { git = "https://github.com/lightningdevkit/rust-lightning", branch="main", features = ["esplora-async"] }
 #lightning-liquidity = { git = "https://github.com/lightningdevkit/lightning-liquidity", branch="main", features = ["std"] }
 
-#lightning = { path = "../rust-lightning/lightning", features = ["std"] }
-#lightning-invoice = { path = "../rust-lightning/lightning-invoice" }
-#lightning-net-tokio = { path = "../rust-lightning/lightning-net-tokio" }
-#lightning-persister = { path = "../rust-lightning/lightning-persister" }
-#lightning-background-processor = { path = "../rust-lightning/lightning-background-processor", features = ["futures"] }
-#lightning-rapid-gossip-sync = { path = "../rust-lightning/lightning-rapid-gossip-sync" }
-#lightning-transaction-sync = { path = "../rust-lightning/lightning-transaction-sync", features = ["esplora-async"] }
-#lightning-liquidity = { path = "../lightning-liquidity", features = ["std"] }
+lightning-invoice = { path = "../rust-lightning/lightning-invoice" }
+lightning-net-tokio = { path = "../rust-lightning/lightning-net-tokio" }
+lightning-persister = { path = "../rust-lightning/lightning-persister" }
+lightning-background-processor = { path = "../rust-lightning/lightning-background-processor", features = ["futures"] }
+lightning-rapid-gossip-sync = { path = "../rust-lightning/lightning-rapid-gossip-sync" }
+lightning-transaction-sync = { path = "../rust-lightning/lightning-transaction-sync", features = ["esplora-async"] }
+lightning-liquidity = { path = "../lightning-liquidity", features = ["std"] }
 
 bdk = { version = "0.29.0", default-features = false, features = ["std", "async-interface", "use-esplora-async", "sqlite-bundled", "keys-bip39"]}
 
@@ -83,7 +83,8 @@ prost = { version = "0.11.6", default-features = false}
 winapi = { version = "0.3", features = ["winbase"] }
 
 [dev-dependencies]
-lightning = { version = "0.0.121", features = ["std", "_test_utils"] }
+lightning = { path = "../rust-lightning/lightning", features = ["std", "_test_utils"] }
+# 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 }

src/builder.rs

@@ -12,6 +12,7 @@ use crate::logger::{log_error, FilesystemLogger, Logger};
 use crate::message_handler::NodeCustomMessageHandler;
 use crate::payment_store::PaymentStore;
 use crate::peer_store::PeerStore;
+use crate::pj_new_crate::ChannelScheduler;
 use crate::pjoin::LDKPayjoinExecuter;
 use crate::sweep::OutputSweeper;
 use crate::tx_broadcaster::TransactionBroadcaster;
@@ -558,6 +559,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)) => {
@@ -568,6 +570,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(),
@@ -586,6 +589,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(),
@@ -952,6 +956,7 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		Arc::clone(&logger),
 		Arc::clone(&peer_manager),
 		Arc::clone(&channel_manager),
+		Arc::clone(&channel_scheduler)
 	);
 	let payjoin = Arc::new(LDKPayjoin::new(payjoin_executer));
 
@@ -975,6 +980,7 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		chain_monitor,
 		output_sweeper,
 		payjoin,
+		channel_scheduler,
 		peer_manager,
 		keys_manager,
 		network_graph,

src/event.rs

@@ -1,3 +1,4 @@
+use crate::pj_new_crate::{ChannelScheduler, FundingTxParams};
 use crate::types::{Sweeper, Wallet};
 use crate::{
 	hex_utils, ChannelManager, Config, Error, NetworkGraph, PeerInfo, PeerStore, UserChannelId,
@@ -292,7 +293,7 @@ impl Future for EventFuture {
 	}
 }
 
-pub(crate) struct EventHandler<K: KVStore + Sync + Send, L: Deref>
+pub(crate) struct EventHandler<K: KVStore + Sync + Send + 'static, L: Deref>
 where
 	L::Target: Logger,
 {
@@ -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,
@@ -317,7 +319,7 @@ where
 		channel_manager: Arc<ChannelManager<K>>, output_sweeper: Arc<Sweeper<K>>,
 		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>,
+		logger: L, config: Arc<Config>, channel_scheduler: Arc<tokio::sync::Mutex<ChannelScheduler>>,
 	) -> Self {
 		Self {
 			event_queue,
@@ -326,6 +328,7 @@ where
 			output_sweeper,
 			network_graph,
 			payment_store,
+			channel_scheduler,
 			peer_store,
 			logger,
 			runtime,
@@ -340,8 +343,10 @@ where
 				counterparty_node_id,
 				channel_value_satoshis,
 				output_script,
+				user_channel_id,
 				..
 			} => {
+				dbg!("Entered FundingGenerationReady event handler");
 				// Construct the raw transaction with the output that is paid the amount of the
 				// channel.
 				let confirmation_target = ConfirmationTarget::NonAnchorChannelFee;
@@ -350,6 +355,22 @@ where
 				let cur_height = self.channel_manager.current_best_block().height();
 				let locktime = LockTime::from_height(cur_height).unwrap_or(LockTime::ZERO);
 
+				// payjoin scenario
+				let mut channel_scheduler = self.channel_scheduler.lock().await;
+				if channel_scheduler.channel(user_channel_id).is_some() {
+					dbg!("Entered payjoin channel scheduler scenario");
+					let funding_tx_params = FundingTxParams::new(
+						output_script.clone().into_bytes(),
+						confirmation_target,
+						locktime,
+						temporary_channel_id
+					);
+					channel_scheduler.add_funding_tx_params(user_channel_id, funding_tx_params);
+					dbg!("payjoin channel scheduler scenario completed");
+					return {};
+				}
+				dbg!("Didnt enter payjoin channel scheduler scenario");
+
 				// Sign the final funding transaction and broadcast it.
 				match self.wallet.create_funding_transaction(
 					output_script,

src/lib.rs

@@ -109,6 +109,7 @@ pub use error::Error as NodeError;
 use error::Error;
 
 pub use event::Event;
+use pj_new_crate::ChannelScheduler;
 pub use types::{BestBlock, ChannelConfig};
 use payjoin::Uri;
 mod pjoin;
@@ -195,6 +196,7 @@ pub struct Node<K: KVStore + Sync + Send + 'static> {
 	chain_monitor: Arc<ChainMonitor<K>>,
 	output_sweeper: Arc<Sweeper<K>>,
 	payjoin: Arc<LDKPayjoin<K>>,
+	channel_scheduler: Arc<tokio::sync::Mutex<ChannelScheduler>>,
 	peer_manager: Arc<PeerManager<K>>,
 	keys_manager: Arc<KeysManager>,
 	network_graph: Arc<NetworkGraph>,
@@ -667,6 +669,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
@@ -738,18 +741,33 @@ impl<K: KVStore + Sync + Send + 'static> Node<K> {
 
 	/// 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,
+		&self,
+		channel_amount_sats: u64,
+		push_msat: Option<u64>,
+		announce_channel: bool,
 		node_id: PublicKey,
+		address: SocketAddress,
 	) -> Result<String, Error> {
+		let user_channel_id: u128 = rand::thread_rng().gen::<u128>();
 		let channel =
-			ScheduledChannel::new(channel_amount_sats, push_msat, announce_channel, node_id);
-		self.payjoin.schedule(channel).await;
+			ScheduledChannel::new(channel_amount_sats, push_msat, announce_channel, node_id, user_channel_id, None, None);
+		self.channel_scheduler.lock().await.schedule(channel);
+		let announce_channel = true;
+		self.connect_open_channel_payjoin(
+			node_id,
+			address,
+			channel_amount_sats,
+			None,
+			None,
+			announce_channel,
+			user_channel_id
+		)?; // this should be stopped after `ACCEPT_CHANNEL`
 		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> {
+	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);
@@ -760,7 +778,7 @@ impl<K: KVStore + Sync + Send + 'static> Node<K> {
 
 	/// List all scheduled payjoin channels.
 	pub async fn list_scheduled_channels(&self) -> Result<Vec<ScheduledChannel>, Error> {
-		Ok(self.payjoin.list_scheduled_channels().await)
+		Ok(self.channel_scheduler.lock().await.channels.clone())
 	}
 
 	/// Disconnects all peers, stops all running background tasks, and shuts down [`Node`].
@@ -973,6 +991,76 @@ impl<K: KVStore + Sync + Send + 'static> Node<K> {
 		Ok(())
 	}
 
+	///  included `user_channel_id` in inputs
+	pub fn connect_open_channel_payjoin(
+		&self, node_id: PublicKey, address: SocketAddress, channel_amount_sats: u64,
+		push_to_counterparty_msat: Option<u64>, channel_config: Option<Arc<ChannelConfig>>,
+		announce_channel: bool, user_channel_id: u128,
+	) -> Result<UserChannelId, 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 cur_balance = self.wallet.get_balance()?;
+		// if cur_balance.get_spendable() < channel_amount_sats {
+		// 	log_error!(self.logger, "Unable to create channel due to insufficient funds.");
+		// 	return Err(Error::InsufficientFunds);
+		// }
+
+		let peer_info = PeerInfo { node_id, address };
+
+		let con_node_id = peer_info.node_id;
+		let con_addr = peer_info.address.clone();
+		let con_logger = Arc::clone(&self.logger);
+		let con_pm = Arc::clone(&self.peer_manager);
+
+		// We need to use our main runtime here as a local runtime might not be around to poll
+		// connection futures going forward.
+		tokio::task::block_in_place(move || {
+			runtime.block_on(async move {
+				connect_peer_if_necessary(con_node_id, con_addr, con_pm, con_logger).await
+			})
+		})?;
+
+		let channel_config = (*(channel_config.unwrap_or_default())).clone().into();
+		let user_config = UserConfig {
+			channel_handshake_limits: Default::default(),
+			channel_handshake_config: ChannelHandshakeConfig {
+				announced_channel: announce_channel,
+				..Default::default()
+			},
+			channel_config,
+			..Default::default()
+		};
+
+		let push_msat = push_to_counterparty_msat.unwrap_or(0);
+
+		match self.channel_manager.create_channel(
+			peer_info.node_id,
+			channel_amount_sats,
+			push_msat,
+			user_channel_id,
+			None,
+			Some(user_config),
+		) {
+			Ok(_) => {
+				log_info!(
+					self.logger,
+					"Initiated channel creation with peer {}. ",
+					peer_info.node_id
+				);
+				self.peer_store.add_peer(peer_info)?;
+				Ok(UserChannelId(user_channel_id))
+			},
+			Err(e) => {
+				log_error!(self.logger, "Failed to initiate channel creation: {:?}", e);
+				Err(Error::ChannelCreationFailed)
+			},
+		}
+	}
+
 	/// Connect to a node and open a new channel. Disconnects and re-connects are handled automatically
 	///
 	/// Disconnects and reconnects are handled automatically.
@@ -1964,3 +2052,15 @@ async fn do_connect_peer<K: KVStore + Sync + Send + 'static>(
 		},
 	}
 }
+
+
+// 1. user schedule channel
+//		1.1 qrcode created to scan
+// 2. user scan qrcode
+//		2.1 node receives payjoin request 
+//		2.2 http endpoint loops for x amount of time looking for PayjoinProposal
+// 3. node scans if any scheduled channels waiting
+//		3.1 node creates the requested channel 
+// 4. node wait for payjoin channel open requests in FundingGenerationReady state
+//		4.1 node creates funding tx with payjoin incoming transaction
+//		4.2 save in channel scheduler