1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as eventbridge from "@distilled.cloud/aws/eventbridge";
5import * as SQS from "@distilled.cloud/aws/sqs";
6import { describe, expect } from "alchemy-test";
7import * as Data from "effect/Data";
8import * as Effect from "effect/Effect";
9import * as Layer from "effect/Layer";
10import * as Schedule from "effect/Schedule";
11import * as HttpClient from "effect/http/HttpClient";
12import * as HttpClientRequest from "effect/http/HttpClientRequest";
13import EventBridgeTestFunctionLive, {
14 BusAndQueues,
15 BusAndQueuesLive,
16 EventBridgeTestFunction,
17} from "./handler.ts";
18
19const testOptions = { providers: AWS.providers() };
20const { test, beforeAll, afterAll } = Test.make(testOptions);
21const sharedStack = Core.scratchStack(testOptions, "EventBridgeBindings");
22
23// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
24// over 60s on a fresh deploy under parallel-suite load. Budget ~150s.
25const readinessPolicy = Schedule.max([
26 Schedule.fixed("2 seconds"),
27 Schedule.recurs(75),
28]);
29
30let baseUrl: string;
31let customQueueUrl: string;
32let defaultQueueUrl: string;
33let functionArn: string;
34
35class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
36 readonly status: number;
37 readonly body: string;
38}> {}
39
40// Retry transient 5xx from the fixture (cold re-init under parallel load);
41// a genuine 4xx/assertion failure is surfaced immediately.
42const send = (request: HttpClientRequest.HttpClientRequest) =>
43 HttpClient.execute(request).pipe(
44 Effect.flatMap((response) =>
45 response.status >= 500
46 ? response.text.pipe(
47 Effect.flatMap((body) =>
48 Effect.fail(
49 new TransientUpstream({ status: response.status, body }),
50 ),
51 ),
52 )
53 : Effect.succeed(response),
54 ),
55 Effect.retry({
56 while: (e) => e._tag === "TransientUpstream",
57 schedule: Schedule.max([
58 Schedule.exponential("500 millis"),
59 Schedule.recurs(6),
60 ]),
61 }),
62 );
63
64class EventNotDelivered extends Data.TaggedError("EventNotDelivered") {}
65
66interface DeliveredEvent {
67 source: string;
68 detailType: string;
69 detail: { marker: string };
70}
71
72// EventBridge rules on a freshly-created bus take a while to propagate; events
73// published before the rule is active are silently dropped (EventBridge does
74// not retro-deliver). So publish-then-poll in a bounded loop: each iteration
75// re-publishes the marker event and long-polls the sink queue for it.
76// Worst case ~= 18 * (publish + 5s long poll) ~= 95s, within the 120s test
77// timeout.
78const publishUntilReceived = Effect.fn(function* (
79 route: string,
80 queueUrl: string,
81 marker: string,
82) {
83 return yield* Effect.gen(function* () {
84 const publishResponse = yield* send(
85 HttpClientRequest.bodyJsonUnsafe(
86 HttpClientRequest.post(`${baseUrl}${route}`),
87 { marker },
88 ),
89 ).pipe(Effect.flatMap((r) => r.json));
90 expect(publishResponse).toHaveProperty("failedEntryCount", 0);
91
92 const result = yield* SQS.receiveMessage({
93 QueueUrl: queueUrl,
94 MaxNumberOfMessages: 10,
95 WaitTimeSeconds: 5,
96 });
97 const match = (result.Messages ?? [])
98 .flatMap((message) =>
99 message.Body ? [JSON.parse(message.Body) as DeliveredEvent] : [],
100 )
101 .find((body) => body.detail?.marker === marker);
102 if (!match) {
103 yield* Effect.logInfo(
104 `EventBridge test: no matching event on ${route} yet (rule still propagating?)`,
105 );
106 return yield* Effect.fail(new EventNotDelivered());
107 }
108 return match;
109 }).pipe(
110 Effect.retry({
111 while: (e) => e._tag === "EventNotDelivered",
112 schedule: Schedule.max([
113 Schedule.fixed("5 seconds"),
114 Schedule.recurs(17),
115 ]),
116 }),
117 );
118});
119
120describe(
121 "EventBridge Bindings",
122 {
123 tags: [
124 "provider:aws",
125 "provider:aws:eventbridge",
126 "provider:aws:lambda",
127 "provider:aws:sqs",
128 "live",
129 ],
130 },
131 () => {
132 beforeAll(
133 Effect.gen(function* () {
134 yield* Effect.logInfo(
135 "EventBridge test setup: destroying previous resources",
136 );
137 yield* sharedStack.destroy();
138
139 yield* Effect.logInfo("EventBridge test setup: deploying fixture");
140 const { fn, custom, dflt } = yield* sharedStack.deploy(
141 Effect.gen(function* () {
142 const { customQueue, defaultQueue } = yield* BusAndQueues;
143 const fn = yield* EventBridgeTestFunction;
144 return { fn, custom: customQueue, dflt: defaultQueue };
145 }).pipe(
146 Effect.provide(
147 Layer.mergeAll(EventBridgeTestFunctionLive, BusAndQueuesLive),
148 ),
149 ),
150 );
151
152 expect(fn.functionUrl).toBeTruthy();
153 baseUrl = fn.functionUrl!.replace(/\/+$/, "");
154 customQueueUrl = custom.queueUrl;
155 defaultQueueUrl = dflt.queueUrl;
156 functionArn = fn.functionArn;
157
158 yield* Effect.logInfo(
159 `EventBridge test setup: probing readiness at ${baseUrl}/health`,
160 );
161 yield* HttpClient.get(`${baseUrl}/health`).pipe(
162 Effect.flatMap((response) =>
163 response.status === 200
164 ? Effect.succeed(response)
165 : Effect.fail(
166 new Error(`Function not ready: ${response.status}`),
167 ),
168 ),
169 Effect.tapError((error) =>
170 Effect.logWarning(
171 `EventBridge test setup: fixture not ready yet (${String(error)})`,
172 ),
173 ),
174 Effect.retry({ schedule: readinessPolicy }),
175 );
176 yield* Effect.logInfo("EventBridge test setup: fixture ready");
177 }),
178 { timeout: 240_000 },
179 );
180
181 afterAll.skipIf(!!process.env.NO_DESTROY)(
182 sharedStack.destroy().pipe(
183 // Typed wait-until-gone: the custom bus must be deleted after teardown.
184 // The raw afterAll context has no providers layer (only `test.provider`
185 // bodies do), so the out-of-band distilled call must be wrapped in
186 // `withProviders` to satisfy Credentials/HttpClient/Region.
187 Effect.andThen(
188 Core.withProviders(
189 eventbridge
190 .describeEventBus({ Name: "alchemy-test-eb-bindings" })
191 .pipe(
192 Effect.map(() => false),
193 Effect.catchTag("ResourceNotFoundException", () =>
194 Effect.succeed(true),
195 ),
196 Effect.repeat({
197 schedule: Schedule.spaced("2 seconds"),
198 until: (isGone): boolean => isGone,
199 times: 10,
200 }),
201 Effect.map((gone) => expect(gone).toBe(true)),
202 ),
203 testOptions,
204 "EventBridgeBindings",
205 ),
206 ),
207 ),
208 {
209 // Teardown deletes ~30 resources (invoke permissions, event-source
210 // rules, the Lambda fixture + role, ToggleRule, TestArchive, both
211 // sinks, then the buses). Each delete is bounded — the EventBus
212 // provider force-sweeps AWS's lingering managed archival rule — but
213 // the serial sum needs headroom on a slow AWS day.
214 timeout: 360_000,
215 },
216 );
217
218 describe("PutEvents", () => {
219 test.provider("publishes an event to the custom bus", (_stack) =>
220 Effect.gen(function* () {
221 const response = (yield* send(
222 HttpClientRequest.bodyJsonUnsafe(
223 HttpClientRequest.post(`${baseUrl}/publish-custom`),
224 { marker: "put-events-custom-direct" },
225 ),
226 ).pipe(Effect.flatMap((r) => r.json))) as {
227 failedEntryCount: number;
228 entries: Array<{ EventId?: string }>;
229 };
230
231 expect(response.failedEntryCount).toBe(0);
232 expect(response.entries).toHaveLength(1);
233 expect(response.entries[0].EventId).toBeTruthy();
234 }),
235 );
236
237 test.provider("publishes an event to the default bus", (_stack) =>
238 Effect.gen(function* () {
239 const response = (yield* send(
240 HttpClientRequest.bodyJsonUnsafe(
241 HttpClientRequest.post(`${baseUrl}/publish-default`),
242 { marker: "put-events-default-direct" },
243 ),
244 ).pipe(Effect.flatMap((r) => r.json))) as {
245 failedEntryCount: number;
246 entries: Array<{ EventId?: string }>;
247 };
248
249 expect(response.failedEntryCount).toBe(0);
250 expect(response.entries).toHaveLength(1);
251 expect(response.entries[0].EventId).toBeTruthy();
252 }),
253 );
254 });
255
256 describe("EnableRule / DisableRule / DescribeRule", () => {
257 test.provider(
258 "toggles the rule state and observes it via DescribeRule",
259 (_stack) =>
260 Effect.gen(function* () {
261 const response = (yield* send(
262 HttpClientRequest.post(`${baseUrl}/rule-toggle`),
263 ).pipe(Effect.flatMap((r) => r.json))) as {
264 afterDisable: string;
265 afterEnable: string;
266 };
267
268 expect(response.afterDisable).toBe("DISABLED");
269 expect(response.afterEnable).toBe("ENABLED");
270 }),
271 { timeout: 60_000 },
272 );
273 });
274
275 describe("ListRuleNamesByTarget", () => {
276 test.provider(
277 "finds the consume-loop rule targeting the function",
278 (_stack) =>
279 Effect.gen(function* () {
280 const response = (yield* send(
281 HttpClientRequest.bodyJsonUnsafe(
282 HttpClientRequest.post(`${baseUrl}/rule-names-by-target`),
283 { targetArn: functionArn },
284 ),
285 ).pipe(Effect.flatMap((r) => r.json))) as {
286 ruleNames: string[];
287 };
288
289 // The default-bus consume loop's rule targets the Lambda.
290 expect(Array.isArray(response.ruleNames)).toBe(true);
291 expect(response.ruleNames.length).toBeGreaterThanOrEqual(1);
292 }),
293 { timeout: 60_000 },
294 );
295 });
296
297 describe("DescribeEventBus", () => {
298 test.provider("describes the bound custom bus", (_stack) =>
299 Effect.gen(function* () {
300 const response = (yield* send(
301 HttpClientRequest.get(`${baseUrl}/bus-info`),
302 ).pipe(Effect.flatMap((r) => r.json))) as {
303 name: string;
304 arn: string;
305 };
306
307 expect(response.name).toBe("alchemy-test-eb-bindings");
308 expect(response.arn).toContain(":event-bus/alchemy-test-eb-bindings");
309 }),
310 );
311 });
312
313 describe("ListEventBuses", () => {
314 test.provider(
315 "enumerates account buses including the default",
316 (_stack) =>
317 Effect.gen(function* () {
318 const response = (yield* send(
319 HttpClientRequest.get(`${baseUrl}/event-buses`),
320 ).pipe(Effect.flatMap((r) => r.json))) as { names: string[] };
321
322 expect(response.names).toContain("default");
323 expect(response.names).toContain("alchemy-test-eb-bindings");
324 }),
325 );
326 });
327
328 describe("ListRules", () => {
329 test.provider("lists the rules on the bound custom bus", (_stack) =>
330 Effect.gen(function* () {
331 const response = (yield* send(
332 HttpClientRequest.get(`${baseUrl}/rules`),
333 ).pipe(Effect.flatMap((r) => r.json))) as { ruleNames: string[] };
334
335 // The toggle rule and the custom-bus consume-loop rule live here.
336 expect(response.ruleNames).toContain("alchemy-test-eb-toggle");
337 expect(response.ruleNames.length).toBeGreaterThanOrEqual(2);
338 }),
339 );
340 });
341
342 describe("ListTargetsByRule", () => {
343 test.provider("lists the (empty) targets of the toggle rule", (_stack) =>
344 Effect.gen(function* () {
345 const response = (yield* send(
346 HttpClientRequest.get(`${baseUrl}/targets-by-rule`),
347 ).pipe(Effect.flatMap((r) => r.json))) as { targetCount: number };
348
349 expect(response.targetCount).toBe(0);
350 }),
351 );
352 });
353
354 describe("TestEventPattern", () => {
355 test.provider("matches and rejects events against a pattern", (_stack) =>
356 Effect.gen(function* () {
357 const matching = (yield* send(
358 HttpClientRequest.bodyJsonUnsafe(
359 HttpClientRequest.post(`${baseUrl}/test-pattern`),
360 { source: "alchemy.test.pattern" },
361 ),
362 ).pipe(Effect.flatMap((r) => r.json))) as { matches: boolean };
363 expect(matching.matches).toBe(true);
364
365 const nonMatching = (yield* send(
366 HttpClientRequest.bodyJsonUnsafe(
367 HttpClientRequest.post(`${baseUrl}/test-pattern`),
368 { source: "other.source" },
369 ),
370 ).pipe(Effect.flatMap((r) => r.json))) as { matches: boolean };
371 expect(nonMatching.matches).toBe(false);
372 }),
373 );
374 });
375
376 describe("Replays", () => {
377 test.provider("ListReplays enumerates account replays", (_stack) =>
378 Effect.gen(function* () {
379 const response = (yield* send(
380 HttpClientRequest.get(`${baseUrl}/replays`),
381 ).pipe(Effect.flatMap((r) => r.json))) as { count: number };
382
383 expect(typeof response.count).toBe("number");
384 expect(response.count).toBeGreaterThanOrEqual(0);
385 }),
386 );
387
388 test.provider(
389 "StartReplay/DescribeReplay/CancelReplay drive a replay of the archive",
390 (_stack) =>
391 Effect.gen(function* () {
392 const res = yield* send(
393 HttpClientRequest.bodyJsonUnsafe(
394 HttpClientRequest.post(`${baseUrl}/replay`),
395 {},
396 ),
397 );
398 const bodyText = yield* res.text;
399 expect(res.status, bodyText).toBe(200);
400 const response = yield* Effect.sync(
401 () => JSON.parse(bodyText) as { state: string; cancel: string },
402 );
403
404 expect([
405 "STARTING",
406 "RUNNING",
407 "COMPLETED",
408 "CANCELLING",
409 "CANCELLED",
410 "FAILED",
411 ]).toContain(response.state);
412 expect(["cancelled", "not-cancellable"]).toContain(response.cancel);
413 }),
414 { timeout: 60_000 },
415 );
416 });
417
418 describe("EventSource", () => {
419 test.provider(
420 "consume loop receives custom-bus events",
421 (_stack) =>
422 Effect.gen(function* () {
423 const event = yield* publishUntilReceived(
424 "/publish-custom",
425 customQueueUrl,
426 "consume-loop-custom",
427 );
428
429 expect(event.source).toBe("alchemy.test.custom");
430 expect(event.detailType).toBe("TestEvent");
431 expect(event.detail.marker).toBe("consume-loop-custom");
432 }),
433 { timeout: 120_000 },
434 );
435
436 test.provider(
437 "consume loop receives default-bus events",
438 (_stack) =>
439 Effect.gen(function* () {
440 const event = yield* publishUntilReceived(
441 "/publish-default",
442 defaultQueueUrl,
443 "consume-loop-default",
444 );
445
446 expect(event.source).toBe("alchemy.test.default");
447 expect(event.detailType).toBe("TestEvent");
448 expect(event.detail.marker).toBe("consume-loop-default");
449 }),
450 { timeout: 120_000 },
451 );
452 });
453 },
454);