Skip to content

Commit

Permalink
multi: send offer payment
Browse files Browse the repository at this point in the history
  • Loading branch information
@orbitalturtle
orbitalturtle committed Feb 8, 2024
1 parent e3e520c commit 2184d97
Show file tree
Hide file tree
Showing 3 changed files with 330 additions and 52 deletions.
114 changes: 109 additions & 5 deletions src/lib.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,14 +8,15 @@ mod rate_limit;
use crate::lnd::{
features_support_onion_messages, get_lnd_client, string_to_network, LndCfg, LndNodeSigner,
};
use crate::lndk_offers::OfferError;
use crate::lndk_offers::{OfferError, PayInvoiceParams};
use crate::onion_messenger::MessengerUtilities;
use bitcoin::network::constants::Network;
use bitcoin::secp256k1::{Error as Secp256k1Error, PublicKey, Secp256k1};
use home::home_dir;
use lightning::blinded_path::BlindedPath;
use lightning::ln::inbound_payment::ExpandedKey;
use lightning::ln::peer_handler::IgnoringMessageHandler;
use lightning::offers::invoice::Bolt12Invoice;
use lightning::offers::invoice_error::InvoiceError;
use lightning::offers::offer::Offer;
use lightning::onion_message::messenger::{
Expand All @@ -33,6 +34,7 @@ use std::collections::HashMap;
use std::str::FromStr;
use std::sync::{Mutex, Once};
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::time::{sleep, Duration};
use tonic_lnd::lnrpc::GetInfoRequest;
use tonic_lnd::Client;
use triggered::{Listener, Trigger};
Expand Down Expand Up @@ -189,6 +191,7 @@ enum OfferState {

pub struct OfferHandler {
active_offers: Mutex<HashMap<String, OfferState>>,
active_invoices: Mutex<Vec<Bolt12Invoice>>,
pending_messages: Mutex<Vec<PendingOnionMessage<OffersMessage>>>,
pub messenger_utils: MessengerUtilities,
expanded_key: ExpandedKey,
Expand All @@ -214,6 +217,7 @@ impl OfferHandler {

OfferHandler {
active_offers: Mutex::new(HashMap::new()),
active_invoices: Mutex::new(Vec::new()),
pending_messages: Mutex::new(Vec::new()),
messenger_utils,
expanded_key,
Expand All @@ -226,8 +230,101 @@ impl OfferHandler {
cfg: PayOfferParams,
started: Receiver<u32>,
) -> Result<(), OfferError<Secp256k1Error>> {
self.send_invoice_request(cfg, started).await?;
Ok(())
let client_clone = cfg.client.clone();
let offer_id = cfg.offer.clone().to_string();
let validated_amount = self.send_invoice_request(cfg, started).await?;

//// Wait for onion messenger to give us the signal that it's ready. Once the onion messenger drops
//// the channel sender, recv will return None and we'll stop blocking here.
//while (started.recv().await).is_some() {
// println!("Error: we shouldn't receive any messages on this channel");
//}

//let validated_amount = validate_amount(&cfg.offer, cfg.amount).await?;

//// For now we connect directly to the introduction node of the blinded path so we don't need any
//// intermediate nodes here. In the future we'll query for a full path to the introduction node for
//// better sender privacy.
//connect_to_peer(cfg.client.clone(), cfg.blinded_path.introduction_node_id).await?;

//let offer_id = cfg.offer.clone().to_string();
//{
// let mut active_offers = self.active_offers.lock().unwrap();
// if active_offers.contains_key(&offer_id.clone()) {
// return Err(OfferError::AlreadyProcessing);
// }
// active_offers.insert(cfg.offer.to_string().clone(), OfferState::OfferAdded);
//}

//let invoice_request = self
// .create_invoice_request(
// cfg.client.clone(),
// cfg.offer.clone(),
// vec![],
// cfg.network,
// validated_amount,
// )
// .await?;

//if cfg.reply_path.is_none() {
// let info = cfg
// .client
// .lightning()
// .get_info(GetInfoRequest {})
// .await
// .expect("failed to get info")
// .into_inner();

// let pubkey = PublicKey::from_str(&info.identity_pubkey).unwrap();
// cfg.reply_path = Some(self.create_reply_path(cfg.client.clone(), pubkey).await?)
//};
//let contents = OffersMessage::InvoiceRequest(invoice_request);
//let pending_message = PendingOnionMessage {
// contents,
// destination: Destination::BlindedPath(cfg.blinded_path.clone()),
// reply_path: cfg.reply_path,
//};

//{
// let mut pending_messages = self.pending_messages.lock().unwrap();
// pending_messages.push(pending_message);

// let mut active_offers = self.active_offers.lock().unwrap();
// active_offers.insert(offer_id.clone(), OfferState::InvoiceRequestSent);
//}

let invoice = self.wait_for_invoice().await;
{
let mut active_offers = self.active_offers.lock().unwrap();
active_offers.insert(offer_id.clone(), OfferState::InvoiceReceived);
}

let payment_hash = invoice.payment_hash();
let path_info = invoice.payment_paths()[0].clone();

let params = PayInvoiceParams {
path: path_info.1,
cltv_expiry_delta: path_info.0.cltv_expiry_delta,
fee_base_msat: path_info.0.fee_base_msat,
payment_hash: payment_hash.0,
msats: validated_amount,
offer_id,
};

self.pay_invoice(client_clone, params).await
}

// wait_for_invoice waits for the offer creator to respond with an invoice.
pub async fn wait_for_invoice(&self) -> Bolt12Invoice {
loop {
{
let mut active_invoices = self.active_invoices.lock().unwrap();
if active_invoices.len() == 1 {
return active_invoices.pop().unwrap();
}
}
sleep(Duration::from_secs(2)).await;
}
}
}

Expand All @@ -252,13 +349,20 @@ impl OffersMessageHandler for OfferHandler {
// returned payment id below to check if we already processed an invoice for
// this payment. Right now it's safe to let this be because we won't try to pay
// a second invoice (if it comes through).
Ok(_payment_id) => Some(OffersMessage::Invoice(invoice)),
Ok(_payment_id) => {
let mut active_invoices = self.active_invoices.lock().unwrap();
active_invoices.push(invoice.clone());
Some(OffersMessage::Invoice(invoice))
}
Err(()) => Some(OffersMessage::InvoiceError(InvoiceError::from_string(
String::from("invoice verification failure"),
))),
}
}
OffersMessage::InvoiceError(_error) => None,
OffersMessage::InvoiceError(error) => {
log::error!("Invoice error received: {}", error);
None
}
}
}

Expand Down
123 changes: 76 additions & 47 deletions src/lndk_offers.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ impl OfferHandler {
&self,
mut cfg: PayOfferParams,
mut started: Receiver<u32>,
) -> Result<(), OfferError<bitcoin::secp256k1::Error>> {
) -> Result<u64, OfferError<bitcoin::secp256k1::Error>> {
// Wait for onion messenger to give us the signal that it's ready. Once the onion messenger drops
// the channel sender, recv will return None and we'll stop blocking here.
while (started.recv().await).is_some() {
Expand Down Expand Up @@ -154,7 +154,7 @@ impl OfferHandler {
pending_messages.push(pending_message);
std::mem::drop(pending_messages);

Ok(())
Ok(validated_amount)
}

// create_invoice_request builds and signs an invoice request, the first step in the BOLT 12 process of paying an offer.
Expand Down Expand Up @@ -254,6 +254,48 @@ impl OfferHandler {
}))?
}
}

