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

"""Data models for db users and authentication."""

import abc
import datetime as dt
import enum
import hashlib
import re
import secrets
from enum import StrEnum
from typing import (
    Annotated,
    Any,
    Literal,
    Optional,
    Self,
    TYPE_CHECKING,
    TypeAlias,
    cast,
)

import pgtrigger
import pydantic
from django.conf import settings
from django.contrib.auth.models import AbstractUser
from django.contrib.auth.models import UserManager as DjangoUserManager
from django.core.exceptions import ImproperlyConfigured, ValidationError
from django.core.validators import MaxLengthValidator, MinLengthValidator
from django.db import models
from django.db.models import (
    CheckConstraint,
    F,
    Q,
    QuerySet,
    UniqueConstraint,
    Value,
)
from django.db.models.functions import Coalesce
from django.urls import reverse
from django.utils import timezone

from debusine.db.models import permissions
from debusine.db.models.permissions import (
    Allow,
    Role,
    get_resource_scope,
    permission_check,
    permission_filter,
    resolve_role,
)
from debusine.db.models.scopes import Scope
from debusine.db.models.workspaces import Workspace
from debusine.db.permissioncontext import PermissionContext
from debusine.server import notifications
from debusine.utils.typing_utils import copy_signature_from

if TYPE_CHECKING:
    from django.http import HttpRequest
    from django_stubs_ext.db.models import TypedModelMeta

    from debusine.db.models.work_requests import WorkRequest
    from debusine.web.views.ui.auth import GroupMembershipUI, GroupUI, UserUI

    # https://github.com/typeddjango/django-stubs/issues/2112
    AbstractUserMeta = TypedModelMeta
else:
    TypedModelMeta = object
    AbstractUserMeta = AbstractUser.Meta


class TokenQuerySet[A](QuerySet["Token", A]):
    """Custom QuerySet for Token."""

    def get_tokens(
        self,
        username: str | None = None,
        key: str | None = None,
        include_expired: bool = False,
    ) -> Self:
        """
        Return all the tokens filtered by a specific owner and/or token.

        To avoid filtering by owner or token set them to None
        """
        tokens = self

        if username:
            tokens = tokens.filter(user__username=username)

        if key:
            token_hash = hashlib.sha256(key.encode()).hexdigest()
            tokens = tokens.filter(hash=token_hash)

        if not include_expired:
            tokens = tokens.exclude(expire_at__lt=timezone.now())

        return tokens

    def get_token_or_none(self, token_key: str) -> Optional["Token"]:
        """Return the token with token_key or None."""
        assert isinstance(token_key, str)

        token_hash = hashlib.sha256(token_key.encode()).hexdigest()

        try:
            return cast(
                Token | None,
                (
                    self.select_related(
                        "worker",
                        "activating_worker",
                        "work_request",
                        "work_request__worker",
                        "work_request__created_by",
                    )
                    .exclude(expire_at__lt=timezone.now())
                    .get(hash=token_hash)
                ),
            )
        except Token.DoesNotExist:
            return None

    def expired(self, at: dt.datetime) -> Self:
        """
        Return queryset with tokens that have expired.

        :param at: datetime to check if the tokens are expired.
        :return: tokens whose ``expire_at`` is before the given datetime.
        """
        return self.filter(expire_at__lt=at)


class TokenManager(models.Manager["Token"]):
    """Manager for Token model."""

    def create_worker_activation(self) -> "Token":
        """Create a token suitable for use in worker activation."""
        # These are currently hardcoded to expire in 15 minutes, which
        # should be enough for most providers to be able to launch an
        # instance.
        return self.create(
            token_type=Token.Types.WORKER_ACTIVATION,
            expire_at=timezone.now() + dt.timedelta(minutes=15),
            enabled=True,
        )

    def create_work_request_token(self, work_request: "WorkRequest") -> "Token":
        """Create a limited token for use in a single work request."""
        # Avoid circular imports
        from debusine.db.models.work_requests import WorkRequest

        if work_request.status not in (
            WorkRequest.Statuses.PENDING,
            WorkRequest.Statuses.RUNNING,
        ):
            raise AssertionError(
                f"Cannot issue a token for a WorkRequest that has status "
                f"{work_request.status}"
            )
        if work_request.worker is None:
            raise AssertionError(
                "Cannot issue a token for a WorkRequest that is not assigned "
                "to a worker."
            )
        # Rotate any existing token. We only permit one task per worker & work
        # request, simply delete any stale tokens.
        # This will need to change to support concurrency > 1
        Token.objects.filter(token_type=Token.Types.WORK_REQUEST).filter(
            Q(work_request=work_request)
            | Q(work_request__worker=work_request.worker)
        ).delete()

        return self.create(
            expire_at=timezone.now() + dt.timedelta(days=7),
            enabled=True,
            token_type=Token.Types.WORK_REQUEST,
            work_request=work_request,
        )

    def get_queryset(self) -> TokenQuerySet[Any]:
        """Use the custom QuerySet."""
        return TokenQuerySet(self.model, using=self._db)


