# 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.

"""
Context for database and permission operations.

This is currently based on ContextVar instead of asgiref.local.Local, because
the behaviour of Local is inconsistent across asgiref versions, and probably
buggy in asgiref 3.7+ (see https://github.com/django/asgiref/issues/473).

Ideally, this should eventually aligned with using the same
thread-local/task-local infrastructure that Django uses.

Note that when using threads, if a new thread is started, it will see the
application context reset to empty values: threads that need application
context values need to implement a way to inherit the caller's values.

Note that similar but more complicated gotchas would apply when
`django.urls.reverse` tries to check if `request.urlconf` was previously set to
a non-default value: threads and tasks that need reverse url resolution need to
be mindful of the inconsistencies in behaviour of the underlying
`asgiref.local.Local` implementation, and of asgiref issue #473.

See also https://code.djangoproject.com/ticket/35807
"""

from collections.abc import Generator
from contextlib import contextmanager
from contextvars import ContextVar
from typing import Optional, TYPE_CHECKING

from debusine.db.permissioncontext import PermissionContext, PermissionUser

if TYPE_CHECKING:
    from debusine.db.models import Scope, Token, Workspace
    from debusine.db.models.lookups import LookupCache
    from debusine.db.models.scopes import ScopeRoles
    from debusine.db.models.workspaces import WorkspaceRoles


class ContextConsistencyError(Exception):
    """Raised if an inconsistency is found when setting application context."""


