diff --git a/lnbits/wallets/lightspark.py b/lnbits/wallets/lightspark.py index 902fc9752..45e8cbb9a 100644 --- a/lnbits/wallets/lightspark.py +++ b/lnbits/wallets/lightspark.py @@ -1,6 +1,7 @@ +import asyncio import hashlib import json -from typing import Any +from typing import Any, AsyncGenerator import httpx from bolt11 import decode as bolt11_decode @@ -238,3 +239,46 @@ class LightsparkSparkWallet(Wallet): if mapped.failed: return False return None + + async def _poll_pending_invoices(self) -> AsyncGenerator[str, None]: + while settings.lnbits_running: + for invoice in list(self.pending_invoices): + try: + status = await self.get_invoice_status(invoice) + if status.paid: + yield invoice + self.pending_invoices.remove(invoice) + elif status.failed: + self.pending_invoices.remove(invoice) + except Exception as exc: + logger.error(f"could not get status of invoice {invoice}: '{exc}' ") + await asyncio.sleep(5) + + async def paid_invoices_stream(self) -> AsyncGenerator[str, None]: + stream_path = "/v1/invoices/stream" + while settings.lnbits_running: + try: + async with self.client.stream("GET", stream_path, timeout=None) as r: + if r.status_code in {404, 405}: + logger.warning( + "Spark sidecar invoice stream not available, " + "falling back to polling." + ) + async for checking_id in self._poll_pending_invoices(): + yield checking_id + return + r.raise_for_status() + logger.info("connected to Spark sidecar invoice stream.") + async for line in r.aiter_lines(): + if not line or not line.startswith("data:"): + continue + data = json.loads(line[5:].strip()) + checking_id = data.get("checking_id") + if checking_id: + yield checking_id + except Exception as exc: + logger.error( + "lost connection to Spark sidecar invoice stream: " + f"'{exc}' retrying in 5 seconds" + ) + await asyncio.sleep(5)