import os
import tempfile
from typing import Any

import pytest
import trio

from py_ipfs_lite.config import Config
from py_ipfs_lite.peer import Peer


@pytest.fixture
def memory_config():
    return Config(
        blockstore_type="memory",
        reprovide_interval_seconds=-1,  # Disable reprovider for quick tests
    )


@pytest.mark.trio
async def test_peer_rejects_bytes_cid(memory_config):
    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])
    await peer.start()
    try:
        with pytest.raises(ValueError, match="Invalid CID string"):
            await peer.get_node(
                b"bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi"
            )
    finally:
        await peer.close()


@pytest.fixture
def fs_config():
    with tempfile.TemporaryDirectory() as tmpdirname:
        yield Config(
            blockstore_type="filesystem",
            blockstore_path=tmpdirname,
            reprovide_interval_seconds=-1,
        )


@pytest.mark.trio
async def test_peer_lifecycle(memory_config):
    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])
    await peer.start()
    assert peer._started is True
    assert len(peer.host.addrs()) > 0
    await peer.close()
    assert peer._started is False


@pytest.mark.trio
async def test_add_get_remove_node(memory_config):
    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])
    await peer.start()

    # 1. Add
    node_data = {"msg": "hello from node test"}
    cid_str = await peer.add_node(node_data, codec="dag-json")
    assert cid_str is not None

    # 2. Get
    fetched = await peer.get_node(cid_str)
    assert fetched == node_data
    assert await peer.has_block(cid_str) is True

    # 3. Remove
    await peer.remove_node(cid_str)
    from libp2p.bitswap.cid import parse_cid

    assert not await peer.blockstore.has(parse_cid(str(cid_str)).buffer)

    await peer.close()


@pytest.mark.trio
async def test_add_get_file(memory_config):
    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])
    await peer.start()

    with tempfile.NamedTemporaryFile(delete=False) as f:
        f.write(b"hello world")
        temp_path = f.name

    try:
        cid_str = await peer.add_file(temp_path)
        assert cid_str is not None

        content_iter = await peer.get_file(cid_str, stream=True)
        chunks = []
        async for chunk in content_iter:
            chunks.append(chunk)
        content = b"".join(chunks)
        assert content == b"hello world"
    finally:
        os.unlink(temp_path)

    await peer.close()


@pytest.mark.trio
async def test_pin_and_gc(memory_config):
    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])
    await peer.start()

    # Add two nodes
    cid1 = await peer.add_node({"name": "pinned"})
    cid2 = await peer.add_node({"name": "unpinned"})

    # Pin cid1
    await peer.add_pin(cid1, recursive=False)

    # GC
    await peer.gc()

    # cid1 should exist, cid2 should be gone
    from libp2p.bitswap.cid import parse_cid

    assert await peer.blockstore.has(parse_cid(str(cid1)).buffer)
    assert not await peer.blockstore.has(parse_cid(str(cid2)).buffer)

    await peer.close()


@pytest.mark.trio
async def test_filesystem_blockstore(fs_config):
    peer = Peer(fs_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])
    await peer.start()

    cid_str = await peer.add_node({"foo": "bar"})
    assert cid_str is not None

    fetched = await peer.get_node(cid_str)
    assert fetched == {"foo": "bar"}

    await peer.close()

    assert os.path.exists(fs_config.blockstore_path)


def test_init_exports():
    from py_ipfs_lite import AddParams, Config, Peer

    assert Peer is not None
    assert Config is not None
    assert AddParams is not None


@pytest.mark.trio
async def test_add_file_with_params(memory_config):
    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])
    await peer.start()

    from py_ipfs_lite.config import AddParams

    params = AddParams(chunker="size-1024")

    with tempfile.NamedTemporaryFile(delete=False) as f:
        f.write(b"a" * 2048)
        temp_path = f.name

    try:
        cid_str = await peer.add_file(temp_path, params=params)
        assert cid_str is not None

        content_iter = await peer.get_file(cid_str, stream=True)
        chunks = []
        async for chunk in content_iter:
            chunks.append(chunk)
        content = b"".join(chunks)
        assert content == b"a" * 2048
    finally:
        os.unlink(temp_path)
        await peer.close()