/// pay_invoice tries to pay the provided invoice.
pub(crate) async fn pay_invoice(
&self,
mut payer: impl InvoicePayer + std::marker::Send + 'static,
params: PayInvoiceParams,
) -> Result<(), OfferError<Secp256k1Error>> {
let resp = payer
.query_routes(
params.path,
params.cltv_expiry_delta,
params.fee_base_msat,
params.msats,
)
.await
.map_err(OfferError::RouteFailure)?;

let _ = payer
.send_to_route(params.payment_hash, resp.routes[0].clone())
.await
.map_err(OfferError::RouteFailure)?;

{
let mut active_offers = self.active_offers.lock().unwrap();
active_offers.insert(params.offer_id, OfferState::InvoicePaymentDispatched);
}

// We'll track the payment until it settles.
payer
.track_payment(params.payment_hash)
.await
.map_err(|_| OfferError::PaymentFailure)
}
}

pub struct PayInvoiceParams {
pub path: BlindedPath,
pub cltv_expiry_delta: u16,
pub fee_base_msat: u32,
pub payment_hash: [u8; 32],
pub msats: u64,
pub offer_id: String,
}

// Checks that the user-provided amount matches the offer.
Expand Down Expand Up @@ -389,32 +431,6 @@ pub async fn connect_to_peer(
Ok(())
}

