1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import * as DynamoDB from "@distilled.cloud/aws/dynamodb";
4import * as Lambda from "@distilled.cloud/aws/lambda";
5import * as SQS from "@distilled.cloud/aws/sqs";
6import { describe, expect } from "alchemy-test";
7import * as Data from "effect/Data";
8import * as Effect from "effect/Effect";
9import * as Schedule from "effect/Schedule";
10import * as Layer from "effect/Layer";
11import DynamoDBStreamFunctionLive, {
12 DynamoDBStreamFunction,
13 TableAndQueue,
14 TableAndQueueLive,
15} from "./stream-handler.ts";
16
17const { test } = Test.make({ providers: AWS.providers() });
18
19describe.skipIf(!!process.env.FAST).sequential(
20 "AWS.DynamoDB.Stream",
21 {
22 tags: [
23 "provider:aws",
24 "provider:aws:dynamodb",
25 "provider:aws:lambda",
26 "provider:aws:sqs",
27 "live",
28 ],
29 },
30 () => {
31 test.provider(
32 "processes real DynamoDB stream records through Lambda",
33 (stack) =>
34 Effect.gen(function* () {
35 yield* Effect.logInfo(
36 "DynamoDB Stream test: destroying previous resources",
37 );
38 yield* stack.destroy();
39
40 yield* Effect.logInfo(
41 "DynamoDB Stream test: deploying stream fixture",
42 );
43 const { table, queue, streamFunction } = yield* stack.deploy(
44 Effect.gen(function* () {
45 const { table, queue } = yield* TableAndQueue;
46
47 const func = yield* DynamoDBStreamFunction;
48
49 return { table, queue, streamFunction: func };
50 }).pipe(
51 Effect.provide(
52 Layer.mergeAll(DynamoDBStreamFunctionLive, TableAndQueueLive),
53 ),
54 ),
55 );
56
57 const streamState = yield* waitForTableStreamSpecification(
58 table.tableName,
59 {
60 StreamEnabled: true,
61 StreamViewType: "NEW_AND_OLD_IMAGES",
62 },
63 );
64 expect(streamState.Table?.StreamSpecification).toEqual({
65 StreamEnabled: true,
66 StreamViewType: "NEW_AND_OLD_IMAGES",
67 });
68 expect(streamState.Table?.LatestStreamArn).toBeDefined();
69
70 yield* waitForEventSourceMappingEnabled(
71 streamFunction.functionName,
72 streamState.Table?.LatestStreamArn!,
73 );
74
75 yield* Effect.logInfo(
76 `DynamoDB Stream test: writing item into ${table.tableName}`,
77 );
78 yield* DynamoDB.putItem({
79 TableName: table.tableName,
80 Item: {
81 pk: { S: "stream#1" },
82 sk: { S: "item#1" },
83 data: { S: "payload" },
84 },
85 });
86
87 const message = yield* waitForQueueMessage(queue.queueUrl);
88 const body = JSON.parse(message.Body!);
89
90 expect(body.eventName).toEqual("INSERT");
91 expect(body.keys.pk.S).toEqual("stream#1");
92 expect(body.keys.sk.S).toEqual("item#1");
93 expect(body.newImage.data.S).toEqual("payload");
94 expect(body.oldImage).toBeUndefined();
95
96 yield* Effect.logInfo("DynamoDB Stream test: destroying fixture");
97 yield* stack.destroy();
98 yield* assertTableIsDeleted(table.tableName);
99 }),
100 // Must stay UNDER the factory's 300s hard-kill wall: a vitest timeout
101 // interrupts the fiber and the `test.provider` ensuring-destroy reclaims
102 // every deployed resource, whereas an external SIGKILL leaks the whole
103 // fixture stack (Table + Queue + Function + Permission + ESM) with no
104 // way to reclaim it (scratch state is in-memory). Typical green run is
105 // ~105s, so 230s leaves interruption + destroy comfortably inside 300s.
106 { timeout: 230_000 },
107 );
108 },
109);
110
111// Out-of-band proof that the trailing destroy deleted the fixture table.
112const assertTableIsDeleted = Effect.fn(function* (tableName: string) {
113 yield* Effect.logInfo(
114 `DynamoDB Stream test: waiting for deletion of ${tableName}`,
115 );
116 yield* DynamoDB.describeTable({
117 TableName: tableName,
118 }).pipe(
119 Effect.flatMap(() => Effect.fail(new TableStillExists())),
120 Effect.retry({
121 while: (e) => e._tag === "TableStillExists",
122 schedule: Schedule.max([
123 Schedule.fixed("2 seconds"),
124 Schedule.recurs(30),
125 ]),
126 }),
127 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
128 );
129});
130
131const waitForEventSourceMappingEnabled = Effect.fn(function* (
132 functionName: string,
133 eventSourceArn: string,
134) {
135 yield* Effect.logInfo(
136 `DynamoDB Stream test: waiting for Lambda event source mapping on ${functionName}`,
137 );
138
139 return yield* Lambda.listEventSourceMappings({
140 FunctionName: functionName,
141 EventSourceArn: eventSourceArn,
142 }).pipe(
143 Effect.flatMap((result) => {
144 const mapping = result.EventSourceMappings?.[0];
145 if (!mapping || mapping.State !== "Enabled") {
146 return Effect.logInfo(
147 `DynamoDB Stream test: event source mapping not ready yet. state=${mapping?.State ?? "missing"}`,
148 ).pipe(Effect.andThen(Effect.fail(new EventSourceMappingNotReady())));
149 }
150 return Effect.logInfo(
151 `DynamoDB Stream test: event source mapping ready (${mapping.UUID})`,
152 ).pipe(Effect.andThen(Effect.succeed(mapping)));
153 }),
154 Effect.retry({
155 while: (error) => error._tag === "EventSourceMappingNotReady",
156 schedule: Schedule.max([
157 Schedule.fixed("2 seconds"),
158 Schedule.recurs(20),
159 ]),
160 }),
161 );
162});
163
164const waitForTableStreamSpecification = Effect.fn(function* (
165 tableName: string,
166 expected: DynamoDB.StreamSpecification,
167) {
168 yield* Effect.logInfo(
169 `DynamoDB Stream test: waiting for stream configuration on ${tableName}`,
170 );
171
172 return yield* DynamoDB.describeTable({
173 TableName: tableName,
174 }).pipe(
175 Effect.flatMap((result) => {
176 const actual = result.Table?.StreamSpecification;
177 if (JSON.stringify(actual) !== JSON.stringify(expected)) {
178 return Effect.logInfo(
179 `DynamoDB Stream test: stream configuration not ready yet. actual=${JSON.stringify(actual)} expected=${JSON.stringify(expected)}`,
180 ).pipe(
181 Effect.andThen(Effect.fail(new TableStreamConfigurationNotReady())),
182 );
183 }
184 return Effect.logInfo(
185 `DynamoDB Stream test: stream configuration ready on ${tableName}`,
186 ).pipe(Effect.andThen(Effect.succeed(result)));
187 }),
188 Effect.retry({
189 while: (error) => error._tag === "TableStreamConfigurationNotReady",
190 schedule: Schedule.max([
191 Schedule.fixed("2 seconds"),
192 Schedule.recurs(20),
193 ]),
194 }),
195 );
196});
197
198const waitForQueueMessage = Effect.fn(function* (queueUrl: string) {
199 yield* Effect.logInfo(
200 `DynamoDB Stream test: waiting for stream output message on ${queueUrl}`,
201 );
202
203 // Even after the EventSourceMapping reports `Enabled`, AWS needs a cold-
204 // start window (typically 30–90s for the first record) before the Lambda
205 // is reliably invoked from a freshly-provisioned DynamoDB Stream shard.
206 // Use SQS long-polling (WaitTimeSeconds=20) and budget ~3 minutes so this
207 // is robust on first-deploy runs without slowing down the happy path.
208 return yield* SQS.receiveMessage({
209 QueueUrl: queueUrl,
210 MaxNumberOfMessages: 1,
211 WaitTimeSeconds: 20,
212 }).pipe(
213 Effect.flatMap((result) => {
214 const message = result.Messages?.[0];
215 if (!message?.Body) {
216 return Effect.logInfo(
217 "DynamoDB Stream test: stream output queue is still empty",
218 ).pipe(Effect.andThen(Effect.fail(new StreamMessageNotReady())));
219 }
220 return Effect.logInfo(
221 `DynamoDB Stream test: received stream output message ${message.MessageId}`,
222 ).pipe(Effect.andThen(Effect.succeed(message)));
223 }),
224 Effect.retry({
225 while: (error) => error._tag === "StreamMessageNotReady",
226 schedule: Schedule.max([
227 Schedule.fixed("5 seconds"),
228 Schedule.recurs(72),
229 ]),
230 }),
231 );
232});
233
234class TableStillExists extends Data.TaggedError("TableStillExists") {}
235
236class EventSourceMappingNotReady extends Data.TaggedError(
237 "EventSourceMappingNotReady",
238) {}
239
240class TableStreamConfigurationNotReady extends Data.TaggedError(
241 "TableStreamConfigurationNotReady",
242) {}
243
244class StreamMessageNotReady extends Data.TaggedError("StreamMessageNotReady") {}