diff --git a/__init__.py b/__init__.py index e9efc6c..873500f 100644 --- a/__init__.py +++ b/__init__.py @@ -36,17 +36,19 @@ def satspay_stop(): def satspay_start(): - from lnbits.tasks import create_permanent_unique_task, create_unique_task + from lnbits.tasks import create_permanent_unique_task paid_invoices_task = create_permanent_unique_task( "ext_satspay_paid_invoices", wait_for_paid_invoices ) onchain_task = create_permanent_unique_task("ext_satspay_onchain", wait_for_onchain) - scheduled_tasks.extend([paid_invoices_task, onchain_task]) - restart_websocket_task() - create_unique_task( - "ext_satspay_restart_address_tracking", restart_address_tracking() + onchain_balance_task = create_permanent_unique_task( + "ext_satspay_restart_address_tracking", restart_address_tracking + ) + scheduled_tasks.extend( + [paid_invoices_task, onchain_task, onchain_balance_task] ) + restart_websocket_task() __all__ = ["db", "satspay_ext", "satspay_start", "satspay_static_files", "satspay_stop"] diff --git a/helpers.py b/helpers.py index 834cf00..00852b0 100644 --- a/helpers.py +++ b/helpers.py @@ -5,7 +5,7 @@ from lnbits.settings import settings from loguru import logger -from .crud import get_or_create_satspay_settings +from .crud import get_or_create_satspay_settings, update_charge from .models import Charge, OnchainBalance @@ -114,9 +114,18 @@ async def check_charge_balance(charge: Charge) -> Charge: charge.paid = charge.balance >= charge.amount + if not charge.paid: + return charge + + # Persist the settled state BEFORE firing the webhook so webhook handlers + # that re-fetch the charge (e.g. GET /satspay/api/v1/charge/{id}) observe + # it as paid. Mirrors the ordering in tasks._handle_ws_message. + charge = await update_charge(charge) + if charge.webhook: resp = await call_webhook(charge) charge.add_extra(resp) + charge = await update_charge(charge) return charge diff --git a/tasks.py b/tasks.py index 49258a5..1d64ad4 100644 --- a/tasks.py +++ b/tasks.py @@ -20,22 +20,49 @@ tracked_addresses: list[str] = [] public_ws_listeners: dict[str, list[WebSocket]] = {} +# How often the on-chain balance is re-checked by REST as a fallback to the +# mempool websocket. The websocket only tracks successfully while the number of +# active addresses stays under mempool's per-connection limit, so this poll is +# the reliable detection path. +ONCHAIN_BALANCE_POLL_SECONDS = 120 + async def restart_address_tracking(): - charges = await get_pending_charges() - for charge in charges: - if ( - charge.onchainaddress - and charge.timestamp.timestamp() + charge.time * 60 > time.time() - ): - charge = await check_charge_balance(charge) - assert charge.onchainaddress - if charge.paid: - charge.add_extra({"payment_method": "onchain"}) - await update_charge(charge) - logger.success(f"Charge {charge.id} marked as paid.") - continue - start_onchain_listener(charge.onchainaddress) + while settings.lnbits_running: + try: + charges = await get_pending_charges() + except Exception as exc: + logger.warning(f"Failed to load pending charges: {exc!s}") + await asyncio.sleep(ONCHAIN_BALANCE_POLL_SECONDS) + continue + + logger.debug( + f"Onchain balance poll: {len(charges)} pending charges, " + f"{len(tracked_addresses)} tracked." + ) + + for charge in charges: + try: + if not charge.onchainaddress: + continue + if charge.timestamp.timestamp() + charge.time * 60 <= time.time(): + # Expired: drop from ws tracking so stale addresses do not + # consume mempool's per-connection address limit. + stop_onchain_listener(charge.onchainaddress) + continue + charge = await check_charge_balance(charge) + assert charge.onchainaddress + if charge.paid: + charge.add_extra({"payment_method": "onchain"}) + await update_charge(charge) + stop_onchain_listener(charge.onchainaddress) + logger.success(f"Charge {charge.id} marked as paid.") + continue + start_onchain_listener(charge.onchainaddress) + except Exception as exc: + logger.warning(f"Error checking charge {charge.id}: {exc!s}") + + await asyncio.sleep(ONCHAIN_BALANCE_POLL_SECONDS) async def wait_for_paid_invoices():