// pay_invoice tries to pay the provided invoice.
pub(crate) async fn pay_invoice(
mut payer: impl InvoicePayer + std::marker::Send + 'static,
path: BlindedPath,
cltv_expiry_delta: u16,
fee_base_msat: u32,
payment_hash: [u8; 32],
msats: u64,
) -> Result<(), OfferError<Secp256k1Error>> {
let resp = payer
.query_routes(path, cltv_expiry_delta, fee_base_msat, msats)
.await
.map_err(OfferError::RouteFailure)?;

let _ = payer
.send_to_route(payment_hash, resp.routes[0].clone())
.await
.map_err(OfferError::RouteFailure)?;

// The payment is still in flight. We'll track it until it settles.
payer
.track_payment(payment_hash)
.await
.map_err(|_| OfferError::PaymentFailure)
}

#[async_trait]
impl PeerConnector for Client {
async fn list_peers(&mut self) -> Result<ListPeersResponse, Status> {
Expand Down Expand Up @@ -583,11 +599,9 @@ impl InvoicePayer for Client {

// Wait for a failed or successful payment.
while let Some(payment) = stream.message().await.map_err(OfferError::TrackFailure)? {
if payment.status == 2 {
// 2 means the payment succeeded
if payment.status() == tonic_lnd::lnrpc::payment::PaymentStatus::Succeeded {
return Ok(());
} else if payment.status == 3 {
// 3 means the payment failed
} else if payment.status() == tonic_lnd::lnrpc::payment::PaymentStatus::Failed {
return Err(OfferError::PaymentFailure);
} else {
continue;
Expand Down Expand Up @@ -969,11 +983,16 @@ mod tests {

let blinded_path = get_blinded_path();
let payment_hash = MessengerUtilities::new().get_secure_random_bytes();
assert!(
pay_invoice(payer_mock, blinded_path, 200, 1, payment_hash, 2000)
.await
.is_ok()
);
let handler = OfferHandler::new();
let params = PayInvoiceParams {
path: blinded_path,
cltv_expiry_delta: 200,
fee_base_msat: 1,
payment_hash: payment_hash,
msats: 2000,
offer_id: get_offer(),
};
assert!(handler.pay_invoice(payer_mock, params,).await.is_ok());
}

#[tokio::test]
Expand All @@ -986,11 +1005,16 @@ mod tests {

let blinded_path = get_blinded_path();
let payment_hash = MessengerUtilities::new().get_secure_random_bytes();
assert!(
pay_invoice(payer_mock, blinded_path, 200, 1, payment_hash, 2000)
.await
.is_err()
);
let handler = OfferHandler::new();
let params = PayInvoiceParams {
path: blinded_path,
cltv_expiry_delta: 200,
fee_base_msat: 1,
payment_hash: payment_hash,
msats: 2000,
offer_id: get_offer(),
};
assert!(handler.pay_invoice(payer_mock, params,).await.is_err());
}

#[tokio::test]
Expand All @@ -1013,10 +1037,15 @@ mod tests {

let blinded_path = get_blinded_path();
let payment_hash = MessengerUtilities::new().get_secure_random_bytes();
assert!(
pay_invoice(payer_mock, blinded_path, 200, 1, payment_hash, 2000)
.await
.is_err()
);
let handler = OfferHandler::new();
let params = PayInvoiceParams {
path: blinded_path,
cltv_expiry_delta: 200,
fee_base_msat: 1,
payment_hash: payment_hash,
msats: 2000,
offer_id: get_offer(),
};
assert!(handler.pay_invoice(payer_mock, params,).await.is_err());
}
}
Loading

0 comments on commit 2184d97

Please sign in to comment.