"""Materialize Workspace authority into project memberships (design D8, D14, D29, D30).

staff_directory remains the projector of Workspace authority. This module reads its
output, plans the reconciliation (membership_reconciliation), and applies the plan to
one project's membership rows in the caller's transaction. It writes only:

- membership rows of the given project;
- role bindings with source=workspace (never source=local);
- suspension holds with source=workspace (never local or legacy);
- the materialized membership status, derived from the holds.

The caller commits. Each write is flushed in plan order, so the owner invariant holds at
every step. Reconciliation is convergent: the same snapshot yields the same plan, and a
converged project yields no operations, so nothing is written and nothing is bumped.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from uuid import UUID, uuid4

from sqlalchemy import delete, func, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from src.domain.project_auth.membership_reconciliation import (
    WORKSPACE,
    CurrentBinding,
    CurrentMembership,
    DesiredIdentity,
    OpKind,
    Plan,
    ProjectionError,
    normalize_email,
    plan_reconciliation,
)
from src.domain.project_auth.models import (
    AuthAssignmentTier,
    ProjectAuthConfig,
    ProjectAuthRole,
    ProjectMembership,
    ProjectMembershipRole,
    ProjectMembershipSuspension,
)
from src.domain.project_auth.staff_directory import SUPER_ADMIN_ROLE, WorkspaceAppIdentity

ADMIN_ROLE_NAME = "admin"


def desired_from_workspace(identities: list[WorkspaceAppIdentity]) -> list[DesiredIdentity]:
    """Project staff_directory output. Only the owner and admin carry generated-app authority."""
    out: list[DesiredIdentity] = []
    for identity in identities:
        if identity.is_owner or identity.app_role == SUPER_ADMIN_ROLE:
            out.append(DesiredIdentity(normalize_email(identity.email), True, False))
        elif identity.app_role == ADMIN_ROLE_NAME:
            out.append(DesiredIdentity(normalize_email(identity.email), False, True))
    return out


def _text(value) -> str:
    return getattr(value, "value", value)


async def _admin_role_id(db: AsyncSession, project_id: UUID) -> UUID | None:
    # The canonical admin preset: codegen-created, admin-tier, not protected. The name only
    # selects the workspace-admin mapping target; it is not an authorization decision.
    return (
        await db.execute(
            select(ProjectAuthRole.id).where(
                ProjectAuthRole.project_id == project_id,
                ProjectAuthRole.name == ADMIN_ROLE_NAME,
                ProjectAuthRole.created_by == "codegen",
                ProjectAuthRole.assignment_tier == AuthAssignmentTier.admin,
                ProjectAuthRole.protected.is_(False),
            )
        )
    ).scalar_one_or_none()


async def _snapshot(db: AsyncSession, project_id: UUID) -> list[CurrentMembership]:
    rows = (
        await db.execute(
            select(ProjectMembership).where(ProjectMembership.project_id == project_id)
        )
    ).scalars().all()
    ids = [row.id for row in rows]
    bindings: dict[UUID, set[CurrentBinding]] = {i: set() for i in ids}
    holds: dict[UUID, set[str]] = {i: set() for i in ids}
    if ids:
        for link in (
            await db.execute(
                select(ProjectMembershipRole).where(ProjectMembershipRole.membership_id.in_(ids))
            )
        ).scalars().all():
            bindings[link.membership_id].add(CurrentBinding(link.role_id, _text(link.source)))
        for hold in (
            await db.execute(
                select(ProjectMembershipSuspension).where(
                    ProjectMembershipSuspension.membership_id.in_(ids)
                )
            )
        ).scalars().all():
            holds[hold.membership_id].add(_text(hold.source))
    return [
        CurrentMembership(
            membership_id=row.id,
            email=row.email_normalized,
            source=_text(row.source),
            is_owner=bool(row.is_owner),
            cloud_app_user_id=row.cloud_app_user_id,
            bindings=frozenset(bindings[row.id]),
            holds=frozenset(holds[row.id]),
        )
        for row in rows
    ]


async def sync_membership_status(db: AsyncSession, membership_id: UUID) -> str:
    """Derive ProjectMembership.status from the hold rows. Holds are the causes."""
    remaining = await db.scalar(
        select(func.count()).select_from(ProjectMembershipSuspension).where(
            ProjectMembershipSuspension.membership_id == membership_id
        )
    )
    status = "suspended" if remaining else "active"
    await db.execute(
        update(ProjectMembership).where(ProjectMembership.id == membership_id).values(status=status)
    )
    return status


async def plan_project_memberships(
    db: AsyncSession,
    project_id: UUID,
    identities: list[WorkspaceAppIdentity],
    *,
    cloud_user_ids: dict[str, str] | None = None,
) -> Plan:
    desired = desired_from_workspace(identities)
    admin_role_id = await _admin_role_id(db, project_id)
    current = await _snapshot(db, project_id)
    return plan_reconciliation(
        desired, current, admin_role_id=admin_role_id, cloud_user_ids=cloud_user_ids
    )


@dataclass
class ApplyResult:
    membership_versions_bumped: int = 0
    operations_applied: int = 0
    touched: set[str] = field(default_factory=set)


async def apply_project_memberships(db: AsyncSession, project_id: UUID, plan: Plan) -> ApplyResult:
    """Apply a plan to one project, flushing in plan order."""
    members = {
        m.email_normalized: m
        for m in (
            await db.execute(
                select(ProjectMembership).where(ProjectMembership.project_id == project_id)
            )
        ).scalars().all()
    }
    result = ApplyResult()

    for op in plan.operations:
        result.operations_applied += 1
        result.touched.add(op.email)
        if op.kind is OpKind.CREATE_MEMBERSHIP:
            row = ProjectMembership(
                id=uuid4(),
                project_id=project_id,
                email_normalized=op.email,
                source=op.source,
                status="active",
                is_owner=bool(op.value),
            )
            db.add(row)
            await db.flush()
            members[op.email] = row
            continue

        row = members[op.email]
        if op.kind is OpKind.UPDATE_OWNER:
            row.is_owner = bool(op.value)
        elif op.kind is OpKind.LINK_CLOUD_USER:
            row.cloud_app_user_id = op.value
        elif op.kind is OpKind.GRANT_ROLE:
            db.add(
                ProjectMembershipRole(membership_id=row.id, role_id=op.role_id, source=op.source)
            )
        elif op.kind is OpKind.REVOKE_ROLE:
            assert op.source == WORKSPACE
            await db.execute(
                delete(ProjectMembershipRole).where(
                    ProjectMembershipRole.membership_id == row.id,
                    ProjectMembershipRole.role_id == op.role_id,
                    ProjectMembershipRole.source == WORKSPACE,
                )
            )
        elif op.kind is OpKind.ADD_SUSPENSION:
            assert op.source == WORKSPACE
            db.add(
                ProjectMembershipSuspension(
                    membership_id=row.id,
                    source=WORKSPACE,
                    reason_code="workspace_authority_removed",
                )
            )
            await db.flush()
            await sync_membership_status(db, row.id)
        elif op.kind is OpKind.REMOVE_SUSPENSION:
            assert op.source == WORKSPACE
            await db.execute(
                delete(ProjectMembershipSuspension).where(
                    ProjectMembershipSuspension.membership_id == row.id,
                    ProjectMembershipSuspension.source == WORKSPACE,
                )
            )
            await db.flush()
            await sync_membership_status(db, row.id)
        await db.flush()

    # One revision per touched membership per reconciliation, not per planner operation.
    for email in result.touched:
        row = members.get(email)
        if row is not None:
            await db.execute(
                update(ProjectMembership)
                .where(ProjectMembership.id == row.id)
                .values(authorization_version=ProjectMembership.authorization_version + 1)
            )
            result.membership_versions_bumped += 1
    return result


async def reconcile_project_memberships(
    db: AsyncSession,
    project_id: UUID,
    identities: list[WorkspaceAppIdentity],
    *,
    cloud_user_ids: dict[str, str] | None = None,
) -> Plan:
    """Plan and apply one project in the caller's transaction. Refuses an unsafe projection."""
    plan = await plan_project_memberships(db, project_id, identities, cloud_user_ids=cloud_user_ids)
    if plan.operations:
        await apply_project_memberships(db, project_id, plan)
    return plan


