# LocalStack Resource Provider Scaffolding v2
from __future__ import annotations

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


class DynamoDBGlobalTableProperties(TypedDict):
    AttributeDefinitions: list[AttributeDefinition] | None
    KeySchema: list[KeySchema] | None
    Replicas: list[ReplicaSpecification] | None
    Arn: str | None
    BillingMode: str | None
    GlobalSecondaryIndexes: list[GlobalSecondaryIndex] | None
    LocalSecondaryIndexes: list[LocalSecondaryIndex] | None
    SSESpecification: SSESpecification | None
    StreamArn: str | None
    StreamSpecification: StreamSpecification | None
    TableId: str | None
    TableName: str | None
    TimeToLiveSpecification: TimeToLiveSpecification | None
    WriteProvisionedThroughputSettings: WriteProvisionedThroughputSettings | None


class AttributeDefinition(TypedDict):
    AttributeName: str | None
    AttributeType: str | None


class KeySchema(TypedDict):
    AttributeName: str | None
    KeyType: str | None


class Projection(TypedDict):
    NonKeyAttributes: list[str] | None
    ProjectionType: str | None


class TargetTrackingScalingPolicyConfiguration(TypedDict):
    TargetValue: float | None
    DisableScaleIn: bool | None
    ScaleInCooldown: int | None
    ScaleOutCooldown: int | None


class CapacityAutoScalingSettings(TypedDict):
    MaxCapacity: int | None
    MinCapacity: int | None
    TargetTrackingScalingPolicyConfiguration: TargetTrackingScalingPolicyConfiguration | None
    SeedCapacity: int | None


class WriteProvisionedThroughputSettings(TypedDict):
    WriteCapacityAutoScalingSettings: CapacityAutoScalingSettings | None


class GlobalSecondaryIndex(TypedDict):
    IndexName: str | None
    KeySchema: list[KeySchema] | None
    Projection: Projection | None
    WriteProvisionedThroughputSettings: WriteProvisionedThroughputSettings | None


class LocalSecondaryIndex(TypedDict):
    IndexName: str | None
    KeySchema: list[KeySchema] | None
    Projection: Projection | None


class ContributorInsightsSpecification(TypedDict):
    Enabled: bool | None


class ReadProvisionedThroughputSettings(TypedDict):
    ReadCapacityAutoScalingSettings: CapacityAutoScalingSettings | None
    ReadCapacityUnits: int | None


class ReplicaGlobalSecondaryIndexSpecification(TypedDict):
    IndexName: str | None
    ContributorInsightsSpecification: ContributorInsightsSpecification | None
    ReadProvisionedThroughputSettings: ReadProvisionedThroughputSettings | None


class PointInTimeRecoverySpecification(TypedDict):
    PointInTimeRecoveryEnabled: bool | None


class ReplicaSSESpecification(TypedDict):
    KMSMasterKeyId: str | None


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


class KinesisStreamSpecification(TypedDict):
    StreamArn: str | None


class ReplicaSpecification(TypedDict):
    Region: str | None
    ContributorInsightsSpecification: ContributorInsightsSpecification | None
    DeletionProtectionEnabled: bool | None
    GlobalSecondaryIndexes: list[ReplicaGlobalSecondaryIndexSpecification] | None
    KinesisStreamSpecification: KinesisStreamSpecification | None
    PointInTimeRecoverySpecification: PointInTimeRecoverySpecification | None
    ReadProvisionedThroughputSettings: ReadProvisionedThroughputSettings | None
    SSESpecification: ReplicaSSESpecification | None
    TableClass: str | None
    Tags: list[Tag] | None


class SSESpecification(TypedDict):
    SSEEnabled: bool | None
    SSEType: str | None


class StreamSpecification(TypedDict):
    StreamViewType: str | None


class TimeToLiveSpecification(TypedDict):
    Enabled: bool | None
    AttributeName: str | None


REPEATED_INVOCATION = "repeated_invocation"


class DynamoDBGlobalTableProvider(ResourceProvider[DynamoDBGlobalTableProperties]):
    TYPE = "AWS::DynamoDB::GlobalTable"  # Autogenerated. Don't change
    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change

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

        Primary identifier fields:
          - /properties/TableName

        Required properties:
          - KeySchema
          - AttributeDefinitions
          - Replicas

        Create-only properties:
          - /properties/LocalSecondaryIndexes
          - /properties/TableName
          - /properties/KeySchema

        Read-only properties:
          - /properties/Arn
          - /properties/StreamArn
          - /properties/TableId

        IAM permissions required:
          - dynamodb:CreateTable
          - dynamodb:CreateTableReplica
          - dynamodb:Describe*
          - dynamodb:UpdateTimeToLive
          - dynamodb:UpdateContributorInsights
          - dynamodb:UpdateContinuousBackups
          - dynamodb:ListTagsOfResource
          - dynamodb:Query
          - dynamodb:Scan
          - dynamodb:UpdateItem
          - dynamodb:PutItem
          - dynamodb:GetItem
          - dynamodb:DeleteItem
          - dynamodb:BatchWriteItem
          - dynamodb:TagResource
          - dynamodb:EnableKinesisStreamingDestination
          - dynamodb:DisableKinesisStreamingDestination
          - dynamodb:DescribeKinesisStreamingDestination
          - dynamodb:DescribeTableReplicaAutoScaling
          - dynamodb:UpdateTableReplicaAutoScaling
          - dynamodb:TagResource
          - application-autoscaling:DeleteScalingPolicy
          - application-autoscaling:DeleteScheduledAction
          - application-autoscaling:DeregisterScalableTarget
          - application-autoscaling:Describe*
          - application-autoscaling:PutScalingPolicy
          - application-autoscaling:PutScheduledAction
          - application-autoscaling:RegisterScalableTarget
          - kinesis:ListStreams
          - kinesis:DescribeStream
          - kinesis:PutRecords
          - kms:CreateGrant
          - kms:Describe*
          - kms:Get*
          - kms:List*
          - kms:RevokeGrant
          - cloudwatch:PutMetricData

        """
        model = request.desired_state

        if not request.custom_context.get(REPEATED_INVOCATION):
            request.custom_context[REPEATED_INVOCATION] = True

            if not model.get("TableName"):
                model["TableName"] = util.generate_default_name(
                    stack_name=request.stack_name, logical_resource_id=request.logical_resource_id
                )

            create_params = util.select_attributes(
                model,
                [
                    "AttributeDefinitions",
                    "BillingMode",
                    "GlobalSecondaryIndexes",
                    "KeySchema",
                    "LocalSecondaryIndexes",
                    "Replicas",
                    "SSESpecification",
                    "StreamSpecification",
                    "TableName",
                    "WriteProvisionedThroughputSettings",
                ],
            )

            replicas = create_params.pop("Replicas", [])

            if sse_specification := create_params.get("SSESpecification"):
                # rename bool attribute to fit boto call
                sse_specification["Enabled"] = sse_specification.pop("SSEEnabled")

            if stream_spec := model.get("StreamSpecification"):
                create_params["StreamSpecification"] = {
                    "StreamEnabled": True,
                    **stream_spec,
                }

            creation_response = request.aws_client_factory.dynamodb.create_table(**create_params)
            model["Arn"] = creation_response["TableDescription"]["TableArn"]
            model["TableId"] = creation_response["TableDescription"]["TableId"]

            if creation_response["TableDescription"].get("LatestStreamArn"):
                model["StreamArn"] = creation_response["TableDescription"]["LatestStreamArn"]

            replicas_to_create = []
            for replica in replicas:
                create = {
                    "RegionName": replica.get("Region"),
                    "KMSMasterKeyId": replica.get("KMSMasterKeyId"),
                    "ProvisionedThroughputOverride": replica.get("ProvisionedThroughputOverride"),
                    "GlobalSecondaryIndexes": replica.get("GlobalSecondaryIndexes"),
                    "TableClassOverride": replica.get("TableClassOverride"),
                }

                create = {k: v for k, v in create.items() if v is not None}

                replicas_to_create.append({"Create": create})

                request.aws_client_factory.dynamodb.update_table(
                    ReplicaUpdates=replicas_to_create, TableName=model["TableName"]
                )

            # add TTL config
            if ttl_config := model.get("TimeToLiveSpecification"):
                request.aws_client_factory.dynamodb.update_time_to_live(
                    TableName=model["TableName"], TimeToLiveSpecification=ttl_config
                )

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

        status = request.aws_client_factory.dynamodb.describe_table(TableName=model["TableName"])[
            "Table"
        ]["TableStatus"]
        if status == "ACTIVE":
            return ProgressEvent(
                status=OperationStatus.SUCCESS,
                resource_model=model,
                custom_context=request.custom_context,
            )

        elif status == "CREATING":
            return ProgressEvent(
                status=OperationStatus.IN_PROGRESS,
                resource_model=model,
                custom_context=request.custom_context,
            )
        else:
            return ProgressEvent(
                status=OperationStatus.FAILED,
                resource_model=model,
                custom_context=request.custom_context,
                message=f"Table creation failed with status {status}",
            )

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

        IAM permissions required:
          - dynamodb:Describe*
          - application-autoscaling:Describe*
          - cloudwatch:PutMetricData
        """
        raise NotImplementedError

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

        IAM permissions required:
          - dynamodb:Describe*
          - application-autoscaling:DeleteScalingPolicy
          - application-autoscaling:DeleteScheduledAction
          - application-autoscaling:DeregisterScalableTarget
          - application-autoscaling:Describe*
          - application-autoscaling:PutScalingPolicy
          - application-autoscaling:PutScheduledAction
          - application-autoscaling:RegisterScalableTarget
        """

        model = request.desired_state
        if not request.custom_context.get(REPEATED_INVOCATION):
            request.custom_context[REPEATED_INVOCATION] = True
            request.aws_client_factory.dynamodb.delete_table(TableName=model["TableName"])

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

        try:
            request.aws_client_factory.dynamodb.describe_table(TableName=model["TableName"])
            return ProgressEvent(
                status=OperationStatus.IN_PROGRESS,
                resource_model=model,
                custom_context=request.custom_context,
            )
        except Exception as ex:
            if "ResourceNotFoundException" in str(ex):
                return ProgressEvent(
                    status=OperationStatus.SUCCESS,
                    resource_model=model,
                    custom_context=request.custom_context,
                )

            return ProgressEvent(
                status=OperationStatus.FAILED,
                resource_model=model,
                custom_context=request.custom_context,
                message=str(ex),
            )

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

        IAM permissions required:
          - dynamodb:Describe*
          - dynamodb:CreateTableReplica
          - dynamodb:UpdateTable
          - dynamodb:UpdateTimeToLive
          - dynamodb:UpdateContinuousBackups
          - dynamodb:UpdateContributorInsights
          - dynamodb:ListTagsOfResource
          - dynamodb:Query
          - dynamodb:Scan
          - dynamodb:UpdateItem
          - dynamodb:PutItem
          - dynamodb:GetItem
          - dynamodb:DeleteItem
          - dynamodb:BatchWriteItem
          - dynamodb:DeleteTable
          - dynamodb:DeleteTableReplica
          - dynamodb:UpdateItem
          - dynamodb:TagResource
          - dynamodb:UntagResource
          - dynamodb:EnableKinesisStreamingDestination
          - dynamodb:DisableKinesisStreamingDestination
          - dynamodb:DescribeKinesisStreamingDestination
          - dynamodb:DescribeTableReplicaAutoScaling
          - dynamodb:UpdateTableReplicaAutoScaling
          - application-autoscaling:DeleteScalingPolicy
          - application-autoscaling:DeleteScheduledAction
          - application-autoscaling:DeregisterScalableTarget
          - application-autoscaling:Describe*
          - application-autoscaling:PutScalingPolicy
          - application-autoscaling:PutScheduledAction
          - application-autoscaling:RegisterScalableTarget
          - kinesis:ListStreams
          - kinesis:DescribeStream
          - kinesis:PutRecords
          - kms:CreateGrant
          - kms:Describe*
          - kms:Get*
          - kms:List*
          - kms:RevokeGrant
          - cloudwatch:PutMetricData
        """
        raise NotImplementedError
