# Source code for debusine.db.models.collections

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

    import datetime as dt
    import enum
    import re
    from collections.abc import Generator
    from functools import cached_property
    from typing import (
        Any,
        ClassVar,
        Optional,
        Self,
        TYPE_CHECKING,
        TypeAlias,
        cast,
        override,
    )

    import jsonpath_rw
    from django.conf import settings
    from django.contrib.postgres.constraints import ExclusionConstraint
    from django.contrib.postgres.fields import RangeOperators
    from django.core.exceptions import ImproperlyConfigured, ValidationError
    from django.db import models, transaction
    from django.db.models import (
        CheckConstraint,
        Count,
        Deferrable,
        Exists,
        F,
        JSONField,
        Max,
        OuterRef,
        Q,
        QuerySet,
        UniqueConstraint,
        Value,
    )
    from django.db.models.fields.json import KT
    from django.db.models.functions import Coalesce
    from django.urls import reverse
    from django.utils import timezone

    from debusine.artifacts.models import (
        ArtifactCategory,
        BareDataCategory,
        BaseArtifactDataModel,
        CollectionCategory,
        DebusinePromise,
        SINGLETON_COLLECTION_CATEGORIES,
    )
    from debusine.db.constraints import JsonDataUniqueConstraint
    from debusine.db.models.permissions import (
        Allow,
        Role,
        RoleBase,
        Roles,
        permission_check,
        permission_filter,
    )
    from debusine.db.models.scopes import Scope
    from debusine.db.models.workspaces import Workspace
    from debusine.db.permissioncontext import PermissionContext

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

        from debusine.db.models.artifacts import Artifact
        from debusine.db.models.auth import Token, User
        from debusine.db.models.work_requests import WorkRequest
        from debusine.server.collections import CollectionManagerInterface
        from debusine.web.views.ui.collection_relations import CollectionRelationUI
        from debusine.web.views.ui.collections import CollectionItemUI, CollectionUI
    else:
        TypedModelMeta = object

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

    def is_valid_collection_name(value: str) -> bool:
        """Check if value is a valid scope name."""
        return value == "_" or bool(collection_name_regex.match(value))

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

    class _CollectionRetainsArtifacts(models.TextChoices):
        """Choices for Collection.retains_artifacts."""

        NEVER = "never", "Never"
        WORKFLOW = "workflow", "While workflow is running"
        ALWAYS = "always", "Always"

    class CollectionRoleBase(RoleBase):
        """Collection role implementation."""

        implied_by_scope_roles: frozenset[Scope.Roles]
        implied_by_workspace_roles: frozenset[Workspace.Roles]
        implied_by_collection_roles: frozenset["CollectionRoles"]
        implied_by_public: bool

        @override
        def _setup(self) -> None:
            """Set up implications for a newly constructed role."""
            implied_by_scope_roles: set[Scope.Roles] = set()
            implied_by_workspace_roles: set[Workspace.Roles] = set()
            implied_by_collection_roles: set[CollectionRoles] = {
                cast(CollectionRoles, self)
            }
            implied_by_public: bool = False
            for i in self.implied_by:
                match i:
                    case Workspace.Roles():
                        implied_by_scope_roles |= i.implied_by_scope_roles
                        implied_by_workspace_roles |= i.implied_by_workspace_roles
                        implied_by_public = implied_by_public or i.implied_by_public
                    case Role():
                        # Resolve a role passed during class definition into its
                        # enum instance
                        role = self.__class__(i.value)
                        implied_by_scope_roles |= role.implied_by_scope_roles
                        implied_by_workspace_roles |= (
                            role.implied_by_workspace_roles
                        )
                        implied_by_collection_roles |= (
                            role.implied_by_collection_roles
                        )
                        implied_by_public = (
                            implied_by_public or role.implied_by_public
                        )
                    case _:
                        raise ImproperlyConfigured(
                            f"Collection roles do not support implications by {i!r}"
                        )
            self.implied_by_scope_roles = frozenset(implied_by_scope_roles)
            self.implied_by_workspace_roles = frozenset(implied_by_workspace_roles)
            self.implied_by_collection_roles = frozenset(
                implied_by_collection_roles
            )
            self.implied_by_public = implied_by_public

        def q(self, pc: PermissionContext) -> Q:
            """Return a Q expression to select collections with this role."""
            if not pc.user.is_authenticated:
                if self.implied_by_public:
                    return Q(workspace__public=True)
                else:
                    return Q(pk__in=())

            q = Q(
                roles__group__in=pc.groups(),
                roles__role__in=self.implied_by_collection_roles,
            )
            workspace_q = Q(
                roles__group__in=pc.groups(),
                roles__role__in=self.implied_by_workspace_roles,
            ) | Q(
                scope__in=Scope.objects.filter(
                    roles__group__in=pc.groups(),
                    roles__role__in=self.implied_by_scope_roles,
                )
            )
            if self.implied_by_public:
                workspace_q |= Q(public=True)
            q |= Q(workspace__in=Workspace.objects.filter(workspace_q))

            return q

        def implies(self, role: "CollectionRoles") -> bool:
            """Check if this role implies the given one."""
            return (
                self.implied_by_scope_roles <= role.implied_by_scope_roles
                and self.implied_by_workspace_roles
                <= role.implied_by_workspace_roles
                and self.implied_by_collection_roles
                <= role.implied_by_collection_roles
                and self.implied_by_public <= role.implied_by_public
            )

    class CollectionRoles(Roles, CollectionRoleBase, enum.ReprEnum):
        """Available roles for a Collection."""

        OWNER = Role("owner", implied_by=[Workspace.Roles.OWNER])

        VIEWER = Role("viewer", implied_by=[OWNER, Workspace.Roles.VIEWER])

    CollectionRoles.setup()

    class CollectionQuerySet[A](QuerySet["Collection", A]):
        """Custom QuerySet for Collection."""

        def in_current_scope(self) -> Self:
            """Filter to collections in the current scope."""
            from debusine.db.context import context

            return self.filter(workspace__scope=context.require_scope())

        def in_current_workspace(self) -> Self:
            """Filter to collections in the current workspace."""
            from debusine.db.context import context

            return self.filter(workspace=context.require_workspace())

        def with_role(self, pc: PermissionContext, role: CollectionRoles) -> Self:
            """Keep only resources where the user has the given role."""
            return self.filter(
                Exists(self.model.objects.filter(role.q(pc), pk=OuterRef("pk")))
            )

        @permission_filter(
            work_request=Allow.PASS,
            anonymous=Allow.PASS,
            explicit_grants_role=CollectionRoles.VIEWER,
        )
        def can_display(self, pc: PermissionContext) -> Self:
            """Keep only Collections that can be displayed."""
            return self.with_role(pc, CollectionRoles.VIEWER)

        @permission_filter()
        def can_configure(self, pc: PermissionContext) -> Self:
            """Keep only Collections that can be configured."""
            return self.with_role(pc, CollectionRoles.OWNER)

        @permission_filter()
        def can_create_collection_relation(self, pc: PermissionContext) -> Self:
            """Keep only Collections that can have a collection relation added."""
            return self.with_role(pc, CollectionRoles.OWNER)

        @permission_filter()
        def can_delete(self, pc: PermissionContext) -> Self:
            """Keep only Collections that can be deleted."""
            return self.with_role(pc, CollectionRoles.OWNER).exclude(
                category=CollectionCategory.WORKFLOW_INTERNAL
            )

        def exported_suites(self) -> "CollectionQuerySet[Any]":
            """Filter to exported ``debian:suite`` collections."""
            return (
                self.filter(category=CollectionCategory.SUITE)
                .annotate(
                    exported=Coalesce(
                        "data__exported", Value("true"), output_field=JSONField()
                    )
                )
                .exclude(exported=False)
            )

    class CollectionManager(models.Manager["Collection"]):
        """Manager for Collection model."""

        def get_roles_model(self) -> type["CollectionRole"]:
            """Get the model used for role assignment."""
            return CollectionRole

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

        @staticmethod
        def get_or_create_singleton(
            category: CollectionCategory,
            workspace: Workspace,
            *,
            data: dict[str, Any] | None = None,
        ) -> "tuple[Collection, bool]":
            """Create a singleton collection."""
            if category not in SINGLETON_COLLECTION_CATEGORIES:
                raise ValueError(
                    f"'{category}' is not a singleton collection category"
                )
            return Collection.objects.get_or_create(
                name="_", category=category, workspace=workspace, data=data or {}
            )

    class Collection(models.Model):
        """Model representing a collection."""

        Roles: TypeAlias = CollectionRoles
        RetainsArtifacts: TypeAlias = _CollectionRetainsArtifacts

        name = models.CharField(
            max_length=255, validators=[validate_collection_name]
        )
        category = models.CharField(max_length=255)
        full_history_retention_period = models.DurationField(
            null=True,
            blank=True,
            help_text="Period after removing an item from the collection"
            " during which the related artifact is not allowed to expire",
        )
        metadata_only_retention_period = models.DurationField(
            null=True,
            blank=True,
            help_text="Period after removing an item from the collection during"
            " which its metadata is kept in the history of the collection",
        )

        workspace = models.ForeignKey(
            Workspace, on_delete=models.PROTECT, related_name="collections"
        )
        retains_artifacts = models.CharField(
            max_length=8,
            choices=RetainsArtifacts.choices,
            default=RetainsArtifacts.ALWAYS,
            help_text="Controls if artifacts in this collection should be retained"
            " rather than being allowed to expire",
        )

        child_artifacts = models.ManyToManyField(
            "db.Artifact",
            through="db.CollectionItem",
            through_fields=("parent_collection", "artifact"),
            related_name="parent_collections",
        )
        child_collections = models.ManyToManyField(
            "self",
            through="db.CollectionItem",
            through_fields=("parent_collection", "collection"),
            related_name="parent_collections",
            symmetrical=False,
        )

        data = models.JSONField(default=dict, blank=True)

        objects = CollectionManager.from_queryset(CollectionQuerySet)()

        class Meta(TypedModelMeta):
            base_manager_name = "objects"
            constraints = [
                UniqueConstraint(
                    fields=["name", "category", "workspace"],
                    name="%(app_label)s_%(class)s_unique_name_category_workspace",
                ),
                CheckConstraint(
                    check=~Q(name=""), name="%(app_label)s_%(class)s_name_not_empty"
                ),
                CheckConstraint(
                    check=(
                        (
                            ~Q(category__in=sorted(SINGLETON_COLLECTION_CATEGORIES))
                            & ~Q(name__startswith="_")
                        )
                        | Q(
                            category__in=sorted(SINGLETON_COLLECTION_CATEGORIES),
                            name="_",
                        )
                    ),
                    name="%(app_label)s_%(class)s_name_not_reserved",
                ),
                CheckConstraint(
                    check=~Q(category=""),
                    name="%(app_label)s_%(class)s_category_not_empty",
                ),
            ]

        @override
        def __str__(self) -> str:
            """Return id, name, category."""
            # Stringify using a valid lookup syntax
            return f"{self.name}@{self.category}"

        @override
        def __repr__(self) -> str:
            """Return id, name, category."""
            # Stringify using a valid lookup syntax
            return f"<Collection: {self.name}@{self.category} ({self.pk})>"

        def get_absolute_url(self) -> str:
            """Return the canonical URL to display the collection."""
            from debusine.server.scopes import urlconf_scope

            with urlconf_scope(self.workspace.scope.name):
                return reverse(
                    "workspaces:collections:detail",
                    kwargs={
                        "wname": self.workspace.name,
                        "ccat": self.category,
                        "cname": self.name,
                    },
                )

        def get_absolute_url_search(self) -> str:
            """Return the canonical URL to search inside the collection."""
            from debusine.server.scopes import urlconf_scope

            with urlconf_scope(self.workspace.scope.name):
                return reverse(
                    "workspaces:collections:search",
                    kwargs={
                        "wname": self.workspace.name,
                        "ccat": self.category,
                        "cname": self.name,
                    },
                )

        @cached_property
        def manager(self) -> "CollectionManagerInterface":
            """Get collection manager for this collection category."""
            # Local import to avoid circular dependency
            from debusine.server.collections import CollectionManagerInterface

            return CollectionManagerInterface.get_manager_for(self)

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

            return CollectionUI.get(request, self)

        def grant_role(
            self, token: "Token", role: CollectionRoles, *, created_by: "User"
        ) -> "CollectionTokenGrant":
            """Grant a role on this collection to a token."""
            assert token.explicit_grants_only
            grant, _ = CollectionTokenGrant.objects.get_or_create(
                token=token, collection=self, role=role, created_by=created_by
            )
            return grant

        def has_role(self, pc: PermissionContext, role: CollectionRoles) -> bool:
            """Check if the user has the given role on this Collection."""
            from debusine.db.context import context

            # Honor implied_by_public immediately as it's independent from the user
            if role.implied_by_public and self.workspace.public:
                return True

            if not pc.user.is_authenticated:
                return False

            if context.pc_is(pc) and context.workspace == self.workspace:
                # Allowed as implied by scope roles the user has
                if not context.scope_roles.isdisjoint(role.implied_by_scope_roles):
                    return True

                # Allowed as implied by workspace roles the user has
                if not context.workspace_roles.isdisjoint(
                    role.implied_by_workspace_roles
                ):
                    return True

            return (
                Collection.objects.with_role(pc, role).filter(pk=self.pk).exists()
            )

        @permission_check(
            "{user} cannot display collection {resource}",
            work_request=Allow.PASS,
            anonymous=Allow.PASS,
            explicit_grants_role=CollectionRoles.VIEWER,
        )
        def can_display(self, pc: PermissionContext) -> bool:
            """Check if the collection can be displayed."""
            return self.has_role(pc, CollectionRoles.VIEWER)

        @permission_check(
            "{user} cannot configure collection {resource}",
        )
        def can_configure(self, pc: PermissionContext) -> bool:
            """Check if the collection can be configured."""
            return self.has_role(pc, CollectionRoles.OWNER)

        @permission_check(
            "{user} cannot delete collection {resource}",
        )
        def can_delete(self, pc: PermissionContext) -> bool:
            """Check if the collection can be deleted."""
            return (
                self.category != CollectionCategory.WORKFLOW_INTERNAL
                and self.has_role(pc, CollectionRoles.OWNER)
            )

        @permission_check(
            "{user} cannot create collection relation in {resource}",
        )
        def can_create_collection_relation(self, pc: PermissionContext) -> bool:
            """Check if a relation can be created from this collection."""
            return self.has_role(pc, CollectionRoles.OWNER)

        def get_item_stats(
            self,
        ) -> Generator[dict[str, Any]]:
            """Return a dict of item counts for each item type."""
            for rec in (
                self.child_items.values(
                    "child_type", "category", active=Q(removed_at__isnull=True)
                )
                .annotate(count=Count("id"))
                .order_by("child_type", "category", "active")
            ):
                child_type = CollectionItem.Types(rec["child_type"])
                yield {
                    "type": child_type,
                    "category": rec["category"],
                    "active": rec["active"],
                    "count": rec["count"],
                }

        def delete_collection(self) -> None:
            """
            Delete this collection, and all its CollectionItem entries.

            Does not delete the associated Collections or Artifacts.
            """
            self.child_items.all().delete()
            self.delete()

    class CollectionRole(models.Model):
        """Role assignment for collection."""

        resource = models.ForeignKey(
            Collection,
            on_delete=models.CASCADE,
            related_name="roles",
        )

        group = models.ForeignKey(
            "Group",
            on_delete=models.CASCADE,
            related_name="collection_roles",
        )

        role = models.CharField(max_length=16, choices=CollectionRoles.choices)

        class Meta(TypedModelMeta):
            constraints = [
                UniqueConstraint(
                    fields=["resource", "group", "role"],
                    name="%(app_label)s_%(class)s_unique_resource_group_role",
                ),
            ]

    class CollectionItemQuerySet(QuerySet["CollectionItem"]):
        """Custom QuerySet for CollectionItem."""

        def active(self) -> "CollectionItemQuerySet":
            """Filter to only active collection items."""
            return super().filter(removed_at__isnull=True)

        def active_at(self, at: dt.datetime) -> "CollectionItemQuerySet":
            """Filter to collection items that were active at a given time."""
            return self.filter(
                Q(created_at__lte=at),
                Q(removed_at__isnull=True) | Q(removed_at__gt=at),
            )

        def artifacts_in_collection(
            self, parent_collection: Collection, category: ArtifactCategory
        ) -> "CollectionItemQuerySet":
            """
            Filter to artifacts in a given parent collection.

            This is tuned to use the indexes on :py:class:`CollectionItem`.
            """
            return self.filter(
                parent_collection=parent_collection,
                # Technically redundant with parent_collection, but clues
                # PostgreSQL into using the correct index.
                parent_category=parent_collection.category,
                child_type=CollectionItem.Types.ARTIFACT,
                category=category,
            )

    class CollectionItemManager(models.Manager["CollectionItem"]):
        """Manager for CollectionItem model."""

        @staticmethod
        def create_from_bare_data(
            category: BareDataCategory,
            *,
            parent_collection: Collection,
            name: str,
            data: BaseArtifactDataModel | dict[str, Any],
            created_at: dt.datetime | None = None,
            created_by_user: "User",
            created_by_workflow: Optional["WorkRequest"],
            replaced_by: Optional["CollectionItem"] = None,
        ) -> "CollectionItem":
            """Create a CollectionItem from bare data."""
            if isinstance(data, BaseArtifactDataModel):
                data = data.model_dump(mode="json", exclude_unset=True)

            match category:
                case BareDataCategory.PROMISE:
                    # Raise ValueError if data is not valid
                    DebusinePromise(**data)

            kwargs: dict[str, Any] = {}
            if created_at is not None:
                kwargs["created_at"] = created_at
            if replaced_by is not None:
                kwargs["removed_at"] = replaced_by.created_at
                kwargs["removed_by_user"] = replaced_by.created_by_user
                kwargs["removed_by_workflow"] = replaced_by.created_by_workflow

            return CollectionItem.objects.create(
                parent_collection=parent_collection,
                name=name,
                child_type=CollectionItem.Types.BARE,
                category=category,
                data=data,
                created_by_user=created_by_user,
                created_by_workflow=created_by_workflow,
                **kwargs,
            )

        @staticmethod
        def create_from_artifact(
            artifact: "Artifact",
            *,
            parent_collection: Collection,
            name: str,
            data: dict[str, Any],
            created_at: dt.datetime | None = None,
            created_by_user: "User",
            created_by_workflow: Optional["WorkRequest"] = None,
            replaced_by: Optional["CollectionItem"] = None,
        ) -> "CollectionItem":
            """Create a CollectionItem from the artifact."""
            kwargs: dict[str, Any] = {}
            if created_at is not None:
                kwargs["created_at"] = created_at
            if replaced_by is not None:
                kwargs["removed_at"] = replaced_by.created_at
                kwargs["removed_by_user"] = replaced_by.created_by_user
                kwargs["removed_by_workflow"] = replaced_by.created_by_workflow

            return CollectionItem.objects.create(
                parent_collection=parent_collection,
                name=name,
                artifact=artifact,
                child_type=CollectionItem.Types.ARTIFACT,
                category=artifact.category,
                data=data,
                created_by_user=created_by_user,
                created_by_workflow=created_by_workflow,
                **kwargs,
            )

        @staticmethod
        def create_from_collection(
            collection: Collection,
            *,
            parent_collection: Collection,
            name: str,
            data: dict[str, Any],
            created_at: dt.datetime | None = None,
            created_by_user: "User",
            created_by_workflow: Optional["WorkRequest"] = None,
            replaced_by: Optional["CollectionItem"] = None,
        ) -> "CollectionItem":
            """Create a CollectionItem from the collection."""
            kwargs: dict[str, Any] = {}
            if created_at is not None:
                kwargs["created_at"] = created_at
            if replaced_by is not None:
                kwargs["removed_at"] = replaced_by.created_at
                kwargs["removed_by_user"] = replaced_by.created_by_user
                kwargs["removed_by_workflow"] = replaced_by.created_by_workflow

            return CollectionItem.objects.create(
                parent_collection=parent_collection,
                name=name,
                collection=collection,
                child_type=CollectionItem.Types.COLLECTION,
                category=collection.category,
                data=data,
                created_by_user=created_by_user,
                created_by_workflow=created_by_workflow,
                **kwargs,
            )

        def drop_full_history(self, at: dt.datetime) -> None:
            """
            Drop artifacts from collections after full_history_retention_period.

            :param at: datetime to check if the artifacts are old enough.
            """
            self.get_queryset().exclude(removed_at__isnull=True).exclude(
                artifact__isnull=True
            ).filter(
                # https://github.com/typeddjango/django-stubs/issues/1548
                removed_at__lt=(
                    at - F("parent_collection__full_history_retention_period")  # type: ignore[operator] # noqa: E501
                )
            ).update(artifact=None)

        def drop_metadata(self, at: dt.datetime) -> None:
            """
            Delete old collection items.

            After full_history_retention_period + metadata_only_retention_period.

            :param at: datetime to check if the collection item is old enough.
            """
            self.get_queryset().exclude(removed_at__isnull=True).filter(
                # https://github.com/typeddjango/django-stubs/issues/1548
                removed_at__lt=(
                    at  # type: ignore[operator]
                    - F("parent_collection__full_history_retention_period")
                    - F("parent_collection__metadata_only_retention_period")
                )
            ).delete()

    class _CollectionItemTypes(models.TextChoices):
        """Choices for the CollectionItem.type."""

        BARE = "b", "Bare"
        ARTIFACT = "a", "Artifact"
        COLLECTION = "c", "Collection"

    class CollectionItem(models.Model):
        """CollectionItem model."""

        name = models.CharField(max_length=255)

        Types: TypeAlias = _CollectionItemTypes

        child_type = models.CharField(max_length=1, choices=Types.choices)

        # category duplicates the category of the artifact or collection of this
        # item, so when the underlying artifact or collection is deleted the
        # category is retained
        category = models.CharField(max_length=255)

        parent_collection = models.ForeignKey(
            Collection,
            on_delete=models.PROTECT,
            related_name="child_items",
        )
        parent_category = models.CharField(max_length=255)

        collection = models.ForeignKey(
            Collection,
            on_delete=models.PROTECT,
            related_name="collection_items",
            null=True,
        )
        artifact = models.ForeignKey(
            "Artifact",
            on_delete=models.PROTECT,
            related_name="collection_items",
            null=True,
        )

        data = models.JSONField(default=dict, blank=True)

        created_at = models.DateTimeField(default=timezone.now)
        created_by_user = models.ForeignKey(
            settings.AUTH_USER_MODEL,
            on_delete=models.PROTECT,
            related_name="user_created_%(class)s",
        )
        created_by_workflow = models.ForeignKey(
            "WorkRequest",
            on_delete=models.SET_NULL,
            null=True,
            related_name="workflow_created_%(class)s",
        )

        removed_at = models.DateTimeField(blank=True, null=True)
        removed_by_user = models.ForeignKey(
            settings.AUTH_USER_MODEL,
            on_delete=models.PROTECT,
            null=True,
            related_name="user_removed_%(class)s",
        )
        removed_by_workflow = models.ForeignKey(
            "WorkRequest",
            on_delete=models.SET_NULL,
            null=True,
            related_name="workflow_removed_%(class)s",
        )

        objects = CollectionItemManager.from_queryset(CollectionItemQuerySet)()

        class Meta(TypedModelMeta):
            base_manager_name = "objects"
            constraints = [
                JsonDataUniqueConstraint(
                    fields=[
                        "category",
                        "data->>'codename'",
                        "data->>'architecture'",
                        "data->>'variant'",
                        "data->>'backend'",
                        "parent_collection",
                    ],
                    condition=Q(
                        parent_category=CollectionCategory.ENVIRONMENTS,
                        removed_at__isnull=True,
                    ),
                    nulls_distinct=False,
                    name="%(app_label)s_%(class)s_unique_debian_environments",
                ),
                UniqueConstraint(
                    fields=["name", "parent_collection"],
                    condition=Q(removed_at__isnull=True),
                    name="%(app_label)s_%(class)s_unique_active_name",
                ),
                # Prevent direct way to add a collection to itself.
                # It is still possible to add loops of collections. The Manager
                # should avoid it
                CheckConstraint(
                    check=~Q(collection=F("parent_collection")),
                    name="%(app_label)s_%(class)s_distinct_parent_collection",
                ),
                CheckConstraint(
                    name="%(app_label)s_%(class)s_childtype_removedat_consistent",
                    check=(
                        Q(
                            child_type=_CollectionItemTypes.BARE,
                            collection__isnull=True,
                            artifact__isnull=True,
                        )
                        | (
                            Q(
                                child_type=_CollectionItemTypes.ARTIFACT,
                                collection__isnull=True,
                            )
                            & (
                                Q(artifact__isnull=False)
                                | Q(removed_at__isnull=False)
                            )
                        )
                        | (
                            Q(
                                child_type=_CollectionItemTypes.COLLECTION,
                                artifact__isnull=True,
                            )
                            & (
                                Q(collection__isnull=False)
                                | Q(removed_at__isnull=False)
                            )
                        )
                    ),
                ),
            ]
            indexes = [
                models.Index("name", name="%(app_label)s_ci_name_idx"),
                models.Index(
                    F("parent_collection"),
                    KT("data__package"),
                    KT("data__version"),
                    name="%(app_label)s_ci_suite_source_idx",
                    condition=Q(
                        parent_category=CollectionCategory.SUITE,
                        child_type=_CollectionItemTypes.ARTIFACT,
                        category=ArtifactCategory.SOURCE_PACKAGE,
                    ),
                ),
                models.Index(
                    F("parent_collection"),
                    KT("data__srcpkg_name"),
                    KT("data__srcpkg_version"),
                    name="%(app_label)s_ci_suite_binary_source_idx",
                    condition=Q(
                        parent_category=CollectionCategory.SUITE,
                        child_type=_CollectionItemTypes.ARTIFACT,
                        category=ArtifactCategory.BINARY_PACKAGE,
                    ),
                ),
                models.Index(
                    F("parent_collection"),
                    KT("data__path"),
                    name="%(app_label)s_ci_suite_index_path_idx",
                    condition=Q(
                        parent_category=CollectionCategory.SUITE,
                        child_type=_CollectionItemTypes.ARTIFACT,
                        category=ArtifactCategory.REPOSITORY_INDEX,
                    ),
                ),
                models.Index(
                    F("parent_collection"),
                    KT("data__path"),
                    name="%(app_label)s_ci_archive_index_path_idx",
                    condition=Q(
                        parent_category=CollectionCategory.ARCHIVE,
                        child_type=_CollectionItemTypes.ARTIFACT,
                        category=ArtifactCategory.REPOSITORY_INDEX,
                    ),
                ),
                # Optimize generating suite indexes.
                models.Index(
                    F("parent_collection"),
                    KT("data__component"),
                    F("created_at"),
                    F("removed_at"),
                    name="%(app_label)s_ci_suite_all_sources_idx",
                    condition=Q(
                        parent_category=CollectionCategory.SUITE,
                        child_type=_CollectionItemTypes.ARTIFACT,
                        category=ArtifactCategory.SOURCE_PACKAGE,
                    ),
                ),
                models.Index(
                    F("parent_collection"),
                    KT("data__component"),
                    KT("data__architecture"),
                    F("created_at"),
                    F("removed_at"),
                    name="%(app_label)s_ci_suite_all_binaries_idx",
                    condition=Q(
                        parent_category=CollectionCategory.SUITE,
                        child_type=_CollectionItemTypes.ARTIFACT,
                        category=ArtifactCategory.BINARY_PACKAGE,
                    ),
                ),
                # Optimize latest:TASKNAME_PACKAGE_ARCHITECTURE lookups in
                # debian:qa-results collections.
                models.Index(
                    F("parent_collection"),
                    KT("data__task_name"),
                    KT("data__package"),
                    KT("data__architecture"),
                    name="%(app_label)s_ci_qa_results_latest_idx",
                    condition=Q(
                        parent_category=CollectionCategory.QA_RESULTS,
                        removed_at__isnull=True,
                    ),
                ),
            ]

        @override
        def __str__(self) -> str:
            """Return id, name, collection_id, child_type."""
            item_info = (
                f" Artifact id: {self.artifact.id}"
                if self.artifact
                else (
                    f" Collection id: {self.collection.id}"
                    if self.collection
                    else ""
                )
            )

            return (
                f"Id: {self.id} Name: {self.name} "
                f"Parent collection id: {self.parent_collection_id} "
                f"Child type: {self.child_type}"
                f"{item_info}"
            )

        @override
        def save(self, *args: Any, **kwargs: Any) -> None:
            """Populate `parent_category` on save."""
            if not self.parent_category:
                self.parent_category = self.parent_collection.category
            super().save(*args, **kwargs)

        def get_absolute_url(self) -> str:
            """Return the canonical URL to display the item."""
            from debusine.server.scopes import urlconf_scope

            with urlconf_scope(self.parent_collection.workspace.scope.name):
                return reverse(
                    "workspaces:collections:item_detail",
                    kwargs={
                        "wname": self.parent_collection.workspace.name,
                        "ccat": self.parent_collection.category,
                        "cname": self.parent_collection.name,
                        "iid": self.pk,
                        "iname": self.name,
                    },
                )

        @staticmethod
        def expand_variables(
            variables: dict[str, str], reference_data: dict[Any, Any]
        ) -> dict[str, str]:
            """
            Expand JSONPath variables against some data.

            The data will normally come from an Artifact.
            """
            jsonpaths = {}
            for name, path in variables.items():
                if name.startswith("$"):
                    try:
                        jsonpaths[name[1:]] = jsonpath_rw.parse(path)
                    except Exception as e:
                        raise ValueError(e)

            expanded_variables = {}
            for name, jsonpath in jsonpaths.items():
                matches = jsonpath.find(reference_data)
                if len(matches) == 1:
                    expanded_variables[name] = matches[0].value
                elif len(matches) > 1:
                    raise ValueError(
                        "Too many values expanding",
                        variables[f"${name}"],
                        reference_data,
                    )
                else:
                    raise KeyError(variables[f"${name}"], reference_data)

            for name, value in variables.items():
                if not name.startswith("$"):
                    if name in expanded_variables:
                        raise ValueError(
                            f"Cannot set both '${name}' and '{name}' variables"
                        )
                    expanded_variables[name] = value

            return expanded_variables

        @staticmethod
        def expand_name(
            name_template: str, expanded_variables: dict[str, str]
        ) -> str:
            """Format item name following item_template."""
            return name_template.format(**expanded_variables)

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

            return CollectionItemUI.get(request, self)

    class CollectionItemMatchConstraint(models.Model):
        """
        Enforce matching-value constraints on collection items.

        All instances of this model with the same :py:attr:`collection`,
        :py:attr:`constraint_name`, and :py:attr:`key` must have the same
        :py:attr:`value`.
        """

        objects: ClassVar[models.Manager["CollectionItemMatchConstraint"]] = (
            models.Manager["CollectionItemMatchConstraint"]()
        )

        collection = models.ForeignKey(
            Collection,
            on_delete=models.CASCADE,
            related_name="item_match_constraints",
        )
        # This is deliberately a bare ID, not a foreign key: some constraints
        # take into account even items that no longer exist but were in a
        # collection in the past.  Also note that the collection item's parent
        # collection may not be the same as this model's collection field: some
        # collections impose indirect constraints via child collections.
        collection_item_id = models.BigIntegerField()
        constraint_name = models.CharField(max_length=255)
        key = models.TextField()
        value = models.TextField()

        class Meta(TypedModelMeta):
            constraints = [
                ExclusionConstraint(
                    name="%(app_label)s_%(class)s_match_value",
                    expressions=(
                        (F("collection"), RangeOperators.EQUAL),
                        (F("constraint_name"), RangeOperators.EQUAL),
                        (F("key"), RangeOperators.EQUAL),
                        (F("value"), RangeOperators.NOT_EQUAL),
                    ),
                )
            ]
            indexes = [
                models.Index(
                    name="%(app_label)s_cimc_collection_item_idx",
                    fields=["collection_item_id"],
                )
            ]

    class _CollectionRelationTypes(models.TextChoices):
        """Types of relations."""

        SUITE_FORKED_FROM = "suite-forked-from", "Forked from"
        SUITE_BASED_ON = "suite-based-on", "Based on"
        SUITE_REQUIRES = "suite-requires", "Requires"
        SUITE_TARGETING = "suite-targeting", "Targeting"
        SUITE_DEFAULT_QA_RESULTS = "suite-default-qa-results", "Default QA results"

    class CollectionRelation(models.Model):
        """Model relationships between collections."""

        Types: TypeAlias = _CollectionRelationTypes

        source = models.ForeignKey(
            Collection, on_delete=models.CASCADE, related_name="relations"
        )
        source_category = models.CharField(max_length=255)

        target = models.ForeignKey(
            Collection, on_delete=models.CASCADE, related_name="targeted_by"
        )
        target_category = models.CharField(max_length=255)

        type = models.CharField(
            max_length=32, choices=_CollectionRelationTypes.choices
        )

        position = models.PositiveIntegerField(null=True, blank=True)

        created_at = models.DateTimeField(auto_now_add=True)
        created_by_user = models.ForeignKey(
            "User",
            related_name="created_collection_relations",
            on_delete=models.PROTECT,
        )

        updated_at = models.DateTimeField(null=True, blank=True)
        updated_by_user = models.ForeignKey(
            "User",
            related_name="updated_collection_relations",
            on_delete=models.PROTECT,
            null=True,
        )

        class Meta(TypedModelMeta):
            base_manager_name = "objects"

            constraints = [
                # Enforce that `requires` relations have a position, others not
                models.CheckConstraint(
                    name="%(app_label)s_%(class)s_position",
                    check=(
                        Q(
                            type=_CollectionRelationTypes.SUITE_REQUIRES,
                            position__isnull=False,
                        )
                        | Q(
                            type__in=[
                                _CollectionRelationTypes.SUITE_FORKED_FROM,
                                _CollectionRelationTypes.SUITE_BASED_ON,
                                _CollectionRelationTypes.SUITE_TARGETING,
                                _CollectionRelationTypes.SUITE_DEFAULT_QA_RESULTS,
                            ],
                            position__isnull=True,
                        )
                    ),
                ),
                # Enforce that `requires` relations have unique positions per source
                models.UniqueConstraint(
                    name="%(app_label)s_%(class)s_suite_requires_no_duplicate_position",
                    fields=["source", "type", "position"],
                    # this only applies for SUITE_REQUIRES: the other ones have
                    # cardinality max 1 enforced. Not using condition so it can
                    # be deferred.
                    deferrable=Deferrable.DEFERRED,
                ),
                models.UniqueConstraint(
                    name="%(app_label)s_%(class)s_cardinality_checks_max_1",
                    fields=["source", "type"],
                    condition=(
                        Q(type=_CollectionRelationTypes.SUITE_FORKED_FROM)
                        | Q(type=_CollectionRelationTypes.SUITE_BASED_ON)
                        | Q(type=_CollectionRelationTypes.SUITE_TARGETING)
                        | Q(type=_CollectionRelationTypes.SUITE_DEFAULT_QA_RESULTS)
                    ),
                ),
                # Enforce that `requires` relations do not have repeated targets
                models.UniqueConstraint(
                    name="%(app_label)s_%(class)s_suite_requires_no_duplicate_target",
                    fields=["source", "type", "target"],
                    condition=Q(type=_CollectionRelationTypes.SUITE_REQUIRES),
                ),
                models.CheckConstraint(
                    name="%(app_label)s_%(class)s_categories_correct",
                    check=(
                        # SUITE_FORKED_FROM: both must be suites
                        Q(
                            type=_CollectionRelationTypes.SUITE_FORKED_FROM,
                            source_category=CollectionCategory.SUITE,
                            target_category=CollectionCategory.SUITE,
                        )
                        # SUITE_BASED_ON: both must be suites
                        | Q(
                            type=_CollectionRelationTypes.SUITE_BASED_ON,
                            source_category=CollectionCategory.SUITE,
                            target_category=CollectionCategory.SUITE,
                        )
                        # SUITE_REQUIRES: both must be suites
                        | Q(
                            type=_CollectionRelationTypes.SUITE_REQUIRES,
                            source_category=CollectionCategory.SUITE,
                            target_category=CollectionCategory.SUITE,
                        )
                        # SUITE_TARGETING: both must be suites
                        | Q(
                            type=_CollectionRelationTypes.SUITE_TARGETING,
                            source_category=CollectionCategory.SUITE,
                            target_category=CollectionCategory.SUITE,
                        )
                        # SUITE_DEFAULT_QA_RESULTS: collection must be a suite,
                        # target must be qa-results
                        | Q(
                            type=_CollectionRelationTypes.SUITE_DEFAULT_QA_RESULTS,
                            source_category=CollectionCategory.SUITE,
                            target_category=CollectionCategory.QA_RESULTS,
                        )
                    ),
                ),
            ]

        @override
        def __str__(self) -> str:
            """Return string representation of the CollectionRelation."""
            if self.position is not None:
                position_str = f", position {self.position}"
            else:
                position_str = ""

            return (
                f"{self.source.name} → {self.target.name} "
                f"({CollectionRelation.Types(self.type).label}{position_str})"
            )

        @override
        def save(
            self, *args: Any, updated_by_user: "User | None" = None, **kwargs: Any
        ) -> None:
            """Populate denormalized categories and enforce update audit fields."""
            if self.pk is not None:
                # Updating an existing CollectionRelation: check that the User
                # is provided
                if updated_by_user is None:
                    raise ValueError(
                        "updated_by_user must be set "
                        "when updating a CollectionRelation"
                    )

                if "update_fields" in kwargs:
                    # "Model.save.update_fields" is not used in Debusine
                    # at the moment. In order to avoid having to deal with
                    # "update_fields" not having "update_by_user": just check that
                    # is not used in this particular case
                    raise ValueError(
                        "update_fields is not supported when updating "
                        "CollectionRelation"
                    )

                self.updated_by_user = updated_by_user
                self.updated_at = timezone.now()

            self.position = self._position_for_create(
                source=self.source,
                relation_type=_CollectionRelationTypes(self.type),
                wanted_position=self.position,
            )

            self.source_category = self.source.category
            self.target_category = self.target.category

            with transaction.atomic():
                if self.position is not None:
                    CollectionRelation.objects.filter(
                        source=self.source,
                        type=self.type,
                        position__gte=self.position,
                    ).update(position=F("position") + 1)

                super().save(*args, **kwargs)

        @permission_check(
            "{user} cannot delete collection relation {resource}",
        )
        def can_delete(self, pc: PermissionContext) -> bool:
            """
            Check if the collection relation can be deleted.

            It can be deleted if the source collection can be configured.
            """
            return self.source.can_configure(pc)

        @classmethod
        def get_max_position(
            cls, collection: Collection, relation_type: _CollectionRelationTypes
        ) -> int:
            """
            Return the current maximum position for a given relation type.

            Return 0 if no relation of this type exists.
            """
            return (
                cls.objects.filter(
                    source=collection,
                    type=relation_type,
                ).aggregate(max_position=Max("position"))["max_position"]
                or 0
            )

        @classmethod
        def _position_for_create(
            cls,
            *,
            source: Collection,
            relation_type: _CollectionRelationTypes,
            wanted_position: int | None,
        ) -> int | None:
            """
            Return the normalized position for a new relation.

            Keep positions contiguous from 1 to the maximum position.
            """
            if wanted_position is None:
                return None

            current_max_position = cls.get_max_position(
                collection=source, relation_type=relation_type
            )

            if wanted_position < 1:
                return 1
            elif wanted_position > current_max_position + 1:
                return current_max_position + 1
            else:
                return wanted_position

        @override
        def delete(self, *args: Any, **kwargs: Any) -> tuple[int, dict[str, int]]:
            """Delete this relation and shift later positions."""
            with transaction.atomic():
                result = super().delete(*args, **kwargs)

                if self.position is not None:
                    CollectionRelation.objects.filter(
                        source=self.source,
                        type=self.type,
                        position__gt=self.position,
                    ).update(
                        position=F("position") - 1,
                    )

            return result

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

            return CollectionRelationUI.get(request, self)

    class CollectionTokenGrantQuerySet[A](QuerySet["CollectionTokenGrant", A]):
        """Custom QuerySet for CollectionTokenGrant."""

        def granted_resources(self, role: RoleBase) -> CollectionQuerySet[Any]:
            """Find collections granted by this queryset with this role."""
            assert isinstance(role, CollectionRoles)
            role_grants = self.filter(role=role)
            return Collection.objects.filter(
                # An explicit grant is enough.
                Q(token_grants__in=role_grants)
                # Granting access to an archive grants access to all its suites.
                | (
                    Q(category=CollectionCategory.SUITE)
                    & Exists(
                        Collection.objects.filter(
                            category=CollectionCategory.ARCHIVE,
                            workspace=OuterRef("workspace"),
                            token_grants__in=role_grants,
                        )
                    )
                )
            )

    class CollectionTokenGrantManager(models.Manager["CollectionTokenGrant"]):
        """Manager for the CollectionTokenGrant model."""

        @override
        def get_queryset(self) -> CollectionTokenGrantQuerySet[Any]:
            return CollectionTokenGrantQuerySet(self.model, using=self._db)

    class CollectionTokenGrant(models.Model):
        """A grant of a specific role on a collection to a token."""

        collection = models.ForeignKey(
            Collection, on_delete=models.CASCADE, related_name="token_grants"
        )
        token = models.ForeignKey(
            "db.Token", on_delete=models.CASCADE, related_name="collection_grants"
        )
        role = models.CharField(choices=CollectionRoles.choices)

        created_at = models.DateTimeField(default=timezone.now)
        created_by = models.ForeignKey(
            settings.AUTH_USER_MODEL,
            on_delete=models.PROTECT,
            related_name="user_created_%(class)s",
        )

        objects = CollectionTokenGrantManager.from_queryset(
            CollectionTokenGrantQuerySet
        )()

        class Meta(TypedModelMeta):
            constraints = [
                UniqueConstraint(
                    fields=["token", "collection", "role"],
                    name="%(app_label)s_%(class)s_unique",
                )
            ]