Web frontends¶

import json
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import Any, Literal

from cfn_typed_sumo.types import apigateway, dynamodb, ecs, elasticloadbalancingv2 as elbv2, events
from cfn_typed_sumo.types import iam, kinesisfirehose, lambda_, logs, route53, s3, sns, sqs

from cfn_typed import AnyStatement, Expr, Resources, Tag, allow, arn, assume_role_policy, get_att, join, policy_document, ref, sub
from cfn_typed.policy import ANY, arn_equals, aws, service, string_equals
from cfn_typed_sumo.constructs import topic_to_queue
from cfn_typed_sumo.patterns import queue_with_dlq

from collections.abc import Sequence
from dataclasses import dataclass

from cfn_typed_sumo.types import certificatemanager as acm
from cfn_typed_sumo.types import cloudfront, s3

from cfn_typed import Expr, Resources, Tag, allow, get_att, join, policy_document, ref, sub
from cfn_typed.policy import arn_equals, service

CACHING_OPTIMIZED = "658327ea-f89d-4fab-a63d-7e88639e58f6"  # the managed cache policy of the same name


@dataclass(frozen=True)
class SiteDomain:
    """Serve the site on `names` instead of the `*.cloudfront.net` name.

    Give `certificate_arn` to use an existing certificate, or `hosted_zone_id` to have the stack request one and
    validate it by DNS in that hosted zone. CloudFront only accepts certificates from **us-east-1**, so a stack
    that creates one has to be deployed there. `names` are the domain names; the first is the certificate's.
    """

    names: Sequence[str]
    certificate_arn: Expr[str] | None = None
    hosted_zone_id: Expr[str] | None = None

    def __post_init__(self) -> None:
        if not self.names:
            raise ValueError("a domain needs at least one name")
        if (self.certificate_arn is None) == (self.hosted_zone_id is None):
            raise ValueError("give either certificate_arn (use that certificate) or hosted_zone_id (request one)")


def static_site(
    name: str,
    *,
    default_root_object: str = "index.html",
    domain: SiteDomain | None = None,
    tags: list[Expr[Tag]] | None = None,
) -> Resources:
    """A private bucket served by CloudFront through origin access control (OAC).

    Logical IDs: `{name}Bucket`, `{name}BucketPolicy`, `{name}OAC` and `{name}Distribution`, plus
    `{name}Certificate` when `domain` asks for one. The bucket stays private: only this distribution may read
    it, which the policy states with `aws:SourceArn`.
    """
    bucket, oac, dist, cert = f"{name}Bucket", f"{name}OAC", f"{name}Distribution", f"{name}Certificate"
    bucket_props: s3.BucketProps = {
        "PublicAccessBlockConfiguration": {
            "BlockPublicAcls": True, "BlockPublicPolicy": True, "IgnorePublicAcls": True, "RestrictPublicBuckets": True,
        },
        "BucketEncryption": {"ServerSideEncryptionConfiguration": [{"ServerSideEncryptionByDefault": {"SSEAlgorithm": "AES256"}}]},
    }
    config: cloudfront.DistributionDistributionConfig = {
        "Enabled": True,
        "HttpVersion": "http2and3",
        "DefaultRootObject": default_root_object,
        "Origins": [{
            "Id": "site",
            "DomainName": get_att(bucket, "RegionalDomainName"),
            "OriginAccessControlId": get_att(oac, "Id"),
            "S3OriginConfig": {},  # required for an S3 origin; empty because OAC replaces the old identity
        }],
        "DefaultCacheBehavior": {
            "TargetOriginId": "site",
            "ViewerProtocolPolicy": "redirect-to-https",
            "CachePolicyId": CACHING_OPTIMIZED,
            "Compress": True,
        },
    }
    extra: Resources = {}
    if domain:
        certificate_arn: Expr[str]
        if domain.certificate_arn is not None:
            certificate_arn = domain.certificate_arn
        elif domain.hosted_zone_id is not None:
            certificate_arn = ref(cert)
            cert_props: acm.CertificateProps = {
                "DomainName": domain.names[0],
                "ValidationMethod": "DNS",
                "DomainValidationOptions": [{"DomainName": n, "HostedZoneId": domain.hosted_zone_id} for n in domain.names],
            }
            if len(domain.names) > 1:
                cert_props["SubjectAlternativeNames"] = list(domain.names[1:])
            extra[cert] = acm.certificate(**cert_props)
        else:  # SiteDomain refuses this
            raise ValueError("a domain needs a certificate_arn or a hosted_zone_id")
        config["Aliases"] = list(domain.names)
        config["ViewerCertificate"] = {
            "AcmCertificateArn": certificate_arn,
            "SslSupportMethod": "sni-only",
            "MinimumProtocolVersion": "TLSv1.2_2021",
        }
    dist_props: cloudfront.DistributionProps = {"DistributionConfig": config}
    if tags:
        bucket_props["Tags"] = dist_props["Tags"] = tags
    return {
        bucket: s3.bucket(**bucket_props),
        oac: cloudfront.origin_access_control(OriginAccessControlConfig={
            "Name": sub("${AWS::StackName}-" + name),
            "OriginAccessControlOriginType": "s3",
            "SigningBehavior": "always",
            "SigningProtocol": "sigv4",
        }),
        dist: cloudfront.distribution(**dist_props),
        f"{name}BucketPolicy": s3.bucket_policy(
            Bucket=ref(bucket),
            PolicyDocument=policy_document([allow(
                "s3:GetObject", join("", [get_att(bucket, "Arn"), "/*"]),
                sid="AllowCloudFrontRead",
                principal=service("cloudfront.amazonaws.com"),
                condition=arn_equals("aws:SourceArn", sub("arn:${AWS::Partition}:cloudfront::${AWS::AccountId}:distribution/${D}", {"D": ref(dist)})),
            )]),
        ),
        **extra,
    }

