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 SinkTable extends Context.Service<
15 SinkTable,
16 { table: AWS.DynamoDB.Table }
17>()("SinkTable") {}
18
19export const SinkTableLive = Layer.effect(
20 SinkTable,
21 Effect.gen(function* () {
22 const table = yield* AWS.DynamoDB.Table("TableSinkTable", {
23 partitionKey: "pk",
24 sortKey: "sk",
25 attributes: { pk: "S", sk: "S" },
26 });
27 return { table };
28 }),
29);
30
31export class TableSinkFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
32 "TableSinkFunction",
33) {}
34
35export const TableSinkFunctionLive = TableSinkFunction.make(
36 {
37 main,
38 functionUrl: true,
39 // The sink's bounded partial-failure retry can sleep up to ~6s, which
40 // exceeds Lambda's 3s default timeout (see PATTERNS §7).
41 timeout: Duration.seconds(30),
42 },
43 Effect.gen(function* () {
44 const { table } = yield* SinkTable;
45 const sink = yield* AWS.DynamoDB.TableSink(table);
46 const tableName = yield* table.tableName;
47
48 return {
49 fetch: Effect.gen(function* () {
50 const request = yield* HttpServerRequest;
51 const pathname = new URL(request.originalUrl).pathname;
52
53 if (request.method === "GET" && pathname === "/ready") {
54 return yield* HttpServerResponse.json({
55 ok: true,
56 tableName: yield* tableName,
57 });
58 }
59
60 if (request.method === "POST" && pathname === "/sink") {
61 const body = (yield* request.json) as {
62 pk: string;
63 puts?: string[];
64 deletes?: string[];
65 };
66
67 const entries: AWS.DynamoDB.TableSinkEntry[] = [
68 ...(body.puts ?? []).map((sk): AWS.DynamoDB.TableSinkEntry => ({
69 PutRequest: {
70 Item: {
71 pk: { S: body.pk },
72 sk: { S: sk },
73 data: { S: `payload-${sk}` },
74 },
75 },
76 })),
77 ...(body.deletes ?? []).map((sk): AWS.DynamoDB.TableSinkEntry => ({
78 DeleteRequest: {
79 Key: {
80 pk: { S: body.pk },
81 sk: { S: sk },
82 },
83 },
84 })),
85 ];
86
87 // The sink is request-scoped: drain fully before responding.
88 yield* Stream.fromIterable(entries).pipe(Stream.run(sink));
89
90 return yield* HttpServerResponse.json({
91 ok: true,
92 count: entries.length,
93 });
94 }
95
96 return yield* HttpServerResponse.json(
97 { error: "Not found", method: request.method, pathname },
98 { status: 404 },
99 );
100 }).pipe(
101 Effect.tapError(Console.log),
102 Effect.catch(() =>
103 Effect.succeed(
104 HttpServerResponse.text("Internal server error", { status: 500 }),
105 ),
106 ),
107 ),
108 };
109 }).pipe(
110 Effect.provide(
111 Layer.provideMerge(
112 Layer.mergeAll(SinkTableLive, AWS.DynamoDB.TableSinkHttp),
113 Layer.mergeAll(AWS.DynamoDB.BatchWriteItemHttp),
114 ),
115 ),
116 ),
117);
118export default TableSinkFunctionLive;