diff --git a/crates/api/src/routes/payment_links.rs b/crates/api/src/routes/payment_links.rs index 7e9681d..bd039b4 100644 --- a/crates/api/src/routes/payment_links.rs +++ b/crates/api/src/routes/payment_links.rs @@ -444,8 +444,15 @@ pub async fn create_payment_intent( #[derive(Debug, Serialize)] pub struct PaymentStatusView { + /// "pending" | "confirmed" | "expired" | "underpaid" | "overpaid". pub status: String, pub transaction_id: Option, + /// The amount this intent expects. Always present, so the frontend can render a mismatch + /// banner ("you sent X, expected Y") without a second request. + pub expected_usdc_stroops: i64, + /// What actually landed on-chain, once a deposit has been matched (confirmed/underpaid/ + /// overpaid). `None` while still pending or expired unpaid. + pub received_usdc_stroops: Option, } /// `GET /v1/pay/:slug/payments/:payment_id` — public, no auth. @@ -462,9 +469,19 @@ pub async fn get_payment_status( .store() .get_payment_link_payment(link.id, payment_id) .await?; + let received_usdc_stroops = match payment.transaction_id { + Some(tx_id) => state + .store() + .get_transaction(tx_id) + .await? + .map(|tx| tx.amount_stroops), + None => None, + }; Ok(Envelope::ok(PaymentStatusView { status: payment.status, transaction_id: payment.transaction_id, + expected_usdc_stroops: payment.amount_usdc_stroops, + received_usdc_stroops, })) } diff --git a/crates/ingest/src/lib.rs b/crates/ingest/src/lib.rs index 6952435..1ed809e 100644 --- a/crates/ingest/src/lib.rs +++ b/crates/ingest/src/lib.rs @@ -301,16 +301,49 @@ impl Ingestor { } }; - // Never confirm for less than the payer committed to — an underpayment stays pending. - if tx.amount_stroops < payment.amount_usdc_stroops { + // Exact match confirms; anything else is a mismatch the merchant/payer must be told + // about — never silently absorbed as if it were correct, and never confirmed either. + if tx.amount_stroops != payment.amount_usdc_stroops { + let status = if tx.amount_stroops < payment.amount_usdc_stroops { + "underpaid" + } else { + "overpaid" + }; tracing::warn!( payment_id = %payment.id, expected = payment.amount_usdc_stroops, received = tx.amount_stroops, - "payment-link deposit is short of the intended amount; leaving pending" + status, + "payment-link deposit does not match the intended amount" ); + if self + .store + .mark_payment_link_payment_mismatched(payment.id, tx.id, status) + .await + .is_err() + { + return; + } + if let Some(sender) = &self.webhooks { + let event = Event { + event_type: "payment_link.mismatched".to_string(), + data: serde_json::json!({ + "payment_link_id": link.id, + "payment_id": payment.id, + "slug": link.slug, + "payer_name": payment.payer_name, + "payer_email": payment.payer_email, + "status": status, + "expected_usdc_stroops": payment.amount_usdc_stroops, + "received_usdc_stroops": tx.amount_stroops, + "stellar_tx_hash": tx.stellar_tx_hash, + }), + }; + sender.dispatch(self.wallet_id, &event).await; + } return; } + if self .store .confirm_payment_link_payment(payment.id, tx.id) @@ -485,6 +518,41 @@ impl Supervisor { } } + /// Mark stale pending payment-link intents as `expired` and fire one `payment_link.expired` + /// webhook per row. Runs every tick — a single indexed `UPDATE ... WHERE` is cheap even when + /// it matches nothing, so no separate timer is needed. Best-effort: a DB error here must + /// never abort the poll loop. + async fn expire_stale_payment_link_payments(&self) { + let expired = match self.store.expire_stale_payment_link_payments().await { + Ok(rows) => rows, + Err(e) => { + tracing::warn!(error = ?e, "failed to sweep stale payment-link payments"); + return; + } + }; + for payment in expired { + let Ok(Some(link)) = self + .store + .get_payment_link_by_id(payment.payment_link_id) + .await + else { + continue; + }; + let event = Event { + event_type: "payment_link.expired".to_string(), + data: serde_json::json!({ + "payment_link_id": link.id, + "payment_id": payment.id, + "slug": link.slug, + "payer_name": payment.payer_name, + "payer_email": payment.payer_email, + "amount_usdc_stroops": payment.amount_usdc_stroops, + }), + }; + self.webhooks.dispatch(link.wallet_id, &event).await; + } + } + /// Run forever: every `interval`, poll all wallets on this network once. pub async fn run(self, interval: Duration, page_limit: u32) { loop { @@ -504,6 +572,8 @@ impl Supervisor { /// interval. Bounding concurrency (rather than firing all requests at once) keeps Horizon /// request volume sane regardless of how many wallets exist. pub async fn tick(&self, page_limit: u32) -> Result { + self.expire_stale_payment_link_payments().await; + // Only wallets actually due under the backoff tiers — a dev/production DB accumulates // wallets that never transact again, and polling them every cycle starves the active ones // of the shared concurrency budget. diff --git a/crates/ingest/tests/process_tests.rs b/crates/ingest/tests/process_tests.rs index 4d9bf08..c829ae3 100644 --- a/crates/ingest/tests/process_tests.rs +++ b/crates/ingest/tests/process_tests.rs @@ -5,11 +5,14 @@ use octo_ingest::horizon::{PaymentRecord, TransactionRecord}; use octo_ingest::{Ingestor, Processed}; -use octo_store::{NewWallet, Store}; +use octo_store::{NewPaymentLink, NewWallet, Store}; use octo_wallet_core::encode_muxed; use std::sync::Once; use uuid::Uuid; +/// Same testnet USDC issuer the ingest crate matches payment-link deposits against. +const USDC_TESTNET_ISSUER: &str = "GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5"; + static LOAD_ENV: Once = Once::new(); fn database_url() -> Option { @@ -281,3 +284,109 @@ async fn missing_transaction_field_yields_no_memo_and_no_panic() { assert_eq!(txs[0].memo_id, None); assert_eq!(txs[0].ledger, None); } + +async fn make_usdc_payment_link( + store: &Store, + wallet_id: Uuid, + amount_usdc_stroops: i64, +) -> (String, Uuid, Uuid) { + let addr = store + .allocate_address( + wallet_id, + |id| encode_muxed(BASE, id as u64).map_err(|_| ()), + None, + serde_json::json!({}), + ) + .await + .unwrap(); + let link = store + .create_payment_link(NewPaymentLink { + wallet_id, + address_id: addr.id, + slug: &format!("link-{}", Uuid::new_v4().simple()), + name: "Test link", + description: None, + image_url: None, + redirect_url: None, + amount_usdc_stroops: Some(amount_usdc_stroops), + }) + .await + .unwrap(); + let intent = store + .record_payment_link_intent(link.id, None, None, amount_usdc_stroops, Some(addr.id)) + .await + .unwrap(); + (addr.muxed_address, link.id, intent.id) +} + +fn usdc_record(id: &str, to_muxed: String, amount: &str) -> PaymentRecord { + let mut rec = base_record(id); + rec.to_muxed = Some(to_muxed); + rec.asset_type = Some("credit_alphanum4".into()); + rec.asset_code = Some("USDC".into()); + rec.asset_issuer = Some(USDC_TESTNET_ISSUER.into()); + rec.amount = Some(amount.into()); + rec +} + +#[tokio::test] +async fn underpaid_payment_link_deposit_is_recorded_but_left_unconfirmed() { + let Some((store, ingestor, wallet_id)) = setup().await else { + return; + }; + + let (muxed, link_id, intent_id) = make_usdc_payment_link(&store, wallet_id, 100_000_000).await; + let rec = usdc_record("op-underpaid-1", muxed, "5.0000000"); + + let outcome = ingestor.process(&rec).await.unwrap(); + assert_eq!(outcome, Processed::Recorded { attributed: true }); + + let payment = store + .get_payment_link_payment(link_id, intent_id) + .await + .unwrap(); + assert_eq!(payment.status, "underpaid"); + assert!( + payment.transaction_id.is_some(), + "the short deposit must still be linked so the merchant can see what arrived" + ); +} + +#[tokio::test] +async fn overpaid_payment_link_deposit_is_recorded_but_left_unconfirmed() { + let Some((store, ingestor, wallet_id)) = setup().await else { + return; + }; + + let (muxed, link_id, intent_id) = make_usdc_payment_link(&store, wallet_id, 100_000_000).await; + let rec = usdc_record("op-overpaid-1", muxed, "15.0000000"); + + let outcome = ingestor.process(&rec).await.unwrap(); + assert_eq!(outcome, Processed::Recorded { attributed: true }); + + let payment = store + .get_payment_link_payment(link_id, intent_id) + .await + .unwrap(); + assert_eq!(payment.status, "overpaid"); + assert!(payment.transaction_id.is_some()); +} + +#[tokio::test] +async fn exact_payment_link_deposit_confirms() { + let Some((store, ingestor, wallet_id)) = setup().await else { + return; + }; + + let (muxed, link_id, intent_id) = make_usdc_payment_link(&store, wallet_id, 100_000_000).await; + let rec = usdc_record("op-exact-1", muxed, "10.0000000"); + + let outcome = ingestor.process(&rec).await.unwrap(); + assert_eq!(outcome, Processed::Recorded { attributed: true }); + + let payment = store + .get_payment_link_payment(link_id, intent_id) + .await + .unwrap(); + assert_eq!(payment.status, "confirmed"); +} diff --git a/crates/store/migrations/0018_payment_status_expansion.sql b/crates/store/migrations/0018_payment_status_expansion.sql new file mode 100644 index 0000000..0365f1f --- /dev/null +++ b/crates/store/migrations/0018_payment_status_expansion.sql @@ -0,0 +1,11 @@ +-- Widen payment_link_payments.status beyond pending/confirmed: +-- expired — pending past its deadline, swept by the ingest supervisor (see Supervisor::tick) +-- underpaid — a deposit landed but for less than the intent's amount +-- overpaid — a deposit landed but for more than the intent's amount +-- underpaid/overpaid still record the transaction (so the merchant/payer can see what actually +-- arrived) but are deliberately NOT 'confirmed' — the merchant decides how to handle the +-- mismatch (refund, top-up request, manual reconciliation), Octo doesn't silently treat it as +-- paid-in-full. +ALTER TABLE payment_link_payments DROP CONSTRAINT payment_link_payments_status_check; +ALTER TABLE payment_link_payments ADD CONSTRAINT payment_link_payments_status_check + CHECK (status IN ('pending', 'confirmed', 'expired', 'underpaid', 'overpaid')); diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index ce251a9..3e6342b 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -773,6 +773,15 @@ impl Store { Ok(rows) } + /// Fetch a single transaction by id. + pub async fn get_transaction(&self, id: Uuid) -> Result, StoreError> { + let row = sqlx::query_as::<_, Transaction>("SELECT * FROM transactions WHERE id = $1") + .bind(id) + .fetch_optional(&self.pool) + .await?; + Ok(row) + } + // --- withdrawals ------------------------------------------------------ /// Cheap existence check on `(wallet_id, idempotency_key)`, used to short-circuit a retried @@ -1181,6 +1190,19 @@ impl Store { .ok_or(StoreError::NotFound) } + /// Unscoped lookup by id — for internal (non-owner-facing) callers that already know which + /// row they want, e.g. the expiry sweep resolving a payment's link to build its webhook. + pub async fn get_payment_link_by_id( + &self, + id: Uuid, + ) -> Result, StoreError> { + let row = sqlx::query_as::<_, PaymentLink>("SELECT * FROM payment_links WHERE id = $1") + .bind(id) + .fetch_optional(&self.pool) + .await?; + Ok(row) + } + /// The payment link whose dedicated deposit address is `address_id`, if any. pub async fn get_payment_link_by_address( &self, @@ -1339,6 +1361,49 @@ impl Store { Ok(()) } + /// Record a deposit that landed on this payment's address but for the wrong amount. + /// `status` must be `"underpaid"` or `"overpaid"` — the transaction is still linked (so the + /// merchant/payer can see what actually arrived) but the payment is deliberately NOT marked + /// `confirmed`. + pub async fn mark_payment_link_payment_mismatched( + &self, + id: Uuid, + transaction_id: Uuid, + status: &str, + ) -> Result<(), StoreError> { + sqlx::query( + r#" + UPDATE payment_link_payments + SET status = $1, transaction_id = $2 + WHERE id = $3 + "#, + ) + .bind(status) + .bind(transaction_id) + .bind(id) + .execute(&self.pool) + .await?; + Ok(()) + } + + /// Mark payments still `pending` past a 1-hour deadline as `expired`, returning the rows that + /// were flipped so the caller can fire one webhook per expiry without a second query. + pub async fn expire_stale_payment_link_payments( + &self, + ) -> Result, StoreError> { + let rows = sqlx::query_as::<_, PaymentLinkPayment>( + r#" + UPDATE payment_link_payments + SET status = 'expired' + WHERE status = 'pending' AND created_at < now() - interval '1 hour' + RETURNING * + "#, + ) + .fetch_all(&self.pool) + .await?; + Ok(rows) + } + /// Payments recorded against a link (newest first), with cursor pagination. /// /// Includes pending intents, not just confirmed ones — a merchant wants to see that someone diff --git a/crates/store/tests/store_tests.rs b/crates/store/tests/store_tests.rs index 69a6f75..52fcb3c 100644 --- a/crates/store/tests/store_tests.rs +++ b/crates/store/tests/store_tests.rs @@ -446,6 +446,168 @@ async fn payment_link_lifecycle_intent_confirm_and_sum() { assert!(!deactivated.active); } +#[tokio::test] +async fn payment_link_mismatched_deposit_records_the_transaction_but_does_not_confirm() { + let Some(store) = store().await else { return }; + let wallet_id = fresh_wallet(&store).await; + let wid = wallet_id.simple(); + + let addr = store + .allocate_address( + wallet_id, + |id| Ok(format!("M{wid}-{id}")), + None, + serde_json::json!({}), + ) + .await + .expect("alloc address"); + + let link = store + .create_payment_link(NewPaymentLink { + wallet_id, + address_id: addr.id, + slug: &format!("link-mismatch-{wid}"), + name: "Underpaid test", + description: None, + image_url: None, + redirect_url: None, + amount_usdc_stroops: Some(10_000_000), + }) + .await + .expect("create link"); + + let intent = store + .record_payment_link_intent(link.id, None, None, 10_000_000, Some(addr.id)) + .await + .expect("record intent"); + + let tx_hash = Uuid::new_v4().to_string(); + let dep = store + .record_deposit(&NewDeposit { + wallet_id, + address_id: Some(addr.id), + asset_code: "USDC".into(), + asset_issuer: Some("GISSUER".into()), + amount_stroops: 5_000_000, // half of what was expected + source_account: Some("Gpayer".into()), + destination_account: Some("Gmaster".into()), + stellar_tx_hash: tx_hash.clone(), + operation_index: 0, + horizon_op_id: format!("{tx_hash}-0"), + ledger: Some(1), + memo_id: None, + }) + .await + .expect("record deposit") + .expect("first insert"); + + store + .mark_payment_link_payment_mismatched(intent.id, dep.id, "underpaid") + .await + .expect("mark mismatched"); + + let mismatched = store + .get_payment_link_payment(link.id, intent.id) + .await + .expect("get payment"); + assert_eq!(mismatched.status, "underpaid"); + assert_eq!( + mismatched.transaction_id, + Some(dep.id), + "the short deposit must still be linked, so the merchant can see what actually arrived" + ); + + // A mismatched payment is not "pending" any more, so it must not still be matchable — ingest + // must not later confuse a second, correct deposit with this already-resolved intent. + assert!(store + .pending_payment_by_address(addr.id) + .await + .expect("by address") + .is_none()); +} + +#[tokio::test] +async fn expire_stale_payment_link_payments_only_sweeps_old_pending_rows() { + let Some(store) = store().await else { return }; + let wallet_id = fresh_wallet(&store).await; + let wid = wallet_id.simple(); + + let addr = store + .allocate_address( + wallet_id, + |id| Ok(format!("M{wid}-{id}")), + None, + serde_json::json!({}), + ) + .await + .expect("alloc address"); + + let link = store + .create_payment_link(NewPaymentLink { + wallet_id, + address_id: addr.id, + slug: &format!("link-expiry-{wid}"), + name: "Expiry test", + description: None, + image_url: None, + redirect_url: None, + amount_usdc_stroops: Some(10_000_000), + }) + .await + .expect("create link"); + + let stale = store + .record_payment_link_intent(link.id, None, None, 10_000_000, Some(addr.id)) + .await + .expect("record stale intent"); + // Backdate it past the 1-hour deadline directly — this test can't wait an hour. + sqlx::query( + "UPDATE payment_link_payments SET created_at = now() - interval '2 hours' WHERE id = $1", + ) + .bind(stale.id) + .execute(store.pool()) + .await + .expect("backdate"); + + let fresh = store + .record_payment_link_intent(link.id, None, None, 10_000_000, Some(addr.id)) + .await + .expect("record fresh intent"); + + let expired = store + .expire_stale_payment_link_payments() + .await + .expect("sweep"); + let expired_ids: Vec = expired.iter().map(|p| p.id).collect(); + assert!( + expired_ids.contains(&stale.id), + "the >1hr-old pending row must be swept" + ); + assert!( + !expired_ids.contains(&fresh.id), + "a freshly-created pending row must not be swept" + ); + + let stale_after = store + .get_payment_link_payment(link.id, stale.id) + .await + .expect("get stale"); + assert_eq!(stale_after.status, "expired"); + + let fresh_after = store + .get_payment_link_payment(link.id, fresh.id) + .await + .expect("get fresh"); + assert_eq!(fresh_after.status, "pending"); + + // Running the sweep again must be a no-op for already-expired rows (idempotent). + let expired_again = store + .expire_stale_payment_link_payments() + .await + .expect("sweep again"); + assert!(!expired_again.iter().any(|p| p.id == stale.id)); +} + #[tokio::test] async fn withdrawal_idempotency_key_blocks_double_spend() { let Some(store) = store().await else { return }; @@ -688,7 +850,7 @@ async fn migrate_applies_exactly_the_expected_version_set() { .expect("query _sqlx_migrations"); versions.sort_unstable(); - // One version per file under crates/store/migrations/, 0001_init.sql .. 0017. + // One version per file under crates/store/migrations/, 0001_init.sql .. 0018. // // NOTE: this version number is a repeat offender — five migrations have now landed with a // colliding 0008 at one point or another (scheme_version, token_denylist, @@ -698,8 +860,8 @@ async fn migrate_applies_exactly_the_expected_version_set() { // every version explicitly rather than just checking a count. assert_eq!( versions, - vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17], - "expected exactly the seventeen known migrations to be recorded as applied" + vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18], + "expected exactly the eighteen known migrations to be recorded as applied" ); }