import uuid
from concurrent.futures import ThreadPoolExecutor
from datetime import UTC, datetime
from threading import Barrier

import pytest
from sqlalchemy import create_engine, text
from testcontainers.postgres import PostgresContainer

from tht.decisions import DecisionInput
from tht.phase import effective_decisions
from tht.session.models import PrincipalContext, SessionManifest
from tht.session.store import SessionError


@pytest.fixture(scope="module")
def database_url():
    with PostgresContainer("postgres:16-alpine") as postgres:
        yield postgres.get_connection_url()


@pytest.fixture(scope="module", autouse=True)
def migrated(database_url):
    from tht.session.postgres_repository import migrate

    migrate(database_url)


def _manifest(session_id: str) -> SessionManifest:
    return SessionManifest(
        id=session_id,
        created_at=datetime(2026, 7, 16, 10, 0, tzinfo=UTC),
        question="Which patients had an ablation?",
        database="testdb",
        schema="public",
    )


def _repository(database_url, subject: str, *, is_admin: bool = False):
    from tht.session.postgres_repository import PostgresSessionRepository

    return PostgresSessionRepository(
        database_url,
        PrincipalContext(issuer="portal", subject=subject, is_admin=is_admin),
    )


def test_owner_can_read_own_snapshot_but_not_another_owners(database_url):
    alice = _repository(database_url, "alice")
    bob = _repository(database_url, "bob")
    session_id = str(uuid.uuid4())

    alice.create(_manifest(session_id))
    alice.write_artifact(session_id, "cte_sql:eligible_patients", "SELECT 1")
    alice.append_decisions(
        session_id, [DecisionInput(type="sql_approved", subject="phase:7", detail="SELECT 1")]
    )

    snapshot = alice.get(session_id)
    assert snapshot.principal == alice.principal
    assert snapshot.artifacts == {"cte_sql:eligible_patients": "SELECT 1"}
    assert snapshot.decisions[0].seq == 1
    with pytest.raises(SessionError, match="Sessione non trovata"):
        bob.get(session_id)


def test_legacy_interaction_language_is_atomic_idempotent_and_owner_scoped(database_url):
    session_id = str(uuid.uuid4())
    owner = _repository(database_url, "language-owner")
    foreign = _repository(database_url, "language-foreign")
    owner.create(_manifest(session_id))
    owner.write_artifact(session_id, "sql_final", "SELECT 'paziente' AS nome")
    with pytest.raises(SessionError, match="Sessione non trovata"):
        foreign.ensure_interaction_language(session_id, "en")

    barrier = Barrier(2)

    def pin(language):
        barrier.wait()
        return owner.ensure_interaction_language(session_id, language).manifest.interaction_language

    with ThreadPoolExecutor(max_workers=2) as pool:
        results = list(pool.map(pin, ["it", "en"]))
    assert results[0] == results[1]
    saved = owner.get(session_id)
    repeated = owner.ensure_interaction_language(session_id, "fr")
    assert repeated == saved
    assert saved.artifacts["sql_final"] == "SELECT 'paziente' AS nome"


def test_cli_creation_and_resume_use_postgres_language_authority(database_url, monkeypatch, tmp_path):
    import json
    from types import SimpleNamespace

    from sqlalchemy.engine import make_url
    from typer.testing import CliRunner

    from tht.cli import session_cmd
    from tht.config import DatabaseConfig

    url = make_url(database_url)
    connection = SimpleNamespace(user=url.username, password=url.password, host=url.host,
                                 port=url.port, database=url.database, sslmode="disable",
                                 sslrootcert=None)
    cfg = SimpleNamespace(language="it", session_storage=SimpleNamespace(connection=connection),
                          database=DatabaseConfig(database="testdb", user="u", password="p",
                                                  schema="public"),
                          paths=SimpleNamespace(sessions=tmp_path / "unused"))
    monkeypatch.setattr(session_cmd, "_load_config_or_exit", lambda _: cfg)
    monkeypatch.setenv("THT_PRINCIPAL_ISSUER", "portal")
    monkeypatch.setenv("THT_PRINCIPAL_SUBJECT", "cli-language")

    def invoke(*args):
        result = CliRunner().invoke(session_cmd.session_app, list(args))
        assert result.exit_code == 0, result.output
        return result.stdout

    session_id = json.loads(invoke("new", "Pazienti", "--interaction-language", "en", "--json"))["id"]
    invoke("reopen", session_id)
    assert json.loads(invoke("show", session_id, "--json"))["interaction_language"] == "en"
    legacy = _manifest(str(uuid.uuid4()))
    _repository(database_url, "cli-language").create(legacy)
    invoke("reopen", legacy.id)
    cfg.language = "en"
    invoke("reopen", legacy.id)
    assert json.loads(invoke("show", legacy.id, "--json"))["interaction_language"] == "it"
    assert not cfg.paths.sessions.exists()