class _TokenTypes(models.TextChoices):
    #: Token used for worker activation only
    WORKER_ACTIVATION = "worker-activation", "Worker activation"
    #: Token used by the worker to connect to Debusine
    WORKER = "worker", "Worker"
    #: Limited token for use in a work request
    WORK_REQUEST = "work-request", "Work request"
    #: User authentication token for debusine client
    USER = "user", "User"


class Token(models.Model):
    """
    Database model of a token.

    A token contains a key and other related data. It's used as a shared
    key between debusine server and clients (workers).

    This token model is very similar to rest_framework.authtoken.models.Token.
    The bigger difference is that debusine's token's owner is a CharField,
    the rest_framework owner is a OneToOne foreign key to a user.

    Database-wise we don't store the token itself, but a hash of the token.
    :py:func:`TokenManager.get_token_or_none` can be used to check a provided
    token key against the database.
    """

    Types: TypeAlias = _TokenTypes

    token_type = models.CharField(choices=Types.choices)

    hash = models.CharField(
        max_length=64,
        unique=True,
        verbose_name='Hexadecimal hash, length is 64 chars',
        validators=[MaxLengthValidator(64), MinLengthValidator(64)],
    )
    _worker_id = models.BigIntegerField(
        null=True,
        editable=False,
        db_column="worker_id",
        help_text="Mirror of Worker.token; only for use in constraints",
    )
    _activating_worker_id = models.BigIntegerField(
        null=True,
        editable=False,
        db_column="activating_worker_id",
        help_text=(
            "Mirror of Worker.activation_token; only for use in constraints"
        ),
    )
    user = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        null=True,
        blank=True,
        on_delete=models.PROTECT,
    )
    work_request = models.OneToOneField(
        "WorkRequest",
        null=True,
        blank=True,
        related_name="work_request_token",
        on_delete=models.PROTECT,
    )
    comment = models.CharField(
        max_length=100,
        default='',
        verbose_name='Reason that this token was created',
        blank=True,
    )
    created_at = models.DateTimeField(auto_now_add=True)
    expire_at = models.DateTimeField(null=True, blank=True)

    enabled = models.BooleanField(default=False)

    last_seen_at = models.DateTimeField(
        blank=True,
        null=True,
        help_text="Last time that the token was used",
    )

    key: str

    objects = TokenManager.from_queryset(TokenQuerySet)()

    class Meta(TypedModelMeta):
        base_manager_name = "objects"
        constraints = [
            CheckConstraint(
                name="%(app_label)s_%(class)s_valid_type",
                check=Q(token_type__in=sorted(_TokenTypes)),
            ),
            # The token type must be consistent with its associated fields.
            CheckConstraint(
                name="%(app_label)s_%(class)s_consistent_type",
                check=(
                    Q(
                        token_type=_TokenTypes.WORKER_ACTIVATION,
                        _worker_id__isnull=True,
                        user__isnull=True,
                        work_request_id__isnull=True,
                    )
                    | Q(
                        token_type=_TokenTypes.WORKER,
                        _activating_worker_id__isnull=True,
                        user__isnull=True,
                        work_request_id__isnull=True,
                    )
                    | Q(
                        token_type=_TokenTypes.WORK_REQUEST,
                        _activating_worker_id__isnull=True,
                        user__isnull=True,
                        _worker_id__isnull=True,
                    )
                    | Q(
                        token_type=_TokenTypes.USER,
                        _activating_worker_id__isnull=True,
                        _worker_id__isnull=True,
                        work_request_id__isnull=True,
                    )
                ),
            ),
        ]
        triggers = [
            # Each token type has a corresponding non-nullable foreign key,
            # but we have to wait for the end of the transaction before
            # checking this because in some cases there's also a check
            # constraint on the other side of the foreign key and it
            # therefore requires multiple statements for everything to be
            # consistent.
            #
            # We must fetch the current state of the row at the start of
            # this trigger function, since the value of NEW is fixed when
            # the trigger is scheduled.
            pgtrigger.Trigger(
                name="db_token_required_type_parameters",
                when=pgtrigger.After,
                operation=pgtrigger.Insert | pgtrigger.Update,
                timing=pgtrigger.Deferred,
                func=" ".join(
                    """
                    SELECT * FROM db_token INTO NEW WHERE id = db_token.id;
                    IF NEW.token_type = 'worker-activation'
                       AND NEW.activating_worker_id IS NULL THEN
                        RAISE EXCEPTION
                            'Worker activation token % has no activating worker',
                            NEW.id;
                    ELSIF NEW.token_type = 'worker'
                          AND NEW.worker_id IS NULL THEN
                        RAISE EXCEPTION 'Worker token % has no worker', NEW.id;
                    ELSIF NEW.token_type = 'work-request'
                          AND NEW.work_request_id IS NULL THEN
                        RAISE EXCEPTION
                            'Work request token % has no work request',
                            NEW.id;
                    ELSIF NEW.token_type = 'user' AND NEW.user_id IS NULL THEN
                        RAISE EXCEPTION 'User token % has no user', NEW.id;
                    ELSE
                        RETURN NULL;
                    END IF;
                    """.split()  # noqa: E501
                ),
            )
        ]

    def __str__(self) -> str:
        """Return the hash of the Token."""
        return self.hash

    def save(self, *args: Any, **kwargs: Any) -> None:
        """Save the token. If it's a new token it generates a key."""
        if not self.hash:
            self.key = self._generate_key()
            self.hash = self._generate_hash(self.key)
        super().save(*args, **kwargs)

    def enable(self) -> None:
        """Enable the token and save it."""
        self.enabled = True
        self.save()

    def disable(self) -> None:
        """Disable the token and save it."""
        self.enabled = False
        self.save()

        notifications.notify_worker_token_disabled(self)

    @classmethod
    def _generate_key(cls) -> str:
        """Create and return a key."""
        return secrets.token_hex(32)

    @classmethod
    def _generate_hash(cls, secret: str) -> str:
        """Hash the given secret."""
        return hashlib.sha256(secret.encode()).hexdigest()