@pytest.mark.trio
async def test_gc_concurrency_lock(memory_config):
    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])
    await peer.start()

    with tempfile.NamedTemporaryFile(delete=False) as f:
        f.write(b"large data block " * 1000)
        temp_path = f.name

    try:
        async with trio.open_nursery() as nursery:
            nursery.start_soon(peer.add_file, temp_path)
            nursery.start_soon(peer.gc)
    finally:
        os.unlink(temp_path)
        await peer.close()


@pytest.mark.trio
async def test_add_file_progress_callback(memory_config):
    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])
    await peer.start()

    with tempfile.NamedTemporaryFile(delete=False) as f:
        f.write(b"hello progress callback")
        temp_path = f.name

    progress_updates = []

    def my_progress(written: int, total: int):
        progress_updates.append((written, total))

    try:
        cid_str = await peer.add_file(temp_path, progress_callback=my_progress)
        assert cid_str is not None
        assert len(progress_updates) > 0
        # The last update should have written == total
        assert progress_updates[-1][0] == progress_updates[-1][1]
        assert progress_updates[-1][1] == len(b"hello progress callback")
    finally:
        os.unlink(temp_path)

    await peer.close()


@pytest.mark.trio
async def test_api_parity_methods():
    from libp2p.crypto.ed25519 import create_new_key_pair

    from py_ipfs_lite import (
        default_bootstrap_peers,
        new_in_memory_datastore,
        setup_libp2p,
    )

    # Test helpers
    boot_peers = default_bootstrap_peers()
    assert isinstance(boot_peers, list)
    assert len(boot_peers) > 0

    mem_store = new_in_memory_datastore()
    assert mem_store is not None

    key_pair = create_new_key_pair()
    host, routing = await setup_libp2p(key_pair, ["/ip4/127.0.0.1/tcp/0"], offline=True)
    assert host is not None
    assert routing is None  # Since offline is True


def test_peer_accessors(memory_config):
    from py_ipfs_lite.peer import Peer, PeerSession

    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])

    # Initialize properties manually to avoid full async start() overhead in this test
    peer.blockstore = peer._create_blockstore()
    peer._exchange = "dummy_exchange"

    # Test accessors
    session = peer.session()
    assert isinstance(session, PeerSession)
    assert peer.block_store() is not None
    assert peer.exchange() == "dummy_exchange"


@pytest.mark.trio
async def test_add_file_bytes(memory_config):
    from py_ipfs_lite.peer import Peer

    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])
    await peer.start()

    data = b"hello from bytes"
    cid_str = await peer.add_file(data)
    assert cid_str is not None

    content_iter = await peer.get_file(cid_str, stream=True)
    chunks = []
    async for chunk in content_iter:
        chunks.append(chunk)
    content = b"".join(chunks)
    assert content == b"hello from bytes"
    await peer.close()


@pytest.mark.trio
async def test_add_file_stream(memory_config):
    import io

    from py_ipfs_lite.peer import Peer

    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])
    await peer.start()

    data = b"hello from stream"
    stream = io.BytesIO(data)
    cid_str = await peer.add_file(stream)
    assert cid_str is not None

    content_iter = await peer.get_file(cid_str, stream=True)
    chunks = []
    async for chunk in content_iter:
        chunks.append(chunk)
    content = b"".join(chunks)
    assert content == b"hello from stream"
    await peer.close()


@pytest.mark.trio
async def test_direct_pin_does_not_protect_children(memory_config):
    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])
    await peer.start()
    try:
        child_cid = await peer.add_node({"msg": "child leaf"}, codec="dag-cbor")
        parent_cid = await peer.add_node(
            {"link": {"/": child_cid}, "msg": "parent"}, codec="dag-cbor"
        )

        # Pin directly
        await peer.add_pin(parent_cid, recursive=False)

        # Run GC
        await peer.gc()

        # The parent should survive because it is directly pinned
        parent_data = await peer.get_node(parent_cid)
        assert parent_data["msg"] == "parent"

        # The child should NOT survive because the pin was not recursive
        assert not await peer.has_block(child_cid)
    finally:
        await peer.close()


class _RecordingNetwork:
    """Fake network that records peers whose connections were closed."""

    def __init__(self) -> None:
        self.closed_peers: list[Any] = []

    async def close_peer(self, peer_id: Any) -> None:
        self.closed_peers.append(peer_id)


