blitz_api/app/bitcoind/service.py

import asyncio
import binascii

import zmq
import zmq.asyncio
from aiohttp import client_exceptions
from fastapi.exceptions import HTTPException
from loguru import logger
from starlette import status

from app.api.utils import Event, broadcast_msg
from app.bitcoind.models import (
    BlockchainInfo,
    BlockRpcFunc,
    BtcInfo,
    FeeEstimationMode,
    NetworkInfo,
    RawTransaction,
)
from app.bitcoind.utils import bitcoin_config, bitcoin_rpc_async

_initialized = False


@logger.catch(exclude=(HTTPException,))
async def initialize_bitcoin_repo() -> bool:
    global _initialized
    if _initialized:
        return True

    logger.info("Initializing bitcoin repository")
    # Wait until the bitcoin node is ready to accept RPC calls
    while not _initialized:
        try:
            await get_blockchain_info()
            _initialized = True
            logger.success("Bitcoin repository initialized")
            return True
        except client_exceptions.ClientConnectorError:
            logger.debug("Unable to connect to Bitcoin Core, waiting 5 seconds...")
            await asyncio.sleep(5)
        except HTTPException as e:
            if e.status_code == status.HTTP_425_TOO_EARLY:
                logger.info("Bitcoin Core initializing, waiting 10 seconds...")
                await asyncio.sleep(10)
                continue

            if e.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR:
                logger.error(e.detail)

            logger.debug(
                (
                    "Connected to Bitcoin Core but it seems to be "
                    f"initializing, waiting 2 seconds... \n{e.detail}"
                )
            )

            await asyncio.sleep(2)


@logger.catch(exclude=(HTTPException,))
async def get_blockchain_info() -> BlockchainInfo:
    result = await bitcoin_rpc_async("getblockchaininfo")

    if "error" in result and result["error"] is not None:
        raise HTTPException(result["status"], detail=result["error"])

    return BlockchainInfo.from_rpc(result["result"])


@logger.catch(exclude=(HTTPException,))
async def estimate_fee(
    target_conf: int = 6,
    mode: FeeEstimationMode = FeeEstimationMode.CONSERVATIVE,
) -> int:
    result = await bitcoin_rpc_async("estimatesmartfee", [target_conf, mode])

    if "error" in result and result["error"] is not None:
        raise HTTPException(result["status"], detail=result["error"])

    if "errors" in result["result"]:
        errors = "Bitcoin Core returned error(s):\n"
        for e in result["result"]["errors"]:
            errors += f"{e}\n"

        raise HTTPException(
            status.HTTP_500_INTERNAL_SERVER_ERROR, detail=errors[0 : len(errors) - 1]
        )

    # returned in BTC by Bitcoin Core => convert to msat
    rate_btc = result["result"]["feerate"]

    return rate_btc * 100000000


@logger.catch(exclude=(HTTPException,))
async def get_network_info() -> NetworkInfo:
    result = await bitcoin_rpc_async("getnetworkinfo")

    if "error" in result and result["error"] is not None:
        raise HTTPException(result["status"], detail=result["error"])

    return NetworkInfo.from_rpc(result["result"])


@logger.catch(exclude=(HTTPException,))
async def get_raw_transaction(txid: str) -> RawTransaction:
    result = await bitcoin_rpc_async("getrawtransaction", [txid, 1])

    if "error" not in result or result["error"] is None:
        return RawTransaction.from_rpc(result["result"])

    if "No such mempool or blockchain transaction." in result["error"]:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail=result["error"])

    if "-txindex option" in result["error"]:
        raise HTTPException(
            status.HTTP_400_BAD_REQUEST,
            detail="-txindex option for Bitcoin Core not enabled",
        )

    if "must be of length 64" in result["error"]:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail=result["error"])

    raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=result["error"])


@logger.catch(exclude=(HTTPException,))
async def get_btc_info() -> BtcInfo:
    binfo = await get_blockchain_info()
    ninfo = await get_network_info()

    if binfo is None or ninfo is None:
        logger.error(f"Blockchain Info or Network Info not available: {binfo} {ninfo}")
        raise HTTPException(
            status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=(
                "Unable to get blockchain or network data. "
                "See the log files for more information"
            ),
        )

    return BtcInfo.from_rpc(binfo, ninfo)


@logger.catch(exclude=(HTTPException,))
async def handle_block_sub_redis(verbosity: int = 1) -> str:
    ctx = zmq.asyncio.Context()
    zmq_socket = ctx.socket(zmq.SUB)
    zmq_socket.setsockopt(zmq.RCVHWM, 0)
    zmq_socket.setsockopt_string(zmq.SUBSCRIBE, bitcoin_config.zmq_block_rpc)
    zmq_socket.connect(bitcoin_config.zmq_url)

    while True:
        hash = ""
        _, body, _ = await zmq_socket.recv_multipart()
        if bitcoin_config.zmq_block_rpc == BlockRpcFunc.HASHBLOCK:
            hash = binascii.hexlify(body).decode("utf-8")
        elif bitcoin_config.zmq_block_rpc == BlockRpcFunc.RAWBLOCK:
            r1 = await bitcoin_rpc_async("getbestblockhash", [])
            if "result" not in r1 or r1.get("error") is not None:
                logger.error(
                    f"getbestblockhash failed, skipping block: {r1.get('error')}"
                )
                continue
            hash = r1["result"]
        else:
            raise NotImplementedError(
                f"ZMQ block function {bitcoin_config.zmq_block_rpc} not supported"
            )

        r = await bitcoin_rpc_async("getblock", [hash, verbosity])
        # strict JSON-RPC 2.0: an errored reply has no "result" key
        if "result" not in r or r.get("error") is not None:
            logger.error(f"getblock failed, skipping block: {r.get('error')}")
            continue

        await broadcast_msg(Event.BTC_NEW_BLOC, r["result"])


@logger.catch(exclude=(HTTPException,))
async def register_bitcoin_zmq_sub():
    asyncio.create_task(handle_block_sub_redis())


@logger.catch(exclude=(HTTPException,))
async def _handle_gather_bitcoin_status():
    last_info = {}
    while True:
        try:
            info = await get_btc_info()
            if info is None:
                continue

            info.verification_progress = round(info.verification_progress, 2)
        except HTTPException as e:
            logger.error(e.detail)
            await asyncio.sleep(2)
            continue

        if last_info != info:
            # only send data if anything has changed
            await broadcast_msg(Event.BTC_INFO, info.model_dump())
            last_info = info

        await asyncio.sleep(2)


@logger.catch(exclude=(HTTPException,))
async def register_bitcoin_status_gatherer():
    asyncio.create_task(_handle_gather_bitcoin_status())