diff --git a/app/repositories/lightning.py b/app/repositories/lightning.py index f393fef..5aae485 100644 --- a/app/repositories/lightning.py +++ b/app/repositories/lightning.py @@ -1,5 +1,8 @@ +import asyncio + from app.models.lightning import Invoice, LightningStatus, LnInfo, Payment from app.utils import SSE, lightning_config, send_sse_message +from decouple import config if lightning_config.ln_node == "lnd": from app.repositories.ln_impl.lnd import ( @@ -7,7 +10,7 @@ if lightning_config.ln_node == "lnd": get_implementation_name, get_ln_info_impl, get_wallet_balance_impl, - register_lightning_listener_impl, + listen_invoices, send_payment_impl, ) else: @@ -16,10 +19,14 @@ else: get_implementation_name, get_ln_info_impl, get_wallet_balance_impl, - register_lightning_listener_impl, + listen_invoices, send_payment_impl, ) +GATHER_INFO_INTERVALL = config("gather_ln_info_interval", default=5, cast=float) + +_CACHE = {"wallet_balance": None} + async def get_ln_status() -> LightningStatus: ln_info = await get_ln_info_impl() @@ -41,18 +48,46 @@ async def send_payment( pay_req: str, timeout_seconds: int, fee_limit_msat: int ) -> Payment: res = await send_payment_impl(pay_req, timeout_seconds, fee_limit_msat) - update_wallet_balance_via_sse() + _update_wallet_balance() return res -async def update_wallet_balance_via_sse(): - res = await get_wallet_balance_impl() - await send_sse_message(SSE.WALLET_BALANCE, res) - - async def get_ln_info() -> LnInfo: return await get_ln_info_impl() async def register_lightning_listener(): - await register_lightning_listener_impl() + loop = asyncio.get_event_loop() + loop.create_task(_handle_info_listener()) + loop.create_task(_handle_invoice_listener()) + + +async def _handle_info_listener(): + last_info = None + while True: + info = await get_ln_info_impl() + + if last_info != info: + status = LightningStatus.from_grpc(get_implementation_name(), info) + await send_sse_message(SSE.LN_STATUS, status.dict()) + last_info = info + + await asyncio.sleep(GATHER_INFO_INTERVALL) + + +async def _handle_invoice_listener(): + async for i in listen_invoices(): + await send_sse_message(SSE.LN_INVOICE_STATUS, i.dict()) + _update_wallet_balance() + + +def _update_wallet_balance(): + async def _perform_update(): + await asyncio.sleep(1.1) + wb = await get_wallet_balance_impl() + if _CACHE["wallet_balance"] != wb: + await send_sse_message(SSE.WALLET_BALANCE, wb.dict()) + _CACHE["wallet_balance"] = wb + + loop = asyncio.get_event_loop() + loop.create_task(_perform_update()) diff --git a/app/repositories/ln_impl/clightning.py b/app/repositories/ln_impl/clightning.py index 7478c04..a443ae7 100644 --- a/app/repositories/ln_impl/clightning.py +++ b/app/repositories/ln_impl/clightning.py @@ -25,5 +25,5 @@ async def get_ln_info_impl() -> LnInfo: raise NotImplementedError("c-lightning not yet implemented") -async def register_lightning_listener_impl(): +async def listen_invoices() -> Invoice: raise NotImplementedError("c-lightning not yet implemented") diff --git a/app/repositories/ln_impl/lnd.py b/app/repositories/ln_impl/lnd.py index 98a3bd3..11d53ab 100644 --- a/app/repositories/ln_impl/lnd.py +++ b/app/repositories/ln_impl/lnd.py @@ -17,14 +17,9 @@ from app.models.lightning import ( from app.utils import SSE from app.utils import lightning_config as lncfg from app.utils import send_sse_message -from decouple import config from fastapi.exceptions import HTTPException from starlette import status -GATHER_INFO_INTERVALL = config("gather_ln_info_interval", default=5, cast=float) - -_CACHE = {"wallet_balance": None} - def get_implementation_name() -> str: return "LND" @@ -81,7 +76,6 @@ async def send_payment_impl( async for response in lncfg.router_stub.SendPaymentV2(r): p = payment_from_grpc(response) await send_sse_message(SSE.LN_PAYMENT_STATUS, p.dict()) - _update_wallet_balance() return p except grpc.aio._call.AioRpcError as error: if ( @@ -103,52 +97,10 @@ async def get_ln_info_impl() -> LnInfo: return ln_info_from_grpc(response) -async def register_lightning_listener_impl(): - loop = asyncio.get_event_loop() - loop.create_task(_handle_invoice_listener()) - loop.create_task(_handle_get_info_gatherer_impl()) - - -async def _handle_invoice_listener(): +async def listen_invoices() -> Invoice: request = ln.InvoiceSubscription() - try: async for r in lncfg.lnd_stub.SubscribeInvoices(request): - i = invoice_from_grpc(r) - await send_sse_message(SSE.LN_INVOICE_STATUS, i.dict()) - if i.state == InvoiceState.settled: - # Wallet balance was updated, send to connected clients - _update_wallet_balance() + yield invoice_from_grpc(r) except error: print(error) - - -async def _handle_get_info_gatherer_impl(): - last_info = None - while True: - info = await get_ln_info_impl() - - if last_info != info: - await send_sse_message(SSE.LN_INFO, info.dict()) - last_info = info - - # LND doesn't provide a subscription for payment events - # This means, if the user pays a via command line or another app - # connected directly to LND, no event is fired to other clients. - # As a workaround we'll poll the wallet balance regularly and send - # it, if it's different. Related: https://github.com/lightningnetwork/lnd/pull/1962 - _update_wallet_balance() - - await asyncio.sleep(GATHER_INFO_INTERVALL) - - -def _update_wallet_balance(): - async def _perform_update(): - await asyncio.sleep(1.1) - wb = await get_wallet_balance_impl() - if _CACHE["wallet_balance"] != wb: - await send_sse_message(SSE.WALLET_BALANCE, wb.dict()) - _CACHE["wallet_balance"] = wb - - loop = asyncio.get_event_loop() - loop.create_task(_perform_update()) diff --git a/app/utils.py b/app/utils.py index deaf15a..8d9ae82 100644 --- a/app/utils.py +++ b/app/utils.py @@ -145,6 +145,7 @@ class SSE: BTC_INFO = "btc_info" LN_INFO = "ln_info" + LN_STATUS = "ln_status" LN_INVOICE_STATUS = "ln_invoice_status" LN_PAYMENT_STATUS = "ln_payment_status"