class Context:
    """
    Storage for Debusine application context.

    Setting the application context needs to be done in a specific order:

    1. ``context.set_scope()``
    2. ``context.set_permission_context()``
    3. ``workspace.set_current()``

    This reflects the order in which information is available in the middleware
    chain, where scope is needed first in order to activate URL resolution,
    which is a prerequisite of the auth layer which will then populate the user.
    """

    # https://github.com/django/asgiref/issues/473

    # Use slots to catch typos in functions that set context variables
    __slots__ = [
        "_lookup_cache",
        "_permission_context",
        "_scope",
        "_scope_roles",
        "_workspace",
        "_workspace_roles",
    ]

    def __init__(self) -> None:
        """Initialize the default values."""
        super().__init__()

        self._lookup_cache: ContextVar[LookupCache | None] = ContextVar(
            "lookup_cache", default=None
        )

        self._permission_context: ContextVar[PermissionContext | None] = (
            ContextVar("permission_context", default=None)
        )

        self._scope: ContextVar[Scope | None] = ContextVar(
            "scope", default=None
        )

        self._scope_roles: ContextVar[frozenset[ScopeRoles] | None] = (
            ContextVar("scope_roles", default=None)
        )

        self._workspace: ContextVar[Workspace | None] = ContextVar(
            "workspace", default=None
        )
        self._workspace_roles: ContextVar[frozenset[WorkspaceRoles] | None] = (
            ContextVar("workspace_roles", default=None)
        )

    @property
    def permission_checks_disabled(self) -> bool:
        """
        Return True if permission checks should always succeed.

        This is used to skip permission checks in situations like management
        commands or test fixture setup.
        """
        if (pc := self._permission_context.get()) is None:
            return False
        return pc.permission_checks_disabled

    @contextmanager
    def disable_permission_checks(self) -> Generator[None]:
        """Set admin mode for the duration of the context manager."""
        if self._permission_context.get() is None:
            orig = self._permission_context.set(
                PermissionContext.for_ui_anonymous().with_permission_checks_disabled()
            )
        else:
            orig = self._permission_context.set(
                self.pc.with_permission_checks_disabled()
            )
        try:
            yield
        finally:
            self._permission_context.reset(orig)

    @property
    def pc(self) -> PermissionContext:
        """Return the current permission context."""
        res = self._permission_context.get()
        if res is None:
            raise ContextConsistencyError(
                "permission context has not yet been set"
            )
        return res

    def pc_is(self, pc: PermissionContext) -> bool:
        """Return True if the current permission context, if set, matches pc."""
        return self._permission_context.get() == pc

    def set_permission_context_scoped(
        self, scope: "Scope", pc: PermissionContext
    ) -> None:
        """Set the permission context, for scoped operations."""
        from debusine.db.models.scopes import ScopeRoles

        if (old_scope := self.scope) is not None:
            raise ContextConsistencyError(
                f"Scope was already set to {old_scope}"
            )

        if (old := self._permission_context.get()) is not None:
            raise ContextConsistencyError(
                f"Permission context was already set to {old}"
            )

        self._scope.set(scope)
        self._scope_roles.set(None)
        self._permission_context.set(pc)
        self._workspace.set(None)
        self._workspace_roles.set(None)

        # TODO: Scope visibility checks should happen here. Currently scopes
        # are always visible, so the only thing needed is this comment as a
        # placeholder

        self._scope_roles.set(
            ScopeRoles.from_iterable(scope.get_group_roles(pc))
        )

    def set_permission_context_scopeless(self, pc: PermissionContext) -> None:
        """Set the permission context, for unscoped operations."""
        if self.scope is not None:
            raise ContextConsistencyError(
                f"Scope was already set to {self.scope}"
            )

        if (old := self._permission_context.get()) is not None:
            raise ContextConsistencyError(
                f"Permission context was already set to {old}"
            )

        self._permission_context.set(pc)
        self._scope_roles.set(None)
        self._workspace.set(None)
        self._workspace_roles.set(None)

    @property
    def scope(self) -> Optional["Scope"]:
        """Get the current scope."""
        return self._scope.get()

    def require_scope(self) -> "Scope":
        """
        Get the current scope.

        :raises ContextConsistencyError: if the scope is not set
        """
        if (scope := self.scope) is None:
            raise ContextConsistencyError("scope is not set in context")
        return scope

    @property
    def user(self) -> PermissionUser:
        """Return the current user."""
        return self.pc.user

    @property
    def user_token(self) -> Optional["Token"]:
        """Get the current user token."""
        return self.pc.user_token

    @property
    def external_user_token(self) -> Optional["Token"]:
        """Get the current external user token."""
        return self.pc.external_user_token

    @property
    def workspace(self) -> Optional["Workspace"]:
        """Get the current workspace."""
        return self._workspace.get()

    def require_workspace(self) -> "Workspace":
        """
        Get the current workspace.

        :raises ContextConsistencyError: if the workspace is not set
        """
        if (workspace := self.workspace) is None:
            raise ContextConsistencyError("workspace is not set in context")
        return workspace

    @property
    def scope_roles(self) -> frozenset["ScopeRoles"]:
        """Get the roles the current user has on the current Scope."""
        if (roles := self._scope_roles.get()) is not None:
            return roles
        raise ContextConsistencyError("scope has not been set")

    @property
    def workspace_roles(self) -> frozenset["WorkspaceRoles"]:
        """Get the roles the current user has on the current Workspace."""
        if (roles := self._workspace_roles.get()) is not None:
            return roles
        raise ContextConsistencyError("workspace is not set")

    @property
    def lookup_cache(self) -> "LookupCache":
        """Get the current context's lookup cache."""
        # Avoid circular import.
        from debusine.db.models.lookups import LookupCache

        if (lookup_cache := self._lookup_cache.get()) is not None:
            return lookup_cache
        self._lookup_cache.set(lookup_cache := LookupCache())
        return lookup_cache

    def clear_lookup_cache(self) -> None:
        """Clear the current context's lookup cache."""
        self._lookup_cache.set(None)

    def reset(self) -> None:
        """Reset the application context to default values."""
        self._permission_context.set(None)
        self._scope.set(None)
        self._scope_roles.set(None)
        self._workspace.set(None)
        self._workspace_roles.set(None)
        self._lookup_cache.set(None)

    @contextmanager
    def local(self, inherit_lookup_cache: bool = False) -> Generator[None]:
        """
        Restore application context when the context manager ends.

        :param inherit_lookup_cache: if True, keep any entries in the
          current lookup cache; if False, set an empty lookup cache.  In
          either case, restore the current lookup cache when the context
          manager exits, discarding any entries made within the context
          manager.
        """
        orig_pc = self._permission_context.get()
        orig_scope = self.scope
        orig_scope_roles = self._scope_roles.get()
        orig_workspace = self.workspace
        orig_workspace_roles = self._workspace_roles.get()
        orig_lookup_cache = self.lookup_cache
        if inherit_lookup_cache:
            self._lookup_cache.set(orig_lookup_cache.copy())
        else:
            self._lookup_cache.set(None)
        try:
            yield
        finally:
            self._permission_context.set(orig_pc)
            self._scope.set(orig_scope)
            self._scope_roles.set(orig_scope_roles)
            self._workspace.set(orig_workspace)
            self._workspace_roles.set(orig_workspace_roles)
            self._lookup_cache.set(orig_lookup_cache)

    @contextmanager
    def management_command(self) -> Generator[None]:
        """
        Set a context for management commands.

        This only sets the context user to the system user, so that group
        management commands can create an appropriate audit log.
        """
        with self.local():
            self.reset()
            pc = PermissionContext.for_maintenance()
            self._permission_context.set(pc)
            yield


context = Context()
