1import * as AWS from "@/AWS";
2import * as Alchemy from "@/index.ts";
3import * as State from "@/State";
4import * as Test from "@/Test/Alchemy";
5import * as Firehose from "@distilled.cloud/aws/firehose";
6import * as S3 from "@distilled.cloud/aws/s3";
7import { describe, expect } from "alchemy-test";
8import * as Data from "effect/Data";
9import * as Effect from "effect/Effect";
10import * as Schedule from "effect/Schedule";
11import * as Stream from "effect/Stream";
12import * as HttpClient from "effect/http/HttpClient";
13import * as HttpClientRequest from "effect/http/HttpClientRequest";
14import FirehoseApiFunctionLive, {
15 BucketAndDeliveryStream,
16 FirehoseApiFunction,
17} from "./handler.ts";
18
19const providers = AWS.providers();
20const state = State.localState();
21const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
22 providers,
23 state,
24});
25
26const Stack = Alchemy.Stack(
27 "firehose-bindings",
28 { providers, state },
29 Effect.gen(function* () {
30 // Share the bucket/delivery-stream between the deployed function and the
31 // stack outputs so the test can verify ingest out-of-band via distilled.
32 const { bucket, deliveryStream } = yield* BucketAndDeliveryStream;
33 const fn = yield* FirehoseApiFunction;
34 return {
35 url: fn.functionUrl.as<string>(),
36 deliveryStreamName: deliveryStream.deliveryStreamName.as<string>(),
37 bucketName: bucket.bucketName.as<string>(),
38 };
39 }).pipe(Effect.provide(FirehoseApiFunctionLive)),
40);
41
42const stack = beforeAll(
43 Effect.gen(function* () {
44 // A prior interrupted run may leave local state pointing at a partially
45 // deleted Lambda/IAM fixture. Start from a clean stack; provider deletes
46 // are idempotent, and deploy then proves stale-state recovery as well.
47 yield* destroy(Stack);
48 return yield* deploy(Stack);
49 }),
50 { timeout: 240_000 },
51);
52afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack), { timeout: 120_000 });
53
54// Lambda Function URLs cold-start (DNS, init) and a fresh role's IAM grants
55// (eventual consistency) can both take a while on the first hit. Retrying on
56// any non-200 lets the first request wait through that window; warm calls
57// return on the first try and never retry.
58const readinessSchedule = Schedule.max([
59 Schedule.fixed("2 seconds"),
60 Schedule.recurs(75),
61]);
62
63const urlOf = (baseUrl: string, path: string) =>
64 `${baseUrl.replace(/\/+$/, "")}${path}`;
65
66const getJson = (baseUrl: string, path: string) =>
67 HttpClient.execute(HttpClientRequest.get(urlOf(baseUrl, path))).pipe(
68 Effect.flatMap((response) =>
69 response.status === 200
70 ? response.json
71 : Effect.fail(new Error(`Request failed: ${response.status}`)),
72 ),
73 Effect.retry({ schedule: readinessSchedule }),
74 );
75
76const postJson = (baseUrl: string, path: string, body: unknown) =>
77 HttpClient.execute(
78 HttpClientRequest.bodyJsonUnsafe(
79 HttpClientRequest.post(urlOf(baseUrl, path)),
80 body,
81 ),
82 ).pipe(
83 Effect.flatMap((response) =>
84 response.status === 200
85 ? response.json
86 : Effect.fail(new Error(`Request failed: ${response.status}`)),
87 ),
88 Effect.retry({ schedule: readinessSchedule }),
89 );
90
91describe(
92 "Firehose Bindings",
93 {
94 tags: [
95 "provider:aws",
96 "provider:aws:firehose",
97 "provider:aws:lambda",
98 "provider:aws:s3",
99 "live",
100 ],
101 },
102 () => {
103 describe("PutRecord", () => {
104 // `test.provider` supplies the AWS environment for the out-of-band
105 // describeDeliveryStream verification via distilled.
106 test.provider("writes a single record and the stream stays ACTIVE", () =>
107 Effect.gen(function* () {
108 const { url, deliveryStreamName } = yield* stack;
109 const response = yield* postJson(url, "/put-record", {
110 data: `put-record-${crypto.randomUUID()}`,
111 });
112 expect((response as any).RecordId).toBeTruthy();
113
114 // Ingest success is the assertion — S3 delivery is asynchronous
115 // (buffering ≥ 60s), so verify the stream is ACTIVE out-of-band
116 // instead of waiting for objects to land.
117 const described = yield* Firehose.describeDeliveryStream({
118 DeliveryStreamName: deliveryStreamName,
119 });
120 expect(
121 described.DeliveryStreamDescription.DeliveryStreamStatus,
122 ).toEqual("ACTIVE");
123 }),
124 );
125 });
126
127 describe("PutRecordBatch", () => {
128 test(
129 "writes a batch of records with zero failures",
130 Effect.gen(function* () {
131 const { url } = yield* stack;
132 const response = yield* postJson(url, "/put-record-batch", {
133 records: [
134 `batch-1-${crypto.randomUUID()}`,
135 `batch-2-${crypto.randomUUID()}`,
136 `batch-3-${crypto.randomUUID()}`,
137 ],
138 });
139 expect((response as any).FailedPutCount).toBe(0);
140 const entries = (response as any).RequestResponses ?? [];
141 expect(entries.length).toBe(3);
142 for (const entry of entries) {
143 expect(entry.RecordId).toBeTruthy();
144 expect(entry.ErrorCode).toBeUndefined();
145 }
146 }),
147 );
148 });
149
150 describe("ListDeliveryStreams", () => {
151 test(
152 "lists delivery streams and includes the fixture stream",
153 Effect.gen(function* () {
154 const { url, deliveryStreamName } = yield* stack;
155 const response = (yield* getJson(url, "/list-streams")) as {
156 DeliveryStreamNames: string[];
157 HasMoreDeliveryStreams: boolean;
158 };
159 expect(Array.isArray(response.DeliveryStreamNames)).toBe(true);
160 expect(response.DeliveryStreamNames).toContain(deliveryStreamName);
161 }),
162 );
163 });
164
165 describe("DeliveryStreamSink", () => {
166 // `test.provider` supplies the AWS environment for the out-of-band
167 // describeDeliveryStream verification via distilled.
168 test.provider("streams records through the sink helper", () =>
169 Effect.gen(function* () {
170 const { url, deliveryStreamName } = yield* stack;
171 const response = yield* postJson(url, "/sink", {
172 records: [
173 `sink-1-${crypto.randomUUID()}`,
174 `sink-2-${crypto.randomUUID()}`,
175 `sink-3-${crypto.randomUUID()}`,
176 ],
177 });
178 expect((response as any).ok).toBe(true);
179 expect((response as any).count).toBe(3);
180
181 // Ingest success is the assertion — S3 delivery is asynchronous
182 // (buffering ≥ 60s; see the gated slow test below for arrival proof).
183 const described = yield* Firehose.describeDeliveryStream({
184 DeliveryStreamName: deliveryStreamName,
185 });
186 expect(
187 described.DeliveryStreamDescription.DeliveryStreamStatus,
188 ).toEqual("ACTIVE");
189 }),
190 );
191
192 test(
193 "splits more than 500 records into multiple PutRecordBatch calls",
194 Effect.gen(function* () {
195 const { url } = yield* stack;
196 // 501 records > the PutRecordBatch limit of 500, so the batched sink
197 // must split the chunk into 2 sequential API calls (500 + 1). Any
198 // per-record ServiceUnavailable failures are retried by the sink
199 // engine before the handler responds.
200 const marker = crypto.randomUUID();
201 const response = yield* postJson(url, "/sink", {
202 records: Array.from(
203 { length: 501 },
204 (_, i) => `sink-${marker}-${i}`,
205 ),
206 });
207 expect((response as any).ok).toBe(true);
208 expect((response as any).count).toBe(501);
209 }),
210 { timeout: 120_000 },
211 );
212 });
213
214 // Even with zero buffering, Firehose treats the interval and size as hints
215 // and has exceeded the default suite's 90-second platform budget in live
216 // runs. Keep the end-to-end proof opt-in, but run its independent checks
217 // concurrently so they share one bounded delivery window.
218 describe.concurrent.skipIf(!process.env.AWS_TEST_SLOW)(
219 "S3 delivery (slow)",
220 () => {
221 class MarkerNotDeliveredYet extends Data.TaggedError(
222 "MarkerNotDeliveredYet",
223 ) {}
224
225 const waitForMarker = (bucketName: string, marker: string) => {
226 const findMarker = Effect.gen(function* () {
227 const listing = yield* S3.listObjectsV2({
228 Bucket: bucketName,
229 Prefix: "records/",
230 });
231 for (const object of listing.Contents ?? []) {
232 if (object.Key === undefined) {
233 continue;
234 }
235 const got = yield* S3.getObject({
236 Bucket: bucketName,
237 Key: object.Key,
238 });
239 // The body read surfaces plain `Error` (streaming transport) — a
240 // mid-delivery read hiccup is just "not delivered yet".
241 const text = yield* Stream.mkString(
242 Stream.decodeText(got.Body!),
243 ).pipe(
244 Effect.catch(() => Effect.fail(new MarkerNotDeliveredYet())),
245 );
246 if (text.includes(marker)) {
247 return object.Key;
248 }
249 }
250 return yield* new MarkerNotDeliveredYet();
251 });
252
253 return findMarker.pipe(
254 Effect.retry({
255 while: (e) => e._tag === "MarkerNotDeliveredYet",
256 schedule: Schedule.max([
257 Schedule.fixed("10 seconds"),
258 Schedule.recurs(12),
259 ]),
260 }),
261 );
262 };
263
264 test.provider(
265 "delivers buffered records to the destination bucket",
266 () =>
267 Effect.gen(function* () {
268 const { url, bucketName } = yield* stack;
269 // Randomness is only a payload correlation marker, never a
270 // physical resource name. It prevents a prior run's object from
271 // satisfying this delivery proof.
272 const marker = `put-record-delivery-${crypto.randomUUID()}`;
273 yield* postJson(url, "/put-record", {
274 data: marker,
275 });
276 expect(yield* waitForMarker(bucketName, marker)).toBeTruthy();
277 }),
278 { timeout: 180_000 },
279 );
280
281 test.provider(
282 "sink records land in the destination bucket (marker-anchored)",
283 () =>
284 Effect.gen(function* () {
285 const { url, bucketName } = yield* stack;
286 const marker = `sink-delivery-${crypto.randomUUID()}`;
287 yield* postJson(url, "/sink", {
288 records: [`${marker}-1`, `${marker}-2`],
289 });
290 expect(yield* waitForMarker(bucketName, marker)).toBeTruthy();
291 }),
292 { timeout: 180_000 },
293 );
294 },
295 );
296 },
297);