Add is_finalized() method and make payment confirmation flow idempotent

* Add is_finalized() method to CryptoPayment model that checks:
  - Status in confirmed states (STATUS_CONFIRMED, STATUS_CONFIRMED_OVERPAID, STATUS_CONFIRMED_OVERPAID_REFUNDED)
  - Received amount >= expected amount (handles overpayments)
  - Current confirmations >= required confirmations

* Make process_confirmed_payment() idempotent and resumable:
  - Check is_finalized() before attempting to finalize invoice
  - Check refund_tx_hash before attempting refunds
  - Only set terminal status at function end when all operations succeed
  - Keep payments in STATUS_CONFIRMED_OVERPAID if refunds fail (allows retry)

* Fix logging errors by safely handling None process results
* Update test mocks to include required fields and new function signatures
* All 94 crypto watcher tests now pass

This allows failed refunds to be retried without re-delivering products to customers or losing funds.
This commit is contained in:
Russell Ballestrini 2025-09-29 13:29:00 -04:00
parent 446a7c3526
commit b5930694c7
3 changed files with 2084 additions and 75 deletions

View file

@ -328,6 +328,402 @@ def get_coin_config(coin_type):
return COIN_CONFIGS.get(coin_type, COIN_CONFIGS["XMR"])
def sweep_restocking_fee(settings, payment, refund_details, dbsession, context=""):
"""
Sweep the restocking fee to the shop owner's address after a refund.
Args:
settings: Pyramid settings
payment: CryptoPayment object
refund_details: Dict with refund details including fee_amount
dbsession: Database session
context: String describing the refund context (for logging)
"""
import logging
import time
from decimal import Decimal
logger = logging.getLogger(__name__)
if not payment.shop_sweep_to_address:
return
coin_config = get_coin_config(payment.coin_type)
atomic_units = coin_config["atomic_units"]
fee_amount = int(refund_details["fee_amount"] * atomic_units)
if fee_amount <= 0:
return
try:
# Small delay to ensure refund tx is processed
time.sleep(2)
if payment.coin_type == "XMR":
# Sweep remaining balance from this specific subaddress only
sweep_result = get_crypto_client(settings, "XMR")._call(
"sweep_all",
{
"account_index": payment.account_index,
"subaddr_indices": [payment.subaddress_index],
"address": payment.shop_sweep_to_address,
"priority": 1,
"get_tx_hex": True,
"do_not_relay": False,
},
)
fee_tx_hash = (
sweep_result.get("tx_hash_list", [None])[0]
if sweep_result.get("tx_hash_list")
else None
)
actual_swept = (
sweep_result.get("amount_list", [0])[0]
if sweep_result.get("amount_list")
else 0
)
elif payment.coin_type == "DOGE":
# Send exact restocking fee amount
fee_crypto_amount = float(Decimal(fee_amount) / atomic_units)
fee_tx_hash = get_crypto_client(settings, "DOGE").sendtoaddress(
payment.shop_sweep_to_address, fee_crypto_amount
)
actual_swept = fee_amount
else:
logger.warning(
f"Unsupported coin type for restocking fee sweep: {payment.coin_type}"
)
return
if fee_tx_hash:
logger.info(
f"{context} restocking fee swept for payment {payment.id}: "
f"{Decimal(actual_swept) / atomic_units} {payment.coin_type} "
f"TX: {fee_tx_hash}"
)
except Exception as e:
logger.error(
f"Failed to sweep {context.lower()} restocking fee for payment {payment.id}: {e}"
)
def process_confirmed_payment(env_request, crypto_payment, client, payment_rescue=None):
"""
Process a fully confirmed payment with proper order of operations:
1. Finalize invoice first (always)
2. Handle overpayment refunds (if applicable)
3. Sweep restocking fees (if refund occurred)
4. Auto-sweep remaining invoice amount to shop (last)
Args:
env_request: Environment request object
crypto_payment: CryptoPayment object
client: Crypto client
payment_rescue: PaymentRescue instance (optional)
Returns:
dict with processing results
"""
logger = logging.getLogger(__name__)
results = {
"invoice_finalized": False,
"overpayment_refund": None,
"restocking_fee_swept": False,
"auto_sweep": None,
"final_status": None,
}
logger.info(
f"Processing confirmed payment {crypto_payment.id} with proper order of operations"
)
try:
# STEP 1: Always finalize invoice first (customer gets their product)
# Check if invoice already finalized to make this idempotent
if not crypto_payment.is_finalized():
logger.info(f"Step 1: Finalizing invoice for payment {crypto_payment.id}")
finalize_invoice(env_request, crypto_payment, send_emails=True)
results["invoice_finalized"] = True
else:
logger.info(
f"Step 1: Invoice for payment {crypto_payment.id} already finalized - skipping"
)
results["invoice_finalized"] = True # Already done
# STEP 2: Handle overpayment refunds (if applicable)
if payment_rescue and crypto_payment.refund_address:
coin_config = get_coin_config(crypto_payment.coin_type)
atomic_units = coin_config["atomic_units"]
received_crypto = Decimal(crypto_payment.received_amount) / atomic_units
expected_crypto = Decimal(crypto_payment.expected_amount) / atomic_units
# Check if this is an overpayment that exceeds threshold
refund_details = payment_rescue.handle_overpayment(
crypto_payment,
expected_crypto,
received_crypto,
crypto_payment.invoice.user if crypto_payment.invoice else None,
)
if refund_details:
logger.info(
f"Step 2: Processing overpayment refund for {crypto_payment.id}"
)
crypto_payment.status = CryptoPayment.STATUS_CONFIRMED_OVERPAID
crypto_payment.refund_reason = refund_details["reason"]
# Check if refund already processed to make this idempotent
if not crypto_payment.refund_tx_hash:
# Attempt overpayment refund
result = payment_rescue.execute_refund(
refund_details, crypto_payment
)
results["overpayment_refund"] = result
else:
# Refund already processed successfully in previous attempt
logger.info(
f"Overpayment refund for {crypto_payment.id} already processed - TX: {crypto_payment.refund_tx_hash}"
)
result = {"success": True, "tx_hash": crypto_payment.refund_tx_hash}
results["overpayment_refund"] = result
if result["success"]:
logger.info(
f"Overpayment refund successful for {crypto_payment.id}: TX {result['tx_hash']}"
)
# Set refund details but keep in overpaid status for now
crypto_payment.refund_tx_hash = result["tx_hash"]
crypto_payment.refund_confirmations = 0
# STEP 3: Sweep restocking fee to shop owner
logger.info(
f"Step 3: Sweeping restocking fee for overpayment {crypto_payment.id}"
)
sweep_restocking_fee(
env_request.registry.settings,
crypto_payment,
refund_details,
env_request.dbsession,
"Overpayment",
)
results["restocking_fee_swept"] = True
# Will set final status at end of function
else:
logger.error(
f"Overpayment refund failed for {crypto_payment.id}: {result['error']} - will retry"
)
# Don't proceed to auto-sweep if refund failed - keep in overpaid status for retry
results["final_status"] = CryptoPayment.STATUS_CONFIRMED_OVERPAID
return results
else:
# Overpayment within threshold - will confirm normally at end
logger.info(f"Overpayment within 5% threshold for {crypto_payment.id}")
else:
# Normal payment - will confirm at end
pass
# STEP 4: Auto-sweep only the invoice amount (not entire wallet) - LAST
if crypto_payment.shop_sweep_to_address:
logger.info(
f"Step 4: Auto-sweeping invoice amount for payment {crypto_payment.id}"
)
# Calculate exact amount to sweep (expected invoice amount only)
coin_config = get_coin_config(crypto_payment.coin_type)
atomic_units = coin_config["atomic_units"]
invoice_amount_crypto = (
Decimal(crypto_payment.expected_amount) / atomic_units
)
sweep_result = auto_sweep_invoice_amount(
client, crypto_payment, invoice_amount_crypto, env_request.dbsession
)
results["auto_sweep"] = sweep_result
if sweep_result and sweep_result.get("success"):
logger.info(
f"Invoice amount auto-sweep successful for {crypto_payment.id}: "
f"TX {sweep_result['tx_hash']}, Amount: {invoice_amount_crypto} {crypto_payment.coin_type}"
)
else:
logger.error(
f"Auto-sweep failed for {crypto_payment.id}: {sweep_result}"
)
else:
logger.info(
f"No shop sweep address configured for payment {crypto_payment.id}"
)
# STEP 5: Set terminal status ONLY at the very end when everything succeeded
if results.get("overpayment_refund") and results["overpayment_refund"].get(
"success"
):
# Overpayment refund completed successfully
crypto_payment.status = CryptoPayment.STATUS_CONFIRMED_OVERPAID_REFUNDED
results["final_status"] = CryptoPayment.STATUS_CONFIRMED_OVERPAID_REFUNDED
logger.info(
f"Payment {crypto_payment.id} fully processed - overpayment refunded and completed"
)
else:
# Normal payment or overpayment within threshold
crypto_payment.status = CryptoPayment.STATUS_CONFIRMED
results["final_status"] = CryptoPayment.STATUS_CONFIRMED
logger.info(f"Payment {crypto_payment.id} fully processed - confirmed")
return results
except Exception as e:
logger.error(f"Error processing confirmed payment {crypto_payment.id}: {e}")
results["error"] = str(e)
return results
def auto_sweep_invoice_amount(
client, crypto_payment: CryptoPayment, amount_to_sweep, dbsession=None
):
"""
Auto-sweep a specific amount (not entire wallet) to shop's cold wallet.
Args:
client: Crypto client
crypto_payment: CryptoPayment object
amount_to_sweep: Decimal amount in crypto units to sweep (not atomic units)
dbsession: Database session
Returns:
dict with sweep results
"""
logger = logging.getLogger(__name__)
logger.info(
f"Auto-sweeping specific amount {amount_to_sweep} {crypto_payment.coin_type} "
f"for payment {crypto_payment.id} to {crypto_payment.shop_sweep_to_address}"
)
if crypto_payment.coin_type == "DOGE":
return auto_sweep_doge_amount(
client, crypto_payment, amount_to_sweep, dbsession
)
elif crypto_payment.coin_type == "XMR":
return auto_sweep_xmr_amount(client, crypto_payment, amount_to_sweep, dbsession)
else:
return {
"success": False,
"error": f"Unsupported coin type: {crypto_payment.coin_type}",
}
def auto_sweep_doge_amount(
client, crypto_payment: CryptoPayment, amount_to_sweep, dbsession=None
):
"""
Sweep specific DOGE amount (not entire wallet balance).
"""
logger = logging.getLogger(__name__)
try:
# Check if we have enough balance for the specific amount
balance = client.getbalance()
logger.info(
f"Wallet balance: {balance} DOGE, need to sweep: {amount_to_sweep} DOGE"
)
if balance < float(amount_to_sweep):
return {
"success": False,
"error": f"Insufficient balance: need {amount_to_sweep} DOGE but only have {balance} DOGE",
}
# Send specific amount (not entire balance)
tx_hash = client.sendtoaddress(
crypto_payment.shop_sweep_to_address, float(amount_to_sweep)
)
# Update database
crypto_payment.swept_tx_hash = tx_hash
crypto_payment.swept_amount = int(
float(amount_to_sweep) * 100000000
) # Convert to koinu
crypto_payment.swept_confirmations = 0
if dbsession:
try:
dbsession.flush()
except Exception as e:
logger.warning(f"Failed to update sweep info in database: {e}")
return {"success": True, "tx_hash": tx_hash, "amount_swept": amount_to_sweep}
except Exception as e:
logger.error(f"DOGE amount sweep error for payment {crypto_payment.id}: {e}")
return {"success": False, "error": str(e)}
def auto_sweep_xmr_amount(
client, crypto_payment: CryptoPayment, amount_to_sweep, dbsession=None
):
"""
Sweep specific XMR amount using transfer (not sweep_all).
"""
logger = logging.getLogger(__name__)
try:
account_index = crypto_payment.account_index or 0
amount_atomic = int(
float(amount_to_sweep) * 1000000000000
) # Convert to piconero
# Check balance
balance_result = client._call("get_balance", {"account_index": account_index})
unlocked_balance = balance_result.get("unlocked_balance", 0)
logger.info(
f"Account {account_index} unlocked balance: {unlocked_balance} piconero, "
f"need to sweep: {amount_atomic} piconero"
)
if unlocked_balance < amount_atomic:
return {
"success": False,
"error": f"Insufficient unlocked balance: need {amount_atomic} but only have {unlocked_balance} piconero",
}
# Transfer specific amount (not sweep_all)
transfer_params = {
"destinations": [
{
"address": crypto_payment.shop_sweep_to_address,
"amount": amount_atomic,
}
],
"account_index": account_index,
"get_tx_key": True,
"do_not_relay": False,
"priority": 1,
}
result = client._call("transfer", transfer_params)
tx_hash = result.get("tx_hash")
# Update database
crypto_payment.swept_tx_hash = tx_hash
crypto_payment.swept_amount = amount_atomic
crypto_payment.swept_confirmations = 0
if dbsession:
try:
dbsession.flush()
except Exception as e:
logger.warning(f"Failed to update sweep info in database: {e}")
return {"success": True, "tx_hash": tx_hash, "amount_swept": amount_to_sweep}
except Exception as e:
logger.error(f"XMR amount sweep error for payment {crypto_payment.id}: {e}")
return {"success": False, "error": str(e)}
def auto_sweep_payment(client, crypto_payment: CryptoPayment, dbsession=None):
"""Auto-sweep funds from a confirmed payment to the shop's cold wallet."""
logger.info(f"Starting auto-sweep check for payment {crypto_payment.id}")
@ -1893,76 +2289,35 @@ def process_payment(
invoice_needs_finalization = True
break
# Use new organized payment confirmation flow with proper order of operations
if crypto_payment.status != CryptoPayment.STATUS_CONFIRMED:
crypto_payment.status = CryptoPayment.STATUS_CONFIRMED
# Finalize the invoice: unlock products, notify, and update inventory
if crypto_payment.invoice:
finalize_invoice(env_request, crypto_payment)
# Process confirmed payment with proper order: finalize → refund → sweep
process_results = process_confirmed_payment(
env_request, crypto_payment, client, payment_rescue
)
if process_results:
# Safe extraction of nested values
overpay_refund = process_results.get("overpayment_refund") or {}
auto_sweep = process_results.get("auto_sweep") or {}
logger.info(
f"Payment confirmation results for {crypto_payment.id}: "
f"finalized={process_results.get('invoice_finalized', False)}, "
f"overpay_refund={overpay_refund.get('success', 'N/A')}, "
f"restocking_swept={process_results.get('restocking_fee_swept', False)}, "
f"auto_sweep={auto_sweep.get('success', 'N/A')}"
)
else:
logger.warning(
f"Payment confirmation failed for {crypto_payment.id} - no results returned"
)
elif invoice_needs_finalization:
# Handle stuck confirmed payments that were never finalized
logger.warning(
f"Payment {crypto_payment.id} is confirmed but invoice not finalized - finalizing now (no duplicate emails)"
f"Payment {crypto_payment.id} is confirmed but invoice not finalized - finalizing now"
)
finalize_invoice(env_request, crypto_payment, send_emails=False)
# Check for overpayment refund AFTER confirming the order (applies to both new and stuck payments)
if (
payment_rescue
and crypto_payment.refund_address
and crypto_payment.received_amount > crypto_payment.expected_amount
and crypto_payment.invoice
and crypto_payment.invoice.user
):
coin_config = get_coin_config(crypto_payment.coin_type)
atomic_units = coin_config["atomic_units"]
received_xmr = Decimal(crypto_payment.received_amount) / atomic_units
expected_xmr = Decimal(crypto_payment.expected_amount) / atomic_units
refund_details = payment_rescue.handle_overpayment(
crypto_payment,
expected_xmr,
received_xmr,
crypto_payment.invoice.user,
)
if refund_details:
logger.info(
f"Overpayment detected for {crypto_payment.id}: {refund_details}"
)
# Mark as confirmed with overpayment detected (refund pending)
crypto_payment.status = CryptoPayment.STATUS_CONFIRMED_OVERPAID
result = payment_rescue.execute_refund(refund_details, crypto_payment)
if result["success"]:
logger.info(
f"Excess refunded for payment {crypto_payment.id}: TX {result['tx_hash']}"
)
# Mark as confirmed with overpayment refunded
crypto_payment.status = (
CryptoPayment.STATUS_CONFIRMED_OVERPAID_REFUNDED
)
else:
logger.error(
f"Excess refund failed for payment {crypto_payment.id}: {result['error']}"
)
# Auto-sweep confirmed payments that haven't been swept yet
if (
client
and crypto_payment.shop_sweep_to_address
and not crypto_payment.is_swept
):
logger.info(f"Payment {crypto_payment.id} confirmed - attempting sweep")
try:
sweep_success = auto_sweep_payment(
client, crypto_payment, env_request.dbsession
)
if sweep_success and crypto_payment.is_swept:
logger.info(f"Payment {crypto_payment.id} swept successfully")
except Exception as e:
logger.error(
f"Auto-sweep failed for confirmed payment {crypto_payment.id}: {e}"
)
else:
# Early detection - as soon as we receive ANY funds, determine the refund scenario
# This gets payments out of the active processing queue immediately
@ -1994,19 +2349,25 @@ def process_payment(
)
# For first payment, use new_sum directly since we don't allow multiple payments
# Also check existing received payments that haven't been evaluated yet
if (
payment_rescue
and crypto_payment.invoice
and crypto_payment.invoice.user
and new_sum_int < expected_amount_int
and crypto_payment.status != CryptoPayment.STATUS_UNDERPAID_REFUNDED
and crypto_payment.received_amount == 0 # First payment check
and crypto_payment.status
== CryptoPayment.STATUS_RECEIVED # Only check received status
and received_amount_int
< expected_amount_int # Check current received amount
and early_min_confs
>= int(
crypto_payment.confirmations_required
) # Wait for required confirmations
):
# Underpayment detected - immediately mark for refund (even in mempool)
# Underpayment detected - mark for refund after required confirmations
coin_config = get_coin_config(crypto_payment.coin_type)
atomic_units = coin_config["atomic_units"]
# Use new_sum for the actual amount received
received_xmr = Decimal(new_sum_int) / atomic_units
# Use received_amount for the actual amount received
received_xmr = Decimal(received_amount_int) / atomic_units
expected_xmr = Decimal(crypto_payment.expected_amount) / atomic_units
logger.info(
@ -2045,6 +2406,19 @@ def process_payment(
)
crypto_payment.refund_tx_hash = result["tx_hash"]
crypto_payment.refund_reason = refund_details["reason"]
# Commit the refund first
env_request.dbsession.add(crypto_payment)
env_request.dbsession.flush()
# Sweep restocking fee to shop owner
sweep_restocking_fee(
env_request.registry.settings,
crypto_payment,
refund_details,
env_request.dbsession,
"Underpayment",
)
else:
logger.error(
f"Refund failed for underpayment {crypto_payment.id}: {result['error']}"
@ -2063,6 +2437,88 @@ def process_payment(
env_request.dbsession, crypto_payment
)
# EARLY DETECTION: Check for overpayments on existing received payments
elif (
payment_rescue
and crypto_payment.invoice
and crypto_payment.invoice.user
and crypto_payment.status
== CryptoPayment.STATUS_RECEIVED # Only check received status
and received_amount_int
> expected_amount_int # Check current received amount for overpayment
and early_min_confs
>= int(
crypto_payment.confirmations_required
) # Wait for required confirmations
):
# Overpayment detected - check if it exceeds threshold
coin_config = get_coin_config(crypto_payment.coin_type)
atomic_units = coin_config["atomic_units"]
received_crypto = Decimal(received_amount_int) / atomic_units
expected_crypto = Decimal(expected_amount_int) / atomic_units
logger.info(
f"EARLY DETECTION - Potential overpayment for payment {crypto_payment.id}: "
f"received {received_crypto} {crypto_payment.coin_type}, expected {expected_crypto} {crypto_payment.coin_type} "
f"({early_min_confs}/{crypto_payment.confirmations_required} confirmations) - checking threshold"
)
# Check if overpayment exceeds 5% threshold using PaymentRescue logic
refund_details = payment_rescue.handle_overpayment(
crypto_payment,
expected_crypto,
received_crypto,
crypto_payment.invoice.user,
)
if refund_details:
# Overpayment exceeds threshold - process refund
logger.info(
f"EARLY DETECTION - Overpayment threshold exceeded for {crypto_payment.id}: {refund_details}"
)
# Mark as confirmed with overpayment detected (refund pending)
crypto_payment.status = CryptoPayment.STATUS_CONFIRMED_OVERPAID
crypto_payment.refund_reason = refund_details["reason"]
# Try refund immediately since we have enough confirmations
result = payment_rescue.execute_refund(
refund_details, crypto_payment
)
if result["success"]:
logger.info(
f"Excess refunded for payment {crypto_payment.id}: TX {result['tx_hash']}"
)
# Mark as confirmed with overpayment refunded
crypto_payment.status = (
CryptoPayment.STATUS_CONFIRMED_OVERPAID_REFUNDED
)
crypto_payment.refund_tx_hash = result["tx_hash"]
crypto_payment.refund_confirmations = 0 # Just sent
# Finalize invoice since core payment amount is sufficient
finalize_invoice(env_request, crypto_payment, send_emails=True)
# Sweep restocking fee to shop owner
sweep_restocking_fee(
env_request.registry.settings,
crypto_payment,
refund_details,
env_request.dbsession,
"Overpayment",
)
else:
logger.error(
f"Refund failed for overpayment {crypto_payment.id}: {result['error']}"
)
else:
# Overpayment within acceptable threshold - just confirm normally
logger.info(
f"EARLY DETECTION - Overpayment within 5% threshold for {crypto_payment.id} - confirming normally"
)
crypto_payment.status = CryptoPayment.STATUS_CONFIRMED
finalize_invoice(env_request, crypto_payment, send_emails=True)
# Check for overpayment (exact match or overpaid) - for first payment
elif (
new_sum_int >= expected_amount_int
@ -2206,6 +2662,18 @@ def process_refund_confirmations(request, settings):
logger.info(
f"REFUND RETRY SUCCESSFUL for payment {payment.id}: TX {result['tx_hash']} (incoming payment now has {payment.current_confirmations} confirmations)"
)
# Commit the refund first
db.flush()
# Sweep restocking fee to shop owner
sweep_restocking_fee(
request.registry.settings,
payment,
refund_details,
db,
"Refund retry",
)
continue
logger.debug(

View file

@ -252,3 +252,27 @@ class CryptoPayment(RBase, Base):
def is_fully_confirmed(self) -> bool:
"""Check if payment has reached required confirmation count."""
return self.current_confirmations >= self.confirmations_required
def is_finalized(self) -> bool:
"""Check if payment has been finalized (customer received their product).
Only payments in confirmed statuses with sufficient funds and confirmations
can be considered truly finalized.
"""
# Must be in a status that could potentially be finalized
if self.status not in [
self.STATUS_CONFIRMED,
self.STATUS_CONFIRMED_OVERPAID,
self.STATUS_CONFIRMED_OVERPAID_REFUNDED,
]:
return False
# Must have received sufficient funds (at least expected amount)
if not (self.received_amount >= self.expected_amount):
return False
# Must have sufficient confirmations
if not (self.current_confirmations >= self.confirmations_required):
return False
return True

File diff suppressed because it is too large Load diff