1import * as DynamoDB from "alchemy/AWS/DynamoDB";
2import * as Lambda from "alchemy/AWS/Lambda";
3import * as S3 from "alchemy/AWS/S3";
4import * as SQS from "alchemy/AWS/SQS";
5import * as RemovalPolicy from "alchemy/RemovalPolicy";
6import { Stack } from "alchemy/Stack";
7import * as Console from "effect/Console";
8import * as Context from "effect/Context";
9import * as Data from "effect/Data";
10import * as Effect from "effect/Effect";
11import * as Layer from "effect/Layer";
12import * as Stream from "effect/Stream";
13
14import type { Job } from "./Job.ts";
15
16export class PutJobError extends Data.TaggedError("PutJobError")<{
17 readonly message: string;
18 readonly cause?: unknown;
19}> {}
20
21export class GetJobError extends Data.TaggedError("GetJobError")<{
22 readonly message: string;
23 readonly cause?: unknown;
24}> {}
25
26export class JobStorage extends Context.Service<
27 JobStorage,
28 {
29 putJob(job: Job): Effect.Effect<Job, PutJobError>;
30 getJob(jobId: string): Effect.Effect<Job | undefined, GetJobError>;
31 }
32>()("JobStorage") {}
33
34export const JobStorageDynamoDB = Layer.provideMerge(
35 Layer.effect(
36 JobStorage,
37 Effect.gen(function* () {
38 const stack = yield* Stack;
39 const table = yield* DynamoDB.Table("JobsTable", {
40 partitionKey: "id",
41 attributes: {
42 id: "S",
43 },
44 });
45 const queue = yield* SQS.Queue("JobsQueue").pipe(
46 RemovalPolicy.retain(stack.stage === "prod"),
47 );
48
49 const getItem = yield* DynamoDB.GetItem(table);
50 const putItem = yield* DynamoDB.PutItem(table);
51 const sink = yield* SQS.QueueSink(queue);
52
53 yield* DynamoDB.consumeTableChanges(table, {
54 streamViewType: "NEW_AND_OLD_IMAGES",
55 startingPosition: "LATEST",
56 batchSize: 10,
57 }, (stream) =>
58 stream.pipe(
59 Stream.map((record) => ({
60 MessageBody: JSON.stringify({
61 eventName: record.eventName,
62 keys: record.dynamodb.Keys,
63 newImage: record.dynamodb.NewImage,
64 oldImage: record.dynamodb.OldImage,
65 }),
66 })),
67 Stream.run(sink),
68 Effect.orDie,
69 ),
70 );
71
72 const putJob = (job: Job) =>
73 putItem({
74 Item: {
75 id: { S: job.id },
76 content: { S: job.content },
77 },
78 }).pipe(
79 Effect.map(() => job),
80 Effect.tapError(Console.log),
81 Effect.catchCause((cause) =>
82 Effect.fail(
83 new PutJobError({
84 message: `Failed to store job "${job.id}": ${cause}`,
85 cause,
86 }),
87 ),
88 ),
89 );
90
91 const getJob = (jobId: string) =>
92 getItem({
93 Key: {
94 id: { S: jobId },
95 },
96 }).pipe(
97 Effect.flatMap((item) =>
98 item.Item
99 ? Effect.try({
100 try: () =>
101 ({
102 id: item.Item?.id?.S ?? jobId,
103 content: item.Item?.content?.S ?? "",
104 }) as Job,
105 catch: (cause) =>
106 new GetJobError({
107 message: `Failed to parse job "${jobId}": ${cause}`,
108 cause,
109 }),
110 })
111 : Effect.succeed(undefined),
112 ),
113 Effect.tapError(Console.log),
114 Effect.catchCause((cause) =>
115 Effect.fail(
116 new GetJobError({
117 message: `Failed to load job "${jobId}": ${cause}`,
118 cause,
119 }),
120 ),
121 ),
122 );
123
124 return JobStorage.of({
125 putJob,
126 getJob,
127 });
128 }),
129 ),
130 Layer.mergeAll(Lambda.TableEventSource, SQS.QueueSinkHttp).pipe(
131 Layer.provideMerge(
132 Layer.mergeAll(
133 DynamoDB.GetItemHttp,
134 DynamoDB.PutItemHttp,
135 SQS.SendMessageBatchHttp,
136 ),
137 ),
138 ),
139);
140
141export const JobStorageS3 = Layer.provideMerge(
142 Layer.effect(
143 JobStorage,
144 Effect.gen(function* () {
145 const stack = yield* Stack;
146 const bucket = yield* S3.Bucket("JobsBucket");
147 const queue = yield* SQS.Queue("JobsQueue").pipe(
148 RemovalPolicy.retain(stack.stage === "prod"),
149 );
150
151 const getObject = yield* S3.GetObject(bucket);
152 const putObject = yield* S3.PutObject(bucket);
153 const sink = yield* SQS.QueueSink(queue);
154
155 const putJob = (job: Job) =>
156 putObject({
157 Key: job.id,
158 Body: JSON.stringify(job),
159 }).pipe(
160 Effect.map(() => job),
161 Effect.tapError(Console.log),
162 Effect.catchCause((cause) =>
163 Effect.fail(
164 new PutJobError({
165 message: `Failed to store job "${job.id}": ${cause}`,
166 cause,
167 }),
168 ),
169 ),
170 );
171
172 const getJob = (jobId: string) =>
173 getObject({
174 Key: jobId,
175 }).pipe(
176 Effect.catchTag("NoSuchKey", () => Effect.succeed(undefined)),
177 Effect.flatMap(
178 (item) =>
179 item?.Body?.pipe(
180 Stream.decodeText,
181 Stream.mkString,
182 Effect.flatMap((body) =>
183 Effect.try({
184 try: () => JSON.parse(body) as Job,
185 catch: (cause) =>
186 new GetJobError({
187 message: `Failed to parse job "${jobId}": ${cause}`,
188 cause,
189 }),
190 }),
191 ),
192 ) ?? Effect.succeed(undefined),
193 ),
194 Effect.tapError(Console.log),
195 Effect.catchCause((cause) =>
196 Effect.fail(
197 new GetJobError({
198 message: `Failed to load job "${jobId}": ${cause}`,
199 cause,
200 }),
201 ),
202 ),
203 );
204
205 yield* S3.consumeBucketEvents(bucket, (stream) =>
206 stream.pipe(
207 Stream.flatMap((item) =>
208 Stream.fromEffect(getJob(item.key).pipe(Effect.orDie)),
209 ),
210 Stream.filter((job): job is Job => job !== undefined),
211 Stream.map((job) => ({ MessageBody: JSON.stringify(job) })),
212 Stream.run(sink),
213 Effect.orDie,
214 ),
215 );
216
217 return JobStorage.of({
218 putJob,
219 getJob,
220 });
221 }),
222 ),
223 Layer.mergeAll(Lambda.BucketEventSource, SQS.QueueSinkHttp).pipe(
224 Layer.provideMerge(
225 Layer.mergeAll(
226 S3.GetObjectHttp,
227 S3.PutObjectHttp,
228 SQS.SendMessageBatchHttp,
229 ),
230 ),
231 ),
232);