SYSTEM_USER_NAME = "_system"


def system_user() -> "User":
    """Return the `_system` user."""
    return User.objects.get(username=SYSTEM_USER_NAME)


class UserQuerySet[A](QuerySet["User", A]):
    """Custom QuerySet for User."""

    @permission_filter(work_request=Allow.PASS)
    def can_display(
        self,
        pc: PermissionContext,  # noqa: ARG002
    ) -> "UserQuerySet[A]":
        """Keep only Users that can be displayed."""
        return self

    @permission_filter()
    def can_manage(self, pc: PermissionContext) -> "UserQuerySet[A]":
        """Keep only Users that the given user can manage."""
        assert pc.user.pk is not None  # Enforced by decorator
        return self.filter(pk=pc.user.pk)


class UserManager[A](DjangoUserManager["User"]):
    """Manager for User model."""

    # We cannot use from_queryset or we hit this problem:
    # https://stackoverflow.com/questions/68367703/adding-custom-queryset-to-usermodel-causes-makemigrations-exception  # noqa: E501
    # Therefore we need to proxy all permission filters here

    def can_display(self, pc: PermissionContext) -> "UserQuerySet[A]":
        """Keep only Users that can be displayed."""
        return self.get_queryset().can_display(pc)

    def can_manage(self, pc: PermissionContext) -> "UserQuerySet[A]":
        """Keep only Users that the given user can manage."""
        return self.get_queryset().can_manage(pc)

    def get_queryset(self) -> UserQuerySet[Any]:
        """Use the custom QuerySet."""
        return UserQuerySet(self.model, using=self._db)


