1import * as AWS from "@/AWS";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Result from "effect/Result";
6import * as Stream from "effect/Stream";
7import { HttpServerRequest } from "effect/http/HttpServerRequest";
8import * as HttpServerResponse from "effect/http/HttpServerResponse";
9
10export class EventBridgeTestFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
11 "EventBridgeTestFunction",
12) {}
13
14/**
15 * Shared infrastructure for the EventBridge bindings fixture:
16 * - a custom event bus (PutEvents + consumeBusEvents against a named bus)
17 * - one SQS sink queue per consume loop (custom bus + default bus) so the
18 * test can observe delivered events out-of-band via `sqs.receiveMessage`
19 * - a target-less rule on the custom bus for the Enable/DisableRule toggles
20 * - an archive on the custom bus for the replay bindings
21 */
22export class BusAndQueues extends Context.Service<
23 BusAndQueues,
24 {
25 bus: AWS.EventBridge.EventBus;
26 customQueue: AWS.SQS.Queue;
27 defaultQueue: AWS.SQS.Queue;
28 toggleRule: AWS.EventBridge.Rule;
29 archive: AWS.EventBridge.Archive;
30 }
31>()("EventBridgeBusAndQueues") {}
32
33export const BusAndQueuesLive = Layer.effect(
34 BusAndQueues,
35 Effect.gen(function* () {
36 const bus = yield* AWS.EventBridge.EventBus("TestBus", {
37 name: "alchemy-test-eb-bindings",
38 // A crashed run can leak event-source rules whose state rows are
39 // gone; without forceDestroy those orphans block deleteEventBus
40 // with EventBusHasRules forever.
41 forceDestroy: true,
42 });
43 const customQueue = yield* AWS.SQS.Queue("CustomBusSink");
44 const defaultQueue = yield* AWS.SQS.Queue("DefaultBusSink");
45 const toggleRule = yield* AWS.EventBridge.Rule("ToggleRule", {
46 name: "alchemy-test-eb-toggle",
47 eventBusName: bus.eventBusName,
48 eventPattern: { source: ["alchemy.test.toggle"] },
49 });
50 const archive = yield* AWS.EventBridge.Archive("TestArchive", {
51 name: "alchemy-test-eb-archive",
52 eventSourceArn: bus.eventBusArn,
53 retention: "1 day",
54 });
55 return { bus, customQueue, defaultQueue, toggleRule, archive };
56 }),
57);
58
59export default EventBridgeTestFunction.make(
60 {
61 main: import.meta.url,
62 functionUrl: true,
63 },
64 Effect.gen(function* () {
65 const { bus, customQueue, defaultQueue, toggleRule, archive } =
66 yield* BusAndQueues;
67
68 const putEventsCustom = yield* AWS.EventBridge.PutEvents(bus);
69 const putEventsDefault = yield* AWS.EventBridge.PutEvents();
70 const customSink = yield* AWS.SQS.QueueSink(customQueue);
71 const defaultSink = yield* AWS.SQS.QueueSink(defaultQueue);
72 const enableRule = yield* AWS.EventBridge.EnableRule(toggleRule);
73 const disableRule = yield* AWS.EventBridge.DisableRule(toggleRule);
74 const describeToggleRule = yield* AWS.EventBridge.DescribeRule(toggleRule);
75 const listRuleNamesByTarget =
76 yield* AWS.EventBridge.ListRuleNamesByTarget();
77 const startReplay = yield* AWS.EventBridge.StartReplay(archive);
78 const describeReplay = yield* AWS.EventBridge.DescribeReplay();
79 const cancelReplay = yield* AWS.EventBridge.CancelReplay();
80 const listReplays = yield* AWS.EventBridge.ListReplays();
81 const describeCustomBus = yield* AWS.EventBridge.DescribeEventBus(bus);
82 const listEventBuses = yield* AWS.EventBridge.ListEventBuses();
83 const listCustomBusRules = yield* AWS.EventBridge.ListRules(bus);
84 const listToggleRuleTargets =
85 yield* AWS.EventBridge.ListTargetsByRule(toggleRule);
86 const testEventPattern = yield* AWS.EventBridge.TestEventPattern();
87
88 // Consume loop on the CUSTOM bus: matching events are forwarded to the
89 // custom sink queue where the test observes them out-of-band.
90 yield* AWS.EventBridge.consumeBusEvents(
91 bus,
92 { source: ["alchemy.test.custom"] },
93 (events: Stream.Stream<AWS.EventBridge.EventRecord>) =>
94 events.pipe(
95 Stream.map((event) => ({
96 MessageBody: JSON.stringify({
97 source: event.source,
98 detailType: event["detail-type"],
99 detail: event.detail,
100 }),
101 })),
102 Stream.run(customSink),
103 Effect.orDie,
104 ),
105 );
106
107 // Consume loop on the DEFAULT bus (no bus argument).
108 yield* AWS.EventBridge.consumeBusEvents(
109 { source: ["alchemy.test.default"] },
110 (events: Stream.Stream<AWS.EventBridge.EventRecord>) =>
111 events.pipe(
112 Stream.map((event) => ({
113 MessageBody: JSON.stringify({
114 source: event.source,
115 detailType: event["detail-type"],
116 detail: event.detail,
117 }),
118 })),
119 Stream.run(defaultSink),
120 Effect.orDie,
121 ),
122 );
123
124 return {
125 fetch: Effect.gen(function* () {
126 const request = yield* HttpServerRequest;
127 const url = new URL(request.originalUrl);
128 const pathname = url.pathname;
129
130 if (request.method === "GET" && pathname === "/health") {
131 return yield* HttpServerResponse.json({ ok: true });
132 }
133
134 if (request.method === "POST" && pathname === "/publish-custom") {
135 const body = (yield* request.json) as unknown as { marker: string };
136 const result = yield* putEventsCustom({
137 Entries: [
138 {
139 Source: "alchemy.test.custom",
140 DetailType: "TestEvent",
141 Detail: JSON.stringify({ marker: body.marker }),
142 },
143 ],
144 });
145 return yield* HttpServerResponse.json({
146 failedEntryCount: result.FailedEntryCount ?? 0,
147 entries: result.Entries ?? [],
148 });
149 }
150
151 if (request.method === "POST" && pathname === "/publish-default") {
152 const body = (yield* request.json) as unknown as { marker: string };
153 const result = yield* putEventsDefault({
154 Entries: [
155 {
156 Source: "alchemy.test.default",
157 DetailType: "TestEvent",
158 Detail: JSON.stringify({ marker: body.marker }),
159 },
160 ],
161 });
162 return yield* HttpServerResponse.json({
163 failedEntryCount: result.FailedEntryCount ?? 0,
164 entries: result.Entries ?? [],
165 });
166 }
167
168 if (request.method === "POST" && pathname === "/rule-toggle") {
169 // Disable, observe, re-enable, observe — exercises DisableRule,
170 // DescribeRule, and EnableRule end-to-end.
171 yield* disableRule();
172 const afterDisable = (yield* describeToggleRule()).State;
173 yield* enableRule();
174 const afterEnable = (yield* describeToggleRule()).State;
175 return yield* HttpServerResponse.json({ afterDisable, afterEnable });
176 }
177
178 if (request.method === "POST" && pathname === "/rule-names-by-target") {
179 const body = (yield* request.json) as unknown as {
180 targetArn: string;
181 };
182 const result = yield* listRuleNamesByTarget({
183 TargetArn: body.targetArn,
184 });
185 return yield* HttpServerResponse.json({
186 ruleNames: result.RuleNames ?? [],
187 });
188 }
189
190 if (request.method === "GET" && pathname === "/bus-info") {
191 // DescribeEventBus scoped to the bound custom bus.
192 const info = yield* describeCustomBus();
193 return yield* HttpServerResponse.json({
194 name: info.Name,
195 arn: info.Arn,
196 });
197 }
198
199 if (request.method === "GET" && pathname === "/event-buses") {
200 // ListEventBuses is account-level; the default bus always exists.
201 const result = yield* listEventBuses();
202 return yield* HttpServerResponse.json({
203 names: (result.EventBuses ?? []).flatMap((b) =>
204 b.Name ? [b.Name] : [],
205 ),
206 });
207 }
208
209 if (request.method === "GET" && pathname === "/rules") {
210 // ListRules scoped to the bound custom bus — sees the toggle rule
211 // and the consume-loop rule.
212 const result = yield* listCustomBusRules();
213 return yield* HttpServerResponse.json({
214 ruleNames: (result.Rules ?? []).flatMap((r) =>
215 r.Name ? [r.Name] : [],
216 ),
217 });
218 }
219
220 if (request.method === "GET" && pathname === "/targets-by-rule") {
221 // ListTargetsByRule scoped to the (target-less) toggle rule.
222 const result = yield* listToggleRuleTargets();
223 return yield* HttpServerResponse.json({
224 targetCount: (result.Targets ?? []).length,
225 });
226 }
227
228 if (request.method === "POST" && pathname === "/test-pattern") {
229 const body = (yield* request.json) as unknown as { source: string };
230 const event = JSON.stringify({
231 id: "1",
232 source: body.source,
233 "detail-type": "TestEvent",
234 account: "123456789012",
235 region: "us-east-1",
236 time: new Date().toISOString(),
237 detail: {},
238 });
239 const pattern = JSON.stringify({ source: ["alchemy.test.pattern"] });
240 const matching = yield* testEventPattern({
241 EventPattern: pattern,
242 Event: event,
243 });
244 return yield* HttpServerResponse.json({
245 matches: matching.Result ?? false,
246 });
247 }
248
249 if (request.method === "GET" && pathname === "/replays") {
250 const result = yield* listReplays();
251 return yield* HttpServerResponse.json({
252 count: (result.Replays ?? []).length,
253 });
254 }
255
256 if (request.method === "POST" && pathname === "/replay") {
257 const replayName = "alchemy-test-eb-replay";
258 // Replay an empty window from the recent past onto the archive's
259 // source bus (the binding's default destination). Replays cannot
260 // be deleted, so re-runs hit ResourceAlreadyExistsException —
261 // treated as success and observed via DescribeReplay below.
262 const now = Date.now();
263 const started = yield* Effect.result(
264 startReplay({
265 ReplayName: replayName,
266 EventStartTime: new Date(now - 10 * 60 * 1000),
267 EventEndTime: new Date(now - 5 * 60 * 1000),
268 }).pipe(
269 Effect.catchTag("ResourceAlreadyExistsException", () =>
270 Effect.succeed(undefined),
271 ),
272 ),
273 );
274 if (Result.isFailure(started)) {
275 return yield* HttpServerResponse.json(
276 {
277 error: "startReplay",
278 failure: JSON.stringify(started.failure),
279 },
280 { status: 400 },
281 );
282 }
283 const described = yield* Effect.result(
284 describeReplay({ ReplayName: replayName }),
285 );
286 if (Result.isFailure(described)) {
287 return yield* HttpServerResponse.json(
288 {
289 error: "describeReplay",
290 failure: JSON.stringify(described.failure),
291 },
292 { status: 400 },
293 );
294 }
295 // Cancelling a finished replay fails with the typed
296 // IllegalStatusException — either outcome proves the binding.
297 const cancel = yield* Effect.result(
298 cancelReplay({ ReplayName: replayName }).pipe(
299 Effect.map(() => "cancelled" as const),
300 Effect.catchTag("IllegalStatusException", () =>
301 Effect.succeed("not-cancellable" as const),
302 ),
303 ),
304 );
305 if (Result.isFailure(cancel)) {
306 return yield* HttpServerResponse.json(
307 {
308 error: "cancelReplay",
309 failure: JSON.stringify(cancel.failure),
310 },
311 { status: 400 },
312 );
313 }
314 return yield* HttpServerResponse.json({
315 state: described.success.State,
316 cancel: cancel.success,
317 });
318 }
319
320 return yield* HttpServerResponse.json(
321 { error: "Not found", method: request.method, pathname },
322 { status: 404 },
323 );
324 }).pipe(Effect.orDie),
325 };
326 }).pipe(
327 Effect.provide(
328 Layer.provideMerge(
329 Layer.mergeAll(
330 AWS.Lambda.EventSource,
331 AWS.EventBridge.PutEventsHttp,
332 AWS.EventBridge.EnableRuleHttp,
333 AWS.EventBridge.DisableRuleHttp,
334 AWS.EventBridge.DescribeRuleHttp,
335 AWS.EventBridge.ListRuleNamesByTargetHttp,
336 AWS.EventBridge.StartReplayHttp,
337 AWS.EventBridge.DescribeReplayHttp,
338 AWS.EventBridge.CancelReplayHttp,
339 AWS.EventBridge.ListReplaysHttp,
340 AWS.EventBridge.DescribeEventBusHttp,
341 AWS.EventBridge.ListEventBusesHttp,
342 AWS.EventBridge.ListRulesHttp,
343 AWS.EventBridge.ListTargetsByRuleHttp,
344 AWS.EventBridge.TestEventPatternHttp,
345 AWS.SQS.QueueSinkHttp,
346 BusAndQueuesLive,
347 ),
348 Layer.mergeAll(AWS.SQS.SendMessageBatchHttp),
349 ),
350 ),
351 ),
352);