Messaging and events¶

Fan-out and EventBridge wiring, after the Solutions Constructs sns-sqs, eventbridge-lambda and eventbridge-sqs patterns. In each of them the resource that sends needs a resource policy that names who may send, and forgetting it is silent: the message is accepted by the sender and never arrives.

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

Fan-out¶

def fan_out(name: str, subscribers: Sequence[str], *, tags: list[Expr[Tag]] | None = None) -> Resources:
    """One topic, and for each subscriber its own queue with a dead-letter queue.

    Logical IDs: `{name}Topic`, and per subscriber `s`: `{s}Queue`, `{s}DLQ`, `{s}QueuePolicy` and
    `{s}Queue{name}TopicSubscription` (from the [CDK L2 `topic_to_queue`](cdk-l2.md)). A slow subscriber
    falls behind on its own queue instead of slowing the others.
    """
    topic = f"{name}Topic"
    topic_props: sns.TopicProps = {}
    _tags(topic_props, tags)
    resources: Resources = {topic: sns.topic(**topic_props)}
    for s in subscribers:
        q = queue_with_dlq(s, tags=tags)
        resources |= q.resources | topic_to_queue(q.queue, [topic], raw_message_delivery=True)
    return resources

fan_out("Orders", ["Billing", "Shipping", "Analytics"])

One topic, and a queue with a dead-letter queue for each subscriber. A slow or broken subscriber backs up on its own queue instead of affecting the others, and the dead-letter queue catches what it cannot process. The subscription and the queue policy come from topic_to_queue in CDK L2 constructs, so this page adds the shape and not another copy of the wiring. Raw message delivery is on, so a consumer reads the message itself and not an SNS envelope around it.

EventBridge to a Lambda or a queue¶

def _rule(name: str, *, pattern: Mapping[str, Any] | None, schedule: str | None, target_id: str, target_arn: Expr[str]) -> events.Rule:
    if (pattern is None) == (schedule is None):
        raise ValueError("give exactly one of pattern and schedule")
    props: events.RuleProps = {"State": "ENABLED", "Targets": [{"Id": target_id, "Arn": target_arn}]}
    if pattern is not None:
        props["EventPattern"] = pattern
    if schedule is not None:
        props["ScheduleExpression"] = schedule
    return events.rule(**props)


def event_to_lambda(
    name: str, function_logical: str, *, pattern: Mapping[str, Any] | None = None, schedule: str | None = None,
) -> Resources:
    """An EventBridge rule that invokes a Lambda: `{name}Rule` and `{name}InvokePermission`.

    Give an event `pattern` or a `schedule` such as `"rate(5 minutes)"`. Lambda needs a permission naming this
    rule as the caller; the rule's target alone does not grant it.
    """
    rule = f"{name}Rule"
    return {
        rule: _rule(name, pattern=pattern, schedule=schedule, target_id=function_logical, target_arn=get_att(function_logical, "Arn")),
        f"{name}InvokePermission": lambda_.permission(
            Action="lambda:InvokeFunction",
            FunctionName=ref(function_logical),
            Principal="events.amazonaws.com",
            SourceArn=get_att(rule, "Arn"),
        ),
    }


def event_to_queue(
    name: str, queue: str, *, pattern: Mapping[str, Any] | None = None, schedule: str | None = None,
) -> Resources:
    """An EventBridge rule that sends to an SQS queue: `{name}Rule` and `{name}Policy`.

    The queue policy lets this rule, and only this rule, send. A queue encrypted with a customer-managed
    key would also need a key policy statement for `events.amazonaws.com`.
    """
    rule = f"{name}Rule"
    return {
        rule: _rule(name, pattern=pattern, schedule=schedule, target_id=queue, target_arn=get_att(queue, "Arn")),
        f"{name}Policy": sqs.queue_policy(
            Queues=[ref(queue)],
            PolicyDocument=policy_document([allow(
                "sqs:SendMessage", get_att(queue, "Arn"),
                principal=service("events.amazonaws.com"),
                condition=arn_equals("aws:SourceArn", get_att(rule, "Arn")),
            )]),
        ),
    }

event_to_lambda("OrderCreated", "Processor", pattern={"source": ["shop"], "detail-type": ["OrderCreated"]})
event_to_queue("Nightly", "ReportQueue", schedule="cron(0 2 * * ? *)")

Exactly one of pattern and schedule. The rule is either an event match or a clock, and giving both or neither is a ValueError before anything is deployed.
The target alone grants nothing. A Lambda target needs a lambda:InvokeFunction permission, and an SQS target needs a queue policy. Both name this rule's ARN, so no other rule can use them.
Overlap with schedule. schedule is the same pattern for a clock only. This one also takes an event pattern, and has a queue variant.
Encrypted queues. A queue that uses a customer-managed KMS key also needs a key policy statement for the sending service. That is a key policy, so it belongs with the key, not here.