1import * as AWS from "@/AWS";
2import * as Context from "effect/Context";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Stream from "effect/Stream";
7import { HttpServerRequest } from "effect/http/HttpServerRequest";
8import * as HttpServerResponse from "effect/http/HttpServerResponse";
9
10export class FirehoseApiFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
11 "FirehoseApiFunction",
12) {}
13
14export class BucketAndDeliveryStream extends Context.Service<
15 BucketAndDeliveryStream,
16 {
17 bucket: AWS.S3.Bucket;
18 deliveryStream: AWS.Firehose.DeliveryStream;
19 }
20>()("BucketAndDeliveryStream") {}
21
22export const BucketAndDeliveryStreamLive = Layer.effect(
23 BucketAndDeliveryStream,
24 Effect.gen(function* () {
25 const bucket = yield* AWS.S3.Bucket("FirehoseFixtureBucket", {
26 forceDestroy: true,
27 });
28
29 const deliveryStream = yield* AWS.Firehose.DeliveryStream(
30 "FixtureDeliveryStream",
31 {
32 destination: {
33 bucketArn: bucket.bucketArn,
34 prefix: "records/",
35 // Direct S3 supports zero buffering. Keep the binding fixture at
36 // zero so end-to-end delivery can be proven inside the normal test
37 // budget instead of imposing Firehose's one-minute flush delay.
38 bufferingInterval: "0 seconds",
39 bufferingSizeInMBs: 1,
40 },
41 tags: { fixture: "firehose-bindings" },
42 },
43 );
44
45 return { bucket, deliveryStream };
46 }),
47);
48
49export const FirehoseApiFunctionLive = FirehoseApiFunction.make(
50 {
51 main: import.meta.url,
52 functionUrl: true,
53 timeout: Duration.seconds(30),
54 },
55 Effect.gen(function* () {
56 const { deliveryStream } = yield* BucketAndDeliveryStream;
57
58 const putRecord = yield* AWS.Firehose.PutRecord(deliveryStream);
59 const putRecordBatch = yield* AWS.Firehose.PutRecordBatch(deliveryStream);
60 const sink = yield* AWS.Firehose.DeliveryStreamSink(deliveryStream);
61 const listDeliveryStreams = yield* AWS.Firehose.ListDeliveryStreams();
62
63 return {
64 fetch: Effect.gen(function* () {
65 const request = yield* HttpServerRequest;
66 const url = new URL(request.originalUrl);
67 const pathname = url.pathname;
68
69 if (request.method === "GET" && pathname === "/ready") {
70 return yield* HttpServerResponse.json({ ok: true });
71 }
72
73 if (request.method === "POST" && pathname === "/put-record") {
74 const body = (yield* request.json) as { data: string };
75 return yield* HttpServerResponse.json(
76 yield* putRecord({
77 Record: { Data: new TextEncoder().encode(`${body.data}\n`) },
78 }),
79 );
80 }
81
82 if (request.method === "POST" && pathname === "/put-record-batch") {
83 const body = (yield* request.json) as { records: string[] };
84 return yield* HttpServerResponse.json(
85 yield* putRecordBatch({
86 Records: body.records.map((data) => ({
87 Data: new TextEncoder().encode(`${data}\n`),
88 })),
89 }),
90 );
91 }
92
93 if (request.method === "GET" && pathname === "/list-streams") {
94 return yield* HttpServerResponse.json(
95 yield* listDeliveryStreams({ Limit: 100 }),
96 );
97 }
98
99 if (request.method === "POST" && pathname === "/sink") {
100 const body = (yield* request.json) as { records: string[] };
101 // Sinks are request-scoped: Stream.run drains fully (including the
102 // engine's bounded partial-failure retries) before the handler
103 // responds.
104 yield* Stream.fromIterable(body.records).pipe(
105 Stream.map((data) => ({
106 Data: new TextEncoder().encode(`${data}\n`),
107 })),
108 Stream.run(sink),
109 );
110 return yield* HttpServerResponse.json({
111 ok: true,
112 count: body.records.length,
113 });
114 }
115
116 return yield* HttpServerResponse.json(
117 { error: "Not found", method: request.method, pathname },
118 { status: 404 },
119 );
120 }).pipe(Effect.orDie),
121 };
122 }).pipe(
123 Effect.provide(
124 Layer.provideMerge(
125 // The sink layer consumes the PutRecordBatch binding, so the op
126 // layers are provided *into* the sink group (and merged out for the
127 // fetch routes that call them directly).
128 Layer.mergeAll(
129 AWS.Firehose.DeliveryStreamSinkHttp,
130 BucketAndDeliveryStreamLive,
131 ),
132 Layer.mergeAll(
133 AWS.Firehose.PutRecordHttp,
134 AWS.Firehose.PutRecordBatchHttp,
135 AWS.Firehose.ListDeliveryStreamsHttp,
136 ),
137 ),
138 ),
139 ),
140 // Re-merge so the deploying Stack can `yield* BucketAndDeliveryStream` and
141 // expose the stream/bucket names as deploy-time outputs. Reusing the same
142 // `BucketAndDeliveryStreamLive` reference keeps it a single shared pair.
143).pipe(Layer.provideMerge(BucketAndDeliveryStreamLive));
144
145export default FirehoseApiFunctionLive;