Data flows¶

Change streams, uploads and delivery, after the Solutions Constructs dynamodb-stream-lambda, s3-lambda and kinesisfirehose-s3 patterns. Each connects a source of records to a destination, and each has one wiring detail that is easy to get wrong.

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

A table stream into a Lambda¶

@dataclass(frozen=True)
class StreamTrigger:
    """What `stream_trigger` returns: the mapping, and the statements the function's role needs to read the stream."""

    resources: Resources
    statements: list[AnyStatement]


def stream_trigger(
    name: str,
    function_logical: str,
    table_logical: str,
    *,
    batch_size: int = 100,
    starting_position: Literal["LATEST", "TRIM_HORIZON"] = "LATEST",
    max_retries: int = 3,
) -> StreamTrigger:
    """Invoke a Lambda for the changes of a DynamoDB table: `{name}StreamMapping`.

    The table needs a `StreamSpecification` (see `with_stream`). The function's role needs `statements`, and a
    stream mapping that fails keeps retrying the same batch until the record expires, so retries are bounded
    and a bad batch is split in half to isolate the record that fails.
    """
    mapping = lambda_.event_source_mapping(
        FunctionName=get_att(function_logical, "Arn"),
        EventSourceArn=get_att(table_logical, "StreamArn"),
        StartingPosition=starting_position,
        BatchSize=batch_size,
        BisectBatchOnFunctionError=True,
        MaximumRetryAttempts=max_retries,
    )
    stream = get_att(table_logical, "StreamArn")
    return StreamTrigger(
        {f"{name}StreamMapping": mapping},
        [
            allow(["dynamodb:DescribeStream", "dynamodb:GetRecords", "dynamodb:GetShardIterator"], stream),
            allow("dynamodb:ListStreams", "*"),
        ],
    )


def with_stream(table: dynamodb.Table, view: Literal["KEYS_ONLY", "NEW_IMAGE", "OLD_IMAGE", "NEW_AND_OLD_IMAGES"] = "NEW_AND_OLD_IMAGES") -> dynamodb.Table:
    """The same table with a stream. The table is not modified."""
    return dynamodb.Table(**{**table, "Properties": {**table["Properties"], "StreamSpecification": {"StreamViewType": view}}})

table = with_stream(dynamodb_table("pk"))          # the table, now with a stream
trigger = stream_trigger("Changes", "Processor", "Orders")
role = lambda_role("processor", trigger.statements)

The handle carries the statements. The function's role needs stream read permissions, and the mapping alone does not give them. trigger.statements goes straight into lambda_role.
A failing batch is retried until the record expires (24 hours for a stream) and blocks the shard behind it. The pattern bounds retries and splits a failing batch in two (BisectBatchOnFunctionError) to find the bad record.
with_stream returns a new table. The table from dynamodb_table is not changed.

A bucket that invokes a Lambda¶

def bucket_to_lambda(
    name: str,
    function_logical: str,
    *,
    bucket_name: str,
    events: Sequence[str] = ("s3:ObjectCreated:*",),
    prefix: str | None = None,
    suffix: str | None = None,
) -> Resources:
    """A private bucket that invokes a Lambda when objects arrive: `{name}Bucket` and `{name}InvokePermission`.

    The bucket **must have a fixed name**. The permission has to exist before the bucket can be created (S3 tests
    the notification when it is set), but a permission that names the bucket with `GetAtt` would make the bucket
    wait for itself. So the permission builds the bucket's ARN from the name, and the bucket depends on it.
    """
    bucket, permission = f"{name}Bucket", f"{name}InvokePermission"
    rules: list[Expr[s3.BucketFilterRule]] = [{"Name": k, "Value": v} for k, v in (("prefix", prefix), ("suffix", suffix)) if v]
    notifications: list[Expr[s3.BucketLambdaConfiguration]] = []
    for event in events:
        n: s3.BucketLambdaConfiguration = {"Event": event, "Function": get_att(function_logical, "Arn")}
        if rules:
            n["Filter"] = {"S3Key": {"Rules": rules}}
        notifications.append(n)
    return {
        permission: lambda_.permission(
            Action="lambda:InvokeFunction",
            FunctionName=ref(function_logical),
            Principal="s3.amazonaws.com",
            SourceAccount=ref("AWS::AccountId"),
            SourceArn=arn.bucket(bucket_name),
        ),
        bucket: s3.bucket(
            BucketName=bucket_name,
            PublicAccessBlockConfiguration={
                "BlockPublicAcls": True, "BlockPublicPolicy": True, "IgnorePublicAcls": True, "RestrictPublicBuckets": True,
            },
            NotificationConfiguration={"LambdaConfigurations": notifications},
            depends_on=permission,
        ),
    }

