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