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

import datetime as dt
from functools import partial
from pathlib import Path
from typing import Any, Optional, Self, TYPE_CHECKING, TypedDict

from django.conf import settings
from django.contrib.auth.models import AnonymousUser
from django.core.exceptions import PermissionDenied, ValidationError
from django.db import models, transaction
from django.db.models import Exists, F, OuterRef, Q, QuerySet, UniqueConstraint
from django.db.models.functions import Coalesce
from django.urls import reverse
from django_cte import CTE, with_cte

from debusine.artifacts import LocalArtifact
from debusine.artifacts.models import ArtifactCategory, ArtifactData
from debusine.db.models.files import File
from debusine.db.models.permissions import (
    Allow,
    permission_check,
    permission_filter,
)
from debusine.db.models.workspaces import Workspace
from debusine.db.permissioncontext import PermissionContext
from debusine.utils.typing_utils import copy_signature_from

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

    from debusine.db.models.work_requests import WorkRequest
    from debusine.web.views.ui.artifacts import ArtifactUI, FileInArtifactUI

    ArtifactWithComplete = WithAnnotations["Artifact", "CompleteDict"]

    WithAnnotations  # fake usage for vulture

else:
    TypedModelMeta = object

ARTIFACT_CATEGORY_ICON_NAMES = {
    ArtifactCategory.AUTOPKGTEST: "folder",
    ArtifactCategory.BINARY_PACKAGE: "folder",
    ArtifactCategory.BINARY_PACKAGES: "folder",
    ArtifactCategory.BLHC: "folder",
    ArtifactCategory.DEBDIFF: "folder",
    ArtifactCategory.LINTIAN: "folder",
    ArtifactCategory.PACKAGE_BUILD_LOG: "journal-text",
    ArtifactCategory.SIGNING_INPUT: "folder",
    ArtifactCategory.SIGNING_OUTPUT: "folder",
    ArtifactCategory.SOURCE_PACKAGE: "folder",
    ArtifactCategory.SYSTEM_IMAGE: "folder",
    ArtifactCategory.SYSTEM_TARBALL: "folder",
    ArtifactCategory.UPLOAD: "folder",
    ArtifactCategory.WORK_REQUEST_DEBUG_LOGS: "folder",
}

ARTIFACT_CATEGORY_SHORT_NAMES = {
    ArtifactCategory.AUTOPKGTEST: "autopkgtest",
    ArtifactCategory.BINARY_PACKAGE: "binary package",
    ArtifactCategory.BINARY_PACKAGES: "binary packages",
    ArtifactCategory.BLHC: "blhc report",
    ArtifactCategory.DEBDIFF: "debdiff report",
    ArtifactCategory.LINTIAN: "lintian report",
    ArtifactCategory.PACKAGE_BUILD_LOG: "build log",
    ArtifactCategory.SIGNING_INPUT: "signing input",
    ArtifactCategory.SIGNING_OUTPUT: "signing output",
    ArtifactCategory.SOURCE_PACKAGE: "source package",
    ArtifactCategory.SYSTEM_IMAGE: "system image",
    ArtifactCategory.SYSTEM_TARBALL: "system tar",
    ArtifactCategory.UPLOAD: "package upload",
    ArtifactCategory.WORK_REQUEST_DEBUG_LOGS: "debug log",
}


class CompleteDict(TypedDict):
    """Additional fields returned by ArtifactQuerySet.annotate_complete."""

    complete: bool


