from __future__ import annotations

import os
from collections.abc import Iterator
from concurrent.futures import ThreadPoolExecutor
from threading import Barrier
from urllib.parse import unquote, urlparse
from uuid import uuid4

import psycopg
import pytest
from psycopg import sql

from app.core.config import Settings
from app.core.security import hash_token
from app.db.migrations import upgrade_to_head
from app.repositories.postgres import PortfolioDBConfig, PostgresPortfolioRepository
from app.services.auth_service import AuthService

_LOCAL_DSN_ENV = "PORTFOLIOHUB_TEST_POSTGRES_DSN"
_LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1"}


def _maintenance_dsn() -> str:
    dsn = os.environ.get(_LOCAL_DSN_ENV)
    if not dsn:
        pytest.skip(
            f"{_LOCAL_DSN_ENV} is required for real PostgreSQL integration tests"
        )
    parsed = urlparse(dsn)
    if parsed.scheme not in {"postgres", "postgresql"}:
        pytest.fail("test DSN must use postgres/postgresql")
    if parsed.hostname not in _LOOPBACK_HOSTS or parsed.path.lstrip("/") != "postgres":
        pytest.fail(
            "tests require a loopback PostgreSQL maintenance database named postgres"
        )
    return dsn


def _settings(dsn: str, database: str) -> Settings:
    parsed = urlparse(dsn)
    assert parsed.hostname is not None
    return Settings(
        db_enabled=True,
        timescaledb_host=parsed.hostname,
        timescaledb_port=parsed.port or 5432,
        timescaledb_name=database,
        timescaledb_user=unquote(parsed.username or ""),
        timescaledb_password=unquote(parsed.password or ""),
        timescaledb_sslmode="disable",
        timescaledb_connect_timeout_sec=3,
        timescaledb_application_name="service-token-rename-test",
    )


def _repo(settings: Settings) -> PostgresPortfolioRepository:
    return PostgresPortfolioRepository(
        PortfolioDBConfig(
            host=settings.active_db_host,
            port=settings.active_db_port,
            dbname=settings.active_db_name,
            user=settings.active_db_user,
            password=settings.active_db_password,
            sslmode=settings.active_db_sslmode,
            connect_timeout_sec=settings.active_db_connect_timeout_sec,
            application_name=settings.active_db_application_name,
        )
    )


@pytest.fixture()
def postgres_context() -> Iterator[tuple[Settings, PostgresPortfolioRepository]]:
    maintenance_dsn = _maintenance_dsn()
    database = f"portfolio_hub_token_rename_{uuid4().hex}"
    with (
        psycopg.connect(maintenance_dsn, autocommit=True) as conn,
        conn.cursor() as cur,
    ):
        cur.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(database)))
    settings = _settings(maintenance_dsn, database)
    upgrade_to_head(settings)
    repo = _repo(settings)
    repo.create_user("alice", permission="read")
    try:
        yield settings, repo
    finally:
        repo.close()
        with (
            psycopg.connect(maintenance_dsn, autocommit=True) as conn,
            conn.cursor() as cur,
        ):
            cur.execute(
                "SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
                "WHERE datname = %s AND pid <> pg_backend_pid()",
                (database,),
            )
            cur.execute(
                sql.SQL("DROP DATABASE IF EXISTS {}").format(sql.Identifier(database))
            )


def _stored_row(repo: PostgresPortfolioRepository, token_id: str) -> tuple:
    with repo._connect().cursor() as cur:
        cur.execute(
            """
            SELECT id, user_id, name, token_prefix, token_hash, permission, token_type,
                   expires_at, last_used_at, revoked_at, created_at, runtime_id, daemon_id
            FROM auth_tokens
            WHERE id = %s
            """,
            (token_id,),
        )
        row = cur.fetchone()
    assert row is not None
    return row


def test_rename_persists_only_name_and_preserves_bearer_authentication(
    postgres_context,
) -> None:
    settings, repo = postgres_context
    service = AuthService(repo, settings)
    issued = service.issue_service_token("old-name", "alice")
    before = _stored_row(repo, issued.auth_token.id)

    renamed = service.rename_service_token("old-name", "new-name")
    after = _stored_row(repo, issued.auth_token.id)

    assert renamed.changed is True
    assert renamed.auth_token.id == issued.auth_token.id
    assert after[2] == "new-name"
    assert before[:2] + before[3:] == after[:2] + after[3:]
    assert repo.get_service_token_by_name("old-name") is None
    assert repo.get_service_token_by_name("new-name").id == issued.auth_token.id

    principal = repo.get_token_principal(hash_token(issued.token))
    assert principal is not None
    assert principal.user_id == issued.auth_token.user_id
    assert principal.permission == issued.auth_token.permission
    assert principal.runtime_id == issued.auth_token.runtime_id
    assert principal.daemon_id == issued.auth_token.daemon_id