This is the classic circular dependency, and the reason CDK has to work around it. The bucket's notification names the function, and S3 checks at creation that it may invoke it, so the permission has to exist first. A permission that gives the bucket as SourceArn: !GetAtt Bucket.Arn would make the bucket wait for itself. The way out is a fixed bucket name:

the permission builds the ARN from the name, so it does not refer to the bucket,
the bucket has DependsOn the permission,
and SourceAccount stops any other account's bucket of the same name from invoking the function.

The cost is that a bucket with a fixed name cannot be replaced while it exists. If that matters, turn on EventBridge notifications for the bucket and use event_to_lambda instead.

Firehose into a bucket¶

def firehose_to_s3(
    name: str,
    *,
    bucket_arn: Expr[str],
    prefix: str = "firehose/",
    buffer_seconds: int = 300,
    buffer_mb: int = 64,
    retention_days: int = 30,
) -> Resources:
    """A Firehose delivery stream that writes GZIP files to a bucket, with its role and logs.

    Logical IDs: `{name}Logs`, `{name}LogStream`, `{name}Role` and `{name}Stream`. The role may write only
    to `bucket_arn` and to its own log group. Firehose does not create the log stream, so the pattern does.
    """
    logs_id, stream_id, role = f"{name}Logs", f"{name}LogStream", f"{name}Role"
    return {
        logs_id: logs.log_group(RetentionInDays=retention_days),
        stream_id: logs.log_stream(LogGroupName=ref(logs_id), LogStreamName="DestinationDelivery"),
        role: iam.role(
            AssumeRolePolicyDocument=assume_role_policy(
                service("firehose.amazonaws.com"),
                condition=string_equals("sts:ExternalId", ref("AWS::AccountId")),
            ),
            Policies=[iam.RolePolicyProperty(
                PolicyName=f"{name}-deliver",
                PolicyDocument=policy_document([
                    allow(["s3:GetBucketLocation", "s3:ListBucket", "s3:ListBucketMultipartUploads"], bucket_arn),
                    allow(["s3:AbortMultipartUpload", "s3:GetObject", "s3:PutObject"], join("", [bucket_arn, "/*"])),
                    allow("logs:PutLogEvents", get_att(logs_id, "Arn")),
                ]),
            )],
        ),
        f"{name}Stream": kinesisfirehose.delivery_stream(
            DeliveryStreamType="DirectPut",
            ExtendedS3DestinationConfiguration={
                "BucketARN": bucket_arn,
                "RoleARN": get_att(role, "Arn"),
                "Prefix": prefix,
                "ErrorOutputPrefix": f"{prefix}errors/",
                "CompressionFormat": "GZIP",
                "BufferingHints": {"IntervalInSeconds": buffer_seconds, "SizeInMBs": buffer_mb},
                "CloudWatchLoggingOptions": {"Enabled": True, "LogGroupName": ref(logs_id), "LogStreamName": "DestinationDelivery"},
            },
            depends_on=stream_id,
        ),
    }

Two S3 statements, not one. ListBucket applies to the bucket and PutObject to its objects, so a single statement over both is wrong for one of them. The IAM check says so.
The trust policy has an ExternalId. It is the account id, so only Firehose acting for this account can assume the role.
The log stream is created by the pattern. Firehose writes to a stream that already exists, and does not create one, so without it delivery errors would be silent. The delivery stream waits for it.
Compressed and buffered. GZIP files of at most five minutes or 64 MB, so the bucket gets few large objects and not many tiny ones.