blitz_api/app/lightning/impl/cln_grpc.py

import asyncio
import json
import sys
from typing import AsyncGenerator, List, Optional

import grpc
from fastapi.exceptions import HTTPException
from loguru import logger
from starlette import status

import app.lightning.impl.protos.cln.node_pb2 as ln
import app.lightning.impl.protos.cln.node_pb2_grpc as clnrpc
import app.lightning.impl.protos.cln.primitives_pb2 as lnp
from app.api.config import config
from app.api.utils import Event, broadcast_msg, config_get_hex_str, next_push_id
from app.bitcoind.utils import bitcoin_rpc_async
from app.lightning.exceptions import NodeNotFoundError
from app.lightning.impl.cln_utils import cln_classify_fee_revenue, parse_cln_msat
from app.lightning.impl.ln_base import LightningNodeBase
from app.lightning.models import (
    Channel,
    FeeRevenue,
    ForwardSuccessEvent,
    GenericTx,
    InitLnRepoUpdate,
    Invoice,
    InvoiceState,
    LnInfo,
    LnInitState,
    NewAddressInput,
    OnChainTransaction,
    Payment,
    PaymentRequest,
    SendCoinsInput,
    SendCoinsResponse,
    TxStatus,
    WalletBalance,
)
from app.lightning.utils import (
    alias_or_empty,
    generic_grpc_error_handler,
    raise_for_pay_req_decode_error,
)


@logger.catch(exclude=(HTTPException,))
async def _make_local_call(*args: str):
    # FIXME: this is a hack because some of the commands are not exposed
    # in the CLN grpc interface yet.

    # Pass the command as a discrete argv list (create_subprocess_exec, not
    # _shell) so user-controlled arguments such as the bolt11 in decodepay
    # can never be interpreted as shell syntax.
    testnet = config("BAPI_NETWORK") == "testnet"
    argv = ["lightning-cli", "-k", *(["--testnet"] if testnet else []), *args]
    proc = await asyncio.create_subprocess_exec(
        *argv,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()

    if stderr is not None and stderr != b"":
        err = stderr.decode()
        if "lightning-cli: Connecting to 'lightning-rpc': Permission denied" in err:
            logger.critical(
                (
                    "Unable to connect to lightning-cli: Permission denied. "
                    "Is the lightning-rpc socket readable for the API user?"
                )
            )

            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="Unable to connect to lightning-cli: Permission denied.",
            )

        if "lightning-cli: Moving into" in err and "No such file or directory" in err:
            logger.critical(
                (
                    "Unable to connect to lightning-cli: No such file or directory. "
                    "Is the lightning-rpc socket available to the API user?"
                )
            )

            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=(
                    "Unable to connect to lightning-cli: "
                    "API Can't access lightning-cli.",
                ),
            )

        logger.critical(f"Unable to connect to lightning-cli: {err}")

        raise HTTPException(
            status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=(
                "Unable to connect to lightning-cli: "
                "Unknown error. Please consult the logs."
            ),
        )
    return stdout, stderr


@logger.catch(exclude=(HTTPException,))
def _extract_message(details):
    return details.split('message: "')[1].replace('" }', ".")