def test_named_decision_is_stale_after_postgres_ledger_reopen(database_url):
    repository = _repository(database_url, "phase-owner")
    session_id = str(uuid.uuid4())
    repository.create(_manifest(session_id))
    for phase in range(1, 5):
        repository.append_decisions(session_id, [
            DecisionInput(type="phase_approved", subject=f"phase:{phase}")
        ])
    repository.append_decisions(session_id, [
        DecisionInput(type="cte_approved", subject="named_cte")
    ])
    repository.append_decisions(session_id, [
        DecisionInput(type="phase_reopened", subject="phase:3")
    ])

    snapshot = repository.get(session_id)
    named = next(item for item in snapshot.decisions if item.subject == "named_cte")
    assert named.phase == 5
    assert "named_cte" not in {item.subject for item in effective_decisions(snapshot)}


def test_admin_can_read_another_owners_session(database_url):
    owner = _repository(database_url, "owner")
    admin = _repository(database_url, "admin", is_admin=True)
    session_id = str(uuid.uuid4())
    owner.create(_manifest(session_id))

    assert admin.get(session_id).manifest.id == session_id


def test_owner_admin_and_foreign_principals_have_distinct_mutation_boundaries(database_url):
    session_id = str(uuid.uuid4())
    alice = _repository(database_url, f"alice-{session_id}")
    bob = _repository(database_url, f"bob-{session_id}")
    admin = _repository(database_url, f"admin-{session_id}", is_admin=True)
    manifest = _manifest(session_id)
    alice.create(manifest)
    alice.write_artifact(session_id, "question", "Alice's question")

    assert [snapshot.manifest.id for snapshot in alice.list()] == [session_id]
    assert bob.list() == []
    for mutation in (
        lambda: bob.write_artifact(session_id, "question", "Bob's overwrite"),
        lambda: bob.append_decisions(
            session_id, [DecisionInput(type="concept_clarified", subject="bob")]
        ),
        lambda: bob.save_manifest(manifest.model_copy(update={"name": "Bob's rename"})),
        lambda: bob.delete(session_id),
    ):
        with pytest.raises(SessionError, match="Sessione non trovata"):
            mutation()

    # Admin access is an explicit, audited exception to normal owner isolation.
    admin.write_artifact(session_id, "question", "Reviewed by admin")
    admin.append_decisions(
        session_id, [DecisionInput(type="concept_clarified", subject="admin-review")]
    )

    owner_snapshot = alice.get(session_id)
    assert owner_snapshot.artifacts["question"] == "Reviewed by admin"
    assert [record.subject for record in owner_snapshot.decisions] == ["admin-review"]
    assert admin.get(session_id).manifest.id == session_id


def test_concurrent_postgres_ledger_mutations_keep_every_decision_in_sequence(database_url, monkeypatch):
    from tht.session.postgres_repository import PostgresSessionRepository

    repository = _repository(database_url, "alice")
    session_id = str(uuid.uuid4())
    repository.create(_manifest(session_id))
    barrier = Barrier(2)
    original_lock = PostgresSessionRepository._lock_session

    def enter_lock_together(connection, locked_session_id):
        if locked_session_id == session_id:
            barrier.wait(timeout=5)
        original_lock(connection, locked_session_id)

    monkeypatch.setattr(
        PostgresSessionRepository,
        "_lock_session",
        staticmethod(enter_lock_together),
    )

    def append(subject: str):
        return repository.append_decisions(
            session_id, [DecisionInput(type="concept_clarified", subject=subject)]
        )

    with ThreadPoolExecutor(max_workers=2) as pool:
        first, second = pool.map(append, ("first", "second"))

    snapshot = repository.get(session_id)
    assert {first[0].seq, second[0].seq} == {1, 2}
    assert [record.seq for record in snapshot.decisions] == [1, 2]
    assert {record.subject for record in snapshot.decisions} == {"first", "second"}


