# LocalStack Resource Provider Scaffolding v2
from __future__ import annotations

import json
from pathlib import Path
from typing import TypedDict

import localstack.services.cloudformation.provider_utils as util
from localstack.services.cloudformation.resource_provider import (
    OperationStatus,
    ProgressEvent,
    ResourceProvider,
    ResourceRequest,
)
from localstack.utils.strings import str_to_bool


class EC2SubnetProperties(TypedDict):
    VpcId: str | None
    AssignIpv6AddressOnCreation: bool | None
    AvailabilityZone: str | None
    AvailabilityZoneId: str | None
    CidrBlock: str | None
    EnableDns64: bool | None
    Ipv6CidrBlock: str | None
    Ipv6CidrBlocks: list[str] | None
    Ipv6Native: bool | None
    MapPublicIpOnLaunch: bool | None
    NetworkAclAssociationId: str | None
    OutpostArn: str | None
    PrivateDnsNameOptionsOnLaunch: dict | None
    SubnetId: str | None
    Tags: list[Tag] | None


class Tag(TypedDict):
    Key: str | None
    Value: str | None


REPEATED_INVOCATION = "repeated_invocation"


def generate_subnet_read_payload(
    ec2_client, schema, subnet_ids: list[str] | None = None
) -> list[EC2SubnetProperties]:
    kwargs = {}
    if subnet_ids:
        kwargs["SubnetIds"] = subnet_ids
    subnets = ec2_client.describe_subnets(**kwargs)["Subnets"]

    models = []
    for subnet in subnets:
        subnet_id = subnet["SubnetId"]

        model = EC2SubnetProperties(**util.select_attributes(subnet, schema))

        if "Tags" not in model:
            model["Tags"] = []

        if "EnableDns64" not in model:
            model["EnableDns64"] = False

        private_dns_name_options = model.setdefault("PrivateDnsNameOptionsOnLaunch", {})

        if "HostnameType" not in private_dns_name_options:
            private_dns_name_options["HostnameType"] = "ip-name"

        optional_bool_attrs = ["EnableResourceNameDnsAAAARecord", "EnableResourceNameDnsARecord"]
        for attr in optional_bool_attrs:
            if attr not in private_dns_name_options:
                private_dns_name_options[attr] = False

        network_acl_associations = ec2_client.describe_network_acls(
            Filters=[{"Name": "association.subnet-id", "Values": [subnet_id]}]
        )
        model["NetworkAclAssociationId"] = network_acl_associations["NetworkAcls"][0][
            "NetworkAclId"
        ]
        models.append(model)

    return models


class EC2SubnetProvider(ResourceProvider[EC2SubnetProperties]):
    TYPE = "AWS::EC2::Subnet"  # Autogenerated. Don't change
    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change

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

        Primary identifier fields:
          - /properties/SubnetId

        Required properties:
          - VpcId

        Create-only properties:
          - /properties/VpcId
          - /properties/AvailabilityZone
          - /properties/AvailabilityZoneId
          - /properties/CidrBlock
          - /properties/OutpostArn
          - /properties/Ipv6Native

        Read-only properties:
          - /properties/NetworkAclAssociationId
          - /properties/SubnetId
          - /properties/Ipv6CidrBlocks

        IAM permissions required:
          - ec2:DescribeSubnets
          - ec2:CreateSubnet
          - ec2:CreateTags
          - ec2:ModifySubnetAttribute

        """
        model = request.desired_state
        ec2 = request.aws_client_factory.ec2

        params = util.select_attributes(
            model,
            [
                "AvailabilityZone",
                "AvailabilityZoneId",
                "CidrBlock",
                "Ipv6CidrBlock",
                "Ipv6Native",
                "OutpostArn",
                "VpcId",
            ],
        )
        if model.get("Tags"):
            tags = [{"ResourceType": "subnet", "Tags": model.get("Tags")}]
            params["TagSpecifications"] = tags

        response = ec2.create_subnet(**params)
        model["SubnetId"] = response["Subnet"]["SubnetId"]
        bool_attrs = [
            "AssignIpv6AddressOnCreation",
            "EnableDns64",
            "MapPublicIpOnLaunch",
        ]
        custom_attrs = bool_attrs + ["PrivateDnsNameOptionsOnLaunch"]
        if not any(attr in model for attr in custom_attrs):
            return ProgressEvent(
                status=OperationStatus.SUCCESS,
                resource_model=model,
                custom_context=request.custom_context,
            )

        # update boolean attributes
        for attr in bool_attrs:
            if attr in model:
                kwargs = {attr: {"Value": str_to_bool(model[attr])}}
                ec2.modify_subnet_attribute(SubnetId=model["SubnetId"], **kwargs)

        # determine DNS hostname type on launch
        dns_options = model.get("PrivateDnsNameOptionsOnLaunch")
        if dns_options:
            if isinstance(dns_options, str):
                dns_options = json.loads(dns_options)
            if dns_options.get("HostnameType"):
                ec2.modify_subnet_attribute(
                    SubnetId=model["SubnetId"],
                    PrivateDnsHostnameTypeOnLaunch=dns_options.get("HostnameType"),
                )
        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_model=model,
            custom_context=request.custom_context,
        )

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

        IAM permissions required:
          - ec2:DescribeSubnets
          - ec2:DescribeNetworkAcls
        """
        models = generate_subnet_read_payload(
            ec2_client=request.aws_client_factory.ec2,
            schema=self.SCHEMA["properties"],
            subnet_ids=[request.desired_state["SubnetId"]],
        )

        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_model=models[0],
            custom_context=request.custom_context,
        )

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

        IAM permissions required:
          - ec2:DescribeSubnets
          - ec2:DeleteSubnet
        """
        model = request.desired_state
        ec2 = request.aws_client_factory.ec2

        ec2.delete_subnet(SubnetId=model["SubnetId"])
        return ProgressEvent(status=OperationStatus.SUCCESS, resource_model=model)

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

        IAM permissions required:
          - ec2:DescribeSubnets
          - ec2:ModifySubnetAttribute
          - ec2:CreateTags
          - ec2:DeleteTags
          - ec2:AssociateSubnetCidrBlock
          - ec2:DisassociateSubnetCidrBlock
        """
        raise NotImplementedError

    def list(
        self, request: ResourceRequest[EC2SubnetProperties]
    ) -> ProgressEvent[EC2SubnetProperties]:
        """
        List resources

        IAM permissions required:
          - ec2:DescribeSubnets
          - ec2:DescribeNetworkAcls
        """
        models = generate_subnet_read_payload(
            request.aws_client_factory.ec2, self.SCHEMA["properties"]
        )
        return ProgressEvent(status=OperationStatus.SUCCESS, resource_models=models)
