from __future__ import annotations

# ruff: noqa: E402
import hashlib
import json
import os
from pathlib import Path

import pytest
from sqlalchemy.exc import SQLAlchemyError

os.environ.setdefault(
    "DATABASE_URL",
    "postgresql+asyncpg://test:test@localhost/test",
)

from src.services.db_bounded_executor import (
    CANONICAL_ASSET,
    COMMAND_ID,
    CREATE_INDEX_SQL,
    MIGRATION_FILE,
    MIGRATION_SHA256,
    NORMALIZED_REINDEX_SQL,
    NORMALIZED_REINDEX_SQL_SHA256,
    NORMALIZED_SQL,
    NORMALIZED_SQL_SHA256,
    REINDEX_INDEX_SQL,
    BoundedExecutorContractError,
    evaluate_catalog_snapshot,
    execute_bounded_db_command,
    validate_execution_contract,
)

ROOT = Path(__file__).resolve().parents[1]
SOURCE_REVISION = "a" * 40
IDENTITY = {
    "trace_id": "trace-AIA-SRE-013-db-index",
    "run_id": "run-AIA-SRE-013-db-index",
    "work_item_id": "AIA-SRE-013",
}


def _normalized_migration_sql() -> str:
    source = (ROOT / "migrations" / MIGRATION_FILE).read_text()
    sql_only = " ".join(
        line for line in source.splitlines() if not line.lstrip().startswith("--")
    )
    return " ".join(sql_only.split())


def _absent_catalog() -> dict:
    return {
        "table_exists": True,
        "columns_complete": True,
        "table_kind_supported": True,
        "primary_database": True,
        "database_writable": True,
        "schema_usage": True,
        "schema_create": True,
        "table_owner_privilege_inherited": True,
        "table_select": True,
        "index_count": 0,
        "index_ready": False,
        "index_valid": False,
        "index_non_unique": False,
        "key_count_exact": False,
        "total_attribute_count_exact": False,
        "access_method_exact": False,
        "key_1_ascending": False,
        "key_1_nulls_last": False,
        "key_2_ascending": False,
        "key_2_descending": False,
        "key_2_nulls_first": False,
        "key_2_nulls_last": False,
        "key_1": "",
        "key_2": "",
        "include_1": "",
        "predicate": "",
    }


def _exact_catalog() -> dict:
    return {
        **_absent_catalog(),
        "index_count": 1,
        "index_ready": True,
        "index_valid": True,
        "index_non_unique": True,
        "key_count_exact": True,
        "total_attribute_count_exact": True,
        "access_method_exact": True,
        "key_1_ascending": True,
        "key_1_nulls_last": True,
        "key_2_descending": True,
        "key_2_nulls_first": True,
        "key_1": "project_id",
        "key_2": "queued_at",
        "include_1": "send_status",
        "predicate": """
            ((channel_type = 'telegram'::text) AND
             (((source_envelope #>> '{callback_reply,action}'::text[]) =
               'controlled_apply_result'::text) OR
              ((source_envelope #>> '{alert_notification,status}'::text[]) =
               'alert_notification_sent'::text)))
        """,
    }


def _invalid_exact_catalog() -> dict:
    return {**_exact_catalog(), "index_valid": False}


def _contract(mode: str = "check") -> dict:
    return {
        "mode": mode,
        "command_id": COMMAND_ID,
        "canonical_asset": CANONICAL_ASSET,
        "migration_sha256": MIGRATION_SHA256,
        "normalized_sql_sha256": NORMALIZED_SQL_SHA256,
        "normalized_reindex_sql_sha256": NORMALIZED_REINDEX_SQL_SHA256,
        **IDENTITY,
        "source_revision": SOURCE_REVISION,
        "runtime_source_revision": SOURCE_REVISION,
        "pod_name": "awoooi-api-7bc9867b77-alpha1",
    }


def test_fixed_migration_and_embedded_sql_digests_match() -> None:
    migration = (ROOT / "migrations" / MIGRATION_FILE).read_bytes()

    assert hashlib.sha256(migration).hexdigest() == MIGRATION_SHA256
    assert _normalized_migration_sql() == NORMALIZED_SQL
    assert hashlib.sha256(NORMALIZED_SQL.encode()).hexdigest() == NORMALIZED_SQL_SHA256
    assert (
        hashlib.sha256(NORMALIZED_REINDEX_SQL.encode()).hexdigest()
        == NORMALIZED_REINDEX_SQL_SHA256
    )


