"""Core → Cloud membership projection (Workstream E), pure.

Canonical Core state goes in; a stable projection per membership comes out. Nothing here
performs I/O. The Cloud write is an adapter contract (CloudProjectionWriter) implemented
by Workstream G.

Rules:
- project_roles: distinct names of effective roles across all surviving bindings, whatever
  their source. Protected roles and unclassified roles are never effective through a
  binding, so they are excluded (and reported), never elevated.
- project_permissions: union of effective roles' permissions, plus the protected owner
  role's permissions when is_owner.
- A suspended membership (any hold) projects as suspended, with no roles, no permissions,
  and no owner flag, so no usable authorization claim survives the next token mint.
- legacy_role is a compatibility projection for old consumers only. It is derived from
  effective roles and is never read back as authority.
- Comparison is semantic: arrays are sorted, and drift ignores ordering.
"""

from __future__ import annotations

import hashlib
import json
from collections.abc import Mapping
from dataclasses import dataclass, field
from typing import Protocol
from uuid import UUID

SCHEMA_VERSION = 1

ACTIVE = "active"
SUSPENDED = "suspended"

LEGACY_OWNER = "super_admin"
LEGACY_ADMIN = "admin"
LEGACY_STAFF = "staff"
LEGACY_MEMBER = "member"


@dataclass(frozen=True)
class RoleSnapshot:
    role_id: UUID
    name: str
    assignment_tier: str | None
    protected: bool
    permissions: frozenset[str]


@dataclass(frozen=True)
class BindingSnapshot:
    role_id: UUID
    source: str


@dataclass(frozen=True)
class MembershipSnapshot:
    membership_id: UUID
    email: str
    is_owner: bool
    cloud_app_user_id: str | None
    authorization_version: int
    bindings: frozenset[BindingSnapshot] = field(default_factory=frozenset)
    holds: frozenset[str] = field(default_factory=frozenset)


@dataclass(frozen=True)
class ProjectSnapshot:
    roles: Mapping[UUID, RoleSnapshot]
    protected_owner_permissions: frozenset[str]
    memberships: tuple[MembershipSnapshot, ...]


@dataclass(frozen=True)
class MembershipProjection:
    membership_id: UUID | None
    email: str
    cloud_app_user_id: str | None
    status: str
    is_owner: bool
    project_roles: tuple[str, ...]
    project_permissions: tuple[str, ...]
    excluded_roles: tuple[str, ...]
    legacy_role: str
    authorization_version: int

    def semantic_key(self) -> dict:
        """Everything that defines effective authorization for this identity. Excludes the
        version and the Core row id, so reorderings and bookkeeping never count as drift."""
        return {
            "schema_version": SCHEMA_VERSION,
            "email": self.email,
            "cloud_app_user_id": self.cloud_app_user_id,
            "status": self.status,
            "is_owner": self.is_owner,
            "project_roles": sorted(self.project_roles),
            "project_permissions": sorted(self.project_permissions),
            "excluded_roles": sorted(self.excluded_roles),
            "legacy_role": self.legacy_role,
        }

    def digest(self) -> str:
        body = {
            "semantic": self.semantic_key(),
            "authorization_version": self.authorization_version,
        }
        return hashlib.sha256(
            json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")
        ).hexdigest()


def legacy_role_for(*, is_owner: bool, suspended: bool, effective_tiers: frozenset[str]) -> str:
    """Compatibility mapping for old consumers during M3/M4. New authorization never reads it."""
    if suspended:
        return LEGACY_MEMBER
    if is_owner:
        return LEGACY_OWNER
    if "admin" in effective_tiers:
        return LEGACY_ADMIN
    if "staff" in effective_tiers:
        return LEGACY_STAFF
    return LEGACY_MEMBER


