feat: expire stale payment-link intents and detect amount mismatches

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<Uuid>,
+    /// 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<i64>,
 }
 
 /// `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,
     }))
 }
 

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<usize, IngestError> {
+        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.

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<String> {
@@ -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");
+}

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'));

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<Option<Transaction>, 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<Option<PaymentLink>, 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<Vec<PaymentLinkPayment>, 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