1import * as AWS from "@/AWS";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Stream from "effect/Stream";
6
7export class DynamoDBStreamFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
8 "DynamoDBStreamFunction",
9) {}
10
11export class TableAndQueue extends Context.Service<
12 TableAndQueue,
13 {
14 table: AWS.DynamoDB.Table;
15 queue: AWS.SQS.Queue;
16 }
17>()("TableAndQueue") {}
18
19export const TableAndQueueLive = Layer.effect(
20 TableAndQueue,
21 Effect.gen(function* () {
22 const table = yield* AWS.DynamoDB.Table("StreamSourceTable", {
23 partitionKey: "pk",
24 sortKey: "sk",
25 attributes: {
26 pk: "S",
27 sk: "S",
28 },
29 });
30 const queue = yield* AWS.SQS.Queue("StreamSinkQueue");
31 return {
32 table,
33 queue,
34 };
35 }),
36);
37
38export default DynamoDBStreamFunction.make(
39 {
40 main: import.meta.url,
41 },
42 Effect.gen(function* () {
43 const { table, queue } = yield* TableAndQueue;
44 const sink = yield* AWS.SQS.QueueSink(queue);
45
46 yield* AWS.DynamoDB.consumeTableChanges(
47 table,
48 {
49 streamViewType: "NEW_AND_OLD_IMAGES",
50 startingPosition: "TRIM_HORIZON",
51 batchSize: 10,
52 },
53 (stream) =>
54 stream.pipe(
55 Stream.map((record) => ({
56 MessageBody: JSON.stringify({
57 eventName: record.eventName,
58 keys: record.dynamodb.Keys,
59 newImage: record.dynamodb.NewImage,
60 oldImage: record.dynamodb.OldImage,
61 }),
62 })),
63 Stream.run(sink),
64 Effect.orDie,
65 ),
66 );
67 }).pipe(
68 Effect.provide(
69 Layer.provideMerge(
70 Layer.mergeAll(
71 AWS.Lambda.TableEventSource,
72 AWS.SQS.QueueSinkHttp,
73 TableAndQueueLive,
74 ),
75 Layer.mergeAll(AWS.SQS.SendMessageBatchHttp),
76 ),
77 ),
78 ),
79);