def test_non_superuser_runtime_login_can_assume_the_restricted_runtime_role(database_url):
    admin = create_engine(database_url)
    runtime_url = admin.url.set(
        username="thoth_sessions_test_login", password="runtime-test-only"
    )
    try:
        with admin.begin() as connection:
            connection.exec_driver_sql(
                "CREATE ROLE thoth_sessions_test_login LOGIN NOINHERIT PASSWORD 'runtime-test-only'"
            )
            connection.exec_driver_sql("GRANT thoth_sessions_runtime TO thoth_sessions_test_login")
        repository = _repository(
            runtime_url.render_as_string(hide_password=False), "runtime-user"
        )
        session_id = str(uuid.uuid4())

        repository.create(_manifest(session_id))

        assert repository.get(session_id).manifest.id == session_id
        repository.close()
    finally:
        with admin.begin() as connection:
            connection.exec_driver_sql("DROP ROLE IF EXISTS thoth_sessions_test_login")
        admin.dispose()


def test_delete_cascades_content_and_leaves_content_free_tombstone(database_url):
    repository = _repository(database_url, "alice")
    session_id = str(uuid.uuid4())
    repository.create(_manifest(session_id))
    repository.write_artifact(session_id, "sql_final", "SELECT confidential_value")
    repository.append_decisions(
        session_id,
        [DecisionInput(type="sql_approved", subject="phase:7", detail="SELECT confidential_value")],
    )

    repository.delete(session_id)

    with pytest.raises(SessionError, match="Sessione non trovata"):
        repository.get(session_id)
    engine = create_engine(database_url)
    try:
        with engine.connect() as connection:
            assert connection.execute(
                text("SELECT count(*) FROM thoth_sessions.session_artifacts WHERE session_id = :session_id"),
                {"session_id": session_id},
            ).scalar_one() == 0
            assert connection.execute(
                text("SELECT count(*) FROM thoth_sessions.review_decisions WHERE session_id = :session_id"),
                {"session_id": session_id},
            ).scalar_one() == 0
            columns = connection.execute(
                text(
                    "SELECT column_name FROM information_schema.columns "
                    "WHERE table_schema = 'thoth_sessions' AND table_name = 'audit_log'"
                )
            ).scalars().all()
            tombstone = connection.execute(
                text(
                    "SELECT action, session_id, actor_issuer, actor_subject "
                    "FROM thoth_sessions.audit_log WHERE session_id = :session_id"
                ),
                {"session_id": session_id},
            ).one()
    finally:
        engine.dispose()
    assert (tombstone[0], str(tombstone[1]), *tombstone[2:]) == (
        "session_deleted",
        session_id,
        "portal",
        "alice",
    )
    assert not {"content", "artifact_content", "detail", "metadata"} & set(columns)


def test_session_schema_does_not_create_or_invoke_embeddings(database_url, monkeypatch):
    import tht.session.postgres_repository as repository_module

    monkeypatch.setattr(
        repository_module,
        "_embed",
        lambda *_: pytest.fail("session persistence must not invoke embeddings"),
        raising=False,
    )
    repository = _repository(database_url, "alice")
    session_id = str(uuid.uuid4())
    repository.create(_manifest(session_id))
    repository.write_artifact(session_id, "evidence", '{"sources": []}')

    engine = create_engine(database_url)
    try:
        with engine.connect() as connection:
            columns = connection.execute(
                text(
                    "SELECT column_name FROM information_schema.columns "
                    "WHERE table_schema = 'thoth_sessions'"
                )
            ).scalars().all()
    finally:
        engine.dispose()
    assert all("embedding" not in column for column in columns)