class _StubHost:
    """Fake host exposing only get_network() (used by Peer._ping_peer)."""

    def __init__(self) -> None:
        self.network = _RecordingNetwork()

    def get_network(self) -> _RecordingNetwork:
        return self.network


@pytest.mark.trio
async def test_ping_timeout_does_not_close_connection(monkeypatch):
    """
    A keep-alive ping that times out is ignored so transient slow responses do
    not trigger a reconnect churn loop. Dead connections are evicted by QUIC idle timeout.
    """
    from py_ipfs_lite.peer import Peer

    class _StubPingService:
        def __init__(self, host: Any) -> None:
            self.host = host

        async def ping(self, peer_id: Any, ping_amt: int = 1) -> None:
            raise TimeoutError("ping timed out")

    monkeypatch.setattr("libp2p.host.ping.PingService", _StubPingService)

    peer = object.__new__(Peer)
    peer.host = _StubHost()
    peer.connection_tracker = None

    await peer._ping_peer("peer1", timeout=0.05)

    assert peer.host.get_network().closed_peers == []


@pytest.mark.trio
async def test_ping_success_does_not_close_connection(monkeypatch):
    from py_ipfs_lite.peer import Peer

    class _StubPingService:
        def __init__(self, host: Any) -> None:
            self.host = host

        async def ping(self, peer_id: Any, ping_amt: int = 1) -> None:
            return None

    monkeypatch.setattr("libp2p.host.ping.PingService", _StubPingService)

    peer = object.__new__(Peer)
    peer.host = _StubHost()
    peer.connection_tracker = None

    await peer._ping_peer("peer1", timeout=0.05)

    assert peer.host.get_network().closed_peers == []


@pytest.mark.trio
async def test_ping_error_does_not_close_connection(monkeypatch):
    """
    A keep-alive ping that errors is ignored to prevent churn loops from transient resets.
    """
    from py_ipfs_lite.peer import Peer

    class _StubPingService:
        def __init__(self, host: Any) -> None:
            self.host = host

        async def ping(self, peer_id: Any, ping_amt: int = 1) -> None:
            raise RuntimeError("peer unreachable")

    monkeypatch.setattr("libp2p.host.ping.PingService", _StubPingService)

    peer = object.__new__(Peer)
    peer.host = _StubHost()
    peer.connection_tracker = None

    await peer._ping_peer("peer1", timeout=0.05)

    assert peer.host.get_network().closed_peers == []


@pytest.mark.trio
async def test_peer_close_closes_routing(memory_config):
    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])

    class MockRouting:
        def __init__(self):
            self.closed = False

        async def close(self):
            self.closed = True

    mock_routing = MockRouting()
    await peer.start()

    peer.routing = mock_routing

    await peer.close()

    assert mock_routing.closed is True


@pytest.mark.trio
async def test_peer_context_manager(memory_config):
    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])

    with pytest.raises(ValueError, match="simulated crash"):
        async with peer:
            assert peer._started is True
            raise ValueError("simulated crash")

    assert peer._started is False


@pytest.mark.trio
async def test_dag_json_rejects_nan(memory_config):
    async with Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"]) as peer:
        with pytest.raises(ValueError, match="Out of range float values"):
            await peer.add_node({"value": float("nan")}, codec="dag-json")
        with pytest.raises(ValueError, match="Out of range float values"):
            await peer.add_node({"value": float("inf")}, codec="dag-json")


@pytest.mark.trio
async def test_add_node_raw_rejects_dict(memory_config):
    peer = Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"])
    await peer.start()
    try:
        with pytest.raises(TypeError, match="only supports bytes or str"):
            await peer.add_node({"data": "invalid"}, codec="raw")
    finally:
        await peer.close()


@pytest.mark.trio
async def test_offline_ipns_methods_raise_error():
    from py_ipfs_lite.config import Config
    from py_ipfs_lite.exceptions import RoutingError
    from py_ipfs_lite.peer import Peer

    config = Config(offline=True, blockstore_type="memory")
    async with Peer(config, listen_addrs=["/ip4/127.0.0.1/tcp/0"]) as peer:
        with pytest.raises(RoutingError, match="peer is offline"):
            await peer.resolve_name(
                "12D3KooWCvVxG5SBv5fZNVULQGpJuhBCiRNAABs24QqyxtEYy1Pv"
            )
        with pytest.raises(RoutingError, match="peer is offline"):
            await peer.publish_name("/ipfs/x")


