1import type * as lambda from "aws-lambda";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Stream from "effect/Stream";
5import {
6 EventSource as EventBridgeEventSource,
7 matchesEventPattern,
8 type EventPattern,
9 type EventRecord,
10 type EventRouteProps,
11 type EventSourceService,
12} from "../EventBridge/EventSource.ts";
13import { toLambda as createLambdaRoute } from "../EventBridge/ToLambda.ts";
14import * as Lambda from "./Function.ts";
15
16/**
17 * Narrow an arbitrary Lambda invocation payload to an EventBridge event.
18 */
19export const isEventBridgeEvent = (
20 event: any,
21): event is lambda.EventBridgeEvent<string, any> =>
22 typeof event?.source === "string" &&
23 typeof event?.["detail-type"] === "string";
24
25/**
26 * Lambda runtime implementation for `AWS.EventBridge.consumeBusEvents(...)`.
27 *
28 * This layer does two things:
29 *
30 * 1. It delegates to `EventSourcePolicy` so deployment creates an EventBridge
31 * rule targeting the current Lambda function.
32 * 2. At runtime it filters incoming Lambda events against the original event
33 * pattern and forwards matching events into the supplied `Stream`.
34 * ### Subscribing To The Default Bus
35 * **Example:** Match User Events On The Default Bus
36 * ```typescript
37 * yield* AWS.EventBridge.consumeBusEvents(
38 * {
39 * source: ["app.user"],
40 * "detail-type": ["UserCreated"],
41 * },
42 * (events) =>
43 * Stream.runForEach(events, (event) =>
44 * Effect.log(`new user: ${event.detail.userId}`),
45 * ),
46 * );
47 * ```
48 *
49 * ### Subscribing To A Custom Bus
50 * **Example:** Match Orders On A Named Bus
51 * ```typescript
52 * const bus = yield* AWS.EventBridge.EventBus("OrdersBus", {
53 * name: "orders",
54 * });
55 *
56 * yield* AWS.EventBridge.consumeBusEvents(
57 * bus,
58 * {
59 * source: ["app.orders"],
60 * "detail-type": ["OrderPaid"],
61 * },
62 * (events) =>
63 * Stream.runForEach(events, (event) =>
64 * Effect.log(`paid order: ${event.detail.orderId}`),
65 * ),
66 * );
67 * ```
68 *
69 * ### Explicit Route Names
70 * **Example:** Name The Backing Rule Deterministically
71 * ```typescript
72 * yield* AWS.EventBridge.consumeBusEvents(
73 * "InvoiceEvents",
74 * {
75 * source: ["app.billing"],
76 * "detail-type": ["InvoiceIssued"],
77 * },
78 * {
79 * description: "Deliver invoice events into this Lambda function",
80 * },
81 * (events) =>
82 * Stream.runForEach(events, (event) =>
83 * Effect.log(`invoice: ${event.detail.invoiceId}`),
84 * ),
85 * );
86 * ```
87 *
88 * ### Processing Typed Details
89 * **Example:** Narrow The Event Detail Payload
90 * ```typescript
91 * type UserCreated = {
92 * userId: string;
93 * email: string;
94 * };
95 *
96 * yield* AWS.EventBridge.consumeBusEvents(
97 * {
98 * source: ["app.user"],
99 * "detail-type": ["UserCreated"],
100 * },
101 * (events) =>
102 * Stream.runForEach(
103 * events as Stream.Stream<AWS.EventBridge.EventRecord<UserCreated>>,
104 * (event) => Effect.log(`welcome ${event.detail.email}`),
105 * ),
106 * );
107 * ```
108 *
109 * @binding
110 */
111export const EventSource = Layer.effect(
112 EventBridgeEventSource,
113 Effect.gen(function* () {
114 const host = yield* Lambda.Function;
115
116 return Effect.fn(function* <
117 Detail = unknown,
118 StreamReq = never,
119 Req = never,
120 >(
121 descriptor: {
122 id?: string;
123 bus?: any;
124 pattern: EventPattern;
125 props?: EventRouteProps;
126 },
127 process: (
128 events: Stream.Stream<EventRecord<Detail>, never, StreamReq>,
129 ) => Effect.Effect<void, never, Req>,
130 ) {
131 // Deploy-time: create the backing EventBridge rule + Lambda permission
132 // targeting this function. Skipped once running inside the deployed
133 // Function (the global guard), where the only work is registering the
134 // runtime handler below.
135 if (!globalThis.__ALCHEMY_RUNTIME__) {
136 yield* createLambdaRoute(descriptor, host).pipe(Effect.asVoid);
137 }
138
139 yield* host.listen(
140 Effect.sync(() => (event: any) => {
141 if (
142 isEventBridgeEvent(event) &&
143 matchesEventPattern(descriptor.pattern, event)
144 ) {
145 return process(Stream.succeed(event as EventRecord<Detail>)).pipe(
146 Effect.orDie,
147 );
148 }
149 }),
150 );
151 }) as EventSourceService;
152 }),
153);