"""Migration 560 — the answer library gains an employee-facing face.

The whole risk of this migration is one number: what every row that
already exists gets for ``employee_visible``. The approved call is that
the existing library goes live to all staff, so the column defaults to
TRUE and no backfill statement is written — the server default does it.
That is exactly the kind of decision that is easy to get wrong silently
(a NULL-able column, a default of 0, a default that only applies to rows
inserted afterwards), and impossible to notice afterwards: the employee
page would simply be empty, which looks like "nobody has written any
FAQ entries yet".

So the fixture rebuilds the pre-560 schema rather than testing against
``create_all`` output, where the columns already exist with their model
defaults and would prove nothing about what the ALTER does.

Also pinned: the migration is resumable (every step is guarded, so a
half-applied run can be re-run), and the gap ledger's ``query_norm``
index is UNIQUE — the upsert in ``security_faq_service.record_gap`` is an
ON DUPLICATE KEY statement, so a non-unique index there would silently
turn one ledger line per question into one line per ask.
"""
from __future__ import annotations

import importlib.util
import pathlib
import uuid

import pytest
import sqlalchemy as sa
from alembic.migration import MigrationContext
from alembic.operations import Operations

from ..conftest import _TEST_SYNC_URL

_MIGRATION = (
    pathlib.Path(__file__).resolve().parents[2]
    / "alembic" / "versions" / "560_employee_security_faq.py"
)

ENTRIES = "auditor_answer_library_entries"
GAPS = "auditor_security_faq_gaps"
VOTES = "auditor_security_faq_votes"
NEW_COLUMNS = (
    "employee_visible", "is_featured", "employee_view_count",
    "employee_helpful_count", "employee_unhelpful_count",
)
_PREFIX = "m560"


def _load():
    spec = importlib.util.spec_from_file_location("m560", _MIGRATION)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _run(conn, direction: str = "upgrade") -> None:
    ctx = MigrationContext.configure(conn)
    with Operations.context(ctx):
        getattr(_load(), direction)()


def _has_column(conn, table: str, column: str) -> bool:
    return conn.execute(
        sa.text(
            "SELECT 1 FROM information_schema.columns "
            "WHERE table_schema = DATABASE() AND table_name = :t "
            "AND column_name = :c"
        ),
        {"t": table, "c": column},
    ).scalar() is not None


def _index_non_unique(conn, table: str, index: str) -> int | None:
    return conn.execute(
        sa.text(
            "SELECT NON_UNIQUE FROM information_schema.statistics "
            "WHERE table_schema = DATABASE() AND table_name = :t "
            "AND index_name = :i LIMIT 1"
        ),
        {"t": table, "i": index},
    ).scalar()


def _strip_to_pre_560(conn) -> None:
    """Recreate the schema as it stood at revision 559."""
    for table in (VOTES, GAPS):
        conn.execute(sa.text(f"DROP TABLE IF EXISTS {table}"))
    if _index_non_unique(conn, ENTRIES, "idx_auditor_answer_lib_employee") is not None:
        conn.execute(
            sa.text(f"DROP INDEX idx_auditor_answer_lib_employee ON {ENTRIES}")
        )
    for column in NEW_COLUMNS:
        if _has_column(conn, ENTRIES, column):
            conn.execute(sa.text(f"ALTER TABLE {ENTRIES} DROP COLUMN {column}"))


def _seed_entry(conn, question: str) -> str:
    entry_id = str(uuid.uuid4())
    conn.execute(
        sa.text(
            f"INSERT INTO {ENTRIES} "
            "(id, category, question_zh, question_en, answer_zh, answer_en, "
            " status, usage_count, review_interval_months, created_at, updated_at) "
            "VALUES (:id, 'general', :q, :q, :a, :a, 'active', 0, 12, NOW(), NOW())"
        ),
        {"id": entry_id, "q": question, "a": f"{_PREFIX} answer"},
    )
    return entry_id