@pytest.mark.skip(reason="macOS python 3.12 libcrypto abort issue")
@pytest.mark.trio
async def test_fetch_local_block_with_affinity(memory_config):
    import os
    import tempfile

    from libp2p.bitswap.cid import cid_to_bytes
    from libp2p.bitswap.dag import decode_dag_pb

    from py_ipfs_lite.peer import Peer

    # Create two peers
    async with Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"]) as peer1:
        async with Peer(memory_config, listen_addrs=["/ip4/127.0.0.1/tcp/0"]) as peer2:
            from libp2p.peer.peerinfo import info_from_p2p_addr

            # Connect them
            peer1_addr = peer1.host.addrs()[0]
            await peer2.host.connect(info_from_p2p_addr(peer1_addr))

            # Create a file large enough to have at least 2 leaf blocks (>256KB)
            data = b"x" * (300 * 1024)
            with tempfile.NamedTemporaryFile(delete=False) as f:
                f.write(data)
                temp_path = f.name

            try:
                # Add to peer1
                root_cid_str = await peer1.add_file(temp_path)

                # Find the first child link
                root_data = await peer1._exchange.get_block(root_cid_str)
                links, _ = decode_dag_pb(root_data)
                assert len(links) >= 2, "File should have at least 2 child blocks"

                # Pre-seed ONLY the first child block into peer2's blockstore
                first_child_cid = links[0].cid
                first_child_data = await peer1._exchange.get_block(first_child_cid)
                # Use raw put to blockstore so Bitswap knows about it
                await peer2.blockstore.put(
                    cid_to_bytes(first_child_cid), first_child_data
                )

                # Now have peer2 fetch the whole file
                # Without the fix, it will fetch root from peer1,
                # hit local store for child1 (setting affinity to b'local'),
                # then try to dial b'local' for child2 and crash!
                fetched_data = await peer2.get_file(root_cid_str)
                assert fetched_data == data

            finally:
                os.unlink(temp_path)


@pytest.mark.skip(reason="macOS python 3.12 libcrypto abort issue")
@pytest.mark.trio
async def test_fetch_local_block_with_affinity_batch():
    import os
    import tempfile

    from libp2p.bitswap.cid import cid_to_bytes
    from libp2p.bitswap.dag import decode_dag_pb

    from py_ipfs_lite.config import Config
    from py_ipfs_lite.peer import Peer

    memory_config1 = Config(offline=False, blockstore_type="memory")
    memory_config2 = Config(offline=False, blockstore_type="memory")
    memory_config2.bitswap_batch_fetch = True

    # Create two peers
    async with Peer(memory_config1, listen_addrs=["/ip4/127.0.0.1/tcp/0"]) as peer1:
        async with Peer(memory_config2, listen_addrs=["/ip4/127.0.0.1/tcp/0"]) as peer2:
            from libp2p.peer.peerinfo import info_from_p2p_addr

            # Connect them
            peer1_addr = peer1.host.addrs()[0]
            await peer2.host.connect(info_from_p2p_addr(peer1_addr))

            # Create a file large enough to have at least 2 leaf blocks (>256KB)
            data = b"x" * (300 * 1024)
            with tempfile.NamedTemporaryFile(delete=False) as f:
                f.write(data)
                temp_path = f.name

            try:
                # Add to peer1
                root_cid_str = await peer1.add_file(temp_path)

                # Find the first child link
                root_data = await peer1._exchange.get_block(root_cid_str)
                links, _ = decode_dag_pb(root_data)
                assert len(links) >= 2, "File should have at least 2 child blocks"

                # Pre-seed ONLY the first child block into peer2's blockstore
                first_child_cid = links[0].cid
                first_child_data = await peer1._exchange.get_block(first_child_cid)
                # Use raw put to blockstore so Bitswap knows about it
                await peer2.blockstore.put(
                    cid_to_bytes(first_child_cid), first_child_data
                )

                # Now have peer2 fetch the whole file
                fetched_data = await peer2.get_file(root_cid_str)
                assert fetched_data == data

            finally:
                os.unlink(temp_path)