from __future__ import annotations

import os
from types import SimpleNamespace

import pytest
import pytest_asyncio
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from sqlalchemy import JSON, func, select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://tester:tester@localhost:5432/pmflow_test")

from app.api import pmflow_os_api  # noqa: E402
from app.core.database import get_db  # noqa: E402
from app.dependencies.auth import get_current_user  # noqa: E402
from app.models import (  # noqa: E402
    DevWorkspaceBinding,
    FeedbackIngest,
    FeishuItem,
    GitHubWebhookDelivery,
    ModelCatalogItem,
    ModelEvalRun,
    ModelRecommendationSnapshot,
    PMFlowApproval,
    PMFlowChangeItem,
    PMFlowChangeSet,
    PMFlowEvidenceLink,
    PMFlowLaunchDraft,
    PMFlowThread,
    PMFlowThreadMessage,
    ProgressEvent,
    Project,
    ProjectModelBinding,
    PromptPackage,
    ReleaseValidationRun,
    RepoBinding,
    RevenueEvent,
    ROISnapshot,
    SemanticFact,
    Signal,
    Task,
    TaskCodeLink,
    TruthNode,
    User,
    UserProject,
)
from app.models.base import Base  # noqa: E402


TABLES = [
    User.__table__,
    FeishuItem.__table__,
    Signal.__table__,
    Project.__table__,
    UserProject.__table__,
    PMFlowLaunchDraft.__table__,
    PMFlowThread.__table__,
    PMFlowThreadMessage.__table__,
    PMFlowChangeSet.__table__,
    PMFlowChangeItem.__table__,
    PMFlowApproval.__table__,
    PMFlowEvidenceLink.__table__,
    TruthNode.__table__,
    Task.__table__,
    SemanticFact.__table__,
    ModelCatalogItem.__table__,
    ProjectModelBinding.__table__,
    ModelRecommendationSnapshot.__table__,
    ModelEvalRun.__table__,
    PromptPackage.__table__,
    DevWorkspaceBinding.__table__,
    RepoBinding.__table__,
    TaskCodeLink.__table__,
    ProgressEvent.__table__,
    ReleaseValidationRun.__table__,
    FeedbackIngest.__table__,
    RevenueEvent.__table__,
    ROISnapshot.__table__,
    GitHubWebhookDelivery.__table__,
]


def _use_sqlite_json_types() -> None:
    for table in TABLES:
        for column in table.columns:
            type_name = column.type.__class__.__name__.upper()
            if type_name in {"JSONB", "ARRAY"}:
                column.type = JSON()


@pytest_asyncio.fixture
async def db_env():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    _use_sqlite_json_types()
    async with engine.begin() as conn:
        await conn.run_sync(lambda sync_conn: Base.metadata.create_all(sync_conn, tables=TABLES))
    maker = async_sessionmaker(engine, expire_on_commit=False)
    async with maker() as session:
        yield session
    await engine.dispose()


async def _seed_user_project(session, *, tenant_id: int = 1, role: str = "MEMBER", member: bool = True):
    user = User(
        tenant_id=tenant_id,
        username=f"user-{tenant_id}-{role}-{member}",
        email=f"user-{tenant_id}-{role}-{member}@example.com",
        password_hash="hashed",
        name="Owner",
        role=role,
        assigned_domains=[],
        workload=0,
        status="active",
        is_active=True,
    )
    session.add(user)
    await session.flush()
    project = Project(
        tenant_id=tenant_id,
        name="真实项目",
        title="真实项目",
        description="用于 Change Review 测试",
        owner_id=user.id,
        status="active",
        assigned_role="BUILDER",
        signal_count=0,
    )
    session.add(project)
    await session.flush()
    if member:
        session.add(UserProject(tenant_id=tenant_id, user_id=user.id, project_id=project.id))
    await session.commit()
    return user, project


def _build_app(session, current_user) -> FastAPI:
    app = FastAPI()
    app.include_router(pmflow_os_api.router)

    async def _fake_get_db():
        yield session

    app.dependency_overrides[get_db] = _fake_get_db
    app.dependency_overrides[get_current_user] = lambda: current_user
    return app


def _proposed_items():
    return [
        {
            "group": "new_object",
            "object_type": "Requirement",
            "title": "新增需求：审查真实落库",
            "summary": "把 AI 候选变更写入待审状态机。",
            "diff": "新增真实 ChangeItem。",
            "payload": {"summary": "候选需求"},
            "confidence": 0.91,
            "risk_level": "medium",
        },
        {
            "group": "risk",
            "object_type": "Risk",
            "title": "新增风险：高风险需记录审批",
            "summary": "高风险项本轮先记录 approval。",
            "diff": "新增高风险候选。",
            "payload": {"summary": "候选风险"},
            "confidence": 0.78,
            "risk_level": "high",
        },
        {
            "group": "decision",
            "object_type": "Decision",
            "title": "待确认决策：先审查后 commit",
            "summary": "C1 不直接写项目真相层。",
            "diff": "新增决策候选。",
            "payload": {"summary": "候选决策"},
            "confidence": 0.83,
            "risk_level": "medium",
        },
    ]


async def _create_message_and_extraction(client, project_id: int):
    message = await client.post(
        f"/pmflow-os/projects/{project_id}/threads/thread-project-{project_id}/messages",
        json={"content": "请把本轮对话提取成待审变更", "source": "web", "attachments": []},
    )
    assert message.status_code == 200
    msg = message.json()["data"]
    extraction = await client.post(
        f"/pmflow-os/projects/{project_id}/extractions",
        json={
            "thread_id": msg["threadId"],
            "message_id": msg["id"],
            "scope": "latest_message",
            "headline": "本轮待审变更",
            "summary": "生成 3 个候选变更。",
            "proposed_items": _proposed_items(),
        },
    )
    assert extraction.status_code == 200
    return msg, extraction.json()["data"]