class User(AbstractUser):
    """Debusine user."""

    is_system = models.BooleanField(default=False)

    # mypy seems to struggle changing the manager in a subclass
    objects = UserManager()  # type: ignore[misc]

    class Meta(AbstractUserMeta):
        base_manager_name = "objects"
        constraints = [
            # System users must not have an email address.
            CheckConstraint(
                check=Q(is_system=False) | Q(email=""),
                name="%(app_label)s_%(class)s_non_system_email",
            ),
            # Email addresses of non-system users must be unique.
            UniqueConstraint(
                fields=["email"],
                condition=Q(is_system=False),
                name="%(app_label)s_%(class)s_unique_email",
            ),
            # Enforce non-integer usernames
            CheckConstraint(
                check=~Q(username__regex=r"^\d+$"),
                name="%(app_label)s_%(class)s_non_integer_username",
            ),
        ]

    @permission_check(
        "{user} cannot display user {resource}", work_request=Allow.PASS
    )
    def can_display(self, pc: PermissionContext) -> bool:  # noqa: ARG002, U100
        """Check if the user can be displayed."""
        # Disallow unauthenticated people to enumerate users in the system.
        #
        # In general, we do not assume that user information is automatically
        # public.
        #
        # This could help make life a bit harder, for example, for drive-by
        # spammers trying to enumerate users and their personal information.
        return True

    @permission_check("{user} cannot manage user {resource}")
    def can_manage(self, pc: PermissionContext) -> bool:
        """Check if the user can manage this user."""
        return self.pk == pc.user.pk

    def ui(self, request: "HttpRequest") -> "UserUI":
        """Return a UI helper for this instance."""
        from debusine.web.views.ui.auth import UserUI

        return UserUI.get(request, self)

    def get_absolute_url(self) -> str:
        """Return an absolute URL to this user."""
        return reverse("user:detail", kwargs={"username": self.username})

    def get_absolute_user_id_url(self) -> str:
        """
        Return an absolute numeric URL to this user.

        This is more stable than a username, long-term, for audit purposes.
        """
        return reverse("user:id", kwargs={"pk": self.id})


class Identity(models.Model):
    """
    Identity for a user in a remote user database.

    An Identity is bound if it's associated with a Django user, or unbound if
    no Django user is known for it.
    """

    user = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        related_name="identities",
        null=True,
        on_delete=models.SET_NULL,
    )
    issuer = models.CharField(
        max_length=512,
        help_text="identifier of auhoritative system for this identity",
    )
    subject = models.CharField(
        max_length=512,
        help_text="identifier of the user in the issuer system",
    )
    last_used = models.DateTimeField(
        auto_now=True, help_text="last time this identity has been used"
    )
    claims = models.JSONField(default=dict)

    class Meta(TypedModelMeta):
        constraints = [
            UniqueConstraint(
                fields=["issuer", "subject"],
                name="%(app_label)s_%(class)s_unique_issuer_subject",
            ),
        ]

    def __str__(self) -> str:
        """Return str for the object."""
        return f"{self.issuer}:{self.subject}"


#: Regexp matching the structure of group names
group_name_regex = re.compile(r"^[A-Za-z][A-Za-z0-9+._-]*$")


def is_valid_group_name(value: str) -> bool:
    """Check if value is a valid group name."""
    return bool(group_name_regex.match(value))


def validate_group_name(value: str) -> None:
    """Validate group names."""
    if not is_valid_group_name(value):
        raise ValidationError(
            "%(value)r is not a valid group name", params={"value": value}
        )


class GroupQuerySet[A](QuerySet["Group", A]):
    """Custom QuerySet for Group."""

    @permission_filter()
    def can_display(
        self,
        pc: PermissionContext,  # noqa: ARG002
    ) -> "GroupQuerySet[A]":
        """Keep only groups that can be displayed."""
        return self.all()

    @permission_filter()
    def can_display_audit_log(
        self, pc: PermissionContext
    ) -> "GroupQuerySet[A]":
        """Keep only groups whose audit log can be displayed."""
        return self.can_display(pc)

    @permission_filter()
    def can_manage(self, pc: PermissionContext) -> "GroupQuerySet[A]":
        """Keep only groups that can be managed."""
        # Allow to scope owners
        constraints = Q(
            scope__in=Scope.objects.filter(
                roles__group__users=pc.user,
                roles__role=Scope.Roles.OWNER,
            )
        )
        # Allow to group admins
        constraints |= Q(
            membership__user=pc.user, membership__role=Group.Roles.ADMIN
        )
        return self.filter(constraints).distinct()

    def unused_ephemeral(self) -> "GroupQuerySet[Any]":
        """List ephemeral groups with no roles assigned."""
        constraints = Q()
        # Introspect reverse foreign keys to detect which resources have a
        # *Roles table
        for field in Group._meta.get_fields():
            if (
                field.auto_created
                and not field.concrete
                and field.name.endswith("_roles")
            ):
                # If this resource is currently assigning roles, then the group
                # is still in use
                constraints &= Q(**{f"{field.name}__isnull": True})
        return self.filter(ephemeral=True).filter(constraints)


