"""Database integration for Workspace membership projection (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.membership_projection import reconcile_project_memberships
from src.domain.project_auth.membership_reconciliation import ProjectionError
from src.domain.project_auth.models import ProjectMembership, ProjectMembershipRole
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):
    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})
    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})
    finance_role = uuid4()
    await db.execute(text(
        "insert into project_auth_roles (id, project_id, name, created_by, permissions_json, "
        "assignment_tier) values "
        "(:id, :p, 'finance', 'user', cast('[\"billing\"]' as jsonb), 'staff')"),
        {"id": finance_role, "p": project})
    return project, admin_role, finance_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 _members(db, project):
    rows = await db.execute(
        select(ProjectMembership).where(ProjectMembership.project_id == project)
    )
    return {m.email_normalized: m for m in rows.scalars().all()}


async def _bindings(db, membership_id):
    rows = (await db.execute(select(ProjectMembershipRole).where(
        ProjectMembershipRole.membership_id == membership_id)))
    return sorted((str(r.role_id), r.source) for r in rows.scalars().all())


@pytest.mark.asyncio
async def test_reconcile_converges_and_second_run_is_a_noop(db):
    project, admin_role, _ = await _project(db)
    identities = [_ws("owner@x.test", owner=True), _ws("admin@x.test", admin=True)]
    first = await reconcile_project_memberships(db, project, identities)
    assert first.operations
    second = await reconcile_project_memberships(db, project, identities)
    assert second.operations == ()
    members = await _members(db, project)
    assert members["owner@x.test"].is_owner and not members["admin@x.test"].is_owner
    assert await _bindings(db, members["admin@x.test"].id) == [(str(admin_role), "workspace")]


@pytest.mark.asyncio
async def test_alice_keeps_local_admin_binding_when_workspace_authority_is_removed(db):
    project, admin_role, finance_role = await _project(db)
    owner = [_ws("owner@x.test", owner=True)]
    await reconcile_project_memberships(db, project, owner + [_ws("alice@x.test", admin=True)])
    alice = (await _members(db, project))["alice@x.test"]
    await db.execute(text(
        "insert into project_membership_roles (membership_id, role_id, source) "
        "values (:m, :r, 'local')"), {"m": alice.id, "r": admin_role})
    await db.execute(text(
        "insert into project_membership_roles (membership_id, role_id, source) "
        "values (:m, :r, 'local')"), {"m": alice.id, "r": finance_role})

    result = await reconcile_project_memberships(db, project, owner)
    kinds = [(op.kind, op.email) for op in result.operations]
    assert ("REVOKE_ROLE", "alice@x.test") in [(k.value, e) for k, e in kinds]
    assert not any(k.value == "SUSPEND" for k, _ in kinds)

    alice = (await _members(db, project))["alice@x.test"]
    assert alice.status == "active"
    assert await _bindings(db, alice.id) == sorted([
        (str(admin_role), "local"), (str(finance_role), "local"),
    ])


@pytest.mark.asyncio
async def test_removed_workspace_only_identity_is_revoked_and_suspended(db):
    project, admin_role, _ = await _project(db)
    owner = [_ws("owner@x.test", owner=True)]
    await reconcile_project_memberships(db, project, owner + [_ws("gone@x.test", admin=True)])
    await reconcile_project_memberships(db, project, owner)
    gone = (await _members(db, project))["gone@x.test"]
    assert gone.status == "suspended"
    assert await _bindings(db, gone.id) == []
    assert (await reconcile_project_memberships(db, project, owner)).operations == ()


@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 _members(db, project)
    owners = [e for e, m in members.items() if m.is_owner]
    assert owners == ["new@x.test"]


@pytest.mark.asyncio
async def test_reconciliation_never_touches_another_project(db):
    project_a, admin_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)])
    members_b = await _members(db, project_b)
    assert members_b["alice@x.test"].status == "active"
    assert await _bindings(db, members_b["alice@x.test"].id) == [(str(admin_b), "workspace")]


@pytest.mark.asyncio
async def test_unsafe_projection_writes_nothing(db):
    project, _, _ = await _project(db)
    with pytest.raises(ProjectionError):
        await reconcile_project_memberships(db, project, [_ws("admin@x.test", admin=True)])
    assert await _members(db, project) == {}