# 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,
)