class GroupManager(models.Manager["Group"]):
    """Manager for Group model."""

    def get_queryset(self) -> GroupQuerySet[Any]:
        """Use the custom QuerySet."""
        return GroupQuerySet(self.model, using=self._db)

    def from_scoped_name(self, scope_group: str) -> "Group":
        """Lookup a group from a scopename/groupname string."""
        scope_name, group_name = scope_group.split("/", 1)
        return Group.objects.get(scope__name=scope_name, name=group_name)

    def create_ephemeral(
        self, workspace: Workspace, name: str, owner: User
    ) -> "Group":
        """Create an ephemeral group for the given user."""
        group = self.create(
            scope=workspace.scope,
            workspace=workspace,
            name=name,
            ephemeral=True,
        )
        # Sets the user as an admin of the newly created group
        group.add_user(owner, Group.Roles.ADMIN)
        return group


class GroupRoleBase(permissions.RoleBase):
    """Group role implementation."""

    implied_by_scope_roles: frozenset["Scope.Roles"]
    implied_by_group_roles: frozenset["GroupRoles"]

    def _setup(self) -> None:
        """Set up implications for a newly constructed role."""
        implied_by_scope_roles: set[Scope.Roles] = set()
        implied_by_group_roles: set[GroupRoles] = {cast(GroupRoles, self)}
        for i in self.implied_by:
            match i:
                case Scope.Roles():
                    implied_by_scope_roles |= i.implied_by_scope_roles
                case Role():
                    # Resolve a role passed during class definition into its
                    # enum instance
                    gr = self.__class__(i.value)
                    implied_by_scope_roles |= gr.implied_by_scope_roles
                    implied_by_group_roles |= gr.implied_by_group_roles
                case _:
                    raise ImproperlyConfigured(
                        f"Group roles do not support implications by {type(i)}"
                    )
        self.implied_by_scope_roles = frozenset(implied_by_scope_roles)
        self.implied_by_group_roles = frozenset(implied_by_group_roles)

    def implies(self, role: "GroupRoles") -> bool:
        """Check if this role implies the given one."""
        return self in role.implied_by_group_roles


class GroupRoles(permissions.Roles, GroupRoleBase, enum.ReprEnum):
    """Available roles for a Scope."""

    ADMIN = Role(
        "admin",
        implied_by=[Scope.Roles.OWNER],
        description="Add and remove other users",
    )

    MEMBER = Role(
        "member",
        implied_by=[ADMIN],
        description="Member",
    )


GroupRoles.setup()