class ArtifactQuerySet[A](QuerySet["Artifact", A]):
    """Custom QuerySet for Artifact."""

    def in_current_scope(self) -> Self:
        """Filter to artifacts 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 artifacts in the current workspace."""
        from debusine.db.context import context

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

    def visible_to_current_workspace(self) -> Self:
        """Filter to artifacts that the current workspace inherits from."""
        from debusine.db.context import context

        visible = CTE(
            Workspace.objects.can_display(context.pc),
            name="visible",
        )
        chain = CTE.recursive(
            context.require_workspace().make_inheritance_chain_cte(visible),
            name="chain",
        )
        return self.filter(
            workspace__in=with_cte(
                visible,
                chain,
                select=chain.join(Workspace, pk=chain.col.chain_parent),
            )
        )

    def work_request_can_display(self, work_request: "WorkRequest") -> Self:
        """Keep only artifacts that the work request can display."""
        task = work_request.get_task()
        return self.filter(
            Q(pk__in=task.get_input_artifacts_ids())
            | Q(created_by_work_request=work_request)
        )

    def work_request_can_change(self, work_request: "WorkRequest") -> Self:
        """Keep only artifacts that the work request can change."""
        return self.filter(created_by_work_request=work_request)

    @permission_filter(work_request=Allow.PASS, anonymous=Allow.PASS)
    def can_display(self, pc: PermissionContext) -> Self:
        """Keep only Artifacts that can be displayed."""
        # Delegate to workspace can_display check
        qs = self.filter(workspace__in=Workspace.objects.can_display(pc))
        if pc.work_request is not None:
            qs = qs.work_request_can_display(pc.work_request)
        return qs

    @permission_filter(work_request=Allow.PASS)
    def can_add_files(self, pc: PermissionContext) -> Self:
        """Keep only Artifacts where files can be added."""
        qs = self.filter(
            workspace__in=Workspace.objects.can_create_artifacts(pc)
        )
        if pc.work_request is not None:
            qs = qs.work_request_can_change(pc.work_request)
        return qs

    @permission_filter(work_request=Allow.PASS)
    def can_add_relation(self, pc: PermissionContext) -> Self:
        """Keep only Artifacts where relations can be added."""
        qs = self.filter(
            workspace__in=Workspace.objects.can_create_artifacts(pc)
        )
        if pc.work_request is not None:
            qs = qs.work_request_can_change(pc.work_request)
        return qs

    def annotate_complete(
        self,
    ) -> "ArtifactQuerySet[ArtifactWithComplete]":
        """Annotate artifacts with whether all their files are complete."""
        return self.annotate(
            complete=~Exists(
                FileInArtifact.objects.filter(
                    artifact=OuterRef("pk"), complete=False
                )
            )
        )

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

        :param at: datetime to check if the artifacts are not expired.
        :return: artifacts that expire_at is None (do not expire) or
          expire_at is after the given datetime.
        """
        return self.annotate(
            _effective_expiration_delay=Coalesce(
                "expiration_delay",
                "workspace__default_expiration_delay",
            )
        ).filter(
            Q(_effective_expiration_delay=dt.timedelta(0))
            | Q(
                # https://github.com/typeddjango/django-stubs/issues/1548
                created_at__gt=(
                    at - F("_effective_expiration_delay")  # type: ignore[operator] # noqa: E501
                )
            )
        )

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

        :param at: datetime to check if the artifacts are expired.
        :return: artifacts that expire_at is before the given datetime.
        """
        return (
            self.annotate(
                _effective_expiration_delay=Coalesce(
                    "expiration_delay",
                    "workspace__default_expiration_delay",
                )
            )
            .exclude(_effective_expiration_delay=dt.timedelta(0))
            .filter(
                # https://github.com/typeddjango/django-stubs/issues/1548
                created_at__lte=(
                    at - F("_effective_expiration_delay")  # type: ignore[operator] # noqa: E501
                )
            )
        )

    def part_of_collection_with_retains_artifacts(self) -> Self:
        """
        Return Artifacts where the parent_collections has retains_artifacts set.

        :return: Artifacts that are part of a retains_artifacts collection.
        """
        # Import here to prevent circular imports
        from debusine.db.models.collections import Collection
        from debusine.db.models.work_requests import WorkRequest

        RetainsArtifacts = Collection.RetainsArtifacts
        return self.filter(
            Q(parent_collections__retains_artifacts=RetainsArtifacts.ALWAYS)
            | Q(
                parent_collections__retains_artifacts=RetainsArtifacts.WORKFLOW,
                parent_collections__workflow__status__in={
                    WorkRequest.Statuses.PENDING,
                    WorkRequest.Statuses.RUNNING,
                    WorkRequest.Statuses.BLOCKED,
                },
            )
        )


