"""Real-PostgreSQL persistence and concurrency verification for daemon assignment (Issue #27 / dcy-0004).

Run with PORTFOLIOHUB_RUN_POSTGRES_INTEGRATION=1 against a disposable/local
PostgreSQL endpoint. Normal unit-test environments skip this module.
"""

from __future__ import annotations

import os
import threading
import uuid
from contextlib import ExitStack
from unittest.mock import patch

import pytest
from fastapi.testclient import TestClient

from app.core.config import Settings, get_settings
from app.domain.errors import DaemonInactiveError, DaemonInUseError, DaemonNotFoundError
from app.main import create_app
from app.repositories.postgres import PortfolioDBConfig, PostgresPortfolioRepository

pytestmark = pytest.mark.skipif(
    os.getenv("PORTFOLIOHUB_RUN_POSTGRES_INTEGRATION") != "1",
    reason="set PORTFOLIOHUB_RUN_POSTGRES_INTEGRATION=1 to run real PostgreSQL tests",
)


def _db_values(dbname: str | None = None) -> dict[str, object]:
    return {
        "host": os.getenv("TIMESCALEDB_HOST", "127.0.0.1"),
        "port": int(os.getenv("TIMESCALEDB_PORT", "5432")),
        "dbname": dbname or os.getenv("TIMESCALEDB_NAME", "portfolio_hub_migration_test"),
        "user": os.getenv("TIMESCALEDB_USER", "portfolio_hub_user"),
        "password": os.getenv("TIMESCALEDB_PASSWORD", "change-me"),
        "sslmode": os.getenv("TIMESCALEDB_SSLMODE", "prefer"),
    }


def _repo(dbname: str | None = None) -> PostgresPortfolioRepository:
    values = _db_values(dbname)
    return PostgresPortfolioRepository(
        PortfolioDBConfig(
            host=str(values["host"]),
            port=int(values["port"]),
            dbname=str(values["dbname"]),
            user=str(values["user"]),
            password=str(values["password"]),
            sslmode=str(values["sslmode"]),
        )
    )


def _prepare_daemon_fixture(repo: PostgresPortfolioRepository) -> tuple[str, str, str, str]:
    repo.ensure_schema()
    strategy = repo.create_strategy_set(f"daemon-{uuid.uuid4().hex}", "{}")
    asset_id = str(uuid.uuid4())
    owner_user_id = f"daemon-owner-{uuid.uuid4().hex}"
    with repo._connect().cursor() as cur:
        cur.execute(
            "INSERT INTO users (id, username, permission, is_active) VALUES (%s, %s, 'trade-write', TRUE)",
            (owner_user_id, owner_user_id),
        )
        cur.execute(
            """
            INSERT INTO assets (asset_id, symbol, markets, status)
            VALUES (%s, %s, ARRAY['KRX'], 'active')
            """,
            (asset_id, f"DA{uuid.uuid4().hex[:10]}"),
        )
    daemon = repo.create_daemon(owner_user_id, daemon_slug=f"d-{uuid.uuid4().hex[:10]}")
    return strategy.id, asset_id, owner_user_id, daemon.id


