refactor: move code from lnd impl to repository

This avoids duplicated code with other ln implementations later on.

This also pushes LightningStatus object to the SSE channel
instead of the full LnInfo object.

refs #11

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())

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")

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())

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"