class Group(models.Model):
    """
    Connect users to roles.

    Group names are scoped: groups with the same name in different scopes are
    entirely distinct and unrelated.
    """

    Roles: TypeAlias = GroupRoles

    name = models.CharField(max_length=255, validators=[validate_group_name])
    scope = models.ForeignKey(Scope, null=False, on_delete=models.PROTECT)
    workspace = models.ForeignKey(
        Workspace, null=True, blank=True, on_delete=models.PROTECT
    )
    users = models.ManyToManyField(
        User,
        blank=True,
        through="db.GroupMembership",
        related_name="debusine_groups",
    )
    ephemeral = models.BooleanField(
        default=False, help_text="remove the group if it has no roles assigned"
    )

    objects = GroupManager.from_queryset(GroupQuerySet)()

    class Meta(TypedModelMeta):
        base_manager_name = "objects"
        constraints = [
            UniqueConstraint(
                F("name"),
                F("scope"),
                # Hack to simulate nulls_distinct in Django 4.2
                Coalesce("workspace", Value(-1)),
                name="%(app_label)s_%(class)s_unique_name_scope_workspace",
            ),
            # From Django 5.0
            # UniqueConstraint(
            #     fields=["name", "scope", "workspace"],
            #     name="%(app_label)s_%(class)s_unique_name_scope_workspace",
            #     nulls_distinct=False,
            # ),
        ]

    def __str__(self) -> str:
        """Return the scoped group name."""
        if self.workspace is not None:
            return f"{self.scope.name}/{self.workspace.name}/{self.name}"
        else:
            return f"{self.scope.name}/{self.name}"

    @copy_signature_from(models.Model.save)
    def save(self, *args: Any, **kwargs: Any) -> None:
        """Update audit log on group create/update."""
        old: Group | None = None
        if self.pk:
            old = Group.objects.get(pk=self.pk)
        super().save(*args, **kwargs)
        if old is None:
            self.add_audit_log(GroupAuditLogCreated.create(self))
        else:
            self.add_audit_log(GroupAuditLogUpdated.create_diff(old, self))

    def get_absolute_url(self) -> str:
        """Return an absolute URL to this group."""
        # Prevent circular import
        from debusine.server.scopes import urlconf_scope

        with urlconf_scope(self.scope.name):
            if self.workspace is None:
                return reverse(
                    "groups:detail",
                    kwargs={"group": self.name},
                )
            else:
                return reverse(
                    "workspaces:groups:detail",
                    kwargs={"wname": self.workspace.name, "group": self.name},
                )

    @property
    def name_without_scope(self) -> str:
        """Get the qualified group name without the scope."""
        if self.workspace is not None:
            return f"{self.workspace.name}/{self.name}"
        else:
            return self.name

    def ui(self, request: "HttpRequest") -> "GroupUI":
        """Return a UI helper for this instance."""
        from debusine.web.views.ui.auth import GroupUI

        return GroupUI.get(request, self)

    def get_absolute_url_audit_log(self) -> str:
        """Return an absolute URL to this group's audit log."""
        # Prevent circular import
        from debusine.server.scopes import urlconf_scope

        with urlconf_scope(self.scope.name):
            if self.workspace is None:
                return reverse(
                    "groups:audit-log",
                    kwargs={"group": self.name},
                )
            else:
                return reverse(
                    "workspaces:groups:audit-log",
                    kwargs={"wname": self.workspace.name, "group": self.name},
                )

    @permission_check("{user} cannot display {resource}")
    def can_display(self, pc: PermissionContext) -> bool:  # noqa: ARG002
        """Check if the group can be displayed."""
        return True

    @permission_check("{user} cannot display the audit log of {resource}")
    def can_display_audit_log(self, pc: PermissionContext) -> bool:
        """Check if the group audit log can be displayed."""
        return self.can_display(pc)

    @permission_check("{user} cannot manage {resource}")
    def can_manage(self, pc: PermissionContext) -> bool:
        """Check if the group can be managed."""
        from debusine.db.context import context

        # Shortcut to avoid hitting the database for common cases
        if (
            self.scope == context.scope
            and context.pc_is(pc)
            and not context.scope_roles.isdisjoint(
                GroupRoles.ADMIN.implied_by_scope_roles
            )
        ):
            return True
        return Group.objects.can_manage(pc).filter(pk=self.pk).exists()

    def assign_role(self, resource: models.Model, role: str) -> models.Model:
        """Assign a role on a resource."""
        if get_resource_scope(resource) != self.scope:
            raise ValueError(
                f"{resource.__class__.__name__} {str(resource)!r}"
                f" is not in scope {self.scope}"
            )

        role = resolve_role(resource, role)

        roles_model_cls: models.Model = getattr(
            resource.__class__, "objects"
        ).get_roles_model()
        assignment, _ = getattr(roles_model_cls, "objects").get_or_create(
            resource=resource, group=self, role=role
        )
        return cast(models.Model, assignment)

    def get_user_role(self, user: User) -> GroupRoles:
        """Get the role of an existing group member."""
        try:
            membership = GroupMembership.objects.get(group=self, user=user)
        except GroupMembership.DoesNotExist:
            raise ValueError(f"User {user} is not a member of group {self}")
        return cast(GroupRoles, membership.role)

    def set_user_role(self, user: User, role: Roles) -> "GroupMembership":
        """Set the role of an existing group member."""
        try:
            membership = GroupMembership.objects.get(group=self, user=user)
        except GroupMembership.DoesNotExist:
            raise ValueError(f"User {user} is not a member of group {self}")
        membership.role = role
        membership.save()
        self.add_audit_log(
            GroupAuditLogMemberRoleChanged(user=user.username, role=role)
        )
        return membership

    def add_user(
        self, user: User, role: Roles = Roles.MEMBER
    ) -> "GroupMembership":
        """Add a user to the group."""
        membership, created = GroupMembership.objects.get_or_create(
            group=self, user=user, defaults={"role": role}
        )
        if not created:
            raise ValueError(f"{user} is already a member of group {self}")
        self.add_audit_log(
            GroupAuditLogMemberAdded(user=user.username, role=role)
        )
        return membership

    def remove_user(self, user: User) -> None:
        """Remove an existing member from the group."""
        try:
            membership = GroupMembership.objects.get(group=self, user=user)
        except GroupMembership.DoesNotExist:
            raise ValueError(f"User {user} is not a member of group {self}")
        membership.delete()
        self.add_audit_log(GroupAuditLogMemberRemoved(user=user.username))

    def add_audit_log(
        self, changes: "GroupAuditLogEntry"
    ) -> Optional["GroupAuditLog"]:
        """Add an audit log entry."""
        from debusine.db.context import ContextConsistencyError, context

        if context.user is None or not context.user.is_authenticated:
            if context.permission_checks_disabled:
                return None
            raise ContextConsistencyError("add_audit_log requires a valid user")
        changes.actor = context.user.username
        return GroupAuditLog.objects.create_entry(
            group=self, actor=context.user, changes=changes
        )


