mirror of
https://github.com/Such-Software/wowcheckout.git
synced 2026-10-17 05:51:45 +02:00
jwinterm
f9159f5877
The codebase was forked from xmrcheckout and kept Monero's 10^12 atomic-unit constant everywhere. Wownero uses 11 decimal places, not 12. Effect: _xmr_to_atomic(amount) produced a required_atomic value 10× larger than what wallet-rpc actually reports for the subaddress. Invoices flagged total_paid_atomic correctly (that comes from the daemon) but is_paid = total_paid >= required was always false, so the reconciler never flipped status pending → payment_detected, webhook never fired, Medusa never saw the payment. Fix all four places: reconciler._xmr_to_atomic, btcpay_routes._xmr_to_atomic, btcpay_routes._atomic_to_xmr, btcpay_routes._format_xmr_fixed, routes.py invoice-status response. Extracted a WOW_ATOMIC_UNITS constant so this doesn't get lost again when we touch these files.
437 lines
17 KiB
Python
437 lines
17 KiB
Python
from __future__ import annotations
|
||
|
||
import logging
|
||
import time
|
||
from datetime import datetime, timedelta, timezone
|
||
from decimal import Decimal, ROUND_DOWN
|
||
|
||
from sqlalchemy import and_, or_, text
|
||
from sqlalchemy.exc import SQLAlchemyError
|
||
from sqlalchemy.orm import Session
|
||
|
||
from .btcpay_webhooks import dispatch_btcpay_webhooks
|
||
from .config import INVOICE_RECONCILE_INTERVAL_SECONDS, LATE_PAYMENT_LOOKBACK_HOURS
|
||
from .db import SessionLocal
|
||
from .models import Invoice, InvoiceTransfer, SystemStatus, User
|
||
from .monero_service import MoneroWalletService, TransferDetail
|
||
from .webhooks import dispatch_webhooks
|
||
|
||
logger = logging.getLogger(__name__)
|
||
MONERO_CONNECTIVITY_STATUS_NAME = "monero_connectivity"
|
||
|
||
|
||
def main() -> None:
|
||
level_name = "INFO"
|
||
try:
|
||
level_name = __import__("os").getenv("LOG_LEVEL", "INFO")
|
||
except Exception:
|
||
level_name = "INFO"
|
||
level = getattr(logging, level_name.upper(), logging.INFO)
|
||
logging.basicConfig(level=level)
|
||
|
||
# Silence expected RPC -13 ("No wallet file") spam from the python-monero
|
||
# library. Those happen routinely when close_wallet is called on an idle
|
||
# backend (by _ensure_wallet_open) and are handled by our retry logic;
|
||
# they are not actionable errors and were drowning real log output.
|
||
# Real wallet-rpc problems still bubble up as exceptions which we log
|
||
# explicitly through logger.error / logger.exception.
|
||
logging.getLogger("monero.backends.jsonrpc.wallet").setLevel(logging.CRITICAL)
|
||
|
||
while True:
|
||
status_db: Session | None = None
|
||
try:
|
||
status_db = SessionLocal()
|
||
service = MoneroWalletService()
|
||
_safe_update_monero_connectivity_status(status_db, service)
|
||
_safe_update_reconciler_status(
|
||
status_db,
|
||
started_at=datetime.now(timezone.utc),
|
||
completed_at=None,
|
||
error_message=None,
|
||
)
|
||
_reconcile_invoices(service)
|
||
_safe_update_reconciler_status(
|
||
status_db,
|
||
completed_at=datetime.now(timezone.utc),
|
||
error_message=None,
|
||
)
|
||
except Exception as exc:
|
||
logger.exception("Invoice reconcile failed: %s", exc)
|
||
if status_db is None:
|
||
status_db = SessionLocal()
|
||
_safe_update_monero_connectivity_error(status_db)
|
||
_safe_update_reconciler_status(
|
||
status_db,
|
||
error_message=str(exc),
|
||
)
|
||
finally:
|
||
if status_db is not None:
|
||
status_db.close()
|
||
time.sleep(INVOICE_RECONCILE_INTERVAL_SECONDS)
|
||
|
||
|
||
def _reconcile_invoices(service: MoneroWalletService) -> None:
|
||
db: Session = SessionLocal()
|
||
try:
|
||
now = datetime.now(timezone.utc)
|
||
late_cutoff = now - timedelta(hours=max(0, LATE_PAYMENT_LOOKBACK_HOURS))
|
||
invoices = (
|
||
db.query(Invoice)
|
||
.filter(
|
||
or_(
|
||
Invoice.status.in_(["pending", "payment_detected"]),
|
||
and_(
|
||
Invoice.status == "expired",
|
||
Invoice.expires_at.is_not(None),
|
||
Invoice.expires_at >= late_cutoff,
|
||
),
|
||
)
|
||
)
|
||
.order_by(Invoice.created_at.asc())
|
||
.all()
|
||
)
|
||
user_groups: dict[object, list[Invoice]] = {}
|
||
for invoice in invoices:
|
||
if invoice.user_id is None:
|
||
logger.debug(
|
||
"Skipping invoice without user",
|
||
extra={"invoice_id": str(invoice.id)},
|
||
)
|
||
continue
|
||
user_groups.setdefault(invoice.user_id, []).append(invoice)
|
||
logger.debug(
|
||
"Reconciling %d invoices across %d users",
|
||
len(invoices),
|
||
len(user_groups),
|
||
)
|
||
for user_id, user_invoices in user_groups.items():
|
||
user = db.query(User).filter(User.id == user_id).first()
|
||
if user is None:
|
||
logger.debug(
|
||
"Skipping invoices with missing user",
|
||
extra={"user_id": str(user_id)},
|
||
)
|
||
continue
|
||
if not user.payment_address or not user.view_key_encrypted:
|
||
logger.debug(
|
||
"Skipping invoices without payment address",
|
||
extra={"user_id": str(user.id)},
|
||
)
|
||
continue
|
||
for invoice in user_invoices:
|
||
# Advisory lock prevents duplicate reconciliation during deploys
|
||
lock_id = hash(str(invoice.id)) & 0x7FFFFFFF
|
||
acquired = db.execute(
|
||
text("SELECT pg_try_advisory_lock(:id)"),
|
||
{"id": lock_id},
|
||
).scalar()
|
||
if not acquired:
|
||
continue
|
||
try:
|
||
transfers = _get_transfers_with_retry(
|
||
service, user, invoice.address,
|
||
)
|
||
|
||
total_atomic = 0
|
||
max_confirmations = 0
|
||
for transfer in transfers:
|
||
if transfer.amount_atomic <= 0:
|
||
continue
|
||
total_atomic += transfer.amount_atomic
|
||
if transfer.confirmations > max_confirmations:
|
||
max_confirmations = transfer.confirmations
|
||
logger.debug(
|
||
"Invoice totals",
|
||
extra={
|
||
"invoice_id": str(invoice.id),
|
||
"received_atomic": total_atomic,
|
||
"confirmations": max_confirmations,
|
||
},
|
||
)
|
||
now = datetime.now(timezone.utc)
|
||
previous_confirmations = invoice.confirmations or 0
|
||
total_changed = invoice.total_paid_atomic != total_atomic
|
||
confirmations_changed = previous_confirmations != max_confirmations
|
||
transfers_changed = _sync_invoice_transfers(
|
||
db,
|
||
invoice=invoice,
|
||
transfers=transfers,
|
||
)
|
||
if total_changed or confirmations_changed or transfers_changed:
|
||
if confirmations_changed:
|
||
invoice.confirmations = max_confirmations
|
||
if total_changed:
|
||
invoice.total_paid_atomic = total_atomic
|
||
db.add(invoice)
|
||
db.commit()
|
||
required_atomic = _xmr_to_atomic(invoice.amount_xmr)
|
||
is_paid = total_atomic >= required_atomic
|
||
|
||
expires_at = invoice.expires_at
|
||
if expires_at is not None and expires_at.tzinfo is None:
|
||
expires_at = expires_at.replace(tzinfo=timezone.utc)
|
||
is_after_expiry = bool(expires_at and now >= expires_at)
|
||
|
||
if total_atomic < required_atomic:
|
||
logger.debug(
|
||
"Payment not yet detected",
|
||
extra={
|
||
"invoice_id": str(invoice.id),
|
||
"required_atomic": required_atomic,
|
||
"received_atomic": total_atomic,
|
||
},
|
||
)
|
||
if invoice.status == "pending" and is_after_expiry:
|
||
logger.info(
|
||
"Invoice expired",
|
||
extra={"invoice_id": str(invoice.id), "user_id": str(user.id)},
|
||
)
|
||
invoice.status = "expired"
|
||
db.add(invoice)
|
||
db.commit()
|
||
dispatch_webhooks(db, str(user.id), "invoice.expired", invoice)
|
||
dispatch_btcpay_webhooks(
|
||
db, str(user.id), "InvoiceExpired", invoice
|
||
)
|
||
continue
|
||
|
||
if is_paid and invoice.status in ("pending", "expired"):
|
||
logger.info(
|
||
"Invoice marked payment detected",
|
||
extra={"invoice_id": str(invoice.id), "user_id": str(user.id)},
|
||
)
|
||
previous_status = invoice.status
|
||
invoice.status = "payment_detected"
|
||
if invoice.detected_at is None:
|
||
invoice.detected_at = now
|
||
if previous_status == "expired" or (previous_status == "pending" and is_after_expiry):
|
||
invoice.paid_after_expiry = True
|
||
if invoice.paid_after_expiry_at is None:
|
||
invoice.paid_after_expiry_at = now
|
||
db.add(invoice)
|
||
db.commit()
|
||
dispatch_webhooks(db, str(user.id), "invoice.payment_detected", invoice)
|
||
dispatch_btcpay_webhooks(
|
||
db, str(user.id), "InvoiceReceivedPayment", invoice
|
||
)
|
||
dispatch_btcpay_webhooks(db, str(user.id), "InvoicePaidInFull", invoice)
|
||
dispatch_btcpay_webhooks(
|
||
db, str(user.id), "InvoiceProcessing", invoice
|
||
)
|
||
if max_confirmations >= invoice.confirmation_target and invoice.status != "confirmed":
|
||
logger.info(
|
||
"Invoice confirmed",
|
||
extra={"invoice_id": str(invoice.id), "user_id": str(user.id)},
|
||
)
|
||
invoice.status = "confirmed"
|
||
if invoice.confirmed_at is None:
|
||
invoice.confirmed_at = now
|
||
db.add(invoice)
|
||
db.commit()
|
||
dispatch_webhooks(db, str(user.id), "invoice.confirmed", invoice)
|
||
dispatch_btcpay_webhooks(db, str(user.id), "InvoiceSettled", invoice)
|
||
dispatch_btcpay_webhooks(
|
||
db, str(user.id), "InvoicePaymentSettled", invoice
|
||
)
|
||
except Exception as exc:
|
||
logger.error(
|
||
"Failed to reconcile invoice: %s",
|
||
exc,
|
||
extra={"invoice_id": str(invoice.id), "user_id": str(user.id)},
|
||
)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
db.execute(text("SELECT pg_advisory_unlock(:id)"), {"id": lock_id})
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def _get_transfers_with_retry(
|
||
service: MoneroWalletService,
|
||
user: User,
|
||
address: str,
|
||
retries: int = 3,
|
||
) -> list[TransferDetail]:
|
||
"""Call get_transfers_for_address with retry + backoff on RPC errors."""
|
||
delays = [0.5, 1.0, 2.0]
|
||
last_exc: Exception | None = None
|
||
for attempt in range(retries):
|
||
try:
|
||
return service.get_transfers_for_address(user=user, address=address)
|
||
except Exception as exc:
|
||
last_exc = exc
|
||
if attempt < retries - 1:
|
||
time.sleep(delays[attempt])
|
||
raise last_exc # type: ignore[misc]
|
||
|
||
|
||
def _sync_invoice_transfers(
|
||
db: Session,
|
||
*,
|
||
invoice: Invoice,
|
||
transfers: list[TransferDetail],
|
||
) -> bool:
|
||
existing = (
|
||
db.query(InvoiceTransfer)
|
||
.filter(InvoiceTransfer.invoice_id == invoice.id)
|
||
.all()
|
||
)
|
||
existing_by_txid = {transfer.txid: transfer for transfer in existing if transfer.txid}
|
||
seen_txids: set[str] = set()
|
||
changed = False
|
||
for transfer in transfers:
|
||
txid = transfer.txid
|
||
if not txid:
|
||
continue
|
||
seen_txids.add(txid)
|
||
stored = existing_by_txid.get(txid)
|
||
if stored is None:
|
||
db.add(
|
||
InvoiceTransfer(
|
||
invoice_id=invoice.id,
|
||
txid=txid,
|
||
amount_atomic=transfer.amount_atomic,
|
||
confirmations=transfer.confirmations,
|
||
timestamp=transfer.timestamp,
|
||
address=transfer.address,
|
||
)
|
||
)
|
||
changed = True
|
||
continue
|
||
if (
|
||
stored.amount_atomic != transfer.amount_atomic
|
||
or stored.confirmations != transfer.confirmations
|
||
or stored.timestamp != transfer.timestamp
|
||
or stored.address != transfer.address
|
||
):
|
||
stored.amount_atomic = transfer.amount_atomic
|
||
stored.confirmations = transfer.confirmations
|
||
stored.timestamp = transfer.timestamp
|
||
stored.address = transfer.address
|
||
changed = True
|
||
# Only delete transfers if we got data back — don't wipe records on RPC glitch
|
||
if transfers:
|
||
for stored in existing:
|
||
if stored.txid not in seen_txids:
|
||
db.delete(stored)
|
||
changed = True
|
||
return changed
|
||
|
||
|
||
def _update_reconciler_status(
|
||
db: Session,
|
||
*,
|
||
started_at: datetime | None = None,
|
||
completed_at: datetime | None = None,
|
||
error_message: str | None = None,
|
||
) -> None:
|
||
status_row = db.query(SystemStatus).filter(SystemStatus.name == "reconciler").first()
|
||
if status_row is None:
|
||
status_row = SystemStatus(name="reconciler")
|
||
if started_at is not None:
|
||
status_row.last_reconcile_started_at = started_at
|
||
if completed_at is not None:
|
||
status_row.last_reconcile_completed_at = completed_at
|
||
status_row.last_reconcile_error = error_message
|
||
db.add(status_row)
|
||
db.commit()
|
||
|
||
|
||
def _update_monero_connectivity_status(
|
||
db: Session,
|
||
*,
|
||
wallet_rpc: str,
|
||
daemon: str,
|
||
daemon_height: int | None,
|
||
) -> None:
|
||
status_row = (
|
||
db.query(SystemStatus)
|
||
.filter(SystemStatus.name == MONERO_CONNECTIVITY_STATUS_NAME)
|
||
.first()
|
||
)
|
||
if status_row is None:
|
||
status_row = SystemStatus(name=MONERO_CONNECTIVITY_STATUS_NAME)
|
||
status_row.wallet_rpc = wallet_rpc
|
||
status_row.daemon = daemon
|
||
status_row.daemon_height = daemon_height
|
||
status_row.checked_at = datetime.now(timezone.utc)
|
||
db.add(status_row)
|
||
db.commit()
|
||
|
||
|
||
def _safe_update_monero_connectivity_status(
|
||
db: Session,
|
||
service: MoneroWalletService,
|
||
) -> None:
|
||
try:
|
||
status_payload = service.get_status()
|
||
_update_monero_connectivity_status(
|
||
db,
|
||
wallet_rpc=str(status_payload.get("wallet_rpc", "unreachable")),
|
||
daemon=str(status_payload.get("daemon", "unknown")),
|
||
daemon_height=(
|
||
int(status_payload["daemon_height"])
|
||
if isinstance(status_payload.get("daemon_height"), int)
|
||
else None
|
||
),
|
||
)
|
||
except Exception:
|
||
db.rollback()
|
||
logger.warning("Unable to persist Monero connectivity status", exc_info=True)
|
||
|
||
|
||
def _safe_update_monero_connectivity_error(db: Session) -> None:
|
||
try:
|
||
_update_monero_connectivity_status(
|
||
db,
|
||
wallet_rpc="unreachable",
|
||
daemon="unknown",
|
||
daemon_height=None,
|
||
)
|
||
except SQLAlchemyError:
|
||
db.rollback()
|
||
logger.warning("Unable to persist Monero connectivity fallback", exc_info=True)
|
||
|
||
|
||
def _safe_update_reconciler_status(
|
||
db: Session,
|
||
*,
|
||
started_at: datetime | None = None,
|
||
completed_at: datetime | None = None,
|
||
error_message: str | None = None,
|
||
) -> None:
|
||
try:
|
||
_update_reconciler_status(
|
||
db,
|
||
started_at=started_at,
|
||
completed_at=completed_at,
|
||
error_message=error_message,
|
||
)
|
||
except SQLAlchemyError:
|
||
db.rollback()
|
||
logger.warning("Unable to persist reconciler status heartbeat", exc_info=True)
|
||
|
||
|
||
|
||
|
||
# Wownero uses 11 decimal places (atomic unit = 10^-11 WOW), NOT 12 like
|
||
# Monero. Using 10^12 here makes required_atomic 10× larger than actual paid
|
||
# amount, so is_paid never becomes true and invoices stay pending forever
|
||
# even after detection + confirmations. This entire codebase was forked from
|
||
# xmrcheckout and the constant must be corrected wherever it appears.
|
||
WOW_ATOMIC_UNITS = Decimal("100000000000") # 10^11
|
||
|
||
|
||
def _xmr_to_atomic(amount: Decimal) -> int:
|
||
quantized = (Decimal(amount) * WOW_ATOMIC_UNITS).to_integral_value(
|
||
rounding=ROUND_DOWN
|
||
)
|
||
return int(quantized)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|