1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import * as eventbridge from "@distilled.cloud/aws/eventbridge";
4import * as SQS from "@distilled.cloud/aws/sqs";
5import { expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import * as HttpBody from "effect/http/HttpBody";
10import * as HttpClient from "effect/http/HttpClient";
11import { BusSinkFunction, BusSinkFunctionLive } from "./sink-handler.ts";
12
13const { test } = Test.make({ providers: AWS.providers() });
14
15class FunctionNotReady extends Data.TaggedError("FunctionNotReady") {}
16
17class SinkRequestFailed extends Data.TaggedError("SinkRequestFailed")<{
18 readonly status: number;
19 readonly body: string;
20}> {}
21
22class EventsNotDelivered extends Data.TaggedError("EventsNotDelivered")<{
23 readonly missing: number;
24}> {}
25
26const waitForFunctionReady = (url: string) =>
27 HttpClient.get(url).pipe(
28 Effect.flatMap((response) =>
29 response.status === 200
30 ? (response.json as Effect.Effect<{ queueUrl: string }>)
31 : Effect.fail(new FunctionNotReady()),
32 ),
33 // A freshly-deployed function can briefly serve a 200 before its captured
34 // env vars (the queue URL) have finished propagating, so treat a missing
35 // queueUrl as "not ready yet" and keep polling.
36 Effect.flatMap((json: any) =>
37 typeof json?.queueUrl === "string"
38 ? Effect.succeed({ queueUrl: json.queueUrl as string })
39 : Effect.fail(new FunctionNotReady()),
40 ),
41 Effect.retry({
42 while: (error) => error._tag === "FunctionNotReady",
43 schedule: Schedule.max([
44 Schedule.fixed("2 seconds"),
45 Schedule.recurs(75),
46 ]),
47 }),
48 );
49
50/** POST a marker batch to the fixture's /sink route; fail on non-200. */
51const postSink = Effect.fn(function* (
52 baseUrl: string,
53 body: { markers: string[]; includeMalformed?: boolean },
54) {
55 const response = yield* HttpClient.post(`${baseUrl}/sink`, {
56 body: yield* HttpBody.json(body),
57 });
58 if (response.status !== 200) {
59 const text = yield* response.text;
60 return yield* Effect.fail(
61 new SinkRequestFailed({ status: response.status, body: text }),
62 );
63 }
64 return (yield* response.json) as { ok: boolean; count: number };
65});
66
67/**
68 * Drain the routed queue, collecting `detail.marker` values into `received`,
69 * until every expected marker has been observed. Bounded — fails with
70 * `EventsNotDelivered` when the schedule is exhausted.
71 */
72const waitForMarkers = Effect.fn(function* (
73 queueUrl: string,
74 expected: readonly string[],
75) {
76 const received = new Set<string>();
77 yield* Effect.gen(function* () {
78 const result = yield* SQS.receiveMessage({
79 QueueUrl: queueUrl,
80 MaxNumberOfMessages: 10,
81 WaitTimeSeconds: 2,
82 });
83 for (const message of result.Messages ?? []) {
84 if (message.Body) {
85 const event = JSON.parse(message.Body) as {
86 detail?: { marker?: string };
87 };
88 const marker = event.detail?.marker;
89 if (typeof marker === "string") {
90 received.add(marker);
91 }
92 }
93 if (message.ReceiptHandle) {
94 yield* SQS.deleteMessage({
95 QueueUrl: queueUrl,
96 ReceiptHandle: message.ReceiptHandle,
97 });
98 }
99 }
100 const missing = expected.filter((marker) => !received.has(marker));
101 if (missing.length > 0) {
102 return yield* Effect.fail(
103 new EventsNotDelivered({ missing: missing.length }),
104 );
105 }
106 }).pipe(
107 Effect.retry({
108 while: (e) => e._tag === "EventsNotDelivered",
109 schedule: Schedule.max([
110 Schedule.fixed("2 seconds"),
111 Schedule.recurs(30),
112 ]),
113 }),
114 );
115 return received;
116});
117
118/**
119 * EventBridge rules on a freshly-created bus take a while to propagate;
120 * events published before the rule is active are silently dropped (no
121 * retro-delivery). Publish-then-poll in a bounded loop: each iteration
122 * re-publishes the probe marker through the sink and long-polls the routed
123 * queue for it.
124 */
125const probeUntilRouted = Effect.fn(function* (
126 baseUrl: string,
127 queueUrl: string,
128 marker: string,
129) {
130 yield* Effect.gen(function* () {
131 const response = yield* postSink(baseUrl, { markers: [marker] });
132 expect(response.ok).toBe(true);
133
134 const result = yield* SQS.receiveMessage({
135 QueueUrl: queueUrl,
136 MaxNumberOfMessages: 10,
137 WaitTimeSeconds: 5,
138 });
139 const match = (result.Messages ?? []).some((message) =>
140 message.Body ? message.Body.includes(marker) : false,
141 );
142 if (!match) {
143 yield* Effect.logInfo(
144 "BusSink test: probe event not routed yet (rule still propagating?)",
145 );
146 return yield* Effect.fail(new EventsNotDelivered({ missing: 1 }));
147 }
148 }).pipe(
149 Effect.retry({
150 while: (e) => e._tag === "EventsNotDelivered",
151 schedule: Schedule.max([
152 Schedule.fixed("5 seconds"),
153 Schedule.recurs(17),
154 ]),
155 }),
156 );
157});
158
159test.provider(
160 "BusSink streams entries through a deployed Lambda to EventBridge",
161 (stack) =>
162 Effect.gen(function* () {
163 // Leading destroy: reconcile away any partial deployment left behind by
164 // a previous crashed run (physical names are deterministic constants).
165 yield* stack.destroy();
166
167 const fn = yield* stack.deploy(
168 BusSinkFunction.pipe(Effect.provide(BusSinkFunctionLive)),
169 );
170 const baseUrl = fn.functionUrl!.replace(/\/+$/, "");
171
172 const { queueUrl } = yield* waitForFunctionReady(`${baseUrl}/ready`);
173
174 // Prove the bus -> queue rule is active before asserting on batches.
175 yield* probeUntilRouted(baseUrl, queueUrl, "bussink-probe");
176
177 // 25 entries > the PutEvents limit of 10, so the batched sink must
178 // split the chunk into 3 sequential API calls (10 + 10 + 5).
179 const markers = Array.from(
180 { length: 25 },
181 (_, i) => `bussink-${i}-${crypto.randomUUID()}`,
182 );
183 const response = yield* postSink(baseUrl, { markers });
184 expect(response.ok).toBe(true);
185 expect(response.count).toBe(markers.length);
186
187 yield* waitForMarkers(queueUrl, markers);
188
189 // Partial failure: one entry with malformed Detail is rejected
190 // per-entry (ErrorCode: MalformedDetail). The sink must drop it and
191 // deliver the rest without failing the batch.
192 const partialMarkers = Array.from(
193 { length: 3 },
194 (_, i) => `bussink-partial-${i}-${crypto.randomUUID()}`,
195 );
196 const partial = yield* postSink(baseUrl, {
197 markers: partialMarkers,
198 includeMalformed: true,
199 });
200 expect(partial.ok).toBe(true);
201 expect(partial.count).toBe(partialMarkers.length + 1);
202
203 yield* waitForMarkers(queueUrl, partialMarkers);
204
205 yield* stack.destroy();
206
207 // Typed wait-until-gone: the sink's event bus must be deleted.
208 const gone = yield* eventbridge
209 .describeEventBus({ Name: "alchemy-test-eb-bus-sink" })
210 .pipe(
211 Effect.map(() => false),
212 Effect.catchTag("ResourceNotFoundException", () =>
213 Effect.succeed(true),
214 ),
215 Effect.repeat({
216 schedule: Schedule.spaced("2 seconds"),
217 until: (isGone): boolean => isGone,
218 times: 10,
219 }),
220 );
221 expect(gone).toBe(true);
222 }),
223 {
224 tags: [
225 "provider:aws",
226 "provider:aws:eventbridge",
227 "provider:aws:lambda",
228 "provider:aws:sqs",
229 "live",
230 ],
231 timeout: 240_000,
232 },
233);