# Source code for debusine.client.models

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

    """Models used by debusine client."""

    import datetime as dt
    import re
    from collections.abc import Collection, Iterator, Sequence
    from enum import StrEnum
    from pathlib import Path
    from typing import Annotated, Any, Literal, NewType, Self, TypeVar, override

    import pydantic
    import yaml

    from debusine.artifacts.models import (
        CollectionCategory,
        DebusineTaskConfiguration,
    )
    from debusine.assets import AssetCategory, asset_data_model
    from debusine.utils import calculate_hash

    class StrictBaseModel(pydantic.BaseModel):
        """Stricter pydantic configuration."""

        model_config = pydantic.ConfigDict(validate_assignment=True)

    class Link(StrictBaseModel):
        """
        Structured version of a resource link.

        This is modeled as a subset of
        https://developer.mozilla.org/en-US/docs/Web/HTML/Reference/Elements/link
        """  # noqa: E501

        #: Resource URL
        href: str
        type: str | None = None

        def can_be_str(self) -> bool:
            """Check if the only field set is href."""
            return self.type is None

    class ServerObject(StrictBaseModel):
        """Model representing an object on the server."""

        # See https://en.wikipedia.org/wiki/HATEOAS

        #: URLs to related resources
        links: dict[str, str | Link] = {}

        def get_link_url(self, name: str) -> str:
            """Return the href URL for a link."""
            res = self.links[name]
            if isinstance(res, str):
                return res
            return res.href

    class PaginatedResponse(StrictBaseModel):
        """Paginated response from the API."""

        count: int | None = None
        next: pydantic.AnyUrl | None = None
        previous: pydantic.AnyUrl | None = None
        results: list[dict[str, Any]]

    class WorkRequestRequest(StrictBaseModel):
        """Client send a WorkRequest to the server."""

        task_name: str
        workspace: str | None = None
        task_data: dict[str, Any]
        event_reactions: dict[str, Any]

    class WorkRequestResponse(ServerObject):
        """Server return a WorkRequest to the client."""

        id: int
        url: str
        parent: int | None = None  # Server > 0.14.7 specifies parent
        workflow_root: int | None = None  # Server > 0.14.7 specifies workflow_root
        created_at: dt.datetime
        started_at: dt.datetime | None = None
        completed_at: dt.datetime | None = None
        duration: float | None = None
        status: str
        result: str
        worker: int | None = None
        task_type: str
        task_name: str
        task_data: dict[str, Any]
        dynamic_task_data: dict[str, Any] | None = None
        priority_base: int
        priority_adjustment: int
        artifacts: list[int]
        scope: str | None = None  # Server >= 0.12 specifies scope
        workspace: str
        scheduler_tags_provided: list[str] = []
        scheduler_tags_required: list[str] = []
        token: str | None = None  # Worker receives a work request token

        @override
        def __str__(self) -> str:
            return f'WorkRequest: {self.id}'

    class WorkRequestExternalDebsignRequest(StrictBaseModel):
        """Client sends data from an external `debsign` run to the server."""

        signed_artifact: int

    class OnWorkRequestCompleted(ServerObject):
        """
        Server return an OnWorkRequestCompleted to the client.

        Returned via websocket consumer endpoint.
        """

        work_request_id: int
        completed_at: dt.datetime
        result: str

    class RuntimeParameter(StrEnum):
        """WorkflowTemplate ``runtime_parameter`` special values."""

        ANY = "any"

    RuntimeParameters = (
        dict[str, list[str] | Literal[RuntimeParameter.ANY]]
        | Literal[RuntimeParameter.ANY]
    )

    class WorkflowTemplateDataNew(StrictBaseModel):
        """Data for a WorkflowTemplate to be created."""

        name: str
        task_name: str
        static_parameters: dict[str, Any]
        runtime_parameters: RuntimeParameters = {}
        priority: int = 0
        extra_groups: list[int | str] = []
        restricted: bool = False

    class WorkflowTemplateData(ServerObject):
        """Data for a WorkflowTemplate."""

        id: int
        name: str
        task_name: str
        static_parameters: dict[str, Any]
        runtime_parameters: RuntimeParameters
        priority: int = 0
        extra_groups: list[int | str] = []
        restricted: bool = False

    class CreateWorkflowRequest(StrictBaseModel):
        """Client sends a workflow creation request to the server."""

        template_name: str = ""
        task_name: str = ""
        workspace: str | None = None
        task_data: dict[str, Any]

        @pydantic.model_validator(mode="after")
        def validate_names(self) -> Self:
            """Exactly one of ``template_name`` and ``task_name`` must be set."""
            if bool(self.template_name) == bool(self.task_name):
                raise ValueError(
                    "Exactly one of template_name and task_name must be set"
                )
            return self

    class FileRequest(StrictBaseModel):
        """Declare a FileRequest: client sends it to the server."""

        size: pydantic.NonNegativeInt
        checksums: dict[str, Annotated[str, pydantic.Field(max_length=255)]]
        type: Literal["file"]
        #: A media type, suitable for an HTTP ``Content-Type`` header.
        content_type: str | None = None

        @staticmethod
        def create_from(
            path: Path, *, content_type: str | None = None
        ) -> "FileRequest":
            """Return a FileRequest for the file path."""
            return FileRequest(
                size=path.stat().st_size,
                checksums={"sha256": calculate_hash(path, "sha256").hex()},
                type="file",
                content_type=content_type,
            )

    class FileResponse(ServerObject):
        """Declare a FileResponse: server sends it to the client."""

        size: pydantic.NonNegativeInt
        checksums: dict[str, Annotated[str, pydantic.Field(max_length=255)]]
        type: Literal["file"]
        url: pydantic.AnyUrl
        content_type: str | None = None

    FilesRequestType = NewType("FilesRequestType", dict[str, FileRequest])
    FilesResponseType = NewType("FilesResponseType", dict[str, FileResponse])

    class ArtifactCreateRequest(StrictBaseModel):
        """Declare an ArtifactCreateRequest: client sends it to the server."""

        category: str
        workspace: str | None = None
        files: FilesRequestType = FilesRequestType({})
        data: dict[str, Any] = {}
        work_request: int | None = None
        expire_at: dt.datetime | None = None

    class ArtifactResponse(ServerObject):
        """Declare an ArtifactResponse: server sends it to the client."""

        id: int
        url: str
        scope: str
        workspace: str
        category: str
        created_at: dt.datetime
        data: dict[str, Any]
        download_tar_gz_url: pydantic.AnyUrl
        files_to_upload: list[str]
        expire_at: dt.datetime | None = None
        files: FilesResponseType = FilesResponseType({})

    class RemoteArtifact(ServerObject):
        """Declare RemoteArtifact."""

        id: int
        url: str
        workspace: str

    class AssetCreateRequest(StrictBaseModel):
        """Request for the Asset creation API."""

        category: AssetCategory
        data: dict[str, Any]
        work_request: int | None = None
        workspace: str

    class AssetResponse(ServerObject):
        """Response from an Asset creation / listing API."""

        id: int
        category: AssetCategory
        data: dict[str, Any]
        work_request: int | None = None
        workspace: str

        @pydantic.model_validator(mode="after")
        def validate_data(self) -> Self:
            """Validate data using the correct data model."""
            asset_data_model(self.category, self.data)
            return self

    class AssetsResponse(
        pydantic.RootModel[Sequence[AssetResponse]], StrictBaseModel
    ):
        """A response from the server with multiple AssetResponse objects."""

        model_config = pydantic.ConfigDict(validate_assignment=True)

        @override
        def __iter__(self) -> Iterator[AssetResponse]:  # type: ignore[override]
            """Iterate over individual asset responses."""
            return iter(self.root)

    class AssetPermissionCheckResponse(ServerObject):
        """A response from the asset permission check endpoint."""

        has_permission: bool
        username: str | None
        user_id: int | None
        resource: dict[str, Any] | None

    class RelationType(StrEnum):
        """Possible values for `RelationCreateRequest.type`."""

        EXTENDS = "extends"
        RELATES_TO = "relates-to"
        BUILT_USING = "built-using"

    class RelationCreateRequest(StrictBaseModel):
        """Declare a RelationCreateRequest: client sends it to the server."""

        artifact: int
        target: int
        type: RelationType

    class RelationResponse(ServerObject):
        """Declare a RelationResponse."""

        id: int
        artifact: int
        target: int
        type: RelationType

    class RelationsResponse(
        pydantic.RootModel[Sequence[RelationResponse]], StrictBaseModel
    ):
        """A response from the server with multiple RelationResponse objects."""

        @override
        def __iter__(self) -> Iterator[RelationResponse]:  # type: ignore[override]
            """Iterate over individual relation responses."""
            return iter(self.root)

    class CollectionRelationType(StrEnum):
        """Possible values for `CollectionRelationResponse.type`."""

        SUITE_FORKED_FROM = "suite-forked-from"
        SUITE_BASED_ON = "suite-based-on"
        SUITE_REQUIRES = "suite-requires"
        SUITE_TARGETING = "suite-targeting"
        SUITE_DEFAULT_QA_RESULTS = "suite-default-qa-results"

    class CollectionIdentifierRequest(StrictBaseModel):
        """Identifies a collection in API requests."""

        name: str
        category: CollectionCategory

    class CollectionRelationSetRequest(StrictBaseModel):
        """Used to set all targets for a collection relation type."""

        targets: list[CollectionIdentifierRequest]

    class CollectionIdentifierResponse(StrictBaseModel):
        """Identifies a collection in API responses."""

        name: str
        category: CollectionCategory

    class CollectionRelationSetResponse(StrictBaseModel):
        """Updated targets for a collection relation list."""

        targets: list[CollectionIdentifierResponse]

    class CollectionRelationResponse(StrictBaseModel):
        """Relationship between two collections."""

        source_collection: CollectionIdentifierResponse
        target_collection: CollectionIdentifierResponse
        type: CollectionRelationType
        position: int | None = None

    class LookupChildType(StrEnum):
        """Possible values for `LookupDict.child_type` and `expect_type`."""

        BARE = "bare"
        ARTIFACT = "artifact"
        ARTIFACT_OR_PROMISE = "artifact-or-promise"
        COLLECTION = "collection"
        ANY = "any"

    class LookupResultType(StrEnum):
        """A collection item type returned by a lookup."""

        BARE = "bare"
        ARTIFACT = "artifact"
        COLLECTION = "collection"

    LookupMultiple = list[int | str | dict[str, Any]]
    LookupSingle = int | str

    class LookupSingleRequest(StrictBaseModel):
        """A request from the client to look up a single collection item."""

        lookup: LookupSingle
        work_request: int | None = None
        expect_type: LookupChildType = LookupChildType.ANY
        default_category: str | None = None
        workspace: str

    class LookupMultipleRequest(StrictBaseModel):
        """A request from the client to look up multiple collection items."""

        lookup: LookupMultiple
        work_request: int | None = None
        expect_type: LookupChildType
        default_category: str | None = None
        workspace: str

    class CollectionItemResponse(StrictBaseModel):
        """A collection item in a lookup response."""

        name: str
        data: dict[str, Any]

    class LookupSingleResponse(ServerObject):
        """A response from the server with a single lookup result."""

        result_type: LookupResultType
        collection_item: CollectionItemResponse | None = None
        artifact: int | None = None
        collection: int | None = None
        parent_collection_lookup: LookupSingle | None = None

        @override
        def __eq__(self, other: Any) -> bool:
            """Treat our subclasses equally when evaluating equality."""
            if not isinstance(other, LookupSingleResponse):
                return False
            return self.model_dump() == other.model_dump()

    class LookupSingleResponseArtifact(LookupSingleResponse):
        """
        A response from the server with a single lookup result for an artifact.

        Used to assist type annotations.
        """

        result_type: Literal[LookupResultType.ARTIFACT] = LookupResultType.ARTIFACT
        artifact: int

    class LookupSingleResponseCollection(LookupSingleResponse):
        """
        A response from the server with a single lookup result for a collection.

        Used to assist type annotations.
        """

        result_type: Literal[LookupResultType.COLLECTION] = (
            LookupResultType.COLLECTION
        )
        collection: int

    # TODO: We should use PEP 695 type parameter declarations here instead, but
    # that causes "Overloaded function implementation cannot produce return
    # type" errors on Debusine.lookup_multiple.
    LSR = TypeVar("LSR", bound=LookupSingleResponse, covariant=True)

    class LookupMultipleResponse(
        pydantic.RootModel[Sequence[LSR]], StrictBaseModel
    ):
        """A response from the server with multiple lookup results."""

        @override
        def __iter__(self) -> Iterator[LSR]:  # type: ignore[override]
            """Iterate over individual results."""
            return iter(self.root)

    re_nonce = re.compile(r"^[A-Za-z0-9_-]{8,64}$")
    re_challenge = re.compile(r"^\w{4,10}(?: \w{4,10}){2,7}$", re.ASCII)

    class EnrollPayload(pydantic.BaseModel):
        """Client-provided enrollment payload."""

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

        #: Nonce identifying the client enrollment
        nonce: Annotated[str, pydantic.Field(pattern=re_nonce)]
        #: Human-readable challenge to verify the client in the web UI
        challenge: Annotated[str, pydantic.Field(pattern=re_challenge)]
        #: Scope. Informational, the confirmation page will warn if invalid
        scope: str
        #: Hostname. Informational, shown in the confirmation page and in the
        #: generated token comment
        hostname: str

    re_token = re.compile(r"^[A-Za-z0-9]{8,64}$")

    class EnrollOutcome(StrEnum):
        """User action in response to an enroll confirmation request."""

        CONFIRM = "confirm"
        CANCEL = "cancel"

    class EnrollConfirmPayload(pydantic.BaseModel):
        """Enrollment response from the server."""

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

        outcome: EnrollOutcome
        token: Annotated[str, pydantic.Field(pattern=re_token)] | None = None

    class TaskConfigurationCollection(ServerObject):
        """Response fragment identifying a TaskConfiguration collection."""

        id: int
        name: str
        data: dict[str, Any]

    class TaskConfigurationCollectionContents(ServerObject):
        """Bundle together a TaskConfiguration collection and its items."""

        collection: TaskConfigurationCollection
        items: list[DebusineTaskConfiguration]

    class TaskConfigurationCollectionUpdateResults(ServerObject):
        """Results of a task configuration collection update."""

        added: int
        updated: int
        removed: int
        unchanged: int

    class WorkspaceInheritanceChainElement(StrictBaseModel):
        """An element in an inheritance chain."""

        id: int | None = None
        scope: str | None = None
        workspace: str | None = None

        @pydantic.model_validator(mode="after")
        def not_empty(self) -> Self:
            """Ensure at least one of id or workspace is set."""
            if self.id is None and self.workspace is None:
                raise ValueError("at least id or workspace need to be set")
            return self

        def matches(self, element: "WorkspaceInheritanceChainElement") -> bool:
            """Check if the components of element that are set match this one."""
            return (
                (element.id is None or element.id == self.id)
                and (element.scope is None or element.scope == self.scope)
                and (
                    element.workspace is None or element.workspace == self.workspace
                )
            )

        @classmethod
        def from_string(cls, value: str) -> Self:
            """Resolve a command line argument to a WorkspaceInheritanceChain."""
            if value.isdigit():
                return cls(id=int(value))
            elif "/" in value:
                scope, workspace = value.split("/", 1)
                return cls(scope=scope, workspace=workspace)
            else:
                return cls(workspace=value)

    class WorkspaceInheritanceChain(ServerObject):
        """Ordered list of workspaces inherited by a workspace."""

        chain: list[WorkspaceInheritanceChainElement] = []

        def __add__(self, other: "WorkspaceInheritanceChain") -> Self:
            """Return the concatenation of this chain and another."""
            return self.__class__(chain=self.chain + other.chain)

        def __sub__(self, other: "WorkspaceInheritanceChain") -> Self:
            """Return a copy of this chain without the elements in other."""
            filtered: list[WorkspaceInheritanceChainElement] = []
            for element in self.chain:
                for to_remove in other.chain:
                    if element.matches(to_remove):
                        break
                else:
                    filtered.append(element)

            return self.__class__(chain=filtered)

        @classmethod
        def from_strings(cls, values: Collection[str]) -> Self:
            """Build an inheritance chain from a list of user-provided strings."""
            return cls(
                chain=[
                    WorkspaceInheritanceChainElement.from_string(value)
                    for value in values
                ]
            )

    class CollectionDataNew(StrictBaseModel):
        """Data for a Collection to be created."""

        name: str
        category: str
        full_history_retention_period: int | None = None
        metadata_only_retention_period: int | None = None
        data: dict[str, Any]

    class CollectionData(ServerObject):
        """Data for a Collection."""

        id: int
        name: str
        category: str
        full_history_retention_period: int | None = None
        metadata_only_retention_period: int | None = None
        data: dict[str, Any]
        stats: list[dict[str, Any]] | None = None

    class CollectionRoles(StrEnum):
        """Available roles for a Collection."""

        # The database model also defines OWNER, but we don't have a use case
        # for clients granting that yet.  Omit it for now to avoid confusing
        # people.
        VIEWER = "viewer"

    class CollectionTokenGrantRequest(ServerObject):
        """Request to grant a Collection role to a token or an external username."""

        token_hash: str | None = None
        username: str | None = None
        comment: str = ""
        role: CollectionRoles

        @pydantic.model_validator(mode="after")
        def validate_token_or_username(self) -> Self:
            """At least one of ``token_hash`` and ``username`` must be set."""
            if self.token_hash is None and self.username is None:
                raise ValueError(
                    "At least one of token_hash and username must be set"
                )
            return self

    class CollectionTokenGrantResponse(ServerObject):
        """Response for a token granted a Collection role."""

        token_type: str
        token_hash: str
        token_key: str | None = None
        username: str
        comment: str
        role: CollectionRoles

    class EmitMetricRequest(StrictBaseModel):
        """A request for the server to emit a metric on behalf of the client."""

        metric_type: Literal["counter", "gauge", "summary", "histogram"]
        name: str
        labels: dict[str, str]
        value: float

    # Workarounds for https://github.com/yaml/pyyaml/issues/722
    yaml.SafeDumper.add_multi_representer(
        StrEnum,
        yaml.representer.SafeRepresenter.represent_str,
    )