jmwallet.backends.base

jmwallet.backends.base

Base blockchain backend interface.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

BlockchainBackend

Bases: ABC

Abstract blockchain backend interface. Implementations provide access to blockchain data without requiring Bitcoin Core wallet functionality (avoiding BerkeleyDB issues).

Source code in jmwallet/src/jmwallet/backends/base.py

class BlockchainBackend(ABC):
    """
    Abstract blockchain backend interface.
    Implementations provide access to blockchain data without requiring
    Bitcoin Core wallet functionality (avoiding BerkeleyDB issues).
    """

    supports_descriptor_scan: bool = False
    """Whether this backend supports efficient descriptor-based UTXO scanning.

    Backends that override ``scan_descriptors()`` with a real implementation
    (e.g. Bitcoin Core's descriptor wallet) should set this to ``True``.
    Light-client backends (Neutrino) leave it at the default ``False`` so that
    ``sync_all()`` does not attempt descriptor scanning and fall back with a
    confusing warning.
    """

    supports_watch_address: bool = False
    """Whether this backend requires addresses to be pre-registered via ``add_watch_address()``.

    Light-client backends (Neutrino) must be told which addresses to watch before
    a rescan.  Full-node backends (Bitcoin Core, descriptor wallet) can query any
    address on demand and do not need pre-registration.
    Set to ``True`` only in backends that implement ``add_watch_address()``.
    """

    supports_tx_enumeration: bool = False
    """Whether this backend can incrementally enumerate wallet transactions.

    Backends that override ``list_wallet_transactions_since()`` with a real
    implementation (e.g. Bitcoin Core's ``listsinceblock``) set this to
    ``True``. The jmwalletd transaction monitor uses it to push WebSocket
    notifications for every wallet transaction (deposits, coinjoins, sends).
    Light clients without cheap tx enumeration leave it ``False`` and the
    monitor stays idle for that backend.
    """

    async def add_watch_address(self, address: str) -> None:
        """Register an address for watching.

        Only meaningful for backends with ``supports_watch_address = True``
        (i.e. light-client backends that need explicit address registration
        before a rescan).  Full-node backends can ignore this.
        """

    async def ensure_addresses_scanned(self, addresses: list[str]) -> None:
        """Ensure *addresses* have been scanned over the wallet's full history.

        Light-client backends (Neutrino) only rescan blocks that arrived since
        their last rescan, and only for addresses that were already watched at
        that time. An address added *after* the initial sync therefore needs an
        explicit historical rescan, otherwise outputs paid to it before it was
        watched (e.g. a fidelity bond funded earlier) are never found.

        The default implementation is a no-op: full-node and descriptor-wallet
        backends can query any address on demand (and handle fidelity bonds via
        descriptor import), so they do not need this. Light-client backends
        override it to trigger a rescan from the wallet's scan start height.
        """

    def set_wallet_creation_height(self, height: int | None) -> None:
        """Provide the block height at which the wallet was created.

        Backends can use this as a hint to skip scanning blocks before
        the wallet existed, avoiding unnecessary work during initial sync.
        Only used when no explicit ``scan_start_height`` is configured.

        Passing ``None`` clears any previously set hint.

        The default implementation is a no-op; backends that support
        scan optimisation override this method.
        """

    @abstractmethod
    async def get_utxos(self, addresses: list[str]) -> list[UTXO]:
        """Get UTXOs for given addresses"""

    @abstractmethod
    async def get_address_balance(self, address: str) -> int:
        """Get balance for an address in satoshis"""

    @abstractmethod
    async def broadcast_transaction(self, tx_hex: str) -> str:
        """Broadcast transaction, returns txid"""

    @abstractmethod
    async def get_transaction(self, txid: str) -> Transaction | None:
        """Get transaction by txid"""

    @abstractmethod
    async def estimate_fee(self, target_blocks: int) -> float:
        """Estimate fee in sat/vbyte for target confirmation blocks.

        Returns:
            Fee rate in sat/vB. Can be fractional (e.g., 0.5 sat/vB).
        """

    async def get_mempool_min_fee(self) -> float | None:
        """Get the minimum fee rate (in sat/vB) for transaction to be accepted into mempool.

        This is used as a floor for fee estimation to ensure transactions are
        relayed and accepted into the mempool. Returns None if not supported
        or unavailable (e.g., light clients).

        Returns:
            Minimum fee rate in sat/vB, or None if unavailable.
        """
        return None

    def can_estimate_fee(self) -> bool:
        """Check if this backend can perform fee estimation.

        Full node backends (Bitcoin Core) can estimate fees.
        Light client backends (Neutrino) typically cannot.

        Returns:
            True if backend supports fee estimation, False otherwise.
        """
        return True

    def has_mempool_access(self) -> bool:
        """Check if this backend can access unconfirmed transactions in the mempool.

        Full node backends (Bitcoin Core) and mempool API backends have
        mempool access and can verify transactions immediately after broadcast.

        Light client backends (Neutrino using BIP157/158) cannot access the mempool
        and can only see transactions after they're confirmed in a block. This
        affects broadcast verification strategy - see BroadcastPolicy docs.

        Returns:
            True if backend can see unconfirmed transactions, False otherwise.
        """
        return True

    def can_get_confirmations_by_txid(self) -> bool:
        """Whether get_transaction() can report a confirmation count for an
        arbitrary txid.

        Full node and mempool-API backends resolve any txid and report how
        deeply it is confirmed, so pending-transaction monitors can rely on
        ``get_transaction()`` to detect confirmation. Light clients (Neutrino)
        are mempool-only here: ``get_transaction()`` only surfaces watched
        *unconfirmed* txs (``confirmations=0``) and returns ``None`` once a tx
        confirms, so confirmation must instead be established with
        :meth:`verify_tx_output` against the output's address.

        Returns:
            True if ``get_transaction()`` reports confirmation depth by txid.
        """
        return True

    async def list_wallet_transactions_since(
        self, cursor: str | None
    ) -> tuple[list[WalletTxEntry], str | None]:
        """Incrementally enumerate wallet transactions since ``cursor``.

        ``cursor`` is an opaque backend token (a block hash for Bitcoin Core's
        ``listsinceblock``); pass ``None`` to enumerate from the beginning.
        Returns the (deduplicated) transactions plus a new cursor to pass on
        the next call. Includes mempool (unconfirmed) transactions on every
        call until they confirm past the cursor.

        Default: unsupported (returns nothing and no cursor). Only backends
        with ``supports_tx_enumeration = True`` override this.
        """
        return [], cursor

    @abstractmethod
    async def get_block_height(self) -> int:
        """Get current blockchain height"""

    @abstractmethod
    async def get_block_time(self, block_height: int) -> int:
        """Get block time (unix timestamp) for given height"""

    async def get_median_time_past(self) -> int:
        """Return the median timestamp of the last 11 blocks.

        Bitcoin evaluates time-based transaction locktimes against the median
        time past of the current chain tip, not the local host clock (BIP 113).
        Backends may override this with a native endpoint when available.
        """
        height = await self.get_block_height()
        first_height = max(0, height - 10)
        block_times = await asyncio.gather(
            *(self.get_block_time(block_height) for block_height in range(first_height, height + 1))
        )
        if not block_times:
            msg = "Cannot determine median time past without block timestamps"
            raise RuntimeError(msg)
        return sorted(block_times)[len(block_times) // 2]

    @abstractmethod
    async def get_block_hash(self, block_height: int) -> str:
        """Get block hash for given height"""

    @abstractmethod
    async def get_utxo(self, txid: str, vout: int) -> UTXO | None:
        """Get a specific UTXO from the blockchain UTXO set (gettxout).
        Returns None if the UTXO does not exist or has been spent."""

    async def scan_descriptors(
        self, descriptors: Sequence[str | dict[str, Any]]
    ) -> dict[str, Any] | None:
        """
        Scan the UTXO set using output descriptors.

        This is an efficient alternative to scanning individual addresses,
        especially useful for HD wallets where xpub descriptors with ranges
        can scan thousands of addresses in a single UTXO set pass.

        Example descriptors:
            - "addr(bc1q...)" - single address
            - "wpkh(xpub.../0/*)" - HD wallet addresses (default range 0-1000)
            - {"desc": "wpkh(xpub.../0/*)", "range": [0, 999]} - explicit range

        Args:
            descriptors: List of output descriptors (strings or dicts with range)

        Returns:
            Scan result dict with:
                - success: bool
                - unspents: list of found UTXOs
                - total_amount: sum of all found UTXOs
            Returns None if not supported or on failure.

        Note:
            Not all backends support descriptor scanning. The default implementation
            returns None. Override in backends that support it (e.g., Bitcoin Core).
        """
        # Default: not supported
        return None

    async def verify_utxo_with_metadata(
        self,
        txid: str,
        vout: int,
        scriptpubkey: str,
        blockheight: int,
    ) -> UTXOVerificationResult:
        """
        Verify a UTXO using provided metadata (neutrino_compat feature).

        This method allows light clients to verify UTXOs without needing
        arbitrary blockchain queries by using metadata provided by the peer.

        The implementation should:
        1. Use scriptpubkey to add the UTXO to watch list (for Neutrino)
        2. Use blockheight as a hint for efficient rescan
        3. Verify the UTXO exists with matching scriptpubkey
        4. Return the UTXO value and confirmations

        Default implementation falls back to get_utxo() for full node backends.

        Args:
            txid: Transaction ID
            vout: Output index
            scriptpubkey: Expected scriptPubKey (hex)
            blockheight: Block height where UTXO was confirmed

        Returns:
            UTXOVerificationResult with verification status and UTXO data
        """
        # Default implementation for full node backends
        # Just uses get_utxo() directly since we can query any UTXO
        utxo = await self.get_utxo(txid, vout)

        if utxo is None:
            return UTXOVerificationResult(
                valid=False,
                error="UTXO not found or spent",
            )

        # Verify scriptpubkey matches
        scriptpubkey_matches = utxo.scriptpubkey.lower() == scriptpubkey.lower()

        if not scriptpubkey_matches:
            return UTXOVerificationResult(
                valid=False,
                value=utxo.value,
                confirmations=utxo.confirmations,
                error="ScriptPubKey mismatch",
                scriptpubkey_matches=False,
            )

        return UTXOVerificationResult(
            valid=True,
            value=utxo.value,
            confirmations=utxo.confirmations,
            scriptpubkey_matches=True,
        )

    def requires_neutrino_metadata(self) -> bool:
        """
        Check if this backend requires Neutrino-compatible metadata for UTXO verification.

        Full node backends can verify any UTXO directly.
        Light client backends need scriptpubkey and blockheight hints.

        Returns:
            True if backend requires metadata for verification
        """
        return False

    def can_provide_neutrino_metadata(self) -> bool:
        """
        Check if this backend can provide Neutrino-compatible metadata for its own UTXOs.

        This determines whether to advertise neutrino_compat feature to the network.
        Backends should return True if they can provide extended UTXO format with
        scriptpubkey and blockheight fields for their own wallet UTXOs in !ioauth.

        This is distinct from requires_neutrino_metadata(): a backend may both
        require metadata from counterparties (to verify their UTXOs) AND provide
        metadata about its own UTXOs. Both full node and neutrino backends can
        provide their own UTXO metadata since they know their wallet's scriptpubkeys
        and block heights.

        Returns:
            True if backend can provide scriptpubkey and blockheight for its UTXOs
        """
        # Default: all backends can provide metadata for their own UTXOs
        return True

    async def verify_tx_output(
        self,
        txid: str,
        vout: int,
        address: str,
        start_height: int | None = None,
    ) -> bool:
        """
        Verify that a specific transaction output exists (was broadcast and confirmed).

        This is useful for verifying a transaction was successfully broadcast when
        we know at least one of its output addresses (e.g., our coinjoin destination).

        For full node backends, this uses get_transaction().
        For light clients (neutrino), this uses UTXO lookup with the address hint.

        Args:
            txid: Transaction ID to verify
            vout: Output index to check
            address: The address that should own this output
            start_height: Optional block height hint for light clients (improves performance)

        Returns:
            True if the output exists (transaction was broadcast), False otherwise
        """
        # Default implementation for full node backends
        tx = await self.get_transaction(txid)
        return tx is not None

    async def verify_bonds(
        self,
        bonds: list[BondVerificationRequest],
    ) -> list[BondVerificationResult]:
        """Verify multiple fidelity bond UTXOs in bulk.

        This is the primary method for verifying fidelity bonds. Each backend can
        override this for optimal performance:
        - Bitcoin Core: JSON-RPC batch of gettxout calls (2 HTTP requests total)
        - Neutrino: batch address rescan + individual UTXO lookups
        - Mempool: parallel HTTP requests

        The default implementation calls get_utxo() sequentially with a semaphore.

        Args:
            bonds: List of bond verification requests with pre-computed addresses

        Returns:
            List of verification results, one per input bond (same order)
        """
        if not bonds:
            return []

        current_height = await self.get_block_height()
        semaphore = asyncio.Semaphore(10)

        async def _verify_one(bond: BondVerificationRequest) -> BondVerificationResult:
            async with semaphore:
                try:
                    utxo = await self.get_utxo(bond.txid, bond.vout)
                    if utxo is None:
                        return BondVerificationResult(
                            txid=bond.txid,
                            vout=bond.vout,
                            value=0,
                            confirmations=0,
                            block_time=0,
                            valid=False,
                            error="UTXO not found or spent",
                        )
                    if utxo.confirmations <= 0:
                        return BondVerificationResult(
                            txid=bond.txid,
                            vout=bond.vout,
                            value=utxo.value,
                            confirmations=0,
                            block_time=0,
                            valid=False,
                            error="UTXO unconfirmed",
                        )
                    if (
                        not utxo.scriptpubkey
                        or utxo.scriptpubkey.lower() != bond.scriptpubkey.lower()
                    ):
                        return BondVerificationResult(
                            txid=bond.txid,
                            vout=bond.vout,
                            value=utxo.value,
                            confirmations=utxo.confirmations,
                            block_time=0,
                            valid=False,
                            error="ScriptPubKey mismatch",
                        )
                    # Get the block time for the confirmation block
                    conf_height = current_height - utxo.confirmations + 1
                    block_time = await self.get_block_time(conf_height)
                    return BondVerificationResult(
                        txid=bond.txid,
                        vout=bond.vout,
                        value=utxo.value,
                        confirmations=utxo.confirmations,
                        block_time=block_time,
                        valid=True,
                    )
                except Exception as e:
                    logger.warning(
                        "Bond verification failed for %s:%d: %s",
                        bond.txid,
                        bond.vout,
                        e,
                    )
                    return BondVerificationResult(
                        txid=bond.txid,
                        vout=bond.vout,
                        value=0,
                        confirmations=0,
                        block_time=0,
                        valid=False,
                        error=str(e),
                    )

        results = await asyncio.gather(*[_verify_one(b) for b in bonds])
        return list(results)

    async def close(self) -> None:
        """Close backend connection"""
        pass

Attributes

supports_descriptor_scan: bool = False class-attribute instance-attribute

Whether this backend supports efficient descriptor-based UTXO scanning.

Backends that override scan_descriptors() with a real implementation (e.g. Bitcoin Core's descriptor wallet) should set this to True. Light-client backends (Neutrino) leave it at the default False so that sync_all() does not attempt descriptor scanning and fall back with a confusing warning.

supports_tx_enumeration: bool = False class-attribute instance-attribute

Whether this backend can incrementally enumerate wallet transactions.

Backends that override list_wallet_transactions_since() with a real implementation (e.g. Bitcoin Core's listsinceblock) set this to True. The jmwalletd transaction monitor uses it to push WebSocket notifications for every wallet transaction (deposits, coinjoins, sends). Light clients without cheap tx enumeration leave it False and the monitor stays idle for that backend.

supports_watch_address: bool = False class-attribute instance-attribute

Whether this backend requires addresses to be pre-registered via add_watch_address().

Light-client backends (Neutrino) must be told which addresses to watch before a rescan. Full-node backends (Bitcoin Core, descriptor wallet) can query any address on demand and do not need pre-registration. Set to True only in backends that implement add_watch_address().

Methods:

add_watch_address(address: str) -> None async

Register an address for watching.

Only meaningful for backends with supports_watch_address = True (i.e. light-client backends that need explicit address registration before a rescan). Full-node backends can ignore this.

Source code in jmwallet/src/jmwallet/backends/base.py

async def add_watch_address(self, address: str) -> None:
    """Register an address for watching.

    Only meaningful for backends with ``supports_watch_address = True``
    (i.e. light-client backends that need explicit address registration
    before a rescan).  Full-node backends can ignore this.
    """

broadcast_transaction(tx_hex: str) -> str abstractmethod async

Broadcast transaction, returns txid

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def broadcast_transaction(self, tx_hex: str) -> str:
    """Broadcast transaction, returns txid"""

can_estimate_fee() -> bool

Check if this backend can perform fee estimation.

Full node backends (Bitcoin Core) can estimate fees. Light client backends (Neutrino) typically cannot.

Returns: True if backend supports fee estimation, False otherwise.

Source code in jmwallet/src/jmwallet/backends/base.py

def can_estimate_fee(self) -> bool:
    """Check if this backend can perform fee estimation.

    Full node backends (Bitcoin Core) can estimate fees.
    Light client backends (Neutrino) typically cannot.

    Returns:
        True if backend supports fee estimation, False otherwise.
    """
    return True

can_get_confirmations_by_txid() -> bool

Whether get_transaction() can report a confirmation count for an arbitrary txid.

Full node and mempool-API backends resolve any txid and report how deeply it is confirmed, so pending-transaction monitors can rely on get_transaction() to detect confirmation. Light clients (Neutrino) are mempool-only here: get_transaction() only surfaces watched unconfirmed txs (confirmations=0) and returns None once a tx confirms, so confirmation must instead be established with :meth:verify_tx_output against the output's address.

Returns: True if get_transaction() reports confirmation depth by txid.

Source code in jmwallet/src/jmwallet/backends/base.py

def can_get_confirmations_by_txid(self) -> bool:
    """Whether get_transaction() can report a confirmation count for an
    arbitrary txid.

    Full node and mempool-API backends resolve any txid and report how
    deeply it is confirmed, so pending-transaction monitors can rely on
    ``get_transaction()`` to detect confirmation. Light clients (Neutrino)
    are mempool-only here: ``get_transaction()`` only surfaces watched
    *unconfirmed* txs (``confirmations=0``) and returns ``None`` once a tx
    confirms, so confirmation must instead be established with
    :meth:`verify_tx_output` against the output's address.

    Returns:
        True if ``get_transaction()`` reports confirmation depth by txid.
    """
    return True

can_provide_neutrino_metadata() -> bool

Check if this backend can provide Neutrino-compatible metadata for its own UTXOs.

This determines whether to advertise neutrino_compat feature to the network. Backends should return True if they can provide extended UTXO format with scriptpubkey and blockheight fields for their own wallet UTXOs in !ioauth.

This is distinct from requires_neutrino_metadata(): a backend may both require metadata from counterparties (to verify their UTXOs) AND provide metadata about its own UTXOs. Both full node and neutrino backends can provide their own UTXO metadata since they know their wallet's scriptpubkeys and block heights.

Returns: True if backend can provide scriptpubkey and blockheight for its UTXOs

Source code in jmwallet/src/jmwallet/backends/base.py

def can_provide_neutrino_metadata(self) -> bool:
    """
    Check if this backend can provide Neutrino-compatible metadata for its own UTXOs.

    This determines whether to advertise neutrino_compat feature to the network.
    Backends should return True if they can provide extended UTXO format with
    scriptpubkey and blockheight fields for their own wallet UTXOs in !ioauth.

    This is distinct from requires_neutrino_metadata(): a backend may both
    require metadata from counterparties (to verify their UTXOs) AND provide
    metadata about its own UTXOs. Both full node and neutrino backends can
    provide their own UTXO metadata since they know their wallet's scriptpubkeys
    and block heights.

    Returns:
        True if backend can provide scriptpubkey and blockheight for its UTXOs
    """
    # Default: all backends can provide metadata for their own UTXOs
    return True

close() -> None async

Close backend connection

Source code in jmwallet/src/jmwallet/backends/base.py

async def close(self) -> None:
    """Close backend connection"""
    pass

ensure_addresses_scanned(addresses: list[str]) -> None async

Ensure addresses have been scanned over the wallet's full history.

Light-client backends (Neutrino) only rescan blocks that arrived since their last rescan, and only for addresses that were already watched at that time. An address added after the initial sync therefore needs an explicit historical rescan, otherwise outputs paid to it before it was watched (e.g. a fidelity bond funded earlier) are never found.

The default implementation is a no-op: full-node and descriptor-wallet backends can query any address on demand (and handle fidelity bonds via descriptor import), so they do not need this. Light-client backends override it to trigger a rescan from the wallet's scan start height.

Source code in jmwallet/src/jmwallet/backends/base.py

async def ensure_addresses_scanned(self, addresses: list[str]) -> None:
    """Ensure *addresses* have been scanned over the wallet's full history.

    Light-client backends (Neutrino) only rescan blocks that arrived since
    their last rescan, and only for addresses that were already watched at
    that time. An address added *after* the initial sync therefore needs an
    explicit historical rescan, otherwise outputs paid to it before it was
    watched (e.g. a fidelity bond funded earlier) are never found.

    The default implementation is a no-op: full-node and descriptor-wallet
    backends can query any address on demand (and handle fidelity bonds via
    descriptor import), so they do not need this. Light-client backends
    override it to trigger a rescan from the wallet's scan start height.
    """

estimate_fee(target_blocks: int) -> float abstractmethod async

Estimate fee in sat/vbyte for target confirmation blocks.

Returns: Fee rate in sat/vB. Can be fractional (e.g., 0.5 sat/vB).

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def estimate_fee(self, target_blocks: int) -> float:
    """Estimate fee in sat/vbyte for target confirmation blocks.

    Returns:
        Fee rate in sat/vB. Can be fractional (e.g., 0.5 sat/vB).
    """

get_address_balance(address: str) -> int abstractmethod async

Get balance for an address in satoshis

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def get_address_balance(self, address: str) -> int:
    """Get balance for an address in satoshis"""

get_block_hash(block_height: int) -> str abstractmethod async

Get block hash for given height

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def get_block_hash(self, block_height: int) -> str:
    """Get block hash for given height"""

get_block_height() -> int abstractmethod async

Get current blockchain height

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def get_block_height(self) -> int:
    """Get current blockchain height"""

get_block_time(block_height: int) -> int abstractmethod async

Get block time (unix timestamp) for given height

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def get_block_time(self, block_height: int) -> int:
    """Get block time (unix timestamp) for given height"""

get_median_time_past() -> int async

Return the median timestamp of the last 11 blocks.

Bitcoin evaluates time-based transaction locktimes against the median time past of the current chain tip, not the local host clock (BIP 113). Backends may override this with a native endpoint when available.

Source code in jmwallet/src/jmwallet/backends/base.py

async def get_median_time_past(self) -> int:
    """Return the median timestamp of the last 11 blocks.

    Bitcoin evaluates time-based transaction locktimes against the median
    time past of the current chain tip, not the local host clock (BIP 113).
    Backends may override this with a native endpoint when available.
    """
    height = await self.get_block_height()
    first_height = max(0, height - 10)
    block_times = await asyncio.gather(
        *(self.get_block_time(block_height) for block_height in range(first_height, height + 1))
    )
    if not block_times:
        msg = "Cannot determine median time past without block timestamps"
        raise RuntimeError(msg)
    return sorted(block_times)[len(block_times) // 2]

get_mempool_min_fee() -> float | None async

Get the minimum fee rate (in sat/vB) for transaction to be accepted into mempool.

This is used as a floor for fee estimation to ensure transactions are relayed and accepted into the mempool. Returns None if not supported or unavailable (e.g., light clients).

Returns: Minimum fee rate in sat/vB, or None if unavailable.

Source code in jmwallet/src/jmwallet/backends/base.py

async def get_mempool_min_fee(self) -> float | None:
    """Get the minimum fee rate (in sat/vB) for transaction to be accepted into mempool.

    This is used as a floor for fee estimation to ensure transactions are
    relayed and accepted into the mempool. Returns None if not supported
    or unavailable (e.g., light clients).

    Returns:
        Minimum fee rate in sat/vB, or None if unavailable.
    """
    return None

get_transaction(txid: str) -> Transaction | None abstractmethod async

Get transaction by txid

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def get_transaction(self, txid: str) -> Transaction | None:
    """Get transaction by txid"""

get_utxo(txid: str, vout: int) -> UTXO | None abstractmethod async

Get a specific UTXO from the blockchain UTXO set (gettxout). Returns None if the UTXO does not exist or has been spent.

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def get_utxo(self, txid: str, vout: int) -> UTXO | None:
    """Get a specific UTXO from the blockchain UTXO set (gettxout).
    Returns None if the UTXO does not exist or has been spent."""

get_utxos(addresses: list[str]) -> list[UTXO] abstractmethod async

Get UTXOs for given addresses

Source code in jmwallet/src/jmwallet/backends/base.py

@abstractmethod
async def get_utxos(self, addresses: list[str]) -> list[UTXO]:
    """Get UTXOs for given addresses"""

has_mempool_access() -> bool

Check if this backend can access unconfirmed transactions in the mempool.

Full node backends (Bitcoin Core) and mempool API backends have mempool access and can verify transactions immediately after broadcast.

Light client backends (Neutrino using BIP157/158) cannot access the mempool and can only see transactions after they're confirmed in a block. This affects broadcast verification strategy - see BroadcastPolicy docs.

Returns: True if backend can see unconfirmed transactions, False otherwise.

Source code in jmwallet/src/jmwallet/backends/base.py

def has_mempool_access(self) -> bool:
    """Check if this backend can access unconfirmed transactions in the mempool.

    Full node backends (Bitcoin Core) and mempool API backends have
    mempool access and can verify transactions immediately after broadcast.

    Light client backends (Neutrino using BIP157/158) cannot access the mempool
    and can only see transactions after they're confirmed in a block. This
    affects broadcast verification strategy - see BroadcastPolicy docs.

    Returns:
        True if backend can see unconfirmed transactions, False otherwise.
    """
    return True

list_wallet_transactions_since(cursor: str | None) -> tuple[list[WalletTxEntry], str | None] async

Incrementally enumerate wallet transactions since cursor.

cursor is an opaque backend token (a block hash for Bitcoin Core's listsinceblock); pass None to enumerate from the beginning. Returns the (deduplicated) transactions plus a new cursor to pass on the next call. Includes mempool (unconfirmed) transactions on every call until they confirm past the cursor.

Default: unsupported (returns nothing and no cursor). Only backends with supports_tx_enumeration = True override this.

Source code in jmwallet/src/jmwallet/backends/base.py

async def list_wallet_transactions_since(
    self, cursor: str | None
) -> tuple[list[WalletTxEntry], str | None]:
    """Incrementally enumerate wallet transactions since ``cursor``.

    ``cursor`` is an opaque backend token (a block hash for Bitcoin Core's
    ``listsinceblock``); pass ``None`` to enumerate from the beginning.
    Returns the (deduplicated) transactions plus a new cursor to pass on
    the next call. Includes mempool (unconfirmed) transactions on every
    call until they confirm past the cursor.

    Default: unsupported (returns nothing and no cursor). Only backends
    with ``supports_tx_enumeration = True`` override this.
    """
    return [], cursor

requires_neutrino_metadata() -> bool

Check if this backend requires Neutrino-compatible metadata for UTXO verification.

Full node backends can verify any UTXO directly. Light client backends need scriptpubkey and blockheight hints.

Returns: True if backend requires metadata for verification

Source code in jmwallet/src/jmwallet/backends/base.py

def requires_neutrino_metadata(self) -> bool:
    """
    Check if this backend requires Neutrino-compatible metadata for UTXO verification.

    Full node backends can verify any UTXO directly.
    Light client backends need scriptpubkey and blockheight hints.

    Returns:
        True if backend requires metadata for verification
    """
    return False

scan_descriptors(descriptors: Sequence[str | dict[str, Any]]) -> dict[str, Any] | None async

Scan the UTXO set using output descriptors.

This is an efficient alternative to scanning individual addresses, especially useful for HD wallets where xpub descriptors with ranges can scan thousands of addresses in a single UTXO set pass.

Example descriptors: - "addr(bc1q...)" - single address - "wpkh(xpub.../0/)" - HD wallet addresses (default range 0-1000) - {"desc": "wpkh(xpub.../0/)", "range": [0, 999]} - explicit range

Args: descriptors: List of output descriptors (strings or dicts with range)

Returns: Scan result dict with: - success: bool - unspents: list of found UTXOs - total_amount: sum of all found UTXOs Returns None if not supported or on failure.

Note: Not all backends support descriptor scanning. The default implementation returns None. Override in backends that support it (e.g., Bitcoin Core).

Source code in jmwallet/src/jmwallet/backends/base.py

async def scan_descriptors(
    self, descriptors: Sequence[str | dict[str, Any]]
) -> dict[str, Any] | None:
    """
    Scan the UTXO set using output descriptors.

    This is an efficient alternative to scanning individual addresses,
    especially useful for HD wallets where xpub descriptors with ranges
    can scan thousands of addresses in a single UTXO set pass.

    Example descriptors:
        - "addr(bc1q...)" - single address
        - "wpkh(xpub.../0/*)" - HD wallet addresses (default range 0-1000)
        - {"desc": "wpkh(xpub.../0/*)", "range": [0, 999]} - explicit range

    Args:
        descriptors: List of output descriptors (strings or dicts with range)

    Returns:
        Scan result dict with:
            - success: bool
            - unspents: list of found UTXOs
            - total_amount: sum of all found UTXOs
        Returns None if not supported or on failure.

    Note:
        Not all backends support descriptor scanning. The default implementation
        returns None. Override in backends that support it (e.g., Bitcoin Core).
    """
    # Default: not supported
    return None

set_wallet_creation_height(height: int | None) -> None

Provide the block height at which the wallet was created.

Backends can use this as a hint to skip scanning blocks before the wallet existed, avoiding unnecessary work during initial sync. Only used when no explicit scan_start_height is configured.

Passing None clears any previously set hint.

The default implementation is a no-op; backends that support scan optimisation override this method.

Source code in jmwallet/src/jmwallet/backends/base.py

def set_wallet_creation_height(self, height: int | None) -> None:
    """Provide the block height at which the wallet was created.

    Backends can use this as a hint to skip scanning blocks before
    the wallet existed, avoiding unnecessary work during initial sync.
    Only used when no explicit ``scan_start_height`` is configured.

    Passing ``None`` clears any previously set hint.

    The default implementation is a no-op; backends that support
    scan optimisation override this method.
    """

verify_bonds(bonds: list[BondVerificationRequest]) -> list[BondVerificationResult] async

Verify multiple fidelity bond UTXOs in bulk.

This is the primary method for verifying fidelity bonds. Each backend can override this for optimal performance: - Bitcoin Core: JSON-RPC batch of gettxout calls (2 HTTP requests total) - Neutrino: batch address rescan + individual UTXO lookups - Mempool: parallel HTTP requests

The default implementation calls get_utxo() sequentially with a semaphore.

Args: bonds: List of bond verification requests with pre-computed addresses

Returns: List of verification results, one per input bond (same order)

Source code in jmwallet/src/jmwallet/backends/base.py

async def verify_bonds(
    self,
    bonds: list[BondVerificationRequest],
) -> list[BondVerificationResult]:
    """Verify multiple fidelity bond UTXOs in bulk.

    This is the primary method for verifying fidelity bonds. Each backend can
    override this for optimal performance:
    - Bitcoin Core: JSON-RPC batch of gettxout calls (2 HTTP requests total)
    - Neutrino: batch address rescan + individual UTXO lookups
    - Mempool: parallel HTTP requests

    The default implementation calls get_utxo() sequentially with a semaphore.

    Args:
        bonds: List of bond verification requests with pre-computed addresses

    Returns:
        List of verification results, one per input bond (same order)
    """
    if not bonds:
        return []

    current_height = await self.get_block_height()
    semaphore = asyncio.Semaphore(10)

    async def _verify_one(bond: BondVerificationRequest) -> BondVerificationResult:
        async with semaphore:
            try:
                utxo = await self.get_utxo(bond.txid, bond.vout)
                if utxo is None:
                    return BondVerificationResult(
                        txid=bond.txid,
                        vout=bond.vout,
                        value=0,
                        confirmations=0,
                        block_time=0,
                        valid=False,
                        error="UTXO not found or spent",
                    )
                if utxo.confirmations <= 0:
                    return BondVerificationResult(
                        txid=bond.txid,
                        vout=bond.vout,
                        value=utxo.value,
                        confirmations=0,
                        block_time=0,
                        valid=False,
                        error="UTXO unconfirmed",
                    )
                if (
                    not utxo.scriptpubkey
                    or utxo.scriptpubkey.lower() != bond.scriptpubkey.lower()
                ):
                    return BondVerificationResult(
                        txid=bond.txid,
                        vout=bond.vout,
                        value=utxo.value,
                        confirmations=utxo.confirmations,
                        block_time=0,
                        valid=False,
                        error="ScriptPubKey mismatch",
                    )
                # Get the block time for the confirmation block
                conf_height = current_height - utxo.confirmations + 1
                block_time = await self.get_block_time(conf_height)
                return BondVerificationResult(
                    txid=bond.txid,
                    vout=bond.vout,
                    value=utxo.value,
                    confirmations=utxo.confirmations,
                    block_time=block_time,
                    valid=True,
                )
            except Exception as e:
                logger.warning(
                    "Bond verification failed for %s:%d: %s",
                    bond.txid,
                    bond.vout,
                    e,
                )
                return BondVerificationResult(
                    txid=bond.txid,
                    vout=bond.vout,
                    value=0,
                    confirmations=0,
                    block_time=0,
                    valid=False,
                    error=str(e),
                )

    results = await asyncio.gather(*[_verify_one(b) for b in bonds])
    return list(results)

verify_tx_output(txid: str, vout: int, address: str, start_height: int | None = None) -> bool async

Verify that a specific transaction output exists (was broadcast and confirmed).

This is useful for verifying a transaction was successfully broadcast when we know at least one of its output addresses (e.g., our coinjoin destination).

For full node backends, this uses get_transaction(). For light clients (neutrino), this uses UTXO lookup with the address hint.

Args: txid: Transaction ID to verify vout: Output index to check address: The address that should own this output start_height: Optional block height hint for light clients (improves performance)

Returns: True if the output exists (transaction was broadcast), False otherwise

Source code in jmwallet/src/jmwallet/backends/base.py

async def verify_tx_output(
    self,
    txid: str,
    vout: int,
    address: str,
    start_height: int | None = None,
) -> bool:
    """
    Verify that a specific transaction output exists (was broadcast and confirmed).

    This is useful for verifying a transaction was successfully broadcast when
    we know at least one of its output addresses (e.g., our coinjoin destination).

    For full node backends, this uses get_transaction().
    For light clients (neutrino), this uses UTXO lookup with the address hint.

    Args:
        txid: Transaction ID to verify
        vout: Output index to check
        address: The address that should own this output
        start_height: Optional block height hint for light clients (improves performance)

    Returns:
        True if the output exists (transaction was broadcast), False otherwise
    """
    # Default implementation for full node backends
    tx = await self.get_transaction(txid)
    return tx is not None

verify_utxo_with_metadata(txid: str, vout: int, scriptpubkey: str, blockheight: int) -> UTXOVerificationResult async

Verify a UTXO using provided metadata (neutrino_compat feature).

This method allows light clients to verify UTXOs without needing arbitrary blockchain queries by using metadata provided by the peer.

The implementation should: 1. Use scriptpubkey to add the UTXO to watch list (for Neutrino) 2. Use blockheight as a hint for efficient rescan 3. Verify the UTXO exists with matching scriptpubkey 4. Return the UTXO value and confirmations

Default implementation falls back to get_utxo() for full node backends.

Args: txid: Transaction ID vout: Output index scriptpubkey: Expected scriptPubKey (hex) blockheight: Block height where UTXO was confirmed

Returns: UTXOVerificationResult with verification status and UTXO data

Source code in jmwallet/src/jmwallet/backends/base.py

async def verify_utxo_with_metadata(
    self,
    txid: str,
    vout: int,
    scriptpubkey: str,
    blockheight: int,
) -> UTXOVerificationResult:
    """
    Verify a UTXO using provided metadata (neutrino_compat feature).

    This method allows light clients to verify UTXOs without needing
    arbitrary blockchain queries by using metadata provided by the peer.

    The implementation should:
    1. Use scriptpubkey to add the UTXO to watch list (for Neutrino)
    2. Use blockheight as a hint for efficient rescan
    3. Verify the UTXO exists with matching scriptpubkey
    4. Return the UTXO value and confirmations

    Default implementation falls back to get_utxo() for full node backends.

    Args:
        txid: Transaction ID
        vout: Output index
        scriptpubkey: Expected scriptPubKey (hex)
        blockheight: Block height where UTXO was confirmed

    Returns:
        UTXOVerificationResult with verification status and UTXO data
    """
    # Default implementation for full node backends
    # Just uses get_utxo() directly since we can query any UTXO
    utxo = await self.get_utxo(txid, vout)

    if utxo is None:
        return UTXOVerificationResult(
            valid=False,
            error="UTXO not found or spent",
        )

    # Verify scriptpubkey matches
    scriptpubkey_matches = utxo.scriptpubkey.lower() == scriptpubkey.lower()

    if not scriptpubkey_matches:
        return UTXOVerificationResult(
            valid=False,
            value=utxo.value,
            confirmations=utxo.confirmations,
            error="ScriptPubKey mismatch",
            scriptpubkey_matches=False,
        )

    return UTXOVerificationResult(
        valid=True,
        value=utxo.value,
        confirmations=utxo.confirmations,
        scriptpubkey_matches=True,
    )

BondVerificationRequest

Request to verify a single fidelity bond UTXO.

All fields are derived from the bond proof data. The address and scriptpubkey are pre-computed by the caller using derive_bond_address(utxo_pub, locktime).

Source code in jmwallet/src/jmwallet/backends/base.py

@dataclass
class BondVerificationRequest:
    """Request to verify a single fidelity bond UTXO.

    All fields are derived from the bond proof data. The address and scriptpubkey
    are pre-computed by the caller using ``derive_bond_address(utxo_pub, locktime)``.
    """

    txid: str
    """Transaction ID (hex, big-endian)"""
    vout: int
    """Output index"""
    utxo_pub: bytes
    """33-byte compressed public key from bond proof"""
    locktime: int
    """Locktime from bond proof (Unix timestamp)"""
    address: str
    """Derived P2WSH bech32 address"""
    scriptpubkey: str
    """Derived P2WSH scriptPubKey (hex)"""

Attributes

address: str instance-attribute

Derived P2WSH bech32 address

locktime: int instance-attribute

Locktime from bond proof (Unix timestamp)

scriptpubkey: str instance-attribute

Derived P2WSH scriptPubKey (hex)

txid: str instance-attribute

Transaction ID (hex, big-endian)

utxo_pub: bytes instance-attribute

33-byte compressed public key from bond proof

vout: int instance-attribute

Output index

BondVerificationResult

Result of verifying a single fidelity bond UTXO.

Source code in jmwallet/src/jmwallet/backends/base.py

@dataclass
class BondVerificationResult:
    """Result of verifying a single fidelity bond UTXO."""

    txid: str
    """Transaction ID"""
    vout: int
    """Output index"""
    value: int
    """UTXO value in satoshis (0 if verification failed)"""
    confirmations: int
    """Number of confirmations (0 if unconfirmed or failed)"""
    block_time: int
    """Confirmation timestamp (0 if unconfirmed or failed)"""
    valid: bool
    """Whether the bond UTXO exists, is unspent, and has positive confirmations"""
    error: str | None = None
    """Error description if verification failed"""

Attributes

block_time: int instance-attribute

Confirmation timestamp (0 if unconfirmed or failed)

confirmations: int instance-attribute

Number of confirmations (0 if unconfirmed or failed)

error: str | None = None class-attribute instance-attribute

Error description if verification failed

txid: str instance-attribute

Transaction ID

valid: bool instance-attribute

Whether the bond UTXO exists, is unspent, and has positive confirmations

value: int instance-attribute

UTXO value in satoshis (0 if verification failed)

vout: int instance-attribute

Output index

Transaction

Source code in jmwallet/src/jmwallet/backends/base.py

@dataclass
class Transaction:
    txid: str
    raw: str
    confirmations: int
    block_height: int | None = None
    block_time: int | None = None

Attributes

block_height: int | None = None class-attribute instance-attribute

block_time: int | None = None class-attribute instance-attribute

confirmations: int instance-attribute

raw: str instance-attribute

txid: str instance-attribute

UTXO

Backend-layer UTXO model returned by blockchain backends.

This type intentionally only contains chain-derived fields. Wallet metadata such as freeze state, labels, derivation path, fidelity-bond flags, and locktime are attached later by the wallet service when converting backend UTXOs into wallet-layer UTXOInfo entries.

Source code in jmwallet/src/jmwallet/backends/base.py

@dataclass
class UTXO:
    """Backend-layer UTXO model returned by blockchain backends.

    This type intentionally only contains chain-derived fields. Wallet metadata
    such as freeze state, labels, derivation path, fidelity-bond flags, and
    locktime are attached later by the wallet service when converting backend
    UTXOs into wallet-layer ``UTXOInfo`` entries.
    """

    txid: str
    vout: int
    value: int
    address: str
    confirmations: int
    scriptpubkey: str
    height: int | None = None

Attributes

address: str instance-attribute

confirmations: int instance-attribute

height: int | None = None class-attribute instance-attribute

scriptpubkey: str instance-attribute

txid: str instance-attribute

value: int instance-attribute

vout: int instance-attribute

UTXOVerificationResult

Result of UTXO verification with metadata.

Used by neutrino_compat feature for Neutrino-compatible verification.

Source code in jmwallet/src/jmwallet/backends/base.py

@dataclass
class UTXOVerificationResult:
    """
    Result of UTXO verification with metadata.

    Used by neutrino_compat feature for Neutrino-compatible verification.
    """

    valid: bool
    value: int = 0
    confirmations: int = 0
    error: str | None = None
    scriptpubkey_matches: bool = False

Attributes

confirmations: int = 0 class-attribute instance-attribute

error: str | None = None class-attribute instance-attribute

scriptpubkey_matches: bool = False class-attribute instance-attribute

valid: bool instance-attribute

value: int = 0 class-attribute instance-attribute

WalletTxEntry

A wallet transaction observed by incremental enumeration.

confirmations is 0 while the tx is still in the mempool. category is Bitcoin Core's classification (receive / send / generate / ...), best-effort and informational. raw is the raw transaction hex when the backend can supply it inline (avoiding a follow-up get_transaction round-trip); empty when the caller should fetch it.

Source code in jmwallet/src/jmwallet/backends/base.py

@dataclass
class WalletTxEntry:
    """A wallet transaction observed by incremental enumeration.

    ``confirmations`` is 0 while the tx is still in the mempool. ``category``
    is Bitcoin Core's classification (``receive`` / ``send`` / ``generate`` /
    ...), best-effort and informational. ``raw`` is the raw transaction hex
    when the backend can supply it inline (avoiding a follow-up
    ``get_transaction`` round-trip); empty when the caller should fetch it.
    """

    txid: str
    confirmations: int = 0
    block_height: int | None = None
    category: str = ""
    raw: str = ""

Attributes

block_height: int | None = None class-attribute instance-attribute

category: str = '' class-attribute instance-attribute

confirmations: int = 0 class-attribute instance-attribute

raw: str = '' class-attribute instance-attribute

txid: str instance-attribute