"""Database integration for the Workspace projection trigger and convergence (Workstream D).

Runs only when P2HS_AUTHZ_TEST_DATABASE_URL points at a migrated, non-production database.
Every write is rolled back.
"""

import os
from uuid import uuid4

import pytest
import pytest_asyncio
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from src.domain.project_auth import membership_projection as projection
from src.domain.project_auth.membership_projection import reconcile_project_memberships
from src.domain.project_auth.models import ProjectMembership
from src.domain.project_auth.staff_directory import WorkspaceAppIdentity
from src.domain.projects import models as _project_models  # noqa: F401  registers projects table
from src.domain.workspaces import models as _workspace_models  # noqa: F401

URL = os.environ.get("P2HS_AUTHZ_TEST_DATABASE_URL", "")
pytestmark = pytest.mark.skipif(not URL, reason="P2HS_AUTHZ_TEST_DATABASE_URL not set")


@pytest_asyncio.fixture
async def db():
    engine = create_async_engine(URL)
    async with engine.connect() as conn:
        trans = await conn.begin()
        session = AsyncSession(bind=conn, expire_on_commit=False)
        try:
            yield session
        finally:
            await session.close()
            await trans.rollback()
    await engine.dispose()


async def _project(db, *, auth_config=False):
    ws = uuid4()
    project = uuid4()
    await db.execute(text("insert into workspaces (id, name, slug) values (:id, 'ws', :slug)"),
                     {"id": ws, "slug": f"ws-{ws}"})
    ptype = (await db.execute(text(
        "select enumlabel from pg_enum e join pg_type t on t.oid = e.enumtypid "
        "where t.typname = (select udt_name from information_schema.columns "
        "where table_name = 'projects' and column_name = 'type') order by enumsortorder limit 1"
    ))).scalar_one()
    await db.execute(text(f"insert into projects (id, name, slug, type, workspace_id) "
                          f"values (:id, 'p', :slug, '{ptype}', :ws)"),
                     {"id": project, "slug": f"p-{project}", "ws": ws})
    if auth_config:
        await db.execute(text("insert into project_auth_config (id, project_id) values (:id, :p)"),
                         {"id": uuid4(), "p": project})
    admin_role = uuid4()
    await db.execute(text(
        "insert into project_auth_roles (id, project_id, name, created_by, permissions_json, "
        "assignment_tier, protected) values "
        "(:id, :p, 'admin', 'codegen', cast('[\"staff\"]' as jsonb), "
        "'admin', false)"), {"id": admin_role, "p": project})
    return project, admin_role


def _ws(email, *, owner=False, admin=False):
    if owner:
        return WorkspaceAppIdentity(email, "", "owner", "super_admin", True)
    return WorkspaceAppIdentity(email, "", "admin", "admin", False)


async def _member(db, project, email):
    return (await db.execute(select(ProjectMembership).where(
        ProjectMembership.project_id == project,
        ProjectMembership.email_normalized == email))).scalar_one()


async def _versions(db, project):
    rows = (await db.execute(select(ProjectMembership).where(
        ProjectMembership.project_id == project))).scalars().all()
    return {m.email_normalized: m.authorization_version for m in rows}


@pytest.mark.asyncio
async def test_reconciliation_projection_is_idempotent_and_versions_are_stable(db):
    project, _ = await _project(db)
    identities = [_ws("owner@x.test", owner=True), _ws("dana@x.test", admin=True)]
    await reconcile_project_memberships(db, project, identities)
    before = await _versions(db, project)
    second = await reconcile_project_memberships(db, project, identities)
    assert second.operations == ()
    assert await _versions(db, project) == before


@pytest.mark.asyncio
async def test_one_reconcile_bumps_each_membership_once_not_per_operation(db):
    project, _ = await _project(db)
    owner = [_ws("owner@x.test", owner=True)]
    await reconcile_project_memberships(db, project, owner + [_ws("erin@x.test", admin=True)])
    erin = await _member(db, project, "erin@x.test")
    before = erin.authorization_version
    # Erin leaves: revoke binding and add workspace hold in one transaction.
    await reconcile_project_memberships(db, project, owner)
    await db.refresh(erin)
    assert erin.authorization_version == before + 1


@pytest.mark.asyncio
async def test_owner_transfer_keeps_exactly_one_owner(db):
    project, _ = await _project(db)
    await reconcile_project_memberships(db, project, [_ws("old@x.test", owner=True)])
    await reconcile_project_memberships(
        db, project, [_ws("new@x.test", owner=True), _ws("old@x.test", admin=True)]
    )
    members = (await db.execute(select(ProjectMembership).where(
        ProjectMembership.project_id == project))).scalars().all()
    assert [m.email_normalized for m in members if m.is_owner] == ["new@x.test"]


@pytest.mark.asyncio
async def test_reconciliation_never_touches_another_project(db):
    project_a, _ = await _project(db)
    project_b, admin_b = await _project(db)
    await reconcile_project_memberships(db, project_b, [_ws("owner@x.test", owner=True),
                                                       _ws("alice@x.test", admin=True)])
    await reconcile_project_memberships(db, project_a, [_ws("owner@x.test", owner=True)])
    alice = await _member(db, project_b, "alice@x.test")
    assert alice.status == "active"


@pytest.mark.asyncio
async def test_trigger_isolates_a_refused_project_from_its_siblings(db, monkeypatch):
    project_ok, _ = await _project(db, auth_config=True)
    project_bad, _ = await _project(db, auth_config=True)
    projects = [project_ok, project_bad]

    async def fake_projects(_db, _workspace_id):
        return projects

    async def fake_identities(_db, project_id):
        if project_id == project_bad:
            return [_ws("admin@x.test", admin=True)]  # no owner: must be refused
        return [_ws("owner@x.test", owner=True)]

    monkeypatch.setattr(projection, "projects_for_workspace", fake_projects)
    monkeypatch.setattr(projection, "workspace_identities_for_project", fake_identities)

    outcomes = await projection.reconcile_workspace_projects(db, uuid4())
    by_project = {o.project_id: o for o in outcomes}
    assert by_project[project_ok].error is None and by_project[project_ok].operations > 0
    assert by_project[project_bad].error is not None
    assert (await db.execute(select(ProjectMembership).where(
        ProjectMembership.project_id == project_bad))).first() is None
    assert (await db.execute(select(ProjectMembership).where(
        ProjectMembership.project_id == project_ok))).first() is not None