def build_projection(
    snapshot: ProjectSnapshot, membership: MembershipSnapshot
) -> MembershipProjection:
    suspended = bool(membership.holds)
    excluded: set[str] = set()
    effective: list[RoleSnapshot] = []

    if not suspended:
        for binding in membership.bindings:
            role = snapshot.roles.get(binding.role_id)
            if role is None or role.protected or role.assignment_tier is None:
                excluded.add(role.name if role is not None else str(binding.role_id))
                continue
            effective.append(role)

    names = tuple(sorted({r.name for r in effective}))
    permissions: set[str] = set()
    for role in effective:
        permissions.update(role.permissions)
    is_owner = membership.is_owner and not suspended
    if is_owner:
        permissions.update(snapshot.protected_owner_permissions)

    tiers = frozenset(r.assignment_tier for r in effective if r.assignment_tier)
    return MembershipProjection(
        membership_id=membership.membership_id,
        email=membership.email,
        cloud_app_user_id=membership.cloud_app_user_id,
        status=SUSPENDED if suspended else ACTIVE,
        is_owner=is_owner,
        project_roles=() if suspended else names,
        project_permissions=() if suspended else tuple(sorted(permissions)),
        excluded_roles=tuple(sorted(excluded)) if not suspended else (),
        legacy_role=legacy_role_for(is_owner=is_owner, suspended=suspended, effective_tiers=tiers),
        authorization_version=membership.authorization_version,
    )


def build_project_projections(snapshot: ProjectSnapshot) -> list[MembershipProjection]:
    ordered = sorted(snapshot.memberships, key=lambda m: m.email)
    return [build_projection(snapshot, m) for m in ordered]


@dataclass(frozen=True)
class ProjectionWrite:
    email: str
    kind: str  # upsert | retire | unchanged
    audit: bool


def plan_projection_writes(
    desired: list[MembershipProjection],
    remote: Mapping[str, MembershipProjection],
) -> list[ProjectionWrite]:
    """Compare desired projections with what Cloud holds, keyed by email (D4).

    audit is True only when effective authorization changed. A version-only difference
    refreshes Cloud without an audit event, so routine refreshes never bury the trail."""
    writes: list[ProjectionWrite] = []
    desired_by_email = {d.email: d for d in desired}
    for email in sorted(desired_by_email):
        d = desired_by_email[email]
        current = remote.get(email)
        if current is None:
            writes.append(ProjectionWrite(email, "upsert", audit=True))
        elif current.semantic_key() != d.semantic_key():
            writes.append(ProjectionWrite(email, "upsert", audit=True))
        elif current.authorization_version != d.authorization_version:
            writes.append(ProjectionWrite(email, "upsert", audit=False))
        else:
            writes.append(ProjectionWrite(email, "unchanged", audit=False))
    for email in sorted(set(remote) - set(desired_by_email)):
        current = remote[email]
        writes.append(
            ProjectionWrite(email, "retire", audit=current.status != SUSPENDED)
        )
    return writes


def to_cloud_payload(project_id: str, projection: MembershipProjection) -> dict:
    """The wire shape Cloud stores. Canonical ordering; no Core row ids."""
    body = projection.semantic_key()
    body["authorization_version"] = projection.authorization_version
    body["project_id"] = project_id
    return body


def projection_from_cloud_payload(payload: dict) -> MembershipProjection:
    if payload.get("schema_version") != SCHEMA_VERSION:
        raise ValueError("unsupported_projection_schema")
    return MembershipProjection(
        membership_id=None,
        email=payload["email"],
        cloud_app_user_id=payload.get("cloud_app_user_id"),
        status=payload["status"],
        is_owner=bool(payload["is_owner"]),
        project_roles=tuple(sorted(payload["project_roles"])),
        project_permissions=tuple(sorted(payload["project_permissions"])),
        excluded_roles=tuple(sorted(payload.get("excluded_roles", []))),
        legacy_role=payload["legacy_role"],
        authorization_version=int(payload["authorization_version"]),
    )


class CloudProjectionWriter(Protocol):
    """Adapter contract implemented by Workstream G. Core authenticates with the narrow
    auth_projection.write capability (D24). Generated apps never hold it."""

    async def put_projection(self, project_id: str, payload: dict) -> None: ...

    async def delete_projection(self, project_id: str, email: str) -> None: ...

    async def list_projections(self, project_id: str) -> list[dict]: ...