@pytest.mark.asyncio
async def test_change_review_message_extraction_detail_and_projection(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        msg, detail = await _create_message_and_extraction(client, project.id)
        truth = await client.get(f"/pmflow-os/projects/{project.id}/truth-state")

    assert msg["threadId"].isdigit()
    assert await session.scalar(select(func.count(PMFlowThread.id))) == 1
    assert await session.scalar(select(func.count(PMFlowThreadMessage.id))) == 1
    assert await session.scalar(select(func.count(PMFlowChangeSet.id))) == 1
    assert await session.scalar(select(func.count(PMFlowChangeItem.id))) == 3
    assert await session.scalar(select(func.count(PMFlowEvidenceLink.id))) == 1

    assert detail["changeSet"]["status"] == "open"
    assert len(detail["items"]) == 3
    assert detail["items"][0]["status"] == "pending_review"
    assert detail["evidence"][0]["sourceType"] == "thread_message"
    assert detail["evidence"][0]["sourceId"] == msg["id"]

    truth_payload = truth.json()["data"]
    assert truth_payload["loopStage"] == "review_open"
    assert truth_payload["activeChangeSetId"] == detail["changeSet"]["id"]
    assert len(truth_payload["changeSets"]) == 1
    assert len(truth_payload["changeItems"]) == 3
    assert len(truth_payload["evidence"]) == 1


@pytest.mark.asyncio
async def test_change_item_accept_edit_defer_write_approvals_and_do_not_write_memory(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        _, detail = await _create_message_and_extraction(client, project.id)
        item_ids = [item["id"] for item in detail["items"]]
        accepted = await client.post(f"/pmflow-os/change-items/{item_ids[0]}/accept")
        edited = await client.post(
            f"/pmflow-os/change-items/{item_ids[1]}/edit",
            json={"summary": "人工修订摘要", "diff": "人工修订 diff", "payload_patch": {"owner": "user"}, "reason": "收紧范围"},
        )
        deferred = await client.post(f"/pmflow-os/change-items/{item_ids[2]}/defer", json={"reason": "下一轮再看"})
        truth = await client.get(f"/pmflow-os/projects/{project.id}/truth-state")

    assert accepted.status_code == 200
    assert edited.status_code == 200
    assert deferred.status_code == 200
    assert await session.scalar(select(func.count(PMFlowApproval.id))) == 3
    assert await session.scalar(select(func.count(TruthNode.id))) == 0
    assert await session.scalar(select(func.count(Task.id))) == 0
    assert await session.scalar(select(func.count(SemanticFact.id))) == 0

    change_set = (await session.execute(select(PMFlowChangeSet))).scalar_one()
    assert change_set.status == "reviewed"
    edited_row = (await session.execute(select(PMFlowChangeItem).where(PMFlowChangeItem.id == int(item_ids[1])))).scalar_one()
    assert edited_row.status == "edited"
    assert edited_row.edited_payload["owner"] == "user"
    assert truth.json()["data"]["projectMemory"]["sourceSummary"]["truth_nodes"] == 0
    assert len(truth.json()["data"]["approvals"]) == 3


@pytest.mark.asyncio
async def test_change_item_review_conflict_and_closed_changeset_guard(db_env) -> None:
    session = db_env
    user, project = await _seed_user_project(session)
    app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        _, detail = await _create_message_and_extraction(client, project.id)
        item_id = detail["items"][0]["id"]
        first = await client.post(f"/pmflow-os/change-items/{item_id}/accept")
        second = await client.post(f"/pmflow-os/change-items/{item_id}/accept")

    assert first.status_code == 200
    assert second.status_code == 409
    assert "item_already_reviewed" in second.json()["message"]

    pending = (await session.execute(select(PMFlowChangeItem).where(PMFlowChangeItem.status == "pending_review"))).scalars().first()
    change_set = (await session.execute(select(PMFlowChangeSet).where(PMFlowChangeSet.id == pending.change_set_id))).scalar_one()
    change_set.status = "closed"
    await session.commit()

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        blocked = await client.post(f"/pmflow-os/change-items/{pending.id}/accept")

    assert blocked.status_code == 409
    assert "changeset_not_open" in blocked.json()["message"]


@pytest.mark.asyncio
async def test_change_review_project_permission_boundary(db_env) -> None:
    session = db_env
    owner, project = await _seed_user_project(session)
    outsider, _ = await _seed_user_project(session, tenant_id=1, member=False)
    owner_app = _build_app(session, SimpleNamespace(id=owner.id, tenant_id=1, role="MEMBER"))

    async with AsyncClient(transport=ASGITransport(app=owner_app), base_url="http://test") as client:
        _, detail = await _create_message_and_extraction(client, project.id)

    outsider_app = _build_app(session, SimpleNamespace(id=outsider.id, tenant_id=1, role="MEMBER"))
    async with AsyncClient(transport=ASGITransport(app=outsider_app), base_url="http://test") as client:
        denied = await client.get(f"/pmflow-os/projects/{project.id}/change-sets/{detail['changeSet']['id']}")
        denied_action = await client.post(f"/pmflow-os/change-items/{detail['items'][0]['id']}/accept")

    assert denied.status_code == 403
    assert denied_action.status_code == 403

    cross_tenant_app = _build_app(session, SimpleNamespace(id=outsider.id, tenant_id=2, role="MEMBER"))
    async with AsyncClient(transport=ASGITransport(app=cross_tenant_app), base_url="http://test") as client:
        missing = await client.get(f"/pmflow-os/projects/{project.id}/change-sets/{detail['changeSet']['id']}")

    assert missing.status_code == 404