class ArtifactManager(models.Manager["Artifact"]):
    """Manager for the Artifact model."""

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

    @classmethod
    def create_from_local_artifact(
        cls,
        local_artifact: LocalArtifact[Any],
        workspace: Workspace,
        *,
        created_by_work_request: Optional["WorkRequest"] = None,
    ) -> "Artifact":
        """Return a new Artifact based on a :py:class:`LocalArtifact`."""
        artifact = Artifact.objects.create(
            category=local_artifact.category,
            workspace=workspace,
            data=local_artifact.data.model_dump(),
            created_by_work_request=created_by_work_request,
        )

        for artifact_path, local_path in local_artifact.files.items():
            file = File.from_local_path(local_path)
            file_backend = workspace.scope.upload_file_backend(file)
            file_backend.add_file(local_path, fileobj=file)
            FileInArtifact.objects.create(
                artifact=artifact,
                path=artifact_path,
                file=file,
                complete=True,
                content_type=local_artifact.content_types.get(artifact_path),
            )

        return artifact


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

    category = models.CharField(max_length=255)
    workspace = models.ForeignKey(Workspace, on_delete=models.PROTECT)
    files = models.ManyToManyField(File, through="db.FileInArtifact")
    data = models.JSONField(default=dict, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    expiration_delay = models.DurationField(blank=True, null=True)
    created_by = models.ForeignKey(
        "User", blank=True, null=True, on_delete=models.PROTECT
    )
    created_by_work_request = models.ForeignKey(
        "WorkRequest", blank=True, null=True, on_delete=models.SET_NULL
    )
    original_artifact = models.ForeignKey(
        "Artifact", blank=True, null=True, on_delete=models.SET_NULL
    )

    objects = ArtifactManager.from_queryset(ArtifactQuerySet)()

    class Meta(TypedModelMeta):
        base_manager_name = "objects"
        indexes = [
            # Used by debusine.server.open_metrics.DebusineCollector.  This
            # is imperfect (on debusine.debian.net at the time of writing it
            # only takes the artifact-counting query from ~550ms to ~300ms).
            # Doing any better seems likely to involve a separate
            # trigger-maintained statistics table.
            models.Index(
                "workspace",
                "category",
                name="%(app_label)s_%(class)s_workspace_category",
                include=["id"],
            )
        ]

    def __str__(self) -> str:
        """Return basic information of Artifact."""
        return (
            f"Id: {self.id} "
            f"Category: {self.category} "
            f"Workspace: {self.workspace.id}"
        )

    @copy_signature_from(models.Model.save)
    def save(self, **kwargs: Any) -> None:
        """Wrap save with permission checks."""
        from debusine.db.context import context

        if self._state.adding:
            # Create
            if not self.workspace.can_create_artifacts(context.pc):
                user = context.user or AnonymousUser()
                raise PermissionDenied(
                    f"{user} cannot create artifacts in {self.workspace}"
                )
        else:
            # Update
            ...  # TODO: check for update permissions

        return super().save(**kwargs)

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

        with urlconf_scope(self.workspace.scope.name):
            return reverse(
                "workspaces:artifacts:detail",
                kwargs={"wname": self.workspace.name, "artifact_id": self.id},
            )

    def get_absolute_url_download(self) -> str:
        """Return the canonical URL to download the artifact."""
        from debusine.server.scopes import urlconf_scope

        with urlconf_scope(self.workspace.scope.name):
            return reverse(
                "workspaces:artifacts:download",
                kwargs={"wname": self.workspace.name, "artifact_id": self.id},
            )

    def clean(self) -> None:
        """
        Ensure that data is valid for this artifact category.

        :raise ValidationError: for invalid data.
        """
        if not isinstance(self.data, dict):
            raise ValidationError({"data": "data must be a dictionary"})

        try:
            artifact_cls = LocalArtifact.class_from_category(self.category)
        except ValueError as e:
            raise ValidationError(
                {"category": f"{self.category}: invalid artifact category"}
            ) from e

        try:
            artifact_cls.create_data(self.data)
        except ValueError as e:
            raise ValidationError(
                {"data": f"invalid artifact data: {e}"}
            ) from e

    def work_request_can_display(self, work_request: "WorkRequest") -> bool:
        """Check if the work request can display this artifact."""
        if work_request == self.created_by_work_request:
            return True
        task = work_request.get_task()
        return self.pk in task.get_input_artifacts_ids()

    def work_request_can_change(self, work_request: "WorkRequest") -> bool:
        """Check if the work request can change this artifact."""
        return work_request == self.created_by_work_request

    @permission_check(
        "{user} cannot display artifact {resource.id}",
        work_request=Allow.PASS,
        anonymous=Allow.PASS,
    )
    def can_display(self, pc: PermissionContext) -> bool:
        """Check if the artifact can be displayed."""
        if not self.workspace.can_display(pc):
            return False
        if pc.work_request is None:
            return True
        return self.work_request_can_display(pc.work_request)

    @permission_check(
        "{user} cannot add files to artifact {resource.id}",
        work_request=Allow.PASS,
    )
    def can_add_files(self, pc: PermissionContext) -> bool:
        """Check if files can be added to the artifact."""
        if not self.workspace.can_create_artifacts(pc):
            return False
        if pc.work_request is None:
            return True
        return self.work_request_can_change(pc.work_request)

    @permission_check(
        "{user} cannot add relations to artifact {resource.id}",
        work_request=Allow.PASS,
    )
    def can_add_relation(self, pc: PermissionContext) -> bool:
        """Check if the artifact can be displayed."""
        if not self.workspace.can_create_artifacts(pc):
            return False
        if pc.work_request is None:
            return True
        return self.work_request_can_change(pc.work_request)

    def create_data(self) -> ArtifactData:
        """Instantiate ArtifactData from data."""
        artifact_cls = LocalArtifact.class_from_category(self.category)
        artifact_data = artifact_cls.create_data(self.data)
        assert isinstance(artifact_data, ArtifactData)
        return artifact_data

    def get_label(self, data: ArtifactData | None = None) -> str:
        """
        Return a label for this artifact.

        Optionally reuse an already instantiated data model.
        """
        if data is None:
            data = self.create_data()
        if label := data.get_label():
            return label
        return str(self.category)

    def effective_expiration_delay(self) -> dt.timedelta:
        """Return expiration_delay, inherited if None."""
        expiration_delay = self.expiration_delay
        if self.expiration_delay is None:  # inherit
            expiration_delay = self.workspace.default_expiration_delay
        assert expiration_delay is not None
        return expiration_delay

    @property
    def expire_at(self) -> dt.datetime | None:
        """Return computed expiration date."""
        delay = self.effective_expiration_delay()
        if delay == dt.timedelta(0):
            return None
        return self.created_at + delay

    def expired(self, at: dt.datetime) -> bool:
        """
        Return True if this artifact has expired at a given datetime.

        :param at: datetime to check if the artifact is expired.
        :return bool: True if the artifact's expire_at is on or earlier than
          the parameter at.
        """
        expire_at = self.expire_at
        if expire_at is None:
            return False
        return expire_at <= at

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

        return ArtifactUI.get(request, self)


class FileInArtifact(models.Model):
    """File in artifact."""

    artifact = models.ForeignKey(Artifact, on_delete=models.PROTECT)
    path = models.CharField(max_length=500)
    file = models.ForeignKey(File, on_delete=models.PROTECT)
    # We do a best-effort data migration here, but note that in cases where
    # a file store is shared between multiple workspaces, this field may be
    # left as False for files that were uploaded before this field was
    # added.  (debusine.debian.net only had a single workspace at the time
    # of this migration, so that's not a problem there.)
    complete = models.BooleanField(default=False)
    # TODO: ruff is correct that we shouldn't use null=True on a CharField,
    # but fixing that retroactively is non-trivial.
    content_type = models.CharField(blank=True, null=True)  # noqa: DJ001

    class Meta(TypedModelMeta):
        constraints = [
            UniqueConstraint(
                fields=["artifact", "path"],
                name="%(app_label)s_%(class)s_unique_artifact_path",
            ),
        ]

    def __str__(self) -> str:
        """Return basic information of FileInArtifact."""
        return (
            f"Id: {self.id} Artifact: {self.artifact.id} "
            f"Path: {self.path} File: {self.file.id}"
        )

    def get_absolute_url(self) -> str:
        """Return an absolute URL to view this file."""
        from debusine.server.scopes import urlconf_scope

        with urlconf_scope(self.artifact.workspace.scope.name):
            return reverse(
                "workspaces:artifacts:file-detail",
                kwargs={
                    "wname": self.artifact.workspace.name,
                    "artifact_id": self.artifact.id,
                    "path": self.path,
                },
            )

    def get_absolute_url_raw(self) -> str:
        """Return an absolute URL to view this file as raw."""
        from debusine.server.scopes import urlconf_scope

        with urlconf_scope(self.artifact.workspace.scope.name):
            return reverse(
                "workspaces:artifacts:file-raw",
                kwargs={
                    "wname": self.artifact.workspace.name,
                    "artifact_id": self.artifact.id,
                    "path": self.path,
                },
            )

    def get_absolute_url_download(self) -> str:
        """Return an absolute URL to download this file."""
        from debusine.server.scopes import urlconf_scope

        with urlconf_scope(self.artifact.workspace.scope.name):
            return reverse(
                "workspaces:artifacts:file-download",
                kwargs={
                    "wname": self.artifact.workspace.name,
                    "artifact_id": self.artifact.id,
                    "path": self.path,
                },
            )

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

        return FileInArtifactUI.get(request, self)


class FileUpload(models.Model):
    """File that is being/has been uploaded."""

    file_in_artifact = models.OneToOneField(
        FileInArtifact, on_delete=models.PROTECT
    )
    path = models.CharField(
        max_length=500,
        help_text="Path in the uploads directory",
        unique=True,
    )
    last_activity_at = models.DateTimeField(auto_now_add=True)

    def __str__(self) -> str:
        """Return basic information."""
        return f"{self.id}"

    @classmethod
    def current_size(cls, artifact: Artifact, path_in_artifact: str) -> int:
        """
        Return current file size.

        The current file size might be smaller than the expected size of the
        file if the file has not finished being uploaded.

        Raise ValueError if path_in_artifact does not exist in Artifact or
        if there's no FileUpload object for the specific File.
        """
        try:
            file_in_artifact = FileInArtifact.objects.get(
                artifact=artifact, path=path_in_artifact
            )
        except FileInArtifact.DoesNotExist:
            raise ValueError(
                f'No FileInArtifact for Artifact {artifact.id} '
                f'and path "{path_in_artifact}"'
            )

        try:
            file_upload = FileUpload.objects.get(
                file_in_artifact=file_in_artifact
            )
        except FileUpload.DoesNotExist:
            raise ValueError(
                f"No FileUpload for FileInArtifact {file_in_artifact.id}"
            )

        try:
            size = file_upload.absolute_file_path().stat().st_size
        except FileNotFoundError:
            size = 0

        return size

    def delete(self, *args: Any, **kwargs: Any) -> tuple[int, dict[str, int]]:
        """Schedule deletion of the file in the store."""
        file_path = self.absolute_file_path()

        result = super().delete(*args, **kwargs)

        # If this method is called from a transaction: transaction.on_commit()
        # will call file_path.unlink when the most outer transaction is
        # committed.
        #
        # In the case that the code is running without a transaction:
        # the file_path.unlink will happen now.
        #
        # It's important that file_path.unlink is called only if the
        # DB is updated with the deletion. Otherwise, the file could be
        # deleted from the store but still referenced from the DB.
        transaction.on_commit(partial(file_path.unlink, missing_ok=True))

        return result

    def absolute_file_path(self) -> Path:
        """
        Return the absolute file path of the file.

        The files are stored in settings.DEBUSINE_UPLOAD_DIRECTORY.
        """
        return Path(settings.DEBUSINE_UPLOAD_DIRECTORY) / self.path


class ArtifactRelationQuerySet[A](QuerySet["ArtifactRelation", A]):
    """Custom QuerySet for ArtifactRelation."""

    def in_current_scope(self) -> "ArtifactRelationQuerySet[A]":
        """Filter to artifact relations in the current scope."""
        from debusine.db.context import context

        scope = context.require_scope()
        return self.filter(
            artifact__workspace__scope=scope, target__workspace__scope=scope
        )

    @permission_filter(work_request=Allow.PASS, anonymous=Allow.PASS)
    def can_display(
        self, pc: PermissionContext
    ) -> "ArtifactRelationQuerySet[A]":
        """Keep only ArtifactRelations that can be displayed."""
        # Delegate to workspace can_display check
        workspaces = Workspace.objects.can_display(pc)
        qs = self.filter(
            artifact__workspace__in=workspaces, target__workspace__in=workspaces
        )
        if pc.work_request is not None:
            task = pc.work_request.get_task()
            input_ids = task.get_input_artifacts_ids()
            qs = qs.filter(
                Q(artifact__pk__in=input_ids)
                | Q(artifact__created_by_work_request=pc.work_request)
            )
            qs = qs.filter(
                Q(target__pk__in=input_ids)
                | Q(target__created_by_work_request=pc.work_request)
            )
        return qs

    @permission_filter()
    def can_delete(
        self, pc: PermissionContext
    ) -> "ArtifactRelationQuerySet[A]":
        """Keep only ArtifactRelations that can be deleted."""
        # Delegate to workspace OWNERS
        owned_workspaces = Workspace.objects.with_role(
            pc, Workspace.Roles.OWNER
        )
        visible_workspaces = Workspace.objects.can_display(pc)
        return self.filter(
            artifact__workspace__in=owned_workspaces,
            target__workspace__in=visible_workspaces,
        )


class ArtifactRelationManager(models.Manager["ArtifactRelation"]):
    """Manager for the ArtifactRelation model."""

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


class ArtifactRelation(models.Model):
    """Model relations between artifacts."""

    class Relations(models.TextChoices):
        EXTENDS = "extends", "Extends"
        RELATES_TO = "relates-to", "Relates to"
        BUILT_USING = "built-using", "Built using"

    artifact = models.ForeignKey(
        Artifact, on_delete=models.PROTECT, related_name="relations"
    )
    target = models.ForeignKey(
        Artifact, on_delete=models.PROTECT, related_name="targeted_by"
    )
    type = models.CharField(max_length=11, choices=Relations.choices)

    objects = ArtifactRelationManager.from_queryset(ArtifactRelationQuerySet)()

    class Meta(TypedModelMeta):
        base_manager_name = "objects"
        constraints = [
            UniqueConstraint(
                fields=["artifact", "target", "type"],
                name="%(app_label)s_%(class)s_unique_artifact_target_type",
            )
        ]

    def __str__(self) -> str:
        """Return str for the object."""
        return f"{self.artifact.id} {self.type} {self.target.id}"

    @permission_check(
        "{user} cannot display artifact relation {resource}",
        work_request=Allow.PASS,
        anonymous=Allow.PASS,
    )
    def can_display(self, pc: PermissionContext) -> bool:
        """Check if the artifact can be displayed."""
        if not (
            self.artifact.workspace.can_display(pc)
            and self.target.workspace.can_display(pc)
        ):
            return False
        if pc.work_request is None:
            return True
        if (
            pc.work_request == self.artifact.created_by_work_request
            and pc.work_request == self.target.created_by_work_request
        ):
            return True
        task = pc.work_request.get_task()
        input_ids = task.get_input_artifacts_ids()
        return (
            pc.work_request == self.artifact.created_by_work_request
            or self.artifact.pk in input_ids
        ) and (
            pc.work_request == self.target.created_by_work_request
            or self.target.pk in input_ids
        )

    @permission_check(
        "{user} cannot delete artifact relation {resource}",
    )
    def can_delete(self, pc: PermissionContext) -> bool:
        """Check if the user can delete artifacts relations."""
        return self.artifact.workspace.has_role(
            pc, Workspace.Roles.OWNER
        ) and self.can_display(pc)