def _cleanup(
    repo: PostgresPortfolioRepository,
    portfolio_ids: list[str],
    strategy_id: str,
    asset_id: str,
    owner_user_id: str,
    daemon_id: str,
    extra_user_ids: list[str] | None = None,
) -> None:
    with repo._connect().cursor() as cur:
        for portfolio_id in portfolio_ids:
            cur.execute("DELETE FROM portfolio_ledger WHERE portfolio_id = %s", (portfolio_id,))
            cur.execute("DELETE FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
        cur.execute("DELETE FROM auth_tokens WHERE daemon_id = %s", (daemon_id,))
        cur.execute("DELETE FROM daemons WHERE id = %s", (daemon_id,))
        cur.execute("DELETE FROM strategy_sets WHERE id = %s", (strategy_id,))
        cur.execute("DELETE FROM assets WHERE asset_id = %s", (asset_id,))
        cur.execute("DELETE FROM users WHERE id = %s", (owner_user_id,))
        for user_id in extra_user_ids or []:
            cur.execute("DELETE FROM users WHERE id = %s", (user_id,))
    repo.close()


def _make_client() -> TestClient:
    get_settings.cache_clear()
    settings = Settings(
        allowed_origins_raw="http://localhost:3000",
        admin_tokens_raw="admin-tok",
        trade_write_tokens_raw="user-tok",
    )
    app = create_app(settings_factory=lambda: settings)
    app.dependency_overrides[get_settings] = lambda: settings
    return TestClient(app)


def _with_repo(repo: PostgresPortfolioRepository) -> ExitStack:
    stack = ExitStack()
    stack.enter_context(
        patch("app.api.v1_portfolios.get_portfolio_repository_from_settings", return_value=repo)
    )
    return stack


def _auth(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def test_http_post_returns_daemon_and_list_get_preserves_it() -> None:
    repo = _repo()
    strategy_id, asset_id, owner_user_id, daemon_id = _prepare_daemon_fixture(repo)
    client = _make_client()

    try:
        with _with_repo(repo):
            resp = client.post(
                "/v1/portfolios",
                json={
                    "asset_id": asset_id,
                    "broker": "kis",
                    "strategy_id": strategy_id,
                    "name": "daemon-assignment-check",
                    "user_id": owner_user_id,
                    "managed_by_daemon_id": daemon_id,
                },
                headers=_auth("admin-tok"),
            )
        assert resp.status_code == 201
        body = resp.json()
        assert body["managed_by_daemon_id"] == daemon_id
        portfolio_id = body["portfolio_id"]

        stored = repo.get(portfolio_id)
        assert stored.managed_by_daemon_id == daemon_id

        with _with_repo(repo):
            list_resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
        assert list_resp.status_code == 200
        matched = [p for p in list_resp.json()["portfolios"] if p["portfolio_id"] == portfolio_id]
        assert matched and matched[0]["managed_by_daemon_id"] == daemon_id

        _cleanup(repo, [portfolio_id], strategy_id, asset_id, owner_user_id, daemon_id)
    finally:
        pass


def test_validation_failure_leaves_no_partial_portfolio() -> None:
    repo = _repo()
    strategy_id, asset_id, owner_user_id, daemon_id = _prepare_daemon_fixture(repo)
    portfolio_id = str(uuid.uuid4())
    inactive = _repo()
    try:
        inactive.update_daemon(daemon_id, is_active=False)

        with pytest.raises(DaemonInactiveError):
            repo.create_portfolio(
                portfolio_id,
                owner_user_id=owner_user_id,
                broker="kis",
                status="enabled",
                name="no-partial-inactive",
                asset_id=asset_id,
                cash_balance=0,
                asset_quantity=0,
                asset_price=0,
                profit_buffer=0,
                active_strategy_id=strategy_id,
                avabile_strategies=[strategy_id],
                managed_by_daemon_id=daemon_id,
            )
        with repo._connect().cursor() as cur:
            cur.execute("SELECT COUNT(*) FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
            assert cur.fetchone()[0] == 0

        foreign_user_id = f"foreign-owner-{uuid.uuid4().hex}"
        with repo._connect().cursor() as cur:
            cur.execute(
                "INSERT INTO users (id, username, permission, is_active) VALUES (%s, %s, 'trade-write', TRUE)",
                (foreign_user_id, foreign_user_id),
            )
        foreign_daemon = repo.create_daemon(foreign_user_id, daemon_slug=f"foreign-{uuid.uuid4().hex[:10]}")
        with pytest.raises(DaemonNotFoundError):
            repo.create_portfolio(
                portfolio_id,
                owner_user_id=owner_user_id,
                broker="kis",
                status="enabled",
                name="no-partial-foreign",
                asset_id=asset_id,
                cash_balance=0,
                asset_quantity=0,
                asset_price=0,
                profit_buffer=0,
                active_strategy_id=strategy_id,
                avabile_strategies=[strategy_id],
                managed_by_daemon_id=foreign_daemon.id,
            )
        with repo._connect().cursor() as cur:
            cur.execute("SELECT COUNT(*) FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
            assert cur.fetchone()[0] == 0

        with repo._connect().cursor() as cur:
            cur.execute("DELETE FROM auth_tokens WHERE daemon_id = %s", (foreign_daemon.id,))
            cur.execute("DELETE FROM daemons WHERE id = %s", (foreign_daemon.id,))
            cur.execute("DELETE FROM users WHERE id = %s", (foreign_user_id,))
        _cleanup(repo, [], strategy_id, asset_id, owner_user_id, daemon_id)
    finally:
        inactive.close()


def test_assignment_and_deactivation_race_never_commits_inactive_assignment() -> None:
    repo = _repo()
    strategy_id, asset_id, owner_user_id, daemon_id = _prepare_daemon_fixture(repo)
    portfolio_id = str(uuid.uuid4())
    barrier = threading.Barrier(3)
    outcomes: list[str] = []

    def assign_daemon() -> None:
        worker = _repo()
        try:
            barrier.wait()
            worker.create_portfolio(
                portfolio_id,
                owner_user_id=owner_user_id,
                broker="kis",
                status="enabled",
                name="daemon-race-check",
                asset_id=asset_id,
                cash_balance=0,
                asset_quantity=0,
                asset_price=0,
                profit_buffer=0,
                active_strategy_id=strategy_id,
                avabile_strategies=[strategy_id],
                managed_by_daemon_id=daemon_id,
            )
            outcomes.append("assigned")
        except DaemonInactiveError:
            outcomes.append("assignment-rejected")
        finally:
            worker.close()

    def deactivate_daemon() -> None:
        worker = _repo()
        try:
            barrier.wait()
            worker.update_daemon(daemon_id, is_active=False)
            outcomes.append("deactivated")
        finally:
            worker.close()

    threads = [threading.Thread(target=assign_daemon), threading.Thread(target=deactivate_daemon)]
    for thread in threads:
        thread.start()
    barrier.wait()
    for thread in threads:
        thread.join(timeout=10)
        assert not thread.is_alive()

    assert sorted(outcomes) in (
        ["assigned", "deactivated"],
        ["assignment-rejected", "deactivated"],
    )
    with repo._connect().cursor() as cur:
        cur.execute("SELECT COUNT(*) FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
        portfolio_exists = cur.fetchone()[0] == 1
    if portfolio_exists:
        assert repo.get(portfolio_id).managed_by_daemon_id == daemon_id

    with repo._connect().cursor() as cur:
        if portfolio_exists:
            cur.execute("DELETE FROM portfolio_ledger WHERE portfolio_id = %s", (portfolio_id,))
            cur.execute("DELETE FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
        cur.execute("DELETE FROM auth_tokens WHERE daemon_id = %s", (daemon_id,))
        cur.execute("DELETE FROM daemons WHERE id = %s", (daemon_id,))
        cur.execute("DELETE FROM strategy_sets WHERE id = %s", (strategy_id,))
        cur.execute("DELETE FROM assets WHERE asset_id = %s", (asset_id,))
        cur.execute("DELETE FROM users WHERE id = %s", (owner_user_id,))
    repo.close()


def test_patch_reassignment_persists_new_daemon() -> None:
    repo = _repo()
    strategy_id, asset_id, owner_user_id, daemon_a = _prepare_daemon_fixture(repo)
    daemon_b = repo.create_daemon(owner_user_id, daemon_slug=f"d-{uuid.uuid4().hex[:10]}")
    portfolio_id = str(uuid.uuid4())
    try:
        repo.create_portfolio(
            portfolio_id,
            owner_user_id=owner_user_id,
            broker="kis",
            status="enabled",
            name="reassign-check",
            asset_id=asset_id,
            cash_balance=0,
            asset_quantity=0,
            asset_price=0,
            profit_buffer=0,
            active_strategy_id=strategy_id,
            avabile_strategies=[strategy_id],
            managed_by_daemon_id=daemon_a,
        )
        assert repo.get(portfolio_id).managed_by_daemon_id == daemon_a

        repo.update_portfolio(portfolio_id, managed_by_daemon_id=daemon_b.id)
        assert repo.get(portfolio_id).managed_by_daemon_id == daemon_b.id

        with repo._connect().cursor() as cur:
            cur.execute("DELETE FROM portfolio_ledger WHERE portfolio_id = %s", (portfolio_id,))
            cur.execute("DELETE FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
            cur.execute("DELETE FROM auth_tokens WHERE daemon_id = %s", (daemon_b.id,))
            cur.execute("DELETE FROM daemons WHERE id = %s", (daemon_b.id,))
        _cleanup(repo, [], strategy_id, asset_id, owner_user_id, daemon_a)
    finally:
        pass


def test_referenced_daemon_owner_transfer_is_atomic() -> None:
    repo = _repo()
    strategy_id, asset_id, owner_a, daemon_id = _prepare_daemon_fixture(repo)
    owner_b = f"daemon-owner-{uuid.uuid4().hex}"
    portfolio_id = str(uuid.uuid4())
    try:
        with repo._connect().cursor() as cur:
            cur.execute(
                "INSERT INTO users (id, username, permission, is_active) VALUES (%s, %s, 'trade-write', TRUE)",
                (owner_b, owner_b),
            )
        repo.upsert_broker_credential(
            owner_a,
            "kis",
            account_ref="owner-a-account",
            encrypted_payload="ciphertext",
            daemon_id=daemon_id,
        )
        repo.create_portfolio(
            portfolio_id,
            owner_user_id=owner_a,
            broker="kis",
            status="enabled",
            name="owner-transfer-blocked",
            asset_id=asset_id,
            cash_balance=0,
            asset_quantity=0,
            asset_price=0,
            profit_buffer=0,
            active_strategy_id=strategy_id,
            avabile_strategies=[strategy_id],
            managed_by_daemon_id=daemon_id,
        )

        with pytest.raises(DaemonInUseError):
            repo.update_daemon_owner(daemon_id, owner_user_id=owner_b)

        assert repo.get_daemon_by_id(daemon_id).owner_user_id == owner_a
        assert repo.get(portfolio_id).managed_by_daemon_id == daemon_id
        credential = repo.get_broker_credential_by_daemon(daemon_id, "kis")
        assert credential is not None
        assert credential.user_id == owner_a

        _cleanup(repo, [portfolio_id], strategy_id, asset_id, owner_a, daemon_id, [owner_b])
    finally:
        pass


def test_unreferenced_daemon_owner_transfer_invalidates_credentials() -> None:
    repo = _repo()
    strategy_id, asset_id, owner_a, daemon_id = _prepare_daemon_fixture(repo)
    owner_b = f"daemon-owner-{uuid.uuid4().hex}"
    try:
        with repo._connect().cursor() as cur:
            cur.execute(
                "INSERT INTO users (id, username, permission, is_active) VALUES (%s, %s, 'trade-write', TRUE)",
                (owner_b, owner_b),
            )
        repo.upsert_broker_credential(
            owner_a,
            "kis",
            account_ref="owner-a-account",
            encrypted_payload="ciphertext",
            daemon_id=daemon_id,
        )

        moved = repo.update_daemon_owner(daemon_id, owner_user_id=owner_b)

        assert moved.owner_user_id == owner_b
        assert repo.get_broker_credential_by_daemon(daemon_id, "kis") is None

        _cleanup(repo, [], strategy_id, asset_id, owner_a, daemon_id, [owner_b])
    finally:
        pass


def test_assignment_and_owner_transfer_race_preserves_same_owner_invariant() -> None:
    repo = _repo()
    strategy_id, asset_id, owner_a, daemon_id = _prepare_daemon_fixture(repo)
    owner_b = f"daemon-owner-{uuid.uuid4().hex}"
    portfolio_id = str(uuid.uuid4())
    barrier = threading.Barrier(3)
    outcomes: list[str] = []
    errors: list[BaseException] = []
    with repo._connect().cursor() as cur:
        cur.execute(
            "INSERT INTO users (id, username, permission, is_active) VALUES (%s, %s, 'trade-write', TRUE)",
            (owner_b, owner_b),
        )

    def assign_daemon() -> None:
        worker = _repo()
        try:
            barrier.wait()
            worker.create_portfolio(
                portfolio_id,
                owner_user_id=owner_a,
                broker="kis",
                status="enabled",
                name="owner-transfer-race",
                asset_id=asset_id,
                cash_balance=0,
                asset_quantity=0,
                asset_price=0,
                profit_buffer=0,
                active_strategy_id=strategy_id,
                avabile_strategies=[strategy_id],
                managed_by_daemon_id=daemon_id,
            )
            outcomes.append("assigned")
        except DaemonNotFoundError:
            outcomes.append("assignment-rejected")
        except BaseException as exc:  # pragma: no cover - populated only on failure
            errors.append(exc)
        finally:
            worker.close()

    def transfer_owner() -> None:
        worker = _repo()
        try:
            barrier.wait()
            worker.update_daemon_owner(daemon_id, owner_user_id=owner_b)
            outcomes.append("transferred")
        except DaemonInUseError:
            outcomes.append("transfer-rejected")
        except BaseException as exc:  # pragma: no cover - populated only on failure
            errors.append(exc)
        finally:
            worker.close()

    threads = [threading.Thread(target=assign_daemon), threading.Thread(target=transfer_owner)]
    for thread in threads:
        thread.start()
    barrier.wait()
    for thread in threads:
        thread.join(timeout=10)
        assert not thread.is_alive()

    assert errors == []
    assert sorted(outcomes) in (
        ["assigned", "transfer-rejected"],
        ["assignment-rejected", "transferred"],
    )
    daemon = repo.get_daemon_by_id(daemon_id)
    portfolio = repo.get(portfolio_id) if "assigned" in outcomes else None
    if portfolio is not None:
        assert daemon.owner_user_id == portfolio.owner_user_id == owner_a
        assert portfolio.managed_by_daemon_id == daemon_id
    else:
        assert daemon.owner_user_id == owner_b

    _cleanup(repo, [portfolio_id], strategy_id, asset_id, owner_a, daemon_id, [owner_b])


def test_deactivation_preserves_assignment_and_revokes_bound_service_token() -> None:
    repo = _repo()
    strategy_id, asset_id, owner_user_id, daemon_id = _prepare_daemon_fixture(repo)
    portfolio_id = str(uuid.uuid4())
    try:
        repo.create_portfolio(
            portfolio_id,
            owner_user_id=owner_user_id,
            broker="kis",
            status="enabled",
            name="deactivation-preserves-reference",
            asset_id=asset_id,
            cash_balance=0,
            asset_quantity=0,
            asset_price=0,
            profit_buffer=0,
            active_strategy_id=strategy_id,
            avabile_strategies=[strategy_id],
            managed_by_daemon_id=daemon_id,
        )
        token = repo.create_auth_token(
            owner_user_id,
            name=f"daemon-token-{uuid.uuid4().hex}",
            token_hash=uuid.uuid4().hex,
            token_prefix="pfh",
            permission="trade-write",
            token_type="service",
            daemon_id=daemon_id,
        )

        repo.update_daemon(daemon_id, is_active=False)
        assert repo.revoke_tokens_by_daemon_id(daemon_id) == 1

        assert repo.get(portfolio_id).managed_by_daemon_id == daemon_id
        with repo._connect().cursor() as cur:
            cur.execute("SELECT revoked_at IS NOT NULL FROM auth_tokens WHERE id = %s", (token.id,))
            assert cur.fetchone()[0] is True

        _cleanup(repo, [portfolio_id], strategy_id, asset_id, owner_user_id, daemon_id)
    finally:
        pass