def test_rename_collision_rolls_back_and_revoked_name_is_reusable(
    postgres_context,
) -> None:
    settings, repo = postgres_context
    service = AuthService(repo, settings)
    source = service.issue_service_token("source", "alice")
    target = service.issue_service_token("target", "alice")
    source_before = _stored_row(repo, source.auth_token.id)
    target_before = _stored_row(repo, target.auth_token.id)

    with pytest.raises(
        ValueError, match="이미 존재하는 서비스 토큰 이름입니다: target"
    ):
        service.rename_service_token("source", "target")

    assert _stored_row(repo, source.auth_token.id) == source_before
    assert _stored_row(repo, target.auth_token.id) == target_before

    repo.revoke_auth_token(target.auth_token.user_id, target.auth_token.id)
    renamed = service.rename_service_token("source", "target")
    assert renamed.changed is True
    assert renamed.auth_token.id == source.auth_token.id


def test_same_name_is_noop_and_missing_same_name_fails(postgres_context) -> None:
    settings, repo = postgres_context
    service = AuthService(repo, settings)
    issued = service.issue_service_token("same-name", "alice")
    before = _stored_row(repo, issued.auth_token.id)

    result = service.rename_service_token("same-name", "same-name")

    assert result.changed is False
    assert _stored_row(repo, issued.auth_token.id) == before
    with pytest.raises(ValueError, match="서비스 토큰을 찾을 수 없습니다: missing"):
        service.rename_service_token("missing", "missing")

    revoked = service.issue_service_token("revoked-source", "alice")
    repo.revoke_auth_token(revoked.auth_token.user_id, revoked.auth_token.id)
    revoked_before = _stored_row(repo, revoked.auth_token.id)
    with pytest.raises(
        ValueError, match="서비스 토큰을 찾을 수 없습니다: revoked-source"
    ):
        service.rename_service_token("revoked-source", "new-name")
    assert _stored_row(repo, revoked.auth_token.id) == revoked_before


def test_concurrent_renames_to_same_target_have_one_winner(postgres_context) -> None:
    settings, repo = postgres_context
    service = AuthService(repo, settings)
    first = service.issue_service_token("race-first", "alice")
    second = service.issue_service_token("race-second", "alice")
    barrier = Barrier(2)

    def worker(source: str) -> str:
        worker_repo = _repo(settings)
        try:
            barrier.wait()
            AuthService(worker_repo, settings).rename_service_token(
                source, "race-target"
            )
            return source
        except ValueError as exc:
            return str(exc)
        finally:
            worker_repo.close()

    with ThreadPoolExecutor(max_workers=2) as executor:
        futures = [
            executor.submit(worker, "race-first"),
            executor.submit(worker, "race-second"),
        ]
        results = [future.result(timeout=15) for future in futures]

    winners = [result for result in results if result in {"race-first", "race-second"}]
    conflicts = [
        result
        for result in results
        if result == "이미 존재하는 서비스 토큰 이름입니다: race-target"
    ]
    assert len(winners) == 1
    assert len(conflicts) == 1
    active_target = repo.get_service_token_by_name("race-target")
    assert active_target is not None
    expected_id = (
        first.auth_token.id if winners == ["race-first"] else second.auth_token.id
    )
    assert active_target.id == expected_id


def test_concurrent_rename_and_issue_have_one_winner(postgres_context) -> None:
    settings, repo = postgres_context
    service = AuthService(repo, settings)
    source = service.issue_service_token("race-source", "alice")
    barrier = Barrier(2)

    def rename_worker() -> str:
        worker_repo = _repo(settings)
        try:
            barrier.wait()
            AuthService(worker_repo, settings).rename_service_token(
                "race-source", "race-target"
            )
            return "renamed"
        except ValueError as exc:
            return str(exc)
        finally:
            worker_repo.close()

    def issue_worker() -> str:
        worker_repo = _repo(settings)
        try:
            barrier.wait()
            AuthService(worker_repo, settings).issue_service_token(
                "race-target", "alice"
            )
            return "issued"
        except ValueError as exc:
            return str(exc)
        finally:
            worker_repo.close()

    with ThreadPoolExecutor(max_workers=2) as executor:
        outcomes = [executor.submit(rename_worker), executor.submit(issue_worker)]
        results = [future.result(timeout=15) for future in outcomes]

    winners = [result for result in results if result in {"renamed", "issued"}]
    conflicts = [
        result
        for result in results
        if result == "이미 존재하는 서비스 토큰 이름입니다: race-target"
    ]
    assert len(winners) == 1
    assert len(conflicts) == 1
    active_target = repo.get_service_token_by_name("race-target")
    assert active_target is not None
    if winners == ["renamed"]:
        assert active_target.id == source.auth_token.id
        assert repo.get_service_token_by_name("race-source") is None
    else:
        assert active_target.id != source.auth_token.id
        assert repo.get_service_token_by_name("race-source").id == source.auth_token.id