# 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 collections.abc import Callable, Collection, Iterable
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


@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, Scope, User

    match obj:
        case ArtifactRelation():
            return obj.artifact.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,
        Group,
        Scope,
        User,
        Workspace,
    )

    match obj:
        case ArtifactRelation():
            return obj.artifact.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
