# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""Basic permission check infrastructure."""

import enum
import functools
import itertools
from abc import ABC
from collections.abc import Callable, Collection, Iterable
from dataclasses import dataclass
from typing import (
    Any,
    NamedTuple,
    Optional,
    Protocol,
    Self,
    TYPE_CHECKING,
    cast,
    runtime_checkable,
)

from django.core.exceptions import PermissionDenied
from django.db.models import Model, QuerySet
from django.utils.functional import classproperty

from debusine.db.permissioncontext import PermissionContext

if TYPE_CHECKING:
    from debusine.db.models import Scope, Workspace

type ImpliedByCollection = Collection[Any]


@runtime_checkable
class WithImplies[R: RoleBase](Protocol):
    """Define an object with an implies operator."""

    def implies(self, other: R) -> bool:
        """Implies operator."""


class Role(NamedTuple):
    """Annotated tuple used to define Roles."""

    value: str
    label: str | None = None
    implied_by: ImpliedByCollection = ()
    description: str | None = None


class RoleBase(str):
    """
    Base for role value implementations.

    This is intended to be subclassed and used as value for a role enum. For
    example::

        class ScopeRoleBase(RoleBase):
            ...

        class ScopeRoles(Roles, ScopeRoleBase, enum.ReprEnum):
            ...
    """

    def __new__(
        cls,
        value: str,
        label: str | None = None,  # noqa: ARG004, U100
        implied_by: ImpliedByCollection = (),  # noqa: ARG004, U100
        description: str | None = None,  # noqa: ARG004, U100
    ) -> Self:
        """Construct the string from value, ignoring other args."""
        return super().__new__(cls, value)

    def __init__(
        self,
        value: str,
        label: str | None = None,
        implied_by: ImpliedByCollection = (),
        description: str | None = None,
    ) -> None:
        """
        Set up RoleBase.

        :param value: string value for the role
        :param label: optional label (auto generated from value by default)
        :param implied_by: roles or other elements that can be used to imply
                           this role
        """
        self.label = label or value.capitalize()
        self.implied_by = implied_by
        self.description = description

    def _setup(self) -> None:
        """Set up the role after the enum is fully built."""

    @classmethod
    def from_iterable(cls, roles: Iterable[str]) -> frozenset[Self]:
        """Resolve a sequence of strings to a minimal set of ScopeRoles."""
        assert issubclass(cls, WithImplies)
        all_roles = frozenset(cls(str_role) for str_role in roles)
        if len(all_roles) == 1:
            return all_roles

        implied_roles = set()
        for a, b in itertools.permutations(all_roles, 2):
            if a != b and a.implies(b):
                implied_roles.add(b)
        return frozenset(all_roles - implied_roles)


# Mixin for role enums
class Roles:
    """Base for declaratively defining available roles for a resource."""

    @classmethod
    def setup(cls) -> None:
        """Finalize enum setup."""
        # Roles is used as a mixin for enum.ReprEnum, which is iterable,
        # but we cannot inherit from it without triggering the enum's setup.
        #
        # Overriding type checks until we have a better idea of how to type
        # this.
        for member in cls:  # type: ignore[attr-defined]
            member._setup()

    @classproperty
    def choices(cls) -> list[tuple[str, str]]:
        """Return labeled choices for use by Django fields."""
        res: list[tuple[str, str]] = []
        for entry in cls:  # type: ignore[attr-defined]
            role = cast("RoleBase", entry)
            res.append((str(role), role.label))
        return res


@dataclass(frozen=True)
class ResourceGrant(ABC):
    """An in-memory grant of a role to a matching resource."""

    resource: Model
    role: WithImplies[Any]

    def __post_init__(self) -> None:
        """Ensure that the role matches the resource."""
        roles_enum = getattr(self.resource, "Roles")
        assert isinstance(self.role, roles_enum)


@runtime_checkable
class PermissionCheckPredicate[M: Model](Protocol):
    """Interface of a permission predicate on a resource."""

    __self__: M

    __func__: Callable[[M, PermissionContext], bool]

    def __call__(self, pc: PermissionContext) -> bool:
        """Test the predicate."""