class LnNodeCLNgRPC(LightningNodeBase):
    _initialized = False
    _channel = None
    _cln_stub: clnrpc.NodeStub = None
    # Decoding the payment request take a long time,
    # hence we build a simple cache here.
    _memo_cache = {}
    _block_cache = {}

    def get_implementation_name(self) -> str:
        return "CLN_GRPC"

    @logger.catch(exclude=(HTTPException,))
    async def initialize(self) -> AsyncGenerator[InitLnRepoUpdate, None]:
        logger.info("Establishing a connection to the CLN daemon ...")
        if self._initialized:
            logger.warning(
                (
                    "Connection already initialized. "
                    "This function must not be called twice."
                )
            )
            yield InitLnRepoUpdate(state=LnInitState.DONE)

        try:
            cln_grpc_key = bytes.fromhex(
                config_get_hex_str(
                    str(config("BAPI_CLN_GRPC_KEY")), name="cln_grpc_key"
                )
            )
            cln_grpc_cert = bytes.fromhex(
                config_get_hex_str(
                    str(config("BAPI_CLN_GRPC_CERT")), name="cln_grpc_cert"
                )
            )
            cln_grpc_ca = bytes.fromhex(
                config_get_hex_str(str(config("BAPI_CLN_GRPC_CA")), name="cln_grpc_ca")
            )
            cln_grpc_url = (
                str(config("BAPI_CLN_GRPC_IP"))
                + ":"
                + str(config("BAPI_CLN_GRPC_PORT"))
            )
        except ValueError as e:
            logger.critical(f"Unable to decode BAPI_CLN_GRPC_CERT: {e.args}.")
            sys.exit(0)

        self.creds = grpc.ssl_channel_credentials(
            root_certificates=cln_grpc_ca,
            private_key=cln_grpc_key,
            certificate_chain=cln_grpc_cert,
        )

        opts = (
            ("grpc.ssl_target_name_override", "cln"),
            ("grpc.max_receive_message_length", 1024 * 1024 * 10),
        )

        while not self._initialized:
            logger.trace("iterating ...")
            try:
                if self._channel is None:
                    self._channel = grpc.aio.secure_channel(
                        cln_grpc_url, self.creds, options=opts
                    )
                    self._cln_stub = clnrpc.NodeStub(self._channel)

                await self._cln_stub.Getinfo(ln.GetinfoRequest())
                self._initialized = True
                yield InitLnRepoUpdate(state=LnInitState.DONE)
            except grpc.aio._call.AioRpcError as error:
                details = error.details()
                logger.debug(f"Waiting for CLN daemon... Details {details}")

                if "failed to connect to all addresses" in details:
                    yield InitLnRepoUpdate(
                        state=LnInitState.OFFLINE,
                        msg="Unable to connect to CLN daemon, waiting...",
                    )

                    await self._channel.close()
                    self._channel = self.cln_stub = None
                else:
                    logger.error(f"Unknown error: {details}")
                    raise

                await asyncio.sleep(2)
            except Exception as e:
                logger.error(f"Unknown error: {e}")

        logger.success("Initialization complete.")

    @logger.catch(exclude=(HTTPException,))
    async def get_wallet_balance(self) -> WalletBalance:
        logger.trace("get_wallet_balance() ")

        req = ln.ListfundsRequest()
        res = await self._cln_stub.ListFunds(req)
        onchain_confirmed = onchain_unconfirmed = onchain_total = 0

        for o in res.outputs:
            sat = o.amount_msat.msat / 1000
            if o.status == 0:  # unconfirmed
                onchain_unconfirmed += sat
            elif o.status == 1 and not o.reserved:  # confirmed
                onchain_confirmed += sat
            # 2 is spent => ignore
            # 3 is immature => not sure what to do with this

        onchain_total = onchain_confirmed + onchain_unconfirmed

        chan_local = chan_remote = chan_pending_local = chan_pending_remote = 0
        for c in res.channels:
            our_msat = c.our_amount_msat.msat
            their_msat = c.amount_msat.msat - our_msat

            if c.state == 2:  # ChanneldNormal
                chan_local += our_msat
                chan_remote += their_msat
            else:
                # treat everything else as pending for now
                chan_pending_local += our_msat
                chan_pending_remote += their_msat

        return WalletBalance(
            onchain_confirmed_balance=onchain_confirmed,
            onchain_total_balance=onchain_total,
            onchain_unconfirmed_balance=onchain_unconfirmed,
            channel_local_balance=chan_local,
            channel_remote_balance=chan_remote,
            # TODO: find out how to get these values with CLN
            channel_unsettled_local_balance=0,
            channel_unsettled_remote_balance=0,
            channel_pending_open_local_balance=chan_pending_local,
            channel_pending_open_remote_balance=chan_pending_remote,
        )

    @logger.catch(exclude=(HTTPException,))
    async def _get_block_time(self, block_height: int) -> tuple:
        logger.trace(f"_get_block_time(block_height={block_height}) ")

        if block_height is None or block_height < 0:
            raise ValueError("block_height cannot be None or negative")

        if block_height in self._block_cache:
            return self._block_cache[block_height]

        res = await bitcoin_rpc_async("getblockstats", params=[block_height])
        hash = res["result"]["blockhash"]
        block = await bitcoin_rpc_async("getblock", params=[hash])
        self._block_cache[block_height] = (
            block["result"]["time"],
            block["result"]["mediantime"],
        )
        return self._block_cache[block_height]

    @logger.catch(exclude=(HTTPException,))
    async def list_all_tx(
        self, successful_only: bool, index_offset: int, max_tx: int, reversed: bool
    ) -> List[GenericTx]:
        logger.trace(
            (
                f"list_all_tx(successful_only={successful_only}, "
                f"index_offset={index_offset}, max_tx={max_tx}, reversed={reversed})"
            )
        )

        list_invoice_req = ln.ListinvoicesRequest()
        list_payments_req = ln.ListpaysRequest()

        try:
            res = await asyncio.gather(
                *[
                    self._cln_stub.ListInvoices(list_invoice_req),
                    self.list_on_chain_tx(),
                    self._cln_stub.ListPays(list_payments_req),
                    self.get_ln_info(),
                ]
            )
            tx = []
            for invoice in res[0].invoices:
                i = GenericTx.from_cln_grpc_invoice(invoice)
                if successful_only and i.status == TxStatus.SUCCEEDED:
                    tx.append(i)
                    continue
                tx.append(i)

            for transaction in res[1]:
                t = GenericTx.from_onchain_tx(transaction, res[3].block_height)
                if successful_only and t.status == TxStatus.SUCCEEDED:
                    tx.append(t)
                    continue

                tx.append(t)

            for pay in res[2].pays:
                decoded_bolt11: PaymentRequest = None

                if pay.bolt11 is not None and len(pay.bolt11) > 0:
                    if pay.bolt11 in self._memo_cache:
                        decoded_bolt11 = self._memo_cache[pay.bolt11]
                    else:
                        decoded_bolt11 = await self.decode_pay_request(pay.bolt11)
                        self._memo_cache[pay.bolt11] = decoded_bolt11

                p = GenericTx.from_cln_grpc_payment(
                    pay, decoded_bolt11.description, decoded_bolt11.num_msat
                )

                if successful_only and p.status == TxStatus.SUCCEEDED:
                    tx.append(p)
                    continue

                tx.append(p)

            def sortKey(e: GenericTx):
                return e.time_stamp

            tx.sort(key=sortKey)

            if reversed:
                tx.reverse()

            tx_length = len(tx)
            for invoice in range(tx_length):
                tx[invoice].index = invoice

            if max_tx == 0:
                max_tx = tx_length

            return tx[index_offset : index_offset + max_tx]
        except grpc.aio._call.AioRpcError as error:
            generic_grpc_error_handler(error)

    @logger.catch(exclude=(HTTPException,))
    async def list_invoices(
        self,
        pending_only: bool,
        index_offset: int,
        num_max_invoices: int,
        reversed: bool,
    ) -> List[Invoice]:
        logger.trace("list_invoices() ")

        try:
            req = ln.ListinvoicesRequest()
            res = await self._cln_stub.ListInvoices(req)

            tx = []
            for i in res.invoices:
                if pending_only:
                    if i.status == 0:
                        tx.append(Invoice.from_cln_grpc(i))
                else:
                    tx.append(Invoice.from_cln_grpc(i))

            if reversed:
                tx.reverse()

            if num_max_invoices == 0 or num_max_invoices is None:
                return tx

            return tx[index_offset : index_offset + num_max_invoices]

        except grpc.aio._call.AioRpcError as error:
            generic_grpc_error_handler(error)

    @logger.catch(exclude=(HTTPException,))
    async def list_on_chain_tx(self) -> List[OnChainTransaction]:
        logger.trace("list_on_chain_tx() ")
        info = await self.get_ln_info()  # for current block height
        res = await _make_local_call("bkpr-listincome")

        if not res:
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="Unknown CLN error while listing account income events",
            )

        if len(res) == 0:
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=(
                    "No response from CLN while trying to list account income events"
                ),
            )

        decoded = res[0].decode()
        js = json.loads(decoded)

        txs = {}
        num_events = len(js["income_events"])
        for i in range(0, num_events):
            e = js["income_events"][i]
            if e["account"] != "wallet":
                continue

            if e["tag"] == "deposit" or e["tag"] == "withdrawal":
                tx = OnChainTransaction.from_cln_bkpr(e)
                txs[tx.tx_hash] = tx
            elif e["tag"] == "onchain_fee":
                if e["txid"] in txs:
                    txs[e["txid"]].total_fees = parse_cln_msat(e["debit_msat"]) / 1000

        # TODO: Improve this once CLN reports the block height in bkpr-listincome
        # see https://github.com/ElementsProject/lightning/issues/5694

        # now get the block height for each tx ...
        res = await _make_local_call("bkpr-listaccountevents")
        if not res:
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="Unknown CLN error while listing account events",
            )

        if len(res) == 0:
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="No response from CLN while trying to list account events",
            )

        decoded = res[0].decode()
        js = json.loads(decoded)
        num_events = len(js["events"])
        for i in range(0, num_events):
            e = js["events"][i]
            if e["account"] != "wallet" or e["type"] != "chain":
                continue

            txid = ""
            if e["tag"] == "deposit":
                txid = e["outpoint"].split(":")[0]
            elif e["tag"] == "withdrawal":
                txid = e["txid"]

            if len(txid) == 0:
                continue

            if txid in txs:
                txs[txid].block_height = e["blockheight"]
                txs[txid].num_confirmations = info.block_height - txs[txid].block_height

        return [txs[k] for k in txs.keys()]

    @logger.catch(exclude=(HTTPException,))
    async def list_payments(
        self,
        include_incomplete: bool,
        index_offset: int,
        max_payments: int,
        reversed: bool,
    ):
        logger.trace(
            (
                f"list_payments(include_incomplete={include_incomplete}, "
                f"index_offset{index_offset}, max_payments={max_payments}, "
                f"reversed={reversed})"
            )
        )
        try:
            req = ln.ListpaysRequest()
            res = await self._cln_stub.ListPays(req)

            pays = []
            for p in res.pays:
                if p.status == 2:
                    # always include completed payments
                    pays.append(Payment.from_cln_grpc(p))
                    continue

                if include_incomplete:
                    pays.append(Payment.from_cln_grpc(p))

            if reversed:
                pays.reverse()

            if max_payments == 0 or max_payments is None:
                return pays

            return pays[index_offset : index_offset + max_payments]
        except grpc.aio._call.AioRpcError as error:
            generic_grpc_error_handler(error)

    @logger.catch(exclude=(HTTPException,))
    async def add_invoice(
        self,
        value_msat: int,
        memo: str = "",
        expiry: int = 3600,
        is_keysend: bool = False,
    ) -> Invoice:
        logger.trace(
            (
                f"add_invoice(value_msat={value_msat}, memo={memo}, "
                f"expiry={expiry}, is_keysend={is_keysend})"
            )
        )

        if value_msat < 0:
            raise ValueError("value_msat cannot be negative")

        msat = None
        if value_msat == 0:
            msat = lnp.AmountOrAny(any=True)
        elif value_msat > 0:
            msat = lnp.AmountOrAny(amount=lnp.Amount(msat=value_msat))

        id = next_push_id()
        req = ln.InvoiceRequest(
            amount_msat=msat,
            description=memo,
            label=id,
            expiry=expiry,
        )

        try:
            res = await self._cln_stub.Invoice(req)
            return Invoice(
                payment_request=res.bolt11,
                memo=memo,
                value_msat=value_msat,
                expiry_date=res.expires_at,
                add_index=id,
                state=InvoiceState.OPEN,
            )
        except grpc.aio._call.AioRpcError as error:
            details = error.details()
            logger.debug(details)

            try:
                self._handle_base_cln_error(error)
            except HTTPException:
                raise

            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Unknown CLN error while adding invoice: {details}",
            )

    @logger.catch(exclude=(HTTPException,))
    async def decode_pay_request(self, pay_req: str) -> PaymentRequest:
        logger.trace(f"decode_pay_request(pay_req={pay_req})")

        res = await _make_local_call("decodepay", f"bolt11={pay_req}")

        if not res:
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="Unknown CLN error decoding pay request",
            )

        if len(res) == 0:
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="No response from CLN decoding pay request",
            )

        decoded = res[0].decode()

        raise_for_pay_req_decode_error(decoded)

        return PaymentRequest.from_cln_json(json.loads(decoded))

    @logger.catch(exclude=(HTTPException,))
    async def get_fee_revenue(self) -> FeeRevenue:
        logger.trace("get_fee_revenue()")
        try:
            # status 1 == "settled"
            req = ln.ListforwardsRequest(status=1)
            res = await self._cln_stub.ListForwards(req)
            day, week, month, year, total = cln_classify_fee_revenue(res.forwards)

            return FeeRevenue(day=day, week=week, month=month, year=year, total=total)
        except grpc.aio._call.AioRpcError as error:
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details()
            )

    @logger.catch(exclude=(HTTPException,))
    async def new_address(self, input: NewAddressInput) -> str:
        logger.trace(f"new_address(input={input})")

        try:
            req = ln.NewaddrRequest()
            res = await self._cln_stub.NewAddr(req)

            return res.bech32
        except grpc.aio._call.AioRpcError as error:
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(error)
            )

            return res.bech32
        except grpc.aio._call.AioRpcError as error:
            generic_grpc_error_handler(error)

    @logger.catch(exclude=(HTTPException,))
    async def send_coins(self, input: SendCoinsInput) -> SendCoinsResponse:
        logger.trace(f"send_coins(input={input})")

        fee_rate: lnp.Feerate = None
        if input.sat_per_vbyte is not None and input.sat_per_vbyte > 0:
            fee_rate = lnp.Feerate(perkw=input.sat_per_vbyte)
        elif input.target_conf is not None and input.target_conf == 1:
            fee_rate = lnp.Feerate(urgent=True)
        elif input.target_conf is not None and input.target_conf >= 2:
            fee_rate = lnp.Feerate(normal=True)
        elif input.target_conf is not None and input.target_conf >= 10:
            fee_rate = lnp.Feerate(slow=True)

        try:
            funds = await self._cln_stub.ListFunds(ln.ListfundsRequest())
            if len(funds.outputs) == 0:
                raise HTTPException(
                    status.HTTP_412_PRECONDITION_FAILED,
                    detail=(
                        f"Could not afford {input.amount}sat. No UTXOs available at all"
                    ),
                )

            utxos = []
            max_amt = 0
            for o in funds.outputs:
                utxos.append(lnp.Outpoint(txid=o.txid, outnum=o.output))
                max_amt += o.amount_msat.msat / 1000

            if not input.send_all and max_amt <= input.amount:
                raise HTTPException(
                    status.HTTP_412_PRECONDITION_FAILED,
                    detail=(
                        f"Could not afford {input.amount}sat. "
                        "Not enough funds available"
                    ),
                )

            amt = lnp.AmountOrAll(amount=lnp.Amount(msat=input.amount * 1000))
            if input.send_all:
                amt = lnp.AmountOrAll(all=True)

            req = ln.WithdrawRequest(
                destination=input.address,
                satoshi=amt,
                minconf=input.min_confs,
                feerate=fee_rate,
                utxos=utxos,
            )
            response = await self._cln_stub.Withdraw(req)
            r = SendCoinsResponse.from_cln_grpc(response, input)
            await broadcast_msg(Event.LN_ONCHAIN_PAYMENT_STATUS, r.model_dump())
            return r
        except grpc.aio._call.AioRpcError as error:
            details = error.details()
            logger.debug(details)

            if details and details.find("Could not parse destination address") > -1:
                raise HTTPException(
                    status.HTTP_400_BAD_REQUEST,
                    detail=(
                        "Could not parse destination address, "
                        " destination should be a valid address."
                    ),
                )
            elif (
                details
                and details.find("UTXO") > -1
                and details.find("already reserved") > -1
            ):
                raise HTTPException(
                    status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail=(
                        "Server tried to use a reserved UTXO. "
                        "Please submit an issue to the BlitzAPI repository."
                    ),
                )
            elif details and details.find("insufficient funds available") > -1:
                raise HTTPException(status.HTTP_412_PRECONDITION_FAILED, detail=details)
            else:
                generic_grpc_error_handler(error)

    @logger.catch(exclude=(HTTPException,))
    async def send_payment(
        self,
        pay_req: str,
        timeout_seconds: int,
        fee_limit_msat: int,
        amount_msat: Optional[int] = None,
    ) -> Payment:
        logger.trace(
            (
                f"send_payment(pay_req={pay_req}, timeout_seconds={timeout_seconds}, "
                f"fee_limit_msat={fee_limit_msat}, amount_msat={amount_msat})"
            )
        )

        amt = lnp.Amount(msat=amount_msat) if amount_msat is not None else None
        fee_limit = lnp.Amount(msat=fee_limit_msat)
        req = ln.PayRequest(
            bolt11=pay_req,
            amount_msat=amt,
            maxfee=fee_limit,
            retry_for=timeout_seconds,
        )

        try:
            res = await self._cln_stub.Pay(req)
        except grpc.aio._call.AioRpcError as error:
            details = error.details()
            logger.debug(details)

            if "Ran out of routes to try after" in details:
                attempts = details.split("Ran out of routes to try after ")[1]
                attempts = attempts.split(" attempts")[0]
                raise HTTPException(
                    status.HTTP_422_UNPROCESSABLE_ENTITY,
                    detail=f"Ran out of routes to try after {attempts} attempts.",
                )

            if "Invalid bolt11: " in details:
                raise HTTPException(
                    status.HTTP_400_BAD_REQUEST,
                    detail="invalid bech32 string",
                )

            if "amount_msat parameter required" in details:
                raise HTTPException(
                    status.HTTP_400_BAD_REQUEST,
                    detail="amount must be specified when paying a zero amount invoice",
                )

            if "amount_msat parameter unnecessary" in details:
                raise HTTPException(
                    status.HTTP_400_BAD_REQUEST,
                    detail=(
                        "amount must not be specified when paying "
                        "a non-zero amount invoice"
                    ),
                )

            generic_grpc_error_handler(error)

        return Payment.from_cln_grpc(res)

    @logger.catch(exclude=(HTTPException,))
    async def get_ln_info(self) -> LnInfo:
        logger.trace("get_ln_info()")

        req = ln.GetinfoRequest()
        try:
            res = await self._cln_stub.Getinfo(req)
            return LnInfo.from_cln_grpc(self.get_implementation_name(), res)
        except grpc.aio._call.AioRpcError as error:
            details = error.details()
            logger.debug(details)

            try:
                self._handle_base_cln_error(error)
            except HTTPException:
                raise

            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Unknown CLN error while getting lightning info: {details}",
            )

    @logger.catch(exclude=(HTTPException,))
    async def unlock_wallet(self, password: str) -> bool:
        logger.trace("unlock_wallet(password=wedontlogpasswords)")

        # Core Lightning doesn't lock wallets,
        # so we don't need to do anything here
        return True

    @logger.catch(exclude=(HTTPException,))
    async def listen_invoices(self) -> AsyncGenerator[Invoice, None]:
        logger.trace("listen_invoices()")
        try:
            lastpay_index = 0
            invoices = await self.list_invoices(
                pending_only=False,
                index_offset=0,
                num_max_invoices=9999999999999,
                reversed=False,
            )

            for i in invoices:  # type Invoice
                if i.state == InvoiceState.SETTLED and i.settle_index > lastpay_index:
                    lastpay_index = i.settle_index

            while True:
                req = ln.WaitanyinvoiceRequest(lastpay_index=lastpay_index)
                i = await self._cln_stub.WaitAnyInvoice(req)
                i = Invoice.from_cln_grpc(i)
                lastpay_index = i.settle_index
                yield i
        except grpc.aio._call.AioRpcError as error:
            details = error.details()
            logger.debug(details)

            try:
                self._handle_base_cln_error(error)
            except HTTPException:
                raise

            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Unknown CLN error while listening for invoices: {details}",
            )

    @logger.catch(exclude=(HTTPException,))
    async def listen_forward_events(self) -> ForwardSuccessEvent:
        logger.trace("listen_forward_events()")

        # CLN has no subscription to forwarded events.
        # We must poll instead.

        interval = config("BAPI_GATHER_LN_INFO_INTERVAL", default=2, cast=float)

        # make sure we know how many forwards we have
        # we need to calculate the difference between each iteration
        # status=1 == "settled"
        req = ln.ListforwardsRequest(status=1)
        res = await self._cln_stub.ListForwards(req)
        num_fwd_last_poll = len(res.forwards)
        while True:
            res = await self._cln_stub.ListForwards(req)
            if len(res.forwards) > num_fwd_last_poll:
                fwds = res.forwards[num_fwd_last_poll:]
                for fwd in fwds:
                    yield ForwardSuccessEvent.from_cln_grpc(fwd)

                num_fwd_last_poll = len(res.forwards)
            await asyncio.sleep(interval - 0.1)

    @logger.catch(exclude=(HTTPException,))
    async def connect_peer(self, uri: str) -> bool:
        logger.trace(f"connect_peer(node_URI={uri})")

        try:
            req = ln.ConnectRequest(id=uri)
            await self._cln_stub.ConnectPeer(req)

            return True
        except grpc.aio._call.AioRpcError as error:
            details = error.details()
            logger.warning(details)

            if "All addresses failed" in details:
                m = details.split('message: "')[1]

                raise HTTPException(
                    status.HTTP_400_BAD_REQUEST,
                    detail=m,
                )

            if "no address known for peer" in details:
                raise HTTPException(
                    status.HTTP_400_BAD_REQUEST,
                    detail="Connection establishment: No address known for peer",
                )

            if "Connection timed out" in details:
                raise HTTPException(
                    status.HTTP_504_GATEWAY_TIMEOUT,
                    detail="Connection establishment: Connection timed out.",
                )

            if "Connection refused" in details:
                raise HTTPException(
                    status.HTTP_504_GATEWAY_TIMEOUT,
                    detail="Connection establishment: Connection refused.",
                )

            logger.exception(details)

            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details()
            )

    @logger.catch(exclude=(HTTPException, NodeNotFoundError))
    async def peer_resolve_alias(self, node_pub: bytes) -> str:
        logger.trace(f"peer_resolve_alias(node_pub={node_pub})")

        try:
            request = ln.ListnodesRequest(id=node_pub)
            response = await self._cln_stub.ListNodes(request)

            if len(response.nodes) == 0:
                raise NodeNotFoundError(node_pub.hex())

            return str(response.nodes[0].alias)

        except grpc.aio._call.AioRpcError as error:
            logger.error(error.details())

            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details()
            )

    @logger.catch(exclude=(HTTPException,))
    async def channel_open(
        self, local_funding_amount: int, node_URI: str, target_confs: int
    ) -> str:
        logger.trace(
            (
                f"channel_open(local_funding_amount={local_funding_amount}, "
                f"node_URI={node_URI}, target_confs={target_confs})"
            )
        )

        await self.connect_peer(node_URI)

        fee_rate: lnp.Feerate = None
        if target_confs == 1:
            fee_rate = lnp.Feerate(urgent=True)
        elif target_confs >= 2 and target_confs <= 9:
            fee_rate = lnp.Feerate(normal=True)
        elif target_confs >= 10:
            fee_rate = lnp.Feerate(slow=True)

        try:
            h = bytes.fromhex(node_URI.split("@")[0])
            req = ln.FundchannelRequest(
                id=h,
                amount=lnp.AmountOrAll(amount=lnp.Amount(msat=local_funding_amount)),
                feerate=fee_rate,
            )
        except TypeError as e:
            logger.error(f"channel_open() failed at ln.FundchannelRequest(): {e}")
            raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))

        try:
            res = await self._cln_stub.FundChannel(req)
            return res.txid.hex()

        except grpc.aio._call.AioRpcError as error:
            details = error.details()
            logger.debug(details)

            if "amount: should be a satoshi amount" in details:
                raise HTTPException(
                    status.HTTP_400_BAD_REQUEST,
                    detail="The amount is not a valid satoshi amount.",
                )

            if "Unknown peer" in details:
                raise HTTPException(
                    status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail=(
                        "We where able to connect to the peer but CLN "
                        "can't find it when opening a channel."
                    ),
                )

            if "Owning subdaemon openingd died" in details:
                # https://github.com/ElementsProject/lightning/issues/2798#issuecomment-511205719
                raise HTTPException(
                    status.HTTP_400_BAD_REQUEST,
                    detail=(
                        "Likely the peer didn't like our channel "
                        "opening proposal and disconnected from us. More info:"
                        "https://github.com/ElementsProject/lightning/issues/2798#issuecomment-511205719"
                    ),
                )

            if (
                "Number of pending channels exceed maximum" in details
                or "exceeds maximum chan size of 10 BTC" in details
                or "Could not afford all using all " in details
                or "BTC is below min chan size of" in details
            ):
                raise HTTPException(
                    status.HTTP_400_BAD_REQUEST, detail=_extract_message(details)
                )

            logger.warning(f"UNHANDLED ERROR: {details}")
            raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=details)

    @logger.catch(exclude=(HTTPException,))
    async def channel_list(self) -> List[Channel]:
        logger.trace("channel_list()")

        try:
            res = await self._cln_stub.ListFunds(ln.ListfundsRequest())
            peer_ids = [c.peer_id for c in res.channels]
            peer_res = await asyncio.gather(
                *[alias_or_empty(self.peer_resolve_alias, p) for p in peer_ids],
                return_exceptions=True,
            )

            channels = []
            for c, p in zip(res.channels, peer_res):
                channels.append(Channel.from_cln_grpc(c, p))

            return channels
        except grpc.aio._call.AioRpcError as error:
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details()
            )

    @logger.catch(exclude=(HTTPException,))
    async def channel_close(self, channel_id: int, force_close: bool) -> str:
        logger.trace(
            f"channel_close(channel_id={channel_id}, force_close={force_close})"
        )

        try:
            # on CLN we wait for 2 minutes to negotiate a channel close
            # if peer doesn't respond we force close
            wait_time_before_unilateral_close = 120 if force_close else 0
            req = ln.CloseRequest(
                id=channel_id,
                unilateraltimeout=wait_time_before_unilateral_close,
                feerange=[lnp.Feerate(slow=True), lnp.Feerate(urgent=True)],
            )
            res = await self._cln_stub.Close(req)

            # “mutual”, “unilateral”, “unopened”
            t = res.item_type
            if t == 0 or t == 1:  # mutual, unilateral
                return res.txid.hex()
            elif t == 2:  # unopened
                raise HTTPException(
                    status.HTTP_400_BAD_REQUEST, detail="Channel is not open yet."
                )

            raise HTTPException(
                status.HTTP_400_BAD_REQUEST,
                detail=f"CLN returned unknown close type: {t}",
            )
        except grpc.aio._call.AioRpcError as error:
            if "Channel is in state AWAITING_UNILATERAL" in error.details():
                raise HTTPException(
                    status.HTTP_400_BAD_REQUEST,
                    detail="Channel is awaiting an unilateral close.",
                )

            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error.details()
            )

    @logger.catch(exclude=(HTTPException,))
    def _handle_base_cln_error(self, error: grpc.aio._call.AioRpcError) -> None:
        # This method handles all errors common to all CLN calls
        details = error.details()

        if details and details.find("Received RST_STREAM with error code 8") > -1:
            logger.error(details)
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="CLN is responding with an error. Please check the logs.",
            )