1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import * as DynamoDB from "@distilled.cloud/aws/dynamodb";
4import { expect } from "alchemy-test";
5import * as Data from "effect/Data";
6import * as Effect from "effect/Effect";
7import * as Schedule from "effect/Schedule";
8import * as HttpBody from "effect/http/HttpBody";
9import * as HttpClient from "effect/http/HttpClient";
10import { TableSinkFunction, TableSinkFunctionLive } from "./sink-handler";
11
12const { test } = Test.make({ providers: AWS.providers() });
13
14class FunctionNotReady extends Data.TaggedError("FunctionNotReady") {}
15
16class TableStillExists extends Data.TaggedError("TableStillExists") {}
17
18// Out-of-band proof that the trailing destroy deleted the fixture table.
19const assertTableIsDeleted = Effect.fn(function* (tableName: string) {
20 yield* DynamoDB.describeTable({
21 TableName: tableName,
22 }).pipe(
23 Effect.flatMap(() => Effect.fail(new TableStillExists())),
24 Effect.retry({
25 while: (e) => e._tag === "TableStillExists",
26 schedule: Schedule.max([
27 Schedule.fixed("2 seconds"),
28 Schedule.recurs(30),
29 ]),
30 }),
31 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
32 );
33});
34
35class ItemCountMismatch extends Data.TaggedError("ItemCountMismatch")<{
36 readonly expected: number;
37 readonly actual: number;
38}> {}
39
40const waitForFunctionReady = (url: string) =>
41 HttpClient.get(url).pipe(
42 Effect.flatMap((response) =>
43 response.status === 200
44 ? (response.json as Effect.Effect<{ tableName: string }>)
45 : Effect.fail(new FunctionNotReady()),
46 ),
47 // A freshly-deployed function can briefly serve a 200 before its captured
48 // env vars (the table name) have finished propagating, so treat a missing
49 // tableName as "not ready yet" and keep polling.
50 Effect.flatMap((json: any) =>
51 typeof json?.tableName === "string"
52 ? Effect.succeed({ tableName: json.tableName as string })
53 : Effect.fail(new FunctionNotReady()),
54 ),
55 Effect.retry({
56 while: (error) => error._tag === "FunctionNotReady",
57 schedule: Schedule.max([
58 Schedule.fixed("2 seconds"),
59 Schedule.recurs(75),
60 ]),
61 }),
62 );
63
64/**
65 * Out-of-band read: strongly-consistent query for every sk under `pk`.
66 * Retries briefly in case a cold re-init served the POST a moment before
67 * the last batch call settled.
68 */
69const waitForSortKeys = Effect.fn(function* (
70 tableName: string,
71 pk: string,
72 expected: number,
73) {
74 return yield* Effect.gen(function* () {
75 const result = yield* DynamoDB.query({
76 TableName: tableName,
77 KeyConditionExpression: "pk = :pk",
78 ExpressionAttributeValues: { ":pk": { S: pk } },
79 ConsistentRead: true,
80 });
81 const sortKeys = (result.Items ?? []).flatMap((item) =>
82 item.sk?.S !== undefined ? [item.sk.S] : [],
83 );
84 if (sortKeys.length !== expected) {
85 return yield* Effect.fail(
86 new ItemCountMismatch({ expected, actual: sortKeys.length }),
87 );
88 }
89 return sortKeys;
90 }).pipe(
91 Effect.retry({
92 while: (e) => e._tag === "ItemCountMismatch",
93 schedule: Schedule.max([
94 Schedule.fixed("2 seconds"),
95 Schedule.recurs(15),
96 ]),
97 }),
98 );
99});
100
101test.provider(
102 "TableSink streams put and delete requests through a deployed Lambda",
103 (stack) =>
104 Effect.gen(function* () {
105 yield* stack.destroy();
106
107 const apiFunction = yield* stack.deploy(
108 TableSinkFunction.pipe(Effect.provide(TableSinkFunctionLive)),
109 );
110 const baseUrl = apiFunction.functionUrl!.replace(/\/+$/, "");
111
112 const { tableName } = yield* waitForFunctionReady(`${baseUrl}/ready`);
113
114 const postSink = (body: {
115 pk: string;
116 puts?: string[];
117 deletes?: string[];
118 }) =>
119 Effect.gen(function* () {
120 const response = yield* HttpClient.post(`${baseUrl}/sink`, {
121 body: yield* HttpBody.json(body),
122 });
123 if (response.status !== 200) {
124 return yield* Effect.fail(new FunctionNotReady());
125 }
126 return (yield* response.json) as { ok: boolean; count: number };
127 }).pipe(
128 Effect.retry({
129 while: (error) => error._tag === "FunctionNotReady",
130 schedule: Schedule.max([
131 Schedule.fixed("2 seconds"),
132 Schedule.recurs(30),
133 ]),
134 }),
135 );
136
137 const pk = `sink#${crypto.randomUUID()}`;
138 // 60 put requests > the BatchWriteItem limit of 25, so the batched sink
139 // must split the chunk into 3 sequential API calls (25 + 25 + 10).
140 const items = Array.from(
141 { length: 60 },
142 (_, i) => `item-${String(i).padStart(3, "0")}`,
143 );
144
145 const putResponse = yield* postSink({ pk, puts: items });
146 expect(putResponse.ok).toBe(true);
147 expect(putResponse.count).toBe(items.length);
148
149 // Out-of-band read via distilled proves every entry landed.
150 const landed = yield* waitForSortKeys(tableName, pk, items.length);
151 expect([...landed].sort()).toEqual(items);
152
153 // Stream DeleteRequests through the same sink for the first half.
154 const toDelete = items.slice(0, 30);
155 const deleteResponse = yield* postSink({ pk, deletes: toDelete });
156 expect(deleteResponse.ok).toBe(true);
157 expect(deleteResponse.count).toBe(toDelete.length);
158
159 const remaining = yield* waitForSortKeys(
160 tableName,
161 pk,
162 items.length - toDelete.length,
163 );
164 expect([...remaining].sort()).toEqual(items.slice(30));
165
166 yield* stack.destroy();
167 yield* assertTableIsDeleted(tableName);
168 }),
169 {
170 tags: [
171 "provider:aws",
172 "provider:aws:dynamodb",
173 "provider:aws:lambda",
174 "live",
175 ],
176 timeout: 240_000,
177 },
178);