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