userAdityaa/caravel
1
1
Fork 0
forked from coracle/caravel
Code Pull Requests Activity

chore: prevent duplicate Lightning charges by adding durable invoice-level NWC payment guard

Browse Source
This commit is contained in:
Aditya Chaudhary
2026-04-25 16:42:05 +05:30
parent 1aeb15971d
commit 534d592282
4 changed files with 279 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/migrations/0003_invoice_nwc_payment_guard.sql
+11
View File
@@ -0,0 +1,11 @@
CREATE TABLE IF NOT EXISTS invoice_nwc_payment (
invoice_id TEXT PRIMARY KEY,
tenant_pubkey TEXT NOT NULL,
state TEXT NOT NULL CHECK (state IN ('pending', 'paid')),
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
FOREIGN KEY (tenant_pubkey) REFERENCES tenant(pubkey)
);
CREATE INDEX IF NOT EXISTS idx_invoice_nwc_payment_tenant_pubkey
ON invoice_nwc_payment (tenant_pubkey);
backend/src/billing.rs
+218 -31
View File
@@ -78,6 +78,12 @@ struct CoinbaseSpotPriceData {
amount: String,
}
enum NwcInvoicePaymentOutcome {
Paid,
Fallback(anyhow::Error),
Pending(anyhow::Error),
}
#[derive(Clone)]
pub struct Billing {
nwc_url: String,
@@ -260,6 +266,23 @@ impl Billing {
Ok(())
}
async fn existing_invoice_nwc_payment_outcome(
&self,
invoice_id: &str,
) -> Result<Option<NwcInvoicePaymentOutcome>> {
let state = self.query.get_invoice_nwc_payment_state(invoice_id).await?;
match state.as_deref() {
Some("paid") => Ok(Some(NwcInvoicePaymentOutcome::Paid)),
Some("pending") => Ok(Some(NwcInvoicePaymentOutcome::Pending(anyhow!(
"invoice {invoice_id} has a pending NWC reconciliation; refusing to create a new Lightning charge"
)))),
Some(other) => Err(anyhow!(
"unknown invoice_nwc_payment state '{other}' for invoice {invoice_id}"
)),
None => Ok(None),
}
}
pub async fn handle_webhook(&self, payload: &str, signature: &str) -> Result<()> {
self.verify_webhook_signature(payload, signature)?;
@@ -364,15 +387,21 @@ impl Billing {
// 1. NWC auto-pay: if the tenant has a nwc_url
if !tenant.nwc_url.is_empty() {
match self
.nwc_pay_invoice(amount_due, currency, &tenant.nwc_url)
.await
.nwc_pay_invoice(
invoice_id,
&tenant.pubkey,
amount_due,
currency,
&tenant.nwc_url,
)
.await?
{
Ok(()) => {
self.stripe_pay_invoice_out_of_band(invoice_id).await?;
self.command.clear_tenant_nwc_error(&tenant.pubkey).await?;
NwcInvoicePaymentOutcome::Paid => {
self.mark_invoice_paid_out_of_band_after_nwc(invoice_id, &tenant.pubkey)
.await?;
return Ok(());
}
Err(e) => {
NwcInvoicePaymentOutcome::Fallback(e) => {
let error_msg = format!("{e}");
self.command
.set_tenant_nwc_error(&tenant.pubkey, &error_msg)
@@ -387,6 +416,20 @@ impl Billing {
nwc_error_for_dm = summarize_nwc_error_for_dm(&error_msg);
// Fall through to next option
}
NwcInvoicePaymentOutcome::Pending(e) => {
let error_msg = format!("{e}");
self.command
.set_tenant_nwc_error(&tenant.pubkey, &error_msg)
.await?;
tracing::error!(
error = %e,
tenant_pubkey = %tenant.pubkey,
stripe_customer_id,
invoice_id,
"nwc auto-payment requires reconciliation before retry"
);
return Err(e);
}
}
}
@@ -748,21 +791,28 @@ impl Billing {
}
match self
.nwc_pay_invoice(amount_due, currency, &tenant.nwc_url)
.await
.nwc_pay_invoice(
invoice_id,
&tenant.pubkey,
amount_due,
currency,
&tenant.nwc_url,
)
.await?
{
Ok(()) => {
if let Err(e) = self.stripe_pay_invoice_out_of_band(invoice_id).await {
NwcInvoicePaymentOutcome::Paid => {
if let Err(e) = self
.mark_invoice_paid_out_of_band_after_nwc(invoice_id, &tenant.pubkey)
.await
{
tracing::error!(
error = %e,
invoice_id,
"failed to mark invoice paid out of band"
);
} else {
let _ = self.command.clear_tenant_nwc_error(&tenant.pubkey).await;
}
}
Err(e) => {
NwcInvoicePaymentOutcome::Fallback(e) => {
let error_msg = format!("{e}");
tracing::error!(
error = %e,
@@ -774,6 +824,18 @@ impl Billing {
.set_tenant_nwc_error(&tenant.pubkey, &error_msg)
.await;
}
NwcInvoicePaymentOutcome::Pending(e) => {
let error_msg = format!("{e}");
tracing::error!(
error = %e,
invoice_id,
"outstanding invoice requires NWC reconciliation before retry"
);
let _ = self
.command
.set_tenant_nwc_error(&tenant.pubkey, &error_msg)
.await;
}
}
}
@@ -1004,19 +1066,45 @@ impl Billing {
// --- NWC helpers ---
async fn mark_invoice_paid_out_of_band_after_nwc(
&self,
invoice_id: &str,
tenant_pubkey: &str,
) -> Result<()> {
self.stripe_pay_invoice_out_of_band(invoice_id).await?;
self.command.clear_tenant_nwc_error(tenant_pubkey).await?;
Ok(())
}
async fn nwc_pay_invoice(
&self,
invoice_id: &str,
tenant_pubkey: &str,
amount_due_minor: i64,
currency: &str,
tenant_nwc_url: &str,
) -> Result<()> {
let amount_msats = self.fiat_minor_to_msats(amount_due_minor, currency).await?;
) -> Result<NwcInvoicePaymentOutcome> {
if let Some(existing_outcome) = self
.existing_invoice_nwc_payment_outcome(invoice_id)
.await?
{
return Ok(existing_outcome);
}
let amount_msats = match self.fiat_minor_to_msats(amount_due_minor, currency).await {
Ok(amount_msats) => amount_msats,
Err(error) => return Ok(NwcInvoicePaymentOutcome::Fallback(error)),
};
// Create a bolt11 invoice using the system wallet (self.nwc_url)
let system_uri: NostrWalletConnectURI = self
let system_uri = self
.nwc_url
.parse()
.map_err(|_| anyhow!("invalid system NWC URL"))?;
.parse::<NostrWalletConnectURI>()
.map_err(|_| anyhow!("invalid system NWC URL"));
let system_uri = match system_uri {
Ok(system_uri) => system_uri,
Err(error) => return Ok(NwcInvoicePaymentOutcome::Fallback(error)),
};
let system_nwc = NWC::new(system_uri);
let make_req = MakeInvoiceRequest {
@@ -1026,29 +1114,64 @@ impl Billing {
expiry: None,
};
let invoice_response = system_nwc
.make_invoice(make_req)
.await
.map_err(|e| anyhow!("failed to create invoice: {e}"))?;
let invoice_response = system_nwc.make_invoice(make_req).await;
let invoice_response = match invoice_response {
Ok(invoice_response) => invoice_response,
Err(error) => {
system_nwc.shutdown().await;
return Ok(NwcInvoicePaymentOutcome::Fallback(anyhow!(
"failed to create invoice: {error}"
)));
}
};
system_nwc.shutdown().await;
// Pay the bolt11 invoice using the tenant's wallet
let tenant_uri: NostrWalletConnectURI = tenant_nwc_url
.parse()
.map_err(|_| anyhow!("invalid tenant NWC URL"))?;
let tenant_uri = tenant_nwc_url
.parse::<NostrWalletConnectURI>()
.map_err(|_| anyhow!("invalid tenant NWC URL"));
let tenant_uri = match tenant_uri {
Ok(tenant_uri) => tenant_uri,
Err(error) => return Ok(NwcInvoicePaymentOutcome::Fallback(error)),
};
if !self
.command
.insert_pending_invoice_nwc_payment(invoice_id, tenant_pubkey)
.await?
{
if let Some(existing_outcome) = self
.existing_invoice_nwc_payment_outcome(invoice_id)
.await?
{
return Ok(existing_outcome);
}
return Err(anyhow!(
"invoice_nwc_payment row missing after insert race for invoice {invoice_id}"
));
}
let tenant_nwc = NWC::new(tenant_uri);
let pay_req = NwcPayInvoiceRequest::new(invoice_response.invoice);
tenant_nwc
.pay_invoice(pay_req)
.await
.map_err(|e| anyhow!("failed to pay invoice: {e}"))?;
let pay_result = tenant_nwc.pay_invoice(pay_req).await;
tenant_nwc.shutdown().await;
Ok(())
match pay_result {
Ok(_) => match self.command.mark_invoice_nwc_payment_paid(invoice_id).await {
Ok(()) => Ok(NwcInvoicePaymentOutcome::Paid),
Err(error) => Ok(NwcInvoicePaymentOutcome::Pending(anyhow!(
"invoice {invoice_id} was charged over NWC but failed to persist paid state: {error}"
))),
},
Err(error) => Ok(NwcInvoicePaymentOutcome::Pending(anyhow!(
"invoice {invoice_id} NWC payment attempt requires reconciliation: {error}"
))),
}
}
async fn fiat_minor_to_msats(&self, amount_due_minor: i64, currency: &str) -> Result<u64> {
@@ -1167,7 +1290,7 @@ pub fn fiat_minor_to_msats_from_quote(
mod tests {
use super::{Billing, fiat_minor_to_msats_from_quote};
use crate::models::{
RELAY_STATUS_ACTIVE, RELAY_STATUS_DELINQUENT, RELAY_STATUS_INACTIVE, Relay,
RELAY_STATUS_ACTIVE, RELAY_STATUS_DELINQUENT, RELAY_STATUS_INACTIVE, Relay, Tenant,
};
fn relay_fixture(status: &str, plan: &str) -> Relay {
@@ -1428,4 +1551,68 @@ mod tests {
assert_eq!(billing.stripe_secret_key, "sk_test_dummy");
assert_eq!(billing.stripe_webhook_secret, "whsec_test_dummy");
}
#[tokio::test]
async fn invoice_nwc_payment_guard_persists_pending_and_paid_states() {
let pool = test_pool().await;
let query = Query::new(pool.clone());
let command = Command::new(pool);
let tenant = Tenant {
pubkey: "tenant_pubkey".to_string(),
nwc_url: String::new(),
nwc_error: None,
created_at: 0,
stripe_customer_id: "cus_test_123".to_string(),
stripe_subscription_id: None,
past_due_at: None,
};
command.create_tenant(&tenant).await.expect("create tenant");
assert_eq!(
query
.get_invoice_nwc_payment_state("in_test_123")
.await
.expect("read missing payment state"),
None
);
assert!(
command
.insert_pending_invoice_nwc_payment("in_test_123", &tenant.pubkey)
.await
.expect("insert pending state")
);
assert_eq!(
query
.get_invoice_nwc_payment_state("in_test_123")
.await
.expect("read pending payment state")
.as_deref(),
Some("pending")
);
assert!(
!command
.insert_pending_invoice_nwc_payment("in_test_123", &tenant.pubkey)
.await
.expect("duplicate insert should be ignored")
);
command
.mark_invoice_nwc_payment_paid("in_test_123")
.await
.expect("mark payment paid");
assert_eq!(
query
.get_invoice_nwc_payment_state("in_test_123")
.await
.expect("read paid payment state")
.as_deref(),
Some("paid")
);
}
}
backend/src/command.rs
+40
View File
@@ -313,6 +313,46 @@ impl Command {
Ok(())
}
pub async fn insert_pending_invoice_nwc_payment(
&self,
invoice_id: &str,
tenant_pubkey: &str,
) -> Result<bool> {
let now = chrono::Utc::now().timestamp();
let result = sqlx::query(
"INSERT INTO invoice_nwc_payment (invoice_id, tenant_pubkey, state, created_at, updated_at)
VALUES (?, ?, 'pending', ?, ?)
ON CONFLICT(invoice_id) DO NOTHING",
)
.bind(invoice_id)
.bind(tenant_pubkey)
.bind(now)
.bind(now)
.execute(&self.pool)
.await?;
Ok(result.rows_affected() > 0)
}
pub async fn mark_invoice_nwc_payment_paid(&self, invoice_id: &str) -> Result<()> {
let now = chrono::Utc::now().timestamp();
let result = sqlx::query(
"UPDATE invoice_nwc_payment
SET state = 'paid', updated_at = ?
WHERE invoice_id = ?",
)
.bind(now)
.bind(invoice_id)
.execute(&self.pool)
.await?;
if result.rows_affected() == 0 {
anyhow::bail!("invoice_nwc_payment row missing for invoice_id: {invoice_id}");
}
Ok(())
}
pub async fn set_tenant_past_due(&self, pubkey: &str) -> Result<()> {
let now = chrono::Utc::now().timestamp();
sqlx::query("UPDATE tenant SET past_due_at = ? WHERE pubkey = ?")
backend/src/query.rs
+10
View File
@@ -161,6 +161,16 @@ impl Query {
Ok(row)
}
pub async fn get_invoice_nwc_payment_state(&self, invoice_id: &str) -> Result<Option<String>> {
let state = sqlx::query_scalar::<_, String>(
"SELECT state FROM invoice_nwc_payment WHERE invoice_id = ?",
)
.bind(invoice_id)
.fetch_optional(&self.pool)
.await?;
Ok(state)
}
pub async fn has_active_paid_relays(&self, tenant_id: &str) -> Result<bool> {
let plans = sqlx::query_scalar::<_, String>(
"SELECT plan FROM relay WHERE tenant = ? AND status = 'active'",