def _rest_api(name: str, *, description: str) -> tuple[str, apigateway.RestApi]:
    api = f"{name}Api"
    return api, apigateway.rest_api(Name=sub("${AWS::StackName}-" + name), Description=description)


def _deploy(name: str, api: str, methods: Sequence[str], stage: str) -> Resources:
    """A deployment and stage. The deployment depends on every method, or it can snapshot an empty API."""
    deployment = f"{name}Deployment"
    return {
        deployment: apigateway.deployment(RestApiId=ref(api), depends_on=list(methods)),
        f"{name}Stage": apigateway.stage(RestApiId=ref(api), DeploymentId=ref(deployment), StageName=stage),
    }

def api_to_lambda(name: str, function_logical: str, *, stage: str = "prod") -> Resources:
    """A REST API that proxies every path and method to a Lambda function.

    Logical IDs: `{name}Api`, `{name}Proxy`, `{name}ProxyMethod`, `{name}RootMethod`, `{name}Deployment`,
    `{name}Stage` and `{name}InvokePermission`. The permission is the part people forget: without it
    API Gateway gets a 500 and nothing in the template says why.
    """
    api, rest = _rest_api(name, description=f"{name} proxy to {function_logical}")
    proxy = f"{name}Proxy"
    uri = sub(
        "arn:${AWS::Partition}:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${Fn}/invocations",
        {"Fn": get_att(function_logical, "Arn")},
    )
    integration: apigateway.MethodIntegration = {"Type": "AWS_PROXY", "IntegrationHttpMethod": "POST", "Uri": uri}
    resources: Resources = {
        api: rest,
        proxy: apigateway.resource(RestApiId=ref(api), ParentId=get_att(api, "RootResourceId"), PathPart="{proxy+}"),
        f"{name}ProxyMethod": apigateway.method(
            RestApiId=ref(api), ResourceId=ref(proxy), HttpMethod="ANY", AuthorizationType="NONE", Integration=integration,
        ),
        # `{proxy+}` does not match the root path, so the root needs a method of its own
        f"{name}RootMethod": apigateway.method(
            RestApiId=ref(api), ResourceId=get_att(api, "RootResourceId"), HttpMethod="ANY", AuthorizationType="NONE", Integration=integration,
        ),
        f"{name}InvokePermission": lambda_.permission(
            Action="lambda:InvokeFunction",
            FunctionName=ref(function_logical),
            Principal="apigateway.amazonaws.com",
            SourceArn=sub("arn:${AWS::Partition}:execute-api:${AWS::Region}:${AWS::AccountId}:${A}/*/*/*", {"A": ref(api)}),
        ),
    }
    return resources | _deploy(name, api, [f"{name}ProxyMethod", f"{name}RootMethod"], stage)