class GroupMembership(models.Model):
    """Membership of a user in a group."""

    Roles: TypeAlias = GroupRoles

    group = models.ForeignKey(
        Group, on_delete=models.CASCADE, related_name="membership"
    )
    user = models.ForeignKey(
        User, on_delete=models.CASCADE, related_name="group_memberships"
    )
    role = models.CharField(
        max_length=16, choices=Roles.choices, default=Roles.MEMBER
    )

    class Meta(TypedModelMeta):
        constraints = [
            UniqueConstraint(
                fields=["group", "user"],
                name="%(app_label)s_%(class)s_unique_group_user",
            ),
        ]

    def __str__(self) -> str:
        """Return a string representation."""
        return f"{self.group}:{self.user}"

    def ui(self, request: "HttpRequest") -> "GroupMembershipUI":
        """Return a UI helper for this instance."""
        from debusine.web.views.ui.auth import GroupMembershipUI

        return GroupMembershipUI.get(request, self)

    def get_absolute_url_update(self) -> str:
        """Return an absolute URL to manage this memberhsip."""
        # Prevent circular import
        from debusine.server.scopes import urlconf_scope

        with urlconf_scope(self.group.scope.name):
            if self.group.workspace is None:
                return reverse(
                    "groups:update-member",
                    kwargs={
                        "group": self.group.name,
                        "user": self.user.username,
                    },
                )
            else:
                return reverse(
                    "workspaces:groups:update-member",
                    kwargs={
                        "wname": self.group.workspace.name,
                        "group": self.group.name,
                        "user": self.user.username,
                    },
                )

    def get_absolute_url_remove(self) -> str:
        """Return an absolute URL to remove this memberhsip."""
        # Prevent circular import
        from debusine.server.scopes import urlconf_scope

        with urlconf_scope(self.group.scope.name):
            if self.group.workspace is None:
                return reverse(
                    "groups:remove-member",
                    kwargs={
                        "group": self.group.name,
                        "user": self.user.username,
                    },
                )
            else:
                return reverse(
                    "workspaces:groups:remove-member",
                    kwargs={
                        "wname": self.group.workspace.name,
                        "group": self.group.name,
                        "user": self.user.username,
                    },
                )


class GroupAuditLogEntryTypes(StrEnum):
    """Possible values for GroupAuditLogEntry types."""

    CREATED = "create"
    UPDATED = "update"
    ADD_USER = "add_user"
    REMOVE_USER = "del_user"
    SET_USER_ROLE = "set_role"


class GroupAuditLogEntryBase(pydantic.BaseModel, abc.ABC):
    """Base class for all audit log payloads."""

    model_config = pydantic.ConfigDict(validate_assignment=True, extra="forbid")

    #: Username of the user performing the action
    actor: str = ""

    @abc.abstractmethod
    def description(self) -> str:
        """Format a description for this change."""


class GroupAuditLogCreated(GroupAuditLogEntryBase):
    """Group has been created."""

    type: Literal[GroupAuditLogEntryTypes.CREATED] = (
        GroupAuditLogEntryTypes.CREATED
    )
    name: str
    scope: str
    workspace: str | None = None
    ephemeral: bool

    @classmethod
    def create(cls, group: Group) -> Self:
        """Create an entry populated with group's fields."""
        return cls(
            name=group.name,
            scope=str(group.scope),
            workspace=group.workspace.name if group.workspace else None,
            ephemeral=group.ephemeral,
        )

    def description(self) -> str:
        """Format a description for this change."""
        unknown = "<unknown>"
        group_name = self.name or unknown
        if self.workspace is not None:
            group_name = self.workspace + "/" + group_name
        group_desc = f"group {group_name} in scope {self.scope or unknown}"
        if self.ephemeral:
            return f"{self.actor} created ephemeral {group_desc}"
        else:
            return f"{self.actor} created {group_desc}"