def test_embedded_sql_is_one_fixed_transactionless_create_only() -> None:
    normalized = CREATE_INDEX_SQL.upper()

    assert normalized.count("CREATE INDEX CONCURRENTLY IF NOT EXISTS") == 1
    assert normalized.count(";") == 1
    for forbidden in (
        "DROP ",
        "TRUNCATE ",
        "DELETE ",
        "UPDATE ",
        "INSERT ",
        "ALTER ",
        "BEGIN",
        "COMMIT",
    ):
        assert forbidden not in normalized


def test_embedded_reindex_is_one_fixed_transactionless_recovery_only() -> None:
    normalized = REINDEX_INDEX_SQL.upper()

    assert normalized.count("REINDEX INDEX CONCURRENTLY") == 1
    assert normalized.count(";") == 1
    assert "PUBLIC.IDX_AWOOOP_OUTBOUND_MSG_TELEGRAM_RECEIPT_LATEST" in normalized
    for forbidden in (
        "DROP ",
        "TRUNCATE ",
        "DELETE ",
        "UPDATE ",
        "INSERT ",
        "ALTER ",
        "BEGIN",
        "COMMIT",
    ):
        assert forbidden not in normalized


def test_precheck_requires_immediately_effective_owner_and_plain_table() -> None:
    source = (ROOT / "src/services/db_bounded_executor.py").read_text()

    assert "pg_has_role(current_user, t.table_owner_oid, 'USAGE')" in source
    assert "pg_has_role(current_user, t.table_owner_oid, 'MEMBER')" not in source
    assert "COALESCE(t.table_relkind = 'r', false)" in source


@pytest.mark.parametrize(
    ("field", "value", "error"),
    (
        ("command_id", "arbitrary_sql_v1", "command_id_contract_mismatch"),
        ("canonical_asset", "public.users", "canonical_asset_contract_mismatch"),
        ("migration_sha256", "0" * 64, "migration_sha256_contract_mismatch"),
        (
            "normalized_sql_sha256",
            "f" * 64,
            "normalized_sql_sha256_contract_mismatch",
        ),
        (
            "normalized_reindex_sql_sha256",
            "e" * 64,
            "normalized_reindex_sql_sha256_contract_mismatch",
        ),
        ("mode", "rollback", "mode_not_allowlisted"),
        ("pod_name", "awoooi-worker-1", "runtime_not_allowlisted_api_pod"),
        (
            "runtime_source_revision",
            "b" * 40,
            "runtime_source_revision_mismatch",
        ),
    ),
)
def test_contract_rejects_any_scope_or_runtime_drift(
    field: str,
    value: str,
    error: str,
) -> None:
    payload = _contract()
    payload[field] = value

    with pytest.raises(BoundedExecutorContractError, match=error):
        validate_execution_contract(**payload)


def test_catalog_precheck_allows_absent_index_but_rejects_drift() -> None:
    absent = evaluate_catalog_snapshot(_absent_catalog())
    exact = evaluate_catalog_snapshot(_exact_catalog())
    invalid_exact = evaluate_catalog_snapshot(_invalid_exact_catalog())
    wrong_null_order = evaluate_catalog_snapshot(
        {
            **_invalid_exact_catalog(),
            "key_2_nulls_first": False,
            "key_2_nulls_last": True,
        }
    )
    ascending_order = evaluate_catalog_snapshot(
        {
            **_invalid_exact_catalog(),
            "key_2_ascending": True,
            "key_2_descending": False,
            "key_2_nulls_first": False,
            "key_2_nulls_last": True,
        }
    )
    drift_row = {**_exact_catalog(), "key_2": "created_at"}
    drift = evaluate_catalog_snapshot(drift_row)
    non_owner = evaluate_catalog_snapshot(
        {**_absent_catalog(), "table_owner_privilege_inherited": False}
    )
    select_revoked = evaluate_catalog_snapshot(
        {**_absent_catalog(), "table_select": False}
    )
    partitioned_parent = evaluate_catalog_snapshot(
        {**_absent_catalog(), "table_kind_supported": False}
    )

    assert absent["apply_ready"] is True
    assert absent["index_exists"] is False
    assert exact["index_shape_exact"] is True
    assert exact["apply_ready"] is True
    assert invalid_exact["index_shape_exact"] is False
    assert invalid_exact["index_definition_exact_except_validity"] is True
    assert invalid_exact["reindex_ready"] is True
    assert invalid_exact["catalog_drift"] is False
    assert invalid_exact["apply_ready"] is True
    assert invalid_exact["key_2_variant"] == "descending_nulls_first"
    assert wrong_null_order["key_2_exact"] is False
    assert wrong_null_order["key_2_variant"] == "descending_nulls_last"
    assert ascending_order["key_2_exact"] is False
    assert ascending_order["key_2_variant"] == "ascending_nulls_last"
    assert wrong_null_order["catalog_drift"] is True
    assert drift["catalog_drift"] is True
    assert drift["apply_ready"] is False
    assert non_owner["apply_ready"] is False
    # CREATE INDEX requires table ownership, not a separate SELECT grant.
    assert select_revoked["apply_ready"] is True
    assert partitioned_parent["apply_ready"] is False


