1import { createHash } from "node:crypto";
2import * as Effect from "effect/Effect";
3import type { PolicyStatement } from "../IAM/Policy.ts";
4import type { Queue } from "../SQS/Queue.ts";
5import type { EventBus } from "./EventBus.ts";
6import { Rule, type RuleProps, type RuleTarget } from "./Rule.ts";
7
8interface EventDescriptor {
9 id?: string;
10 bus?: EventBus;
11 pattern: Record<string, any>;
12 props?: Pick<RuleProps, "description" | "state">;
13}
14
15export interface QueueRouteTargetProps extends Pick<
16 RuleTarget,
17 | "Input"
18 | "InputPath"
19 | "InputTransformer"
20 | "RetryPolicy"
21 | "DeadLetterConfig"
22> {
23 sqsParameters?: RuleTarget["SqsParameters"];
24}
25
26/**
27 * Routes matching events from an EventBridge bus to an SQS queue.
28 *
29 * Creates a {@link Rule} targeting the queue and binds a queue policy that
30 * allows `events.amazonaws.com` to send messages from that rule. Usually
31 * reached through the `events(...)` builder rather than called directly.
32 * **Example:** Route Matching Events to an SQS Queue
33 * ```typescript
34 * yield* AWS.EventBridge.events(bus, { source: ["my.app"] }).toQueue(queue);
35 * ```
36 *
37 * @binding
38 */
39export const toQueue = (
40 descriptor: EventDescriptor,
41 queue: Queue,
42 props: QueueRouteTargetProps = {},
43) =>
44 Effect.gen(function* () {
45 const routeId =
46 descriptor.id ?? createRouteId(descriptor, `${queue.LogicalId}Queue`);
47
48 const rule = yield* Rule(routeId, {
49 description: descriptor.props?.description,
50 state: descriptor.props?.state,
51 eventBusName: descriptor.bus?.eventBusName,
52 eventPattern: descriptor.pattern,
53 targets: [
54 {
55 Id: `${queue.LogicalId}Target`,
56 Arn: queue.queueArn as any,
57 Input: props.Input,
58 InputPath: props.InputPath,
59 InputTransformer: props.InputTransformer,
60 RetryPolicy: props.RetryPolicy,
61 DeadLetterConfig: props.DeadLetterConfig,
62 SqsParameters: props.sqsParameters,
63 },
64 ],
65 });
66
67 if (!globalThis.__ALCHEMY_RUNTIME__) {
68 yield* queue.bind`Allow(${rule}, AWS.EventBridge.toQueue(${queue}))`({
69 policyStatements: [
70 {
71 Effect: "Allow",
72 Principal: {
73 Service: "events.amazonaws.com",
74 },
75 Action: ["sqs:SendMessage"],
76 Resource: [queue.queueArn as any],
77 Condition: {
78 ArnEquals: {
79 "aws:SourceArn": [rule.ruleArn as any],
80 },
81 },
82 } satisfies PolicyStatement,
83 ],
84 });
85 }
86
87 return rule;
88 });
89
90const createRouteId = (descriptor: EventDescriptor, suffix: string) =>
91 `EventBridge${createHash("sha1")
92 .update(
93 JSON.stringify({
94 bus: descriptor.bus?.LogicalId ?? "default",
95 pattern: descriptor.pattern,
96 suffix,
97 }),
98 )
99 .digest("hex")
100 .slice(0, 10)}`;