1import type * as lambda from "aws-lambda";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import * as Binding from "../../Binding.ts";
5import type { Cluster } from "../ECS/Cluster.ts";
6import type { Function as LambdaFunction } from "../Lambda/Function.ts";
7import type { Queue } from "../SQS/Queue.ts";
8import type { EventBus } from "./EventBus.ts";
9import type { RuleProps } from "./Rule.ts";
10import {
11 toEcsTask as createEcsTaskRoute,
12 type EcsRouteTargetProps,
13} from "./ToEcsTask.ts";
14import {
15 toLambda as createLambdaRoute,
16 type LambdaRouteTargetProps,
17} from "./ToLambda.ts";
18import {
19 toQueue as createQueueRoute,
20 type QueueRouteTargetProps,
21} from "./ToQueue.ts";
22
23export type EventPattern = Record<string, any>;
24export type EventRecord<Detail = unknown> = lambda.EventBridgeEvent<
25 string,
26 Detail
27>;
28
29export interface EventRouteProps extends Pick<
30 RuleProps,
31 "description" | "state"
32> {}
33
34export interface SubscribeProps extends EventRouteProps {}
35
36export type { EcsRouteTargetProps } from "./ToEcsTask.ts";
37export type { LambdaRouteTargetProps } from "./ToLambda.ts";
38export type { QueueRouteTargetProps } from "./ToQueue.ts";
39
40interface EventDescriptor {
41 id?: string;
42 bus?: EventBus;
43 pattern: EventPattern;
44 props?: EventRouteProps;
45}
46
47/**
48 * Event source connecting an EventBridge {@link EventBus} to the hosting
49 * compute (Lambda function or ServerHost process). Matching events invoke
50 * the host with a stream of {@link EventRecord}s.
51 *
52 * Use it through the {@link consumeBusEvents} helper; the host-specific
53 * implementation layer (e.g. `AWS.Lambda.EventSource`) creates the rule,
54 * grants EventBridge invoke permission, and dispatches events at runtime.
55 * ### Consuming Events
56 * **Example:** Consume Matching Events on a Lambda Function
57 * ```typescript
58 * // init — subscribe to matching events (provide AWS.Lambda.EventSource on the Function)
59 * yield* AWS.EventBridge.consumeBusEvents(
60 * bus,
61 * { source: ["my.app"] },
62 * (events: Stream.Stream<AWS.EventBridge.EventRecord>) =>
63 * events.pipe(
64 * Stream.runForEach((event) =>
65 * Effect.log(event["detail-type"], event.detail),
66 * ),
67 * ),
68 * );
69 * ```
70 *
71 * @binding
72 */
73export interface EventSource extends Binding.Service<
74 EventSource,
75 "AWS.EventBridge.EventSource",
76 EventSourceService
77> {}
78export const EventSource = Binding.Service<EventSource>(
79 "AWS.EventBridge.EventSource",
80);
81
82export type EventSourceService = <
83 Detail = unknown,
84 StreamReq = never,
85 Req = never,
86>(
87 descriptor: EventDescriptor,
88 process: (
89 events: Stream.Stream<EventRecord<Detail>, never, StreamReq>,
90 ) => Effect.Effect<void, never, Req>,
91) => Effect.Effect<void, never, never>;
92
93/**
94 * Build a routing target for an EventBridge event bus. Pass the bus and
95 * pattern (no handler) and chain `.toLambda` / `.toQueue` / `.toEcsTask` to
96 * route matching events to a target resource.
97 *
98 * @example Route matching events to a Lambda function
99 * ```typescript
100 * yield* events(bus, { source: ["my.app"] }).toLambda(fn);
101 * ```
102 *
103 * @example Route matching events to an SQS queue
104 * ```typescript
105 * yield* events(bus, { source: ["my.app"] }).toQueue(queue);
106 * ```
107 *
108 * To consume events locally with a handler, use {@link consumeBusEvents}.
109 */
110export const events = (...args: any[]) => {
111 const descriptor = parseEventDescriptor(args);
112
113 return {
114 toLambda: (fn: LambdaFunction, props: LambdaRouteTargetProps = {}) =>
115 createLambdaRoute(descriptor, fn, props),
116 toQueue: (queue: Queue, props: QueueRouteTargetProps = {}) =>
117 createQueueRoute(descriptor, queue, props),
118 toEcsTask: (cluster: Cluster, props: EcsRouteTargetProps) =>
119 createEcsTaskRoute(descriptor, cluster, props),
120 };
121};
122
123/**
124 * Consume events from an EventBridge event bus with a handler. The handler is
125 * the LAST positional argument; the event bus, pattern, and optional props
126 * precede it.
127 *
128 * @example Consume matching events with a handler
129 * ```typescript
130 * yield* consumeBusEvents(bus, { source: ["my.app"] }, (events) =>
131 * events.pipe(Stream.runForEach((event) => Effect.log(event))),
132 * );
133 * ```
134 *
135 * To route events to another resource instead of consuming them locally, use
136 * {@link events}.
137 */
138export const consumeBusEvents = (...args: any[]) => {
139 // The handler is the LAST positional argument. Peel it off and run the
140 // subscribe body directly.
141 const process = args[args.length - 1] as (
142 events: Stream.Stream<EventRecord, never, never>,
143 ) => Effect.Effect<void, never, never>;
144 const descriptor = parseEventDescriptor(args.slice(0, -1));
145 return EventSource.use((source) => source(descriptor, process));
146};
147
148export const matchesEventPattern = (
149 pattern: EventPattern,
150 event: Record<string, any>,
151): boolean =>
152 Object.entries(pattern).every(([key, expected]) =>
153 matchValue(expected, event[key]),
154 );
155
156const matchValue = (expected: any, actual: any): boolean => {
157 if (Array.isArray(expected)) {
158 return expected.some((value) => matchValue(value, actual));
159 }
160
161 if (expected && typeof expected === "object") {
162 if (actual === null || typeof actual !== "object") {
163 return false;
164 }
165
166 return Object.entries(expected).every(([key, value]) =>
167 matchValue(value, actual[key]),
168 );
169 }
170
171 return actual === expected;
172};
173
174const parseEventDescriptor = (args: any[]): EventDescriptor => {
175 if (typeof args[0] === "string") {
176 if (isEventBus(args[1])) {
177 return {
178 id: args[0],
179 bus: args[1],
180 pattern: args[2],
181 props: args[3],
182 };
183 }
184
185 return {
186 id: args[0],
187 pattern: args[1],
188 props: args[2],
189 };
190 }
191
192 if (isEventBus(args[0])) {
193 return {
194 bus: args[0],
195 pattern: args[1],
196 props: args[2],
197 };
198 }
199
200 return {
201 pattern: args[0],
202 props: args[1],
203 };
204};
205
206const isEventBus = (value: any): value is EventBus =>
207 value &&
208 typeof value === "object" &&
209 "Type" in value &&
210 value.Type === "AWS.EventBridge.EventBus";