class _Result:
    def __init__(self, *, row: dict | None = None, scalar: bool | None = None):
        self.row = row
        self.scalar = scalar

    def mappings(self):
        return self

    def one(self):
        assert self.row is not None
        return self.row

    def scalar_one(self):
        return self.scalar


class _Connection:
    def __init__(self, catalog_rows: list[dict]):
        self.catalog_rows = list(catalog_rows)
        self.statements: list[str] = []
        self.isolation_level = ""

    async def execution_options(self, *, isolation_level: str):
        self.isolation_level = isolation_level
        return self

    async def execute(self, statement):
        sql = str(statement)
        self.statements.append(sql)
        if "WITH target AS" in sql:
            return _Result(row=self.catalog_rows.pop(0))
        if sql.startswith("SET "):
            return _Result()
        if "pg_try_advisory_lock" in sql:
            return _Result(scalar=True)
        if "pg_advisory_unlock" in sql:
            return _Result(scalar=True)
        if "CREATE INDEX CONCURRENTLY IF NOT EXISTS" in sql:
            return _Result()
        if "REINDEX INDEX CONCURRENTLY" in sql:
            return _Result()
        raise AssertionError(f"unexpected SQL: {sql}")


class _ConnectionContext:
    def __init__(self, connection: _Connection):
        self.connection = connection

    async def __aenter__(self):
        return self.connection

    async def __aexit__(self, *_args):
        return False


class _Engine:
    def __init__(self, connection: _Connection):
        self.connection = connection
        self.connect_count = 0

    def connect(self):
        self.connect_count += 1
        return _ConnectionContext(self.connection)


class _CreateFailureConnection(_Connection):
    async def execute(self, statement):
        sql = str(statement)
        if "CREATE INDEX CONCURRENTLY IF NOT EXISTS" in sql:
            self.statements.append(sql)
            raise SQLAlchemyError("sensitive backend detail must not escape")
        return await super().execute(statement)


@pytest.mark.asyncio
async def test_apply_uses_autocommit_fixed_sql_and_same_session_lock_release() -> None:
    connection = _Connection([_absent_catalog(), _absent_catalog(), _exact_catalog()])
    engine = _Engine(connection)

    receipt = await execute_bounded_db_command(
        **_contract("apply"),
        engine=engine,
    )

    assert receipt["terminal"] == "apply_verified_local"
    assert receipt["writes_performed"] is True
    assert receipt["operation"] == "create_index_concurrently"
    assert receipt["domain"] == "database"
    assert receipt["cross_domain_fallback_allowed"] is False
    assert receipt["secret_value_read"] is False
    assert connection.isolation_level == "AUTOCOMMIT"
    assert engine.connect_count == 1
    assert connection.statements[:3] == [
        "SET search_path TO pg_catalog, public",
        "SET lock_timeout TO '30s'",
        "SET statement_timeout TO '10min'",
    ]
    assert sum("CREATE INDEX CONCURRENTLY" in sql for sql in connection.statements) == 1
    assert connection.statements.index(CREATE_INDEX_SQL) < next(
        index
        for index, sql in enumerate(connection.statements)
        if "pg_advisory_unlock" in sql
    )