def resolve_role(resource: Model | type[Model], role: str) -> str:
    """Convert a role in various forms into the right enum."""
    roles_enum = getattr(resource, "Roles")

    if isinstance(role, RoleBase):
        if not isinstance(role, roles_enum):
            raise TypeError(f"{role!r} cannot be converted to {roles_enum!r}")
        return cast(str, role)

    return cast(str, roles_enum(role))


def resolve_roles_list(
    resource: Model | type[Model], roles: str | Collection[str]
) -> list[str]:
    """Convert roles into a list of the right enum."""
    if not hasattr(resource, "Roles"):
        return []
    # Roles are string enums, which are instances of str, and therefore also of
    # Collection. This test intends to catch all roles that are not actual
    # collections of roles
    if isinstance(roles, str):
        return [resolve_role(resource, roles)]
    return [resolve_role(resource, r) for r in roles]


class Allow(enum.StrEnum):
    """
    Handling strategies for permission predicates.

    This is used to specify how to handle work_request and anonymous users.
    """

    #: The tested condition makes the predicate always fail
    NEVER = "never"
    #: The tested condition is ignored and the decision is delegated to the
    #: body of the predicate
    PASS = "pass"


type Predicate[M] = Callable[[M, PermissionContext], bool]


def permission_check[M: Model](
    msg: str,
    *,
    work_request: Allow = Allow.NEVER,
    anonymous: Allow = Allow.NEVER,
    explicit_grants_role: RoleBase | None = None,
) -> Callable[[Predicate[M]], Predicate[M]]:
    """
    Implement common elements of permission checking predicates.

    :param work_request: what to do if a work request token is passed
    :param anonymous: what to do if an anonymous user is passed
    :param explicit_grants_role: if a user token requiring explicit grants
      is passed, check for an explicit grant of this resource with this
      role; if such a token is passed and this parameter is unset, return
      False

    Note that a worker token currently implies that the user is anonymous. This
    may change with #523.

    Predicates should normally also check permissions on any containing
    resources, relying on query caching as needed for performance. This
    provides some defence in depth against omitted checks.
    """

    def wrap(f: Predicate[M]) -> Predicate[M]:
        @functools.wraps(f)
        def wrapper(self: M, pc: PermissionContext) -> bool:
            if pc.permission_checks_disabled:
                return True

            overrides = {
                Allow.NEVER: False,
                Allow.PASS: None,
            }
            if pc.work_request:
                if (override := overrides[work_request]) is not None:
                    return override
            elif pc.is_non_permission_bearing_token:
                return False
            elif pc.explicit_grants_only:
                if explicit_grants_role is not None:
                    # Delegate the decision of which resources should be
                    # allowed to per-model `extra_*_grants` querysets on the
                    # permission context, which should be derived from the
                    # token.
                    grants: QuerySet[Any] = getattr(
                        pc, f"extra_{self._meta.model.__name__.lower()}_grants"
                    )
                    assert hasattr(grants, "granted_resources")
                    granted_resources = grants.granted_resources(
                        role=explicit_grants_role
                    )
                    assert isinstance(granted_resources, QuerySet)
                    return granted_resources.filter(pk=self.pk).exists()
                else:
                    return False
            else:
                if not pc.user.is_authenticated:
                    if (override := overrides[anonymous]) is not None:
                        return override

            return f(self, pc)

        setattr(wrapper, "error_template", msg)
        setattr(wrapper, "work_request", work_request)
        setattr(wrapper, "anonymous", anonymous)
        setattr(wrapper, "explicit_grants_role", explicit_grants_role)
        return wrapper

    return wrap


type PredicateFilter[QS] = Callable[[QS, PermissionContext], QS]


