#!/usr/bin/env python3
# Copyright (c) 2026 The Bitcoin Unlimited developers

"""
Test that reproduces the faster_statushash rscan ordering bug.

The bug: faster_statushash uses rscan (reverse DB scan) to cap entries at
max_elements. The DB key is (scripthash, flags, outpointhash) with no height
component. So rscan returns entries by descending outpointhash (random hash
order), NOT by most-recent block height. When total entries exceed
max_elements, newly confirmed entries may fall outside the rscan window,
causing statushash to remain unchanged even though the address history changed.

This test sets ROSTRUM_STATUSHASH_MAX_ELEMENTS to a small value, creates many
more entries than that, then asserts that each new confirmed transaction
changes the statushash.
"""

import os

# Must be set before the test framework imports and starts processes.
os.environ["ROSTRUM_STATUSHASH_MAX_ELEMENTS"] = "5"

# pylint: disable=wrong-import-position
import asyncio
from test_framework.electrumutil import ElectrumTestFramework
from test_framework.electrumconnection import ElectrumConnection
from test_framework.environment import on_bch, on_nex
from test_framework.utiltx import p2pkh

if on_bch():
    from test_framework.bch.blocktools import create_transaction
    from test_framework.bch.script import anyonecanspend_unlocking_script
elif on_nex():
    from test_framework.nex.blocktools import create_transaction
    from test_framework.nex.script import anyonecanspend_unlocking_script
    from test_framework.nex.nodemessages import TxOut
else:
    raise NotImplementedError()

SCRIPTHASH_SUBSCRIBE = "blockchain.scripthash.subscribe"

NUM_INITIAL_TXS = 50
NUM_ITERATIONS = 3


async def address_to_scripthash(address):
    cli = ElectrumConnection()
    await cli.connect()
    scripthash = await cli.call("blockchain.address.get_scripthash", address)
    cli.disconnect()
    return scripthash


def make_conform_to_ctor(txs):
    for tx in txs:
        tx.rehash()
    return sorted(txs, key=lambda tx: tx.get_rpc_hex_id())


class StatushashBugTest(ElectrumTestFramework):
    """
    Reproduces the faster_statushash bug where new confirmed transactions
    don't change the statushash when entries exceed max_elements.
    """

    async def run_test(self):
        n = self.nodes[0]
        await self.bootstrap_p2p()

        cli = ElectrumConnection()
        await cli.connect()
        try:
            coinbases = await self.mine_blocks(cli, n, 201)

            addr = n.getnewaddress("p2pkh")
            scripthash = await address_to_scripthash(addr)

            self.info(
                "Creating %d transactions to address to exceed max_elements (5)",
                NUM_INITIAL_TXS,
            )
            txs = []
            out_type = (
                TxOut.TYPE_SATOSCRIPT  # pylint: disable=possibly-used-before-assignment
                if on_nex()
                else 0
            )
            for _ in range(NUM_INITIAL_TXS):
                coinbase = coinbases.pop(0)
                tx = create_transaction(
                    prevtx=coinbase,
                    value=10000,
                    n=0,
                    sig=anyonecanspend_unlocking_script(),
                    out=p2pkh(addr),
                    out_type=out_type,
                )
                txs.append(tx)

            txs = make_conform_to_ctor(txs)
            await self.mine_blocks(cli, n, 1, txs)

            self.info("Subscribing to scripthash to get initial statushash")
            old_statushash, _ = await cli.subscribe(SCRIPTHASH_SUBSCRIBE, scripthash)
            assert (
                old_statushash is not None
            ), "Address with history should have a statushash"
            self.info("Initial statushash: %s", old_statushash)

            for i in range(NUM_ITERATIONS):
                self.info("Iteration %d: mining 1 new tx to the address", i + 1)
                coinbase = coinbases.pop(0)
                tx = create_transaction(
                    prevtx=coinbase,
                    value=10000,
                    n=0,
                    sig=anyonecanspend_unlocking_script(),
                    out=p2pkh(addr),
                    out_type=out_type,
                )
                tx.rehash()
                await self.mine_blocks(cli, n, 1, [tx])

                self.info("Re-subscribing to get updated statushash")
                new_statushash, _ = await cli.subscribe(
                    SCRIPTHASH_SUBSCRIBE, scripthash
                )

                assert new_statushash is not None, "Statushash should not be None"
                assert (
                    new_statushash != old_statushash
                ), f"Iteration {i + 1}: statushash should change after new confirmed tx (old={old_statushash}, new={new_statushash})"

                self.info(
                    "Iteration %d: statushash changed %s -> %s",
                    i + 1,
                    old_statushash,
                    new_statushash,
                )
                old_statushash = new_statushash

            self.info("All iterations passed - statushash updated correctly")
        finally:
            cli.disconnect()


if __name__ == "__main__":
    asyncio.run(StatushashBugTest().main())