@pytest.mark.asyncio
async def test_apply_is_idempotent_when_exact_index_already_exists() -> None:
    connection = _Connection([_exact_catalog(), _exact_catalog(), _exact_catalog()])
    engine = _Engine(connection)

    receipt = await execute_bounded_db_command(
        **_contract("apply"),
        engine=engine,
    )

    assert receipt["terminal"] == "apply_verified_local"
    assert receipt["writes_performed"] is False
    assert receipt["operation"] == "no_write"
    assert all("CREATE INDEX CONCURRENTLY" not in sql for sql in connection.statements)


@pytest.mark.asyncio
async def test_apply_recovers_only_exact_invalid_index_with_concurrent_reindex() -> (
    None
):
    connection = _Connection(
        [_invalid_exact_catalog(), _invalid_exact_catalog(), _exact_catalog()]
    )
    engine = _Engine(connection)

    receipt = await execute_bounded_db_command(
        **_contract("apply"),
        engine=engine,
    )

    assert receipt["terminal"] == "apply_verified_local"
    assert receipt["writes_performed"] is True
    assert receipt["operation"] == "reindex_index_concurrently"
    assert connection.statements.count(REINDEX_INDEX_SQL) == 1
    assert all("CREATE INDEX CONCURRENTLY" not in sql for sql in connection.statements)


@pytest.mark.asyncio
async def test_apply_blocks_invalid_index_with_wrong_definition_without_write() -> None:
    wrong = {
        **_invalid_exact_catalog(),
        "key_2_ascending": True,
        "key_2_descending": False,
        "key_2_nulls_first": False,
        "key_2_nulls_last": True,
    }
    connection = _Connection([wrong])
    engine = _Engine(connection)

    receipt = await execute_bounded_db_command(
        **_contract("apply"),
        engine=engine,
    )

    assert receipt["terminal"] == "blocked"
    assert receipt["writes_performed"] is False
    assert receipt["operation"] == "none"
    assert all("REINDEX INDEX" not in sql for sql in connection.statements)
    assert all("CREATE INDEX" not in sql for sql in connection.statements)


@pytest.mark.asyncio
async def test_verify_is_catalog_only_and_never_takes_the_apply_lock() -> None:
    connection = _Connection([_exact_catalog()])
    engine = _Engine(connection)

    receipt = await execute_bounded_db_command(
        **_contract("verify"),
        engine=engine,
    )

    assert receipt["terminal"] == "verified_healthy"
    assert receipt["executor_role"] == "independent_catalog_verifier"
    assert receipt["operation"] == "none"
    assert receipt["writes_performed"] is False
    assert all("advisory_lock" not in sql for sql in connection.statements)


@pytest.mark.asyncio
async def test_failed_create_reports_unknown_write_state_without_backend_detail() -> (
    None
):
    connection = _CreateFailureConnection([_absent_catalog(), _absent_catalog()])
    engine = _Engine(connection)

    receipt = await execute_bounded_db_command(
        **_contract("apply"),
        engine=engine,
    )

    assert receipt["terminal"] == "failed"
    assert receipt["write_attempted"] is True
    assert receipt["writes_performed"] is None
    assert receipt["error_code"] == "database_operation_failed"
    assert "sensitive backend detail" not in json.dumps(receipt)
    assert any("pg_advisory_unlock" in sql for sql in connection.statements)


class _SafeSqlstateFailureConnection(_Connection):
    async def execute(self, statement):
        sql = str(statement)
        if "REINDEX INDEX CONCURRENTLY" in sql:
            self.statements.append(sql)
            error = SQLAlchemyError("backend text must not escape")
            error.orig = type("DriverError", (), {"sqlstate": "55P03"})()
            raise error
        return await super().execute(statement)


@pytest.mark.asyncio
async def test_failed_reindex_returns_safe_sqlstate_without_backend_detail() -> None:
    connection = _SafeSqlstateFailureConnection(
        [_invalid_exact_catalog(), _invalid_exact_catalog()]
    )
    engine = _Engine(connection)

    receipt = await execute_bounded_db_command(
        **_contract("apply"),
        engine=engine,
    )

    assert receipt["terminal"] == "failed"
    assert receipt["operation"] == "reindex_index_concurrently"
    assert receipt["writes_performed"] is None
    assert receipt["database_sqlstate"] == "55P03"
    assert "backend text" not in json.dumps(receipt)