1import * as AWS from "@/AWS";
2import * as Console from "effect/Console";
3import * as Context from "effect/Context";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Stream from "effect/Stream";
8import { HttpServerRequest } from "effect/http/HttpServerRequest";
9import * as HttpServerResponse from "effect/http/HttpServerResponse";
10import path from "pathe";
11
12const main = path.resolve(import.meta.dirname, "sink-handler.ts");
13
14export class TestQueue extends Context.Service<
15 TestQueue,
16 { queue: AWS.SQS.Queue }
17>()("TestQueue") {}
18
19export const TestQueueLive = Layer.effect(
20 TestQueue,
21 Effect.gen(function* () {
22 const queue = yield* AWS.SQS.Queue("QueueSinkQueue");
23 return { queue };
24 }),
25);
26
27export class QueueSinkFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
28 "QueueSinkFunction",
29) {}
30
31export const QueueSinkFunctionLive = QueueSinkFunction.make(
32 {
33 main,
34 functionUrl: true,
35 // The sink's bounded partial-failure retry can sleep up to ~6s, which
36 // exceeds Lambda's 3s default timeout (see PATTERNS §7).
37 timeout: Duration.seconds(30),
38 },
39 Effect.gen(function* () {
40 const { queue } = yield* TestQueue;
41 const sink = yield* AWS.SQS.QueueSink(queue);
42 const queueUrl = yield* queue.queueUrl;
43
44 return {
45 fetch: Effect.gen(function* () {
46 const request = yield* HttpServerRequest;
47 yield* Console.log(request);
48 const pathname = new URL(request.originalUrl).pathname;
49
50 if (request.method === "GET" && pathname === "/ready") {
51 return yield* HttpServerResponse.json({
52 ok: true,
53 queueUrl: yield* queueUrl,
54 });
55 }
56
57 if (request.method === "POST" && pathname === "/sink") {
58 const body = (yield* request.json) as { messages: string[] };
59
60 yield* Stream.fromIterable(body.messages).pipe(
61 Stream.map((message) => ({ MessageBody: message })),
62 Stream.run(sink),
63 );
64
65 return yield* HttpServerResponse.json({
66 ok: true,
67 count: body.messages.length,
68 });
69 }
70
71 return yield* HttpServerResponse.json(
72 { error: "Not found", method: request.method, pathname },
73 { status: 404 },
74 );
75 }).pipe(
76 Effect.tapError(Console.log),
77 Effect.tap(Console.log),
78 Effect.catch(() =>
79 Effect.succeed(
80 HttpServerResponse.text("Internal server error", { status: 500 }),
81 ),
82 ),
83 ),
84 };
85 }).pipe(
86 Effect.provide(
87 Layer.provideMerge(
88 Layer.mergeAll(TestQueueLive, AWS.SQS.QueueSinkHttp),
89 Layer.mergeAll(AWS.SQS.SendMessageBatchHttp),
90 ),
91 ),
92 ),
93);
94export default QueueSinkFunctionLive;