1import * as AWS from "@/AWS";
2import { S3BucketEventSource } from "@/Server/S3BucketEventSource.ts";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Stream from "effect/Stream";
6
7export const INCOMING_PREFIX = "incoming/";
8export const INCOMING_SUFFIX = ".txt";
9export const PROCESSED_PREFIX = "processed/";
10export const RECEIVED_PREFIX = "deliveries/received/";
11export const ACKNOWLEDGED_PREFIX = "deliveries/acknowledged/";
12
13export const ServerEventBucket = AWS.S3.Bucket("ServerEventBucket", {
14 versioning: "Enabled",
15 forceDestroy: true,
16});
17
18export const artifactKey = (
19 key: string,
20 eventName: string,
21 versionId: string,
22) =>
23 `${PROCESSED_PREFIX}${[key, eventName, versionId].map(encodeURIComponent).join("/")}.txt`;
24
25// Preserve the wire body for replay without competing with the ECS consumer.
26const observedReceiveMessage = Layer.effect(
27 AWS.SQS.ReceiveMessage,
28 Effect.gen(function* () {
29 const receiveMessage = yield* AWS.SQS.ReceiveMessage;
30 const bucket = yield* ServerEventBucket;
31 const putObject = yield* AWS.S3.PutObject(bucket);
32 return Effect.fn(function* (queue: AWS.SQS.Queue) {
33 const receive = yield* receiveMessage(queue);
34 return Effect.fn(function* (request: AWS.SQS.ReceiveMessageRequest) {
35 const result = yield* receive(request);
36 yield* Effect.forEach(result.Messages ?? [], (message) =>
37 putObject({
38 Key: `${RECEIVED_PREFIX}${message.MessageId!}.json`,
39 Body: message.Body!,
40 ContentType: "application/json",
41 }).pipe(Effect.orDie),
42 );
43 return result;
44 });
45 });
46 }),
47).pipe(Layer.provide(AWS.SQS.ReceiveMessageHttp));
48
49// The server acknowledges a batch only after the S3 handler completes.
50const observedDeleteMessageBatch = Layer.effect(
51 AWS.SQS.DeleteMessageBatch,
52 Effect.gen(function* () {
53 const deleteMessageBatch = yield* AWS.SQS.DeleteMessageBatch;
54 const bucket = yield* ServerEventBucket;
55 const putObject = yield* AWS.S3.PutObject(bucket);
56 return Effect.fn(function* (queue: AWS.SQS.Queue) {
57 const remove = yield* deleteMessageBatch(queue);
58 return Effect.fn(function* (request: AWS.SQS.DeleteMessageBatchRequest) {
59 const result = yield* remove(request);
60 yield* Effect.forEach(result.Successful ?? [], (entry) =>
61 putObject({
62 Key: `${ACKNOWLEDGED_PREFIX}${entry.Id}.json`,
63 Body: "{}",
64 ContentType: "application/json",
65 }).pipe(Effect.orDie),
66 );
67 return result;
68 });
69 });
70 }),
71).pipe(Layer.provide(AWS.SQS.DeleteMessageBatchHttp));
72
73const serverEvents = S3BucketEventSource.pipe(
74 Layer.provide(
75 Layer.mergeAll(observedReceiveMessage, observedDeleteMessageBatch).pipe(
76 Layer.provide(AWS.S3.PutObjectHttp),
77 ),
78 ),
79);
80
81export default class ServerEventTask extends AWS.ECS.Task<ServerEventTask>()(
82 "S3ServerEventTask",
83 {
84 main: import.meta.filename,
85 image: "oven/bun:1",
86 cpu: 256,
87 memory: 512,
88 container: {
89 linuxParameters: { initProcessEnabled: true },
90 stopTimeout: 2,
91 },
92 runtimePlatform: {
93 cpuArchitecture: "ARM64",
94 operatingSystemFamily: "LINUX",
95 },
96 },
97 Effect.gen(function* () {
98 const bucket = yield* ServerEventBucket;
99 const getObject = yield* AWS.S3.GetObject(bucket);
100 const putObject = yield* AWS.S3.PutObject(bucket);
101
102 yield* AWS.S3.consumeBucketEvents(
103 bucket,
104 {
105 events: ["s3:ObjectCreated:*", "s3:ObjectRemoved:*"],
106 prefix: INCOMING_PREFIX,
107 suffix: INCOMING_SUFFIX,
108 },
109 (events) =>
110 events.pipe(
111 Stream.runForEach((event) =>
112 Effect.gen(function* () {
113 const versionId = event.versionId;
114 if (!versionId) {
115 return yield* Effect.fail(
116 new Error("Versioned S3 event omitted versionId"),
117 );
118 }
119 const key = yield* Effect.sync(() =>
120 artifactKey(event.key, event.type, versionId),
121 );
122 // Redelivery can arrive after the source version was deleted.
123 const recorded = yield* getObject({ Key: key }).pipe(
124 Effect.flatMap(({ Body }) => Stream.runDrain(Body!)),
125 Effect.as(true),
126 Effect.catchTag("NoSuchKey", () => Effect.succeed(false)),
127 );
128 if (recorded) return;
129 const object = event.type.startsWith("s3:ObjectCreated:")
130 ? yield* getObject({ Key: event.key, VersionId: versionId })
131 : undefined;
132 const content = object
133 ? yield* Stream.mkString(Stream.decodeText(object.Body!))
134 : undefined;
135 const body = yield* Effect.sync(() =>
136 JSON.stringify({
137 bucket: event.bucket,
138 key: event.key,
139 eventName: event.type,
140 versionId,
141 sequencer: event.sequencer,
142 size: event.size,
143 eTag: event.eTag,
144 content,
145 readVersionId: object?.VersionId,
146 }),
147 );
148 // The suffix still matches; only the prefix prevents recursion.
149 yield* putObject({
150 Key: key,
151 Body: body,
152 ContentType: "application/json",
153 });
154 }).pipe(Effect.orDie),
155 ),
156 ),
157 );
158
159 return { run: Effect.never };
160 }).pipe(
161 Effect.provide(
162 Layer.mergeAll(serverEvents, AWS.S3.GetObjectHttp, AWS.S3.PutObjectHttp),
163 ),
164 ),
165) {}