class GroupAuditLogUpdated(GroupAuditLogEntryBase):
    """Group has been updated."""

    type: Literal[GroupAuditLogEntryTypes.UPDATED] = (
        GroupAuditLogEntryTypes.UPDATED
    )
    name: tuple[str, str] | None = None
    scope: tuple[str, str] | None = None
    workspace: tuple[str | None, str | None] | None = None
    ephemeral: tuple[bool, bool] | None = None

    @classmethod
    def create_diff(cls, old: Group, new: Group) -> Self:
        """
        Compute the changes between two Group objects.

        :returns: a dict mapping field names to tuples of ``(old, new)`` values
        """
        name: tuple[str, str] | None = None
        scope: tuple[str, str] | None = None
        workspace: tuple[str | None, str | None] | None = None
        ephemeral: tuple[bool, bool] | None = None
        if old.name != new.name:
            name = (old.name, new.name)
        if old.scope != new.scope:
            scope = (str(old.scope), str(new.scope))
        if old.workspace != new.workspace:
            workspace = (
                old.workspace.name if old.workspace else None,
                new.workspace.name if new.workspace else None,
            )
        if old.ephemeral != new.ephemeral:
            ephemeral = (old.ephemeral, new.ephemeral)
        return cls(
            name=name, scope=scope, workspace=workspace, ephemeral=ephemeral
        )

    def description(self) -> str:
        """Format a description for this change."""
        changes: list[str] = []
        for field in "name", "scope", "workspace", "ephemeral":
            change = getattr(self, field)
            if change is None:
                continue
            changes.append(f"{field}: {change[0]!r}→{change[1]!r}")
        if changes:
            return f"{self.actor} updated {'; '.join(changes)}"
        else:
            return f"{self.actor} updated group, no changes detected"


class GroupAuditLogMembershipBase(GroupAuditLogEntryBase):
    """Base for membership changes descriptions."""

    #: Username of the member affected
    user: str


class GroupAuditLogMemberAdded(GroupAuditLogMembershipBase):
    """Member added to the gruoup."""

    type: Literal[GroupAuditLogEntryTypes.ADD_USER] = (
        GroupAuditLogEntryTypes.ADD_USER
    )
    role: GroupRoles

    def description(self) -> str:
        """Format a description for this change."""
        return f"{self.actor} added {self.user} as {self.role}"


class GroupAuditLogMemberRoleChanged(GroupAuditLogMembershipBase):
    """Member role updated."""

    type: Literal[GroupAuditLogEntryTypes.SET_USER_ROLE] = (
        GroupAuditLogEntryTypes.SET_USER_ROLE
    )
    role: GroupRoles

    def description(self) -> str:
        """Format a description for this change."""
        return f"{self.actor} set {self.user}'s role as {self.role}"


class GroupAuditLogMemberRemoved(GroupAuditLogMembershipBase):
    """Member removed from the gruoup."""

    type: Literal[GroupAuditLogEntryTypes.REMOVE_USER] = (
        GroupAuditLogEntryTypes.REMOVE_USER
    )

    def description(self) -> str:
        """Format a description for this change."""
        return f"{self.actor} removed {self.user}"


GroupAuditLogEntry: TypeAlias = Annotated[
    GroupAuditLogCreated
    | GroupAuditLogUpdated
    | GroupAuditLogMemberAdded
    | GroupAuditLogMemberRoleChanged
    | GroupAuditLogMemberRemoved,
    pydantic.Field(discriminator="type"),
]

group_audit_log_entry_adapter: pydantic.TypeAdapter[GroupAuditLogEntry] = (
    pydantic.TypeAdapter(GroupAuditLogEntry)
)


class GroupAuditLogManager(models.Manager["GroupAuditLog"]):
    """Manager for :py:class:`GroupAuditLog`."""

    def create_entry(
        self,
        group: Group,
        actor: User,
        changes: GroupAuditLogEntry,
    ) -> "GroupAuditLog":
        """Create GroupAuditLog entry."""
        return self.create(
            group=group, actor=actor, changes=changes.model_dump()
        )


class GroupAuditLog(models.Model):
    """Audit log entry for group changes."""

    created_at = models.DateTimeField(auto_now_add=True)
    group = models.ForeignKey(
        Group, on_delete=models.CASCADE, related_name="audit_log"
    )
    actor = models.ForeignKey(User, on_delete=models.SET_NULL, null=True)
    changes = models.JSONField()

    objects = GroupAuditLogManager()

    class Meta(TypedModelMeta):
        base_manager_name = "objects"

    @property
    def changes_model(self) -> GroupAuditLogEntry:
        """Return the pydantic model for changes."""
        return group_audit_log_entry_adapter.validate_python(self.changes)


class ClientEnroll(models.Model):
    """Information from a client wanting to enroll."""

    created_at = models.DateTimeField(auto_now_add=True)
    nonce = models.CharField(max_length=64, unique=True)
    payload = models.JSONField(default=dict)