@pytest.fixture
def conn():
    engine = sa.create_engine(_TEST_SYNC_URL)
    with engine.begin() as c:
        try:
            yield c
        finally:
            # Leave the schema at 560 whatever happened above — every
            # later test in this shard reads it through the ORM models.
            _run(c, "upgrade")
            c.execute(
                sa.text(f"DELETE FROM {ENTRIES} WHERE answer_zh = :a"),
                {"a": f"{_PREFIX} answer"},
            )
    engine.dispose()


def test_existing_entries_come_out_visible_to_employees(conn):
    _strip_to_pre_560(conn)
    first = _seed_entry(conn, f"{_PREFIX} 已有条目一")
    second = _seed_entry(conn, f"{_PREFIX} 已有条目二")

    _run(conn, "upgrade")

    rows = conn.execute(
        sa.text(
            "SELECT id, employee_visible, is_featured, employee_view_count, "
            "       employee_helpful_count, employee_unhelpful_count "
            f"FROM {ENTRIES} WHERE id IN (:a, :b)"
        ),
        {"a": first, "b": second},
    ).mappings().all()

    assert len(rows) == 2
    for row in rows:
        # The approved call: the library goes live as it stands.
        assert row["employee_visible"] == 1
        # Curation and engagement start from zero, not from "everything
        # is featured" — a featured-by-default library has no top.
        assert row["is_featured"] == 0
        assert row["employee_view_count"] == 0
        assert row["employee_helpful_count"] == 0
        assert row["employee_unhelpful_count"] == 0


def test_the_gap_dedup_index_is_unique(conn):
    _strip_to_pre_560(conn)
    _run(conn, "upgrade")

    # NON_UNIQUE = 0 means unique. record_gap()'s ON DUPLICATE KEY UPDATE
    # depends on this: without uniqueness every ask inserts a new row and
    # ask_count never rises above 1.
    assert _index_non_unique(conn, GAPS, "uq_auditor_security_faq_gaps_query") == 0
    assert _index_non_unique(conn, VOTES, "uq_auditor_security_faq_votes") == 0


def test_the_upsert_actually_aggregates_on_the_real_schema(conn):
    """The unique index is only interesting if the statement that relies
    on it behaves — assert the aggregation on the migrated table."""
    _strip_to_pre_560(conn)
    _run(conn, "upgrade")

    norm = f"{_PREFIX}-duplicate-question"
    for _ in range(3):
        conn.execute(
            sa.text(
                f"INSERT INTO {GAPS} "
                "(id, query_norm, query_sample, lang, ask_count, "
                " first_asked_at, last_asked_at, status, created_at, updated_at) "
                "VALUES (:id, :norm, :norm, 'zh', 1, NOW(), NOW(), 'open', "
                " NOW(), NOW()) "
                "ON DUPLICATE KEY UPDATE ask_count = ask_count + 1"
            ),
            {"id": str(uuid.uuid4()), "norm": norm},
        )

    rows = conn.execute(
        sa.text(f"SELECT ask_count FROM {GAPS} WHERE query_norm = :n"),
        {"n": norm},
    ).scalars().all()
    conn.execute(
        sa.text(f"DELETE FROM {GAPS} WHERE query_norm = :n"), {"n": norm},
    )
    assert rows == [3]


def test_a_half_applied_run_is_resumable(conn):
    """Every step is guarded, so re-running after a partial failure must
    be a no-op rather than "duplicate column name"."""
    _strip_to_pre_560(conn)
    _run(conn, "upgrade")
    _run(conn, "upgrade")

    for column in NEW_COLUMNS:
        assert _has_column(conn, ENTRIES, column)
    assert _index_non_unique(conn, ENTRIES, "idx_auditor_answer_lib_employee") == 1


def test_downgrade_removes_everything_it_added(conn):
    _strip_to_pre_560(conn)
    _run(conn, "upgrade")
    _run(conn, "downgrade")

    for column in NEW_COLUMNS:
        assert not _has_column(conn, ENTRIES, column)
    for table in (GAPS, VOTES):
        assert conn.execute(
            sa.text(
                "SELECT 1 FROM information_schema.tables "
                "WHERE table_schema = DATABASE() AND table_name = :t"
            ),
            {"t": table},
        ).scalar() is None