async def workspace_identities_for_project(
    db: AsyncSession, project_id: UUID
) -> list[WorkspaceAppIdentity]:
    from src.domain.project_auth.staff_directory import identities_for_workspace
    from src.domain.projects.models import Project
    from src.domain.workspaces.models import Workspace

    project = (
        await db.execute(select(Project).where(Project.id == project_id))
    ).scalar_one_or_none()
    if project is None:
        raise ProjectionError("project_not_found")
    workspace = (
        await db.execute(select(Workspace).where(Workspace.id == project.workspace_id))
    ).scalar_one_or_none()
    if workspace is None:
        raise ProjectionError("workspace_not_found")
    return await identities_for_workspace(db, workspace.id, workspace.owner_id)


async def reconcile_project_from_workspace(db: AsyncSession, project_id: UUID) -> Plan:
    """Event-independent entry point: read Workspace state, then converge one project."""
    identities = await workspace_identities_for_project(db, project_id)
    return await reconcile_project_memberships(db, project_id, identities)


def _projects_with_auth_config():
    """Projects eligible for automatic Core reconciliation: shadow or enabled only (rollout gate)."""
    from src.domain.projects.models import Project

    return select(Project.id).join(
        ProjectAuthConfig, ProjectAuthConfig.project_id == Project.id
    ).where(ProjectAuthConfig.projection_state != "disabled")


async def projects_for_workspace(db: AsyncSession, workspace_id: UUID) -> list[UUID]:
    from src.domain.projects.models import Project

    rows = await db.execute(
        _projects_with_auth_config()
        .where(Project.workspace_id == workspace_id)
        .order_by(Project.id)
    )
    return list(rows.scalars().all())


async def project_ids_page(db: AsyncSession, after: UUID | None, limit: int) -> list[UUID]:
    """Keyset page of projects that have auth configured, for the repair sweep."""
    from src.domain.projects.models import Project

    query = _projects_with_auth_config().order_by(Project.id).limit(limit)
    if after is not None:
        query = query.where(Project.id > after)
    return list((await db.execute(query)).scalars().all())


@dataclass(frozen=True)
class ProjectOutcome:
    project_id: UUID
    operations: int
    error: str | None = None


async def reconcile_workspace_projects(
    db: AsyncSession, workspace_id: UUID
) -> list[ProjectOutcome]:
    """Converge every project of a workspace. Each project runs in its own savepoint, so one
    project's unsafe projection never blocks or partially writes another."""
    outcomes: list[ProjectOutcome] = []
    for project_id in await projects_for_workspace(db, workspace_id):
        try:
            async with db.begin_nested():
                plan = await reconcile_project_from_workspace(db, project_id)
            outcomes.append(ProjectOutcome(project_id, len(plan.operations)))
        except ProjectionError as exc:
            outcomes.append(ProjectOutcome(project_id, 0, error=str(exc)))
    return outcomes