def permission_filter[QS: QuerySet[Any, Any]](
    *,
    work_request: Allow = Allow.NEVER,
    anonymous: Allow = Allow.NEVER,
    explicit_grants_role: RoleBase | None = None,
) -> Callable[[PredicateFilter[QS]], PredicateFilter[QS]]:
    """
    Implement common elements of permission filtering predicates.

    :param work_request: what to do if a work request token is passed
    :param anonymous: what to do if an anonymous user is passed
    :param explicit_grants: what to do if a user token requiring explicit
      grants is passed
    :param explicit_grants_role: if a user token requiring explicit grants
      is passed, filter to resources explicitly granted with this role; if
      such a token is passed and this parameter is unset, return an empty
      queryset

    Note that a worker token currently implies that the user is anonymous. This
    may change with #523.

    Predicates should normally also check permissions on any containing
    resources, relying on query caching as needed for performance.  This
    provides some defence in depth against omitted checks.
    """

    def wrap(f: PredicateFilter[QS]) -> PredicateFilter[QS]:
        @functools.wraps(f)
        def wrapper(self: QS, pc: PermissionContext) -> QS:
            if pc.permission_checks_disabled:
                return self

            overrides = {
                Allow.NEVER: self.none(),
                Allow.PASS: None,
            }
            if pc.work_request:
                if (override := overrides[work_request]) is not None:
                    return override
            elif pc.is_non_permission_bearing_token:
                return self.none()
            elif pc.explicit_grants_only:
                if explicit_grants_role is not None:
                    # Delegate the decision of which resources should be
                    # allowed to per-model `extra_*_grants` querysets on the
                    # permission context, which should be derived from the
                    # token.
                    grants: QuerySet[Any] = getattr(
                        pc, f"extra_{self.model.__name__.lower()}_grants"
                    )
                    assert hasattr(grants, "granted_resources")
                    granted_resources = grants.granted_resources(
                        role=explicit_grants_role
                    )
                    assert isinstance(granted_resources, QuerySet)
                    return self.filter(pk__in=granted_resources)
                else:
                    return self.none()
            else:
                if not pc.user.is_authenticated:
                    if (override := overrides[anonymous]) is not None:
                        return override

            return f(self, pc)

        setattr(wrapper, "work_request", work_request)
        setattr(wrapper, "anonymous", anonymous)
        setattr(wrapper, "explicit_grants_role", explicit_grants_role)
        return wrapper

    return wrap


def format_permission_check_error(
    predicate: Callable[[PermissionContext], bool],
    pc: PermissionContext,
) -> str:
    """Format a permission check error message."""
    assert hasattr(predicate, "error_template")
    error_template = predicate.error_template
    assert isinstance(error_template, str)
    assert hasattr(predicate, "__self__")
    return error_template.format(
        resource=predicate.__self__, user=pc.user, pc=pc
    )


def enforce(predicate: Callable[[PermissionContext], bool]) -> None:
    """Enforce a permission predicate at the model level."""
    from debusine.db.context import context

    if predicate(context.pc):
        return

    raise PermissionDenied(
        format_permission_check_error(predicate, context.pc),
    )


def get_resource_scope(obj: Model) -> Optional["Scope"]:
    """
    Get the Scope object for a resource.

    :returns: None if the resource does not have a scope
    :raises NotImplementedError: for resources not currently supported
    """
    from debusine.db.models import ArtifactRelation, CollectionItem, Scope, User

    match obj:
        case ArtifactRelation():
            return obj.artifact.workspace.scope
        case CollectionItem():
            return obj.parent_collection.workspace.scope
        case Scope():
            return obj
        case User():
            return None
        case _:
            if hasattr(obj, "scope"):
                assert isinstance(obj.scope, Scope)
                return obj.scope

            if hasattr(obj, "workspace"):
                assert isinstance(obj.workspace.scope, Scope)
                return obj.workspace.scope

            raise NotImplementedError(
                f"Cannot get scope for {obj.__class__.__name__} object"
            )


def get_resource_workspace[M: Model](obj: M) -> Optional["Workspace"]:
    """
    Get the Workspace object for a resource.

    :returns: None if the resource does not have a workspace
    :raises NotImplementedError: for resources not currently supported
    """
    from debusine.db.models import (
        ArtifactRelation,
        CollectionItem,
        Group,
        Scope,
        User,
        Workspace,
    )

    match obj:
        case ArtifactRelation():
            return obj.artifact.workspace
        case CollectionItem():
            return obj.parent_collection.workspace
        case Scope() | User() | Group():
            return None
        case Workspace():
            return obj
        case _:
            if hasattr(obj, "workspace"):
                assert isinstance(obj.workspace, Workspace)
                return obj.workspace

            raise NotImplementedError(
                f"Cannot get workspace for {obj.__class__.__name__} object"
            )


def anonymous_access(pc: PermissionContext) -> bool:  # noqa: ARG001
    """Permission predicate that grants access to all."""
    return True
