"""Integration-test fixtures using an isolated Docker Compose infrastructure.

The test infra (postgres + redis + rabbitmq + minio) is brought up automatically
by the session-scoped `infra` fixture and torn down after the session.  This
keeps integration tests from racing against the development service workers
that also consume RabbitMQ queues.

To run against an already-running external infra instead, set
``INTEGRATION_TEST_USE_EXTERNAL_INFRA=1`` before invoking pytest.
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
import time
from collections.abc import Iterator
from pathlib import Path
from typing import TYPE_CHECKING

import httpx
import pytest
import pytest_asyncio
from asgi_lifespan import LifespanManager
from sqlalchemy import text

from contract_check.api.app import create_app
from contract_check.core.config import get_settings
from contract_check.core.db.session import create_session_factory
from contract_check.core.tokens import hash_token

if TYPE_CHECKING:
    pass

pytestmark = pytest.mark.integration

_BOT_SERVICE_TOKEN = "it-test-bot-token"

# Ports must match docker-compose.test.yml.
_TEST_DB_URL = "postgresql+asyncpg://contract_check:contract_check@localhost:25432/contract_check"
_TEST_AMQP_URL = "amqp://contract_check:contract_check@localhost:6672//"
_TEST_S3_URL = "http://localhost:10000"

_DB_URL = _TEST_DB_URL
_AMQP_URL = _TEST_AMQP_URL
_S3_URL = _TEST_S3_URL

_USE_EXTERNAL_INFRA = os.environ.get("INTEGRATION_TEST_USE_EXTERNAL_INFRA", "0") == "1"


def _repo_root() -> Path:
    return Path(__file__).resolve().parents[2]


def _run(cmd: list[str], *, check: bool = True) -> subprocess.CompletedProcess[str]:
    result = subprocess.run(
        cmd,
        cwd=str(_repo_root()),
        env={**os.environ},
        check=False,
        capture_output=True,
        text=True,
    )
    if check and result.returncode != 0:
        raise subprocess.CalledProcessError(
            result.returncode,
            cmd,
            output=result.stdout,
            stderr=result.stderr,
        )
    return result


def _docker_compose_test(args: list[str]) -> list[str]:
    # Use a dedicated project name so the test stack is independent from the
    # development stack.  Without this, `docker compose` treats the running dev
    # services as "orphans" and returns a non-zero exit code.
    return [
        "docker",
        "compose",
        "-p",
        "contract-check-test",
        "-f",
        str(_repo_root() / "docker-compose.test.yml"),
        *args,
    ]


def _wait_for_healthy(service: str, *, deadline_seconds: int = 60) -> None:
    """Poll `docker compose ps` until `service` reports healthy."""
    start = time.monotonic()
    while time.monotonic() - start < deadline_seconds:
        result = _run(
            _docker_compose_test(["ps", service, "--format", "json"]),
            check=False,
        )
        if result.returncode == 0 and result.stdout:
            # `docker compose ps --format json` emits one JSON object per line.
            for line in result.stdout.strip().splitlines():
                try:
                    info = json.loads(line)
                except json.JSONDecodeError:
                    continue
                if info.get("Health") == "healthy":
                    return
                if info.get("State") == "exited" and info.get("ExitCode") == 0:
                    # One-shot containers are also fine.
                    return
        time.sleep(1)
    raise RuntimeError(f"service {service} did not become healthy within {deadline_seconds}s")


def _start_test_infra() -> None:
    if _USE_EXTERNAL_INFRA:
        return

    # Bring up the test stack without --wait: minio-init-test is a one-shot
    # container that exits after creating the bucket, which makes --wait fail.
    _run(_docker_compose_test(["up", "-d"]))

    # Wait for every persistent service to be healthy.
    for service in ("postgres-test", "redis-test", "rabbitmq-test", "minio-test"):
        _wait_for_healthy(service)

    # postgres healthcheck can still race with alembic, so explicitly wait for
    # the DB to accept connections.
    _run(
        _docker_compose_test(
            [
                "exec",
                "-T",
                "postgres-test",
                "sh",
                "-c",
                "until pg_isready -U contract_check -d contract_check; do sleep 1; done",
            ]
        )
    )


def _stop_test_infra() -> None:
    if _USE_EXTERNAL_INFRA:
        return
    # Use --volumes to wipe test data between runs.  Never use -v for external infra.
    _run(_docker_compose_test(["down", "-v"]), check=False)


def _set_env() -> None:
    env_vars = {
        "DATABASE_URL": _DB_URL,
        "RABBITMQ_URL": _AMQP_URL,
        "S3_ENDPOINT_URL": _S3_URL,
        "S3_ACCESS_KEY": "contract_check",
        "S3_SECRET_KEY": "contract_check",
        "S3_BUCKET": "contract-check-docs-test",
        "REDIS_URL": "redis://localhost:27379/0",
        "OLLAMA_HOST": "http://localhost",
        "OLLAMA_API_KEY": "test",
        "JWT_SECRET": "it-test-jwt-secret-not-for-production",
        "JWT_ALGORITHM": "HS256",
        "JWT_ACCESS_TTL_MINUTES": "1440",
        "TELEGRAM_BOT_TOKEN": "it-test-bot-token:it-test-secret",
        # httpx ASGITransport buffers the whole response — infinite SSE streams
        # can only complete via the max-stream deadline (which emits the
        # `timeout` event). Keep it short so streaming tests finish fast.
        "SSE_POLL_INTERVAL_SECONDS": "0.25",
        "SSE_MAX_STREAM_SECONDS": "5",
    }
    for k, v in env_vars.items():
        os.environ[k] = v


@pytest.fixture(scope="session")
def infra() -> Iterator[dict[str, str]]:
    _set_env()
    get_settings.cache_clear()

    _start_test_infra()

    _run(
        [
            sys.executable,
            "-m",
            "alembic",
            "-c",
            str(_repo_root() / "alembic.ini"),
            "upgrade",
            "head",
        ],
        check=True,
    )

    factory = create_session_factory()

    async def _seed() -> None:
        async with factory() as session:
            await session.execute(
                text(
                    "INSERT INTO service_tokens (name, token_hash, adapter) "
                    "VALUES (:name, :hash, 'bot') "
                    "ON CONFLICT (name) DO UPDATE SET "
                    "  token_hash = EXCLUDED.token_hash, revoked = FALSE"
                ),
                {"name": "bot-test", "hash": hash_token(_BOT_SERVICE_TOKEN)},
            )
            await session.commit()

    import asyncio

    asyncio.run(_seed())

    yield {
        "database_url": _DB_URL,
        "rabbitmq_url": _AMQP_URL,
        "s3_endpoint_url": _S3_URL,
        "s3_access_key": "contract_check",
        "s3_secret_key": "contract_check",
        "s3_bucket": "contract-check-docs-test",
        "token": _BOT_SERVICE_TOKEN,
    }

    get_settings.cache_clear()
    _stop_test_infra()


@pytest_asyncio.fixture
async def client(infra: dict[str, str]) -> httpx.AsyncClient:
    get_settings.cache_clear()
    _set_env()
    get_settings()
    app = create_app()
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test"
    ) as c:
        async with LifespanManager(app):
            yield c
    get_settings.cache_clear()


@pytest.fixture
def auth_header(infra: dict[str, str]) -> dict[str, str]:
    return {"Authorization": f"Bearer {infra['token']}"}


async def user_token(client: httpx.AsyncClient, infra: dict[str, str], telegram_id: int) -> str:
    """Call /auth/telegram/bot and return a user JWT for the given telegram_id."""
    r = await client.post(
        "/api/v1/auth/telegram/bot",
        json={"telegram_id": telegram_id},
        headers={"Authorization": f"Bearer {infra['token']}"},
    )
    assert r.status_code == 200, f"auth failed: {r.status_code} {r.text}"
    return r.json()["access_token"]


@pytest.fixture
async def db_session():
    _set_env()
    get_settings.cache_clear()
    factory = create_session_factory()
    async with factory() as session:
        yield session
        await session.close()
    get_settings.cache_clear()