def api_to_dynamodb(name: str, table_logical: str, *, key: str = "pk", stage: str = "prod") -> Resources:
    """`GET /items/{id}` straight from a table, with no Lambda in between.

    API Gateway calls DynamoDB `GetItem` itself, through a role that may read only this table. Logical IDs:
    `{name}Api`, `{name}Role`, `{name}Items`, `{name}Item`, `{name}GetMethod`, `{name}Deployment` and `{name}Stage`.
    """
    api, rest = _rest_api(name, description=f"{name} reads {table_logical}")
    items, item, method, role = f"{name}Items", f"{name}Item", f"{name}GetMethod", f"{name}Role"
    request = json.dumps({"TableName": "${T}", "Key": {key: {"S": "$input.params('id')"}}})
    resources: Resources = {
        api: rest,
        role: iam.role(
            AssumeRolePolicyDocument=assume_role_policy(service("apigateway.amazonaws.com")),
            Policies=[iam.RolePolicyProperty(
                PolicyName=f"{name}-read",
                PolicyDocument=policy_document([allow("dynamodb:GetItem", get_att(table_logical, "Arn"))]),
            )],
        ),
        items: apigateway.resource(RestApiId=ref(api), ParentId=get_att(api, "RootResourceId"), PathPart="items"),
        item: apigateway.resource(RestApiId=ref(api), ParentId=ref(items), PathPart="{id}"),
        method: apigateway.method(
            RestApiId=ref(api), ResourceId=ref(item), HttpMethod="GET", AuthorizationType="NONE",
            RequestParameters={"method.request.path.id": True},
            MethodResponses=[{"StatusCode": "200"}],
            Integration={
                "Type": "AWS",
                "IntegrationHttpMethod": "POST",
                "Uri": sub("arn:${AWS::Partition}:apigateway:${AWS::Region}:dynamodb:action/GetItem"),
                "Credentials": get_att(role, "Arn"),
                "RequestTemplates": {"application/json": sub(request, {"T": ref(table_logical)})},
                "IntegrationResponses": [{"StatusCode": "200"}],
            },
        ),
    }
    return resources | _deploy(name, api, [method], stage)

def lambda_behind_alb(
    name: str,
    function_logical: str,
    *,
    listener: str,
    target_group_name: str,
    path_pattern: str,
    priority: int,
) -> Resources:
    """Send the requests that match `path_pattern` on an existing listener to a Lambda function.

    Logical IDs: `{name}TargetGroup`, `{name}InvokePermission` and `{name}Rule`. A Lambda target group has no
    VPC, port or health check. The target group must wait for the permission, or registering the function fails,
    so the permission cannot refer to the group (a cycle). It builds the group's ARN from the **fixed
    `target_group_name`** instead, which is why the name is required.
    """
    tg, permission = f"{name}TargetGroup", f"{name}InvokePermission"
    return {
        permission: lambda_.permission(
            Action="lambda:InvokeFunction",
            FunctionName=ref(function_logical),
            Principal="elasticloadbalancing.amazonaws.com",
            SourceArn=sub("arn:${AWS::Partition}:elasticloadbalancing:${AWS::Region}:${AWS::AccountId}:targetgroup/" + target_group_name + "/*"),
        ),
        tg: elbv2.target_group(
            Name=target_group_name,
            TargetType="lambda",
            Targets=[{"Id": get_att(function_logical, "Arn")}],
            depends_on=permission,
        ),
        f"{name}Rule": elbv2.listener_rule(
            ListenerArn=ref(listener),
            Priority=priority,
            Conditions=[{"Field": "path-pattern", "PathPatternConfig": {"Values": [path_pattern]}}],
            Actions=[{"Type": "forward", "TargetGroupArn": ref(tg)}],
        ),
    }

CLOUDFRONT_HOSTED_ZONE = "Z2FDTNDATAQYW2"  # the same for every CloudFront distribution


def alias_to_load_balancer(logical_id: str, *, zone_id: Expr[str], record: str, load_balancer: str) -> Resources:
    """An A record in `zone_id` that points at a load balancer in this template."""
    return {logical_id: route53.record_set(
        HostedZoneId=zone_id,
        Name=record,
        Type="A",
        AliasTarget={
            "DNSName": get_att(load_balancer, "DNSName"),
            "HostedZoneId": get_att(load_balancer, "CanonicalHostedZoneID"),
            "EvaluateTargetHealth": True,
        },
    )}


def alias_to_distribution(logical_id: str, *, zone_id: Expr[str], record: str, distribution: str) -> Resources:
    """An A record in `zone_id` that points at a CloudFront distribution in this template."""
    return {logical_id: route53.record_set(
        HostedZoneId=zone_id,
        Name=record,
        Type="A",
        AliasTarget={
            "DNSName": get_att(distribution, "DomainName"),
            "HostedZoneId": CLOUDFRONT_HOSTED_ZONE,
            "EvaluateTargetHealth": False,
        },
    )}