"""マイグレーションの適用（testing.md 必須対象 8）."""

import re
import sqlite3
from pathlib import Path

import pytest

from tiasdata import catalog
from tiasdata.catalog import migrate
from tiasdata.catalog.migrate import Migration, MigrationResult


def _tables(conn: sqlite3.Connection) -> set[str]:
    rows = conn.execute(
        "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'"
    )
    return {name for (name,) in rows}


def _schema(conn: sqlite3.Connection) -> list[tuple]:
    return conn.execute("SELECT type, name, sql FROM sqlite_master ORDER BY type, name").fetchall()


@pytest.mark.parametrize("series", ["catalog", "identity"])
def test_apply_to_empty_db_reaches_latest_version(tmp_path: Path, series: str) -> None:
    conn = catalog.connect(tmp_path / "db.sqlite", create=True)
    try:
        result = catalog.apply_migrations(conn, series)
        latest = catalog.latest_version(series)
        assert latest >= 1
        assert result == MigrationResult(0, latest)
        assert result.applied == latest
        assert conn.execute("PRAGMA user_version").fetchone()[0] == latest
        assert catalog.schema_version(conn) == latest
    finally:
        conn.close()


@pytest.mark.parametrize("series", ["catalog", "identity"])
def test_reapply_changes_nothing(tmp_path: Path, series: str) -> None:
    path = tmp_path / "db.sqlite"
    conn = catalog.connect(path, create=True)
    catalog.apply_migrations(conn, series)
    schema_before = _schema(conn)
    conn.close()

    # 別の接続から再適用しても何も変わらない
    conn = catalog.connect(path)
    try:
        result = catalog.apply_migrations(conn, series)
        latest = catalog.latest_version(series)
        assert result == MigrationResult(latest, latest)
        assert result.applied == 0
        assert _schema(conn) == schema_before
        assert conn.execute("PRAGMA integrity_check").fetchone()[0] == "ok"
    finally:
        conn.close()


def test_reapply_keeps_kinds_rows_and_data(catalog_conn: sqlite3.Connection) -> None:
    catalog_conn.execute(
        "INSERT INTO subjects (subject_id, created_at) VALUES ('S000001', '2026-10-06T00:00:00Z')"
    )
    catalog_conn.commit()
    catalog.apply_migrations(catalog_conn, "catalog")
    assert catalog_conn.execute("SELECT COUNT(*) FROM kinds").fetchone()[0] == 5
    assert catalog_conn.execute("SELECT COUNT(*) FROM subjects").fetchone()[0] == 1


def test_catalog_and_identity_are_separate_series(tmp_path: Path) -> None:
    cat = catalog.connect(tmp_path / "catalog.sqlite", create=True)
    ident = catalog.connect(tmp_path / "identity.sqlite", create=True)
    try:
        catalog.apply_migrations(cat, "catalog")
        catalog.apply_migrations(ident, "identity")
        # 目録に対応表の表は無く，対応表に目録の表は無い
        assert "identity_map" not in _tables(cat)
        assert _tables(ident) == {"identity_map"}
    finally:
        cat.close()
        ident.close()


def test_db_newer_than_library_is_rejected(tmp_path: Path) -> None:
    conn = catalog.connect(tmp_path / "db.sqlite", create=True)
    try:
        catalog.apply_migrations(conn, "catalog")
        newer = catalog.latest_version("catalog") + 1
        conn.execute(f"PRAGMA user_version = {newer}")
        with pytest.raises(ValueError):
            catalog.apply_migrations(conn, "catalog")
        # 版は書き換わらない
        assert catalog.schema_version(conn) == newer
    finally:
        conn.close()


def test_failed_migration_leaves_nothing(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    # 1 ファイル目は正常，2 ファイル目は途中で失敗するマイグレーション
    good = Migration(1, "0001_good.sql", "CREATE TABLE a (x INTEGER);")
    bad = Migration(
        2,
        "0002_bad.sql",
        "CREATE TABLE b (x INTEGER);\nINSERT INTO a VALUES (1);\nTHIS IS NOT SQL;",
    )
    monkeypatch.setattr(migrate, "load_migrations", lambda series: [good, bad])

    conn = catalog.connect(tmp_path / "db.sqlite", create=True)
    try:
        with pytest.raises(sqlite3.Error):
            catalog.apply_migrations(conn, "catalog")
        # 失敗したファイルの変更は残らず，前のファイルまでは適用済み
        assert catalog.schema_version(conn) == 1
        assert _tables(conn) == {"a"}
        assert conn.execute("SELECT COUNT(*) FROM a").fetchone()[0] == 0
        assert not conn.in_transaction
    finally:
        conn.close()

    # 別の接続から見ても同じ（確定した内容だけが残っている）
    conn = catalog.connect(tmp_path / "db.sqlite")
    try:
        assert catalog.schema_version(conn) == 1
        assert _tables(conn) == {"a"}
    finally:
        conn.close()


def test_failed_first_migration_leaves_empty_db(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    bad = Migration(1, "0001_bad.sql", "CREATE TABLE a (x INTEGER);\nTHIS IS NOT SQL;")
    monkeypatch.setattr(migrate, "load_migrations", lambda series: [bad])
    conn = catalog.connect(tmp_path / "db.sqlite", create=True)
    try:
        with pytest.raises(sqlite3.Error):
            catalog.apply_migrations(conn, "catalog")
        assert catalog.schema_version(conn) == 0
        assert _tables(conn) == set()
    finally:
        conn.close()


@pytest.mark.parametrize("series", ["catalog", "identity"])
def test_bundled_migrations_are_numbered_from_one(series: str) -> None:
    migrations = migrate.load_migrations(series)
    assert [m.version for m in migrations] == list(range(1, len(migrations) + 1))
    assert catalog.latest_version(series) == len(migrations)


@pytest.mark.parametrize("series", ["catalog", "identity"])
def test_bundled_migrations_do_not_manage_transaction_or_version(series: str) -> None:
    # apply_migrations() が BEGIN・COMMIT・user_version を付けるため，ファイル側に書かない
    for m in migrate.load_migrations(series):
        code = "\n".join(
            line for line in m.sql.splitlines() if not line.lstrip().startswith("--")
        ).upper()
        assert "USER_VERSION" not in code, m.name
        # トリガ本体の BEGIN … END; は対象外（トランザクションの文だけを見る）
        pattern = (
            r"^\s*(BEGIN(\s+(DEFERRED|IMMEDIATE|EXCLUSIVE))?(\s+TRANSACTION)?"
            r"|COMMIT(\s+TRANSACTION)?|END\s+TRANSACTION|ROLLBACK)\s*;"
        )
        assert re.search(pattern, code, re.M) is None, m.name