# LocalStack Resource Provider Scaffolding v2
from __future__ import annotations

import json

import localstack.services.cloudformation.provider_utils as util
from localstack.services.cloudformation.resource_provider import (
    OperationStatus,
    ProgressEvent,
    ResourceRequest,
)
from localstack.services.sqs.resource_providers.generated.aws_sqs_queue_base import (
    SQSQueueProperties,
    SQSQueueProviderBase,
)

_queue_attribute_list = [
    "ContentBasedDeduplication",
    "DeduplicationScope",
    "DelaySeconds",
    "FifoQueue",
    "FifoThroughputLimit",
    "KmsDataKeyReusePeriodSeconds",
    "KmsMasterKeyId",
    "MaximumMessageSize",
    "MessageRetentionPeriod",
    "ReceiveMessageWaitTimeSeconds",
    "RedriveAllowPolicy",
    "RedrivePolicy",
    "SqsManagedSseEnabled",
    "VisibilityTimeout",
]


class SQSQueueProvider(SQSQueueProviderBase):
    # Values used when a property is removed from a template and needs to be set to its default.
    # If AWS changes their defaults in the future, our parity tests should break.
    DEFAULT_ATTRIBUTE_VALUES = {
        "ReceiveMessageWaitTimeSeconds": "0",
        "DelaySeconds": "0",
        "KmsMasterKeyId": "",
        "RedrivePolicy": "",
        "MessageRetentionPeriod": "345600",
        "MaximumMessageSize": "262144",  # Note: CloudFormation sets this to 256KB on update, but 1MB on create
        "VisibilityTimeout": "30",
        "KmsDataKeyReusePeriodSeconds": "300",
        # Note that "SqsManagedSseEnabled" is deliberately omitted from this list, since AWS
        # doesn't seem to reset it to a default.
    }

    # Similar for FIFO, but only applied when FifoQueue is true (these can't be set otherwise)
    DEFAULT_FIFO_ATTRIBUTE_VALUES = {
        "ContentBasedDeduplication": "false",
        "DeduplicationScope": "messageGroup",
        "FifoThroughputLimit": "perMessageGroupId",
    }

    # Private method for creating a unique queue name, if none is specified.
    def _autogenerated_queue_name(self, request: ResourceRequest[SQSQueueProperties]) -> str:
        queue_name = util.generate_default_name(request.stack_name, request.logical_resource_id)
        if self._is_true_property(request.desired_state.get("FifoQueue")):
            queue_name = f"{queue_name[:-5]}.fifo"
        return queue_name

    def create(
        self,
        request: ResourceRequest[SQSQueueProperties],
    ) -> ProgressEvent[SQSQueueProperties]:
        """
        Create a new resource.

        Primary identifier fields:
          - /properties/QueueUrl

        Create-only properties:
          - /properties/FifoQueue
          - /properties/QueueName

        Read-only properties:
          - /properties/QueueUrl
          - /properties/Arn

        IAM permissions required:
          - sqs:CreateQueue
          - sqs:GetQueueUrl
          - sqs:GetQueueAttributes
          - sqs:ListQueueTags
          - sqs:TagQueue

        """
        # TODO: validations - what validations are needed?
        model = request.desired_state
        sqs = request.aws_client_factory.sqs

        # if no QueueName is specified, automatically generate one
        if not model.get("QueueName"):
            model["QueueName"] = self._autogenerated_queue_name(request)

        attributes = self._compile_sqs_queue_attributes(model)
        result = request.aws_client_factory.sqs.create_queue(
            QueueName=model["QueueName"],
            Attributes=attributes,
            tags={t["Key"]: t["Value"] for t in model.get("Tags", [])},
        )

        # set read-only properties
        model["QueueUrl"] = result["QueueUrl"]
        model["Arn"] = sqs.get_queue_attributes(
            QueueUrl=result["QueueUrl"], AttributeNames=["QueueArn"]
        )["Attributes"]["QueueArn"]

        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_model=model,
            custom_context=request.custom_context,
        )

    def read(
        self,
        request: ResourceRequest[SQSQueueProperties],
    ) -> ProgressEvent[SQSQueueProperties]:
        """
        Fetch resource information

        IAM permissions required:
          - sqs:GetQueueAttributes
          - sqs:ListQueueTags
        """
        raise NotImplementedError

    def delete(
        self,
        request: ResourceRequest[SQSQueueProperties],
    ) -> ProgressEvent[SQSQueueProperties]:
        """
        Delete a resource

        IAM permissions required:
          - sqs:DeleteQueue
          - sqs:GetQueueAttributes
        """
        sqs = request.aws_client_factory.sqs
        try:
            queue_url = sqs.get_queue_url(QueueName=request.previous_state["QueueName"])["QueueUrl"]
            sqs.delete_queue(QueueUrl=queue_url)

        except sqs.exceptions.QueueDoesNotExist:
            return ProgressEvent(
                status=OperationStatus.SUCCESS, resource_model=request.desired_state
            )

        return ProgressEvent(status=OperationStatus.SUCCESS, resource_model=request.desired_state)

    def update(
        self,
        request: ResourceRequest[SQSQueueProperties],
    ) -> ProgressEvent[SQSQueueProperties]:
        """
        Update a resource

        IAM permissions required:
          - sqs:SetQueueAttributes
          - sqs:GetQueueAttributes
          - sqs:ListQueueTags
          - sqs:TagQueue
          - sqs:UntagQueue
        """
        sqs = request.aws_client_factory.sqs
        model = request.desired_state
        prev_model = request.previous_state

        assert request.previous_state is not None

        queue_url = prev_model["QueueUrl"]
        self._populate_missing_attributes_with_defaults(model)
        sqs.set_queue_attributes(
            QueueUrl=queue_url, Attributes=self._compile_sqs_queue_attributes(model)
        )

        (tags_to_remove, tags_to_add_or_update) = util.resource_tags_to_remove_or_update(
            prev_model.get("Tags", []), model.get("Tags", [])
        )
        sqs.untag_queue(QueueUrl=queue_url, TagKeys=tags_to_remove)
        sqs.tag_queue(QueueUrl=queue_url, Tags=tags_to_add_or_update)

        model["QueueUrl"] = queue_url
        model["Arn"] = request.previous_state["Arn"]

        # For QueueName and FifoQueue, always use the value from the previous model. These fields
        # are create-only, so they cannot be changed via an update (even though they might be omitted)
        model["QueueName"] = prev_model.get("QueueName")
        model["FifoQueue"] = prev_model.get("FifoQueue", False)

        return ProgressEvent(OperationStatus.SUCCESS, resource_model=model)

    def _compile_sqs_queue_attributes(self, properties: SQSQueueProperties) -> dict[str, str]:
        """
        SQS is really awkward in how the ``CreateQueue`` operation expects arguments. Most of a Queue's
        attributes are passed as a string values in the "Attributes" dictionary. So we need to compile this
        dictionary here.

        :param properties: the properties passed from cloudformation
        :return: a mapping used for the ``Attributes`` argument of the `CreateQueue` call.
        """
        result = {}

        for k in _queue_attribute_list:
            v = properties.get(k)

            if v is None:
                continue
            elif isinstance(v, str):
                pass
            elif isinstance(v, bool):
                v = str(v).lower()
            elif isinstance(v, dict):
                # RedrivePolicy and RedriveAllowPolicy
                v = json.dumps(v)
            elif isinstance(v, int):
                v = str(v)
            else:
                raise TypeError(f"cannot convert attribute {k}, unhandled type {type(v)}")

            result[k] = v

        return result

    def _populate_missing_attributes_with_defaults(self, properties: SQSQueueProperties) -> None:
        """
        For any attribute that is missing from the desired state, populate it with the default value.
        This is the only way to remove an attribute from an existing SQS queue's configuration.
        :param properties: the properties passed from cloudformation
        """
        for k, v in self.DEFAULT_ATTRIBUTE_VALUES.items():
            properties.setdefault(k, v)

        if self._is_true_property(properties.get("FifoQueue")):
            for k, v in self.DEFAULT_FIFO_ATTRIBUTE_VALUES.items():
                properties.setdefault(k, v)

    def _is_true_property(self, property_value: str | None) -> bool:
        """
        Detect whether a 'true' value is passed in a property. If it's None (property was omitted) or False, or any
        type of string (e.g. a typo such as "Fasle"), then it's not a true value. This extra check is needed because
        the CloudFormation engine doesn't fully validate booleans properties before passing them to the resource provider.
        """
        return (
            property_value == True  # noqa: E712
            or (isinstance(property_value, str) and property_value.lower() == "true")
        )

    def list(
        self,
        request: ResourceRequest[SQSQueueProperties],
    ) -> ProgressEvent[SQSQueueProperties]:
        resources = request.aws_client_factory.sqs.list_queues()
        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_models=[
                SQSQueueProperties(QueueUrl=url) for url in resources.get("QueueUrls", [])
            ],
        )
