Open History to see the last commit
1import * as AWS from "@alchemy.run/aws";
2import * as Cloudflare from "@alchemy.run/cloudflare";
3import { describe, expect } from "@alchemy.run/test";
4import * as Test from "@alchemy.run/test/Alchemy";
5import * as Core from "@alchemy.run/test/Core";
6import * as DynamoDBApi from "@distilled.cloud/aws/dynamodb";
7import * as Data from "effect/Data";
8import * as Effect from "effect/Effect";
9import * as HttpClient from "effect/http/HttpClient";
10import * as HttpClientRequest from "effect/http/HttpClientRequest";
11import * as Layer from "effect/Layer";
12import * as Schedule from "effect/Schedule";
13import LinksWorker, { LinksTable } from "./fixtures/links-worker.ts";
14import DynamoDBTestFunctionLive, { DynamoDBTestFunction } from "./handler";
15
16const testOptions = { providers: AWS.providers() };
17const { test, beforeAll, afterAll } = Test.make(testOptions);
18const sharedStack = Core.scratchStack(testOptions, "DynamoDBBindings");
19
20// Lambda function URL cold-start (DNS, IAM propagation, init) can take
21// well over 60s on a fresh deploy under parallel-suite load — observed
22// up to ~90s when S3 throttling delays the Lambda code upload too.
23// Budget ~150s of readiness polling so we don't fail the whole suite on
24// a slow init.
25const readinessPolicy = Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(75)]);
26
27let baseUrl: string;
28const sourceTableId = "TestTable";
29
30class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
31 readonly status: number;
32 readonly body: string;
33}> {}
34
35// The shared Lambda fixture occasionally answers a transient 5xx under
36// full-suite parallel load — a cold re-init, or a throttled/eventually-
37// consistent DynamoDB call that the handler's `Effect.orDie` surfaces as a 500
38// with a non-JSON ("Internal Server Error") body. Those are not assertion
39// failures: retry the request a few times before surfacing it. A genuine
40// 4xx/assertion failure is returned immediately (only 5xx is retried).
41const send = (request: HttpClientRequest.HttpClientRequest) =>
42 HttpClient.execute(request).pipe(
43 Effect.flatMap((response) =>
44 response.status >= 500
45 ? response.text.pipe(
46 Effect.flatMap((body) =>
47 Effect.fail(new TransientUpstream({ status: response.status, body })),
48 ),
49 )
50 : Effect.succeed(response),
51 ),
52 Effect.retry({
53 while: (e) => e._tag === "TransientUpstream",
54 schedule: Schedule.max([Schedule.exponential("500 millis"), Schedule.recurs(6)]),
55 }),
56 );
57
58describe(
59 "DynamoDB Bindings",
60 {
61 tags: [
62 "provider:aws",
63 "provider:aws:batch",
64 "provider:aws:dynamodb",
65 "provider:aws:lambda",
66 "provider:aws:s3",
67 "live",
68 ],
69 },
70 () => {
71 beforeAll(
72 Effect.gen(function* () {
73 yield* Effect.logInfo("DynamoDB test setup: destroying previous resources");
74 yield* sharedStack.destroy();
75
76 yield* Effect.logInfo("DynamoDB test setup: deploying fixture");
77 const { functionUrl } = yield* sharedStack.deploy(
78 Effect.gen(function* () {
79 return yield* DynamoDBTestFunction;
80 }).pipe(Effect.provide(DynamoDBTestFunctionLive)),
81 );
82
83 expect(functionUrl).toBeTruthy();
84 baseUrl = functionUrl!.replace(/\/+$/, "");
85 const readinessUrl = `${baseUrl}/scan`;
86
87 yield* Effect.logInfo(`DynamoDB test setup: function URL ready (${functionUrl})`);
88 yield* Effect.logInfo(
89 `DynamoDB test setup: probing readiness at ${readinessUrl} (20s budget)`,
90 );
91
92 yield* HttpClient.get(readinessUrl).pipe(
93 Effect.flatMap((response) =>
94 response.status === 200
95 ? Effect.succeed(response)
96 : Effect.fail(new Error(`Function not ready: ${response.status}`)),
97 ),
98 Effect.tap(() => Effect.logInfo("DynamoDB test setup: fixture responded successfully")),
99 Effect.tapError((error) =>
100 Effect.logWarning(`DynamoDB test setup: fixture not ready yet (${String(error)})`),
101 ),
102 Effect.retry({ schedule: readinessPolicy }),
103 );
104 }),
105 { timeout: 240_000 },
106 );
107
108 // Lambda cleanup watches for late log flushes before the dependent tables
109 // can be deleted. Allow time for both phases to confirm deletion.
110 afterAll(sharedStack.destroy(), { timeout: 120_000 });
111
112 describe("PutItem", () => {
113 test.provider("puts an item into the table", (_stack) =>
114 Effect.gen(function* () {
115 const response = yield* send(
116 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/put`), {
117 pk: "put-test#1",
118 sk: "item",
119 data: "test data",
120 }),
121 ).pipe(Effect.flatMap((r) => r.json));
122
123 expect(response).toHaveProperty("success", true);
124 }),
125 );
126 });
127
128 describe("GetItem", () => {
129 test.provider("gets an existing item from the table", (_stack) =>
130 Effect.gen(function* () {
131 yield* send(
132 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/put`), {
133 pk: "get-test#1",
134 sk: "item",
135 data: "get test data",
136 }),
137 );
138
139 const response = yield* HttpClient.get(
140 `${baseUrl}/get?pk=${encodeURIComponent("get-test#1")}&sk=item`,
141 ).pipe(Effect.flatMap((r) => r.json));
142
143 expect(response).toHaveProperty("item");
144 expect((response as any).item.pk.S).toBe("get-test#1");
145 expect((response as any).item.sk.S).toBe("item");
146 expect((response as any).item.data.S).toBe("get test data");
147 }),
148 );
149
150 test.provider("returns undefined for non-existent item", (_stack) =>
151 Effect.gen(function* () {
152 const response = yield* HttpClient.get(
153 `${baseUrl}/get?pk=${encodeURIComponent("non-existent")}&sk=item`,
154 ).pipe(Effect.flatMap((r) => r.json));
155
156 expect((response as any).item).toBeUndefined();
157 }),
158 );
159 });
160
161 describe("DescribeTable", () => {
162 test.provider("describes the bound table", (_stack) =>
163 Effect.gen(function* () {
164 const response = yield* HttpClient.get(`${baseUrl}/describe-table`).pipe(
165 Effect.flatMap((r) => r.json),
166 );
167
168 expect((response as any).table.TableName).toBeTruthy();
169 expect((response as any).table.KeySchema).toEqual([
170 { AttributeName: "pk", KeyType: "HASH" },
171 { AttributeName: "sk", KeyType: "RANGE" },
172 ]);
173 }),
174 );
175 });
176
177 describe("DescribeTimeToLive", () => {
178 test.provider("describes table ttl configuration", (_stack) =>
179 Effect.gen(function* () {
180 const response = yield* HttpClient.get(`${baseUrl}/describe-ttl`).pipe(
181 Effect.flatMap((r) => r.json),
182 );
183
184 expect(response).toHaveProperty("timeToLiveDescription");
185 }),
186 );
187 });
188
189 describe("BatchWriteItem", () => {
190 test.provider("writes multiple items through the bound table", (_stack) =>
191 Effect.gen(function* () {
192 const response = yield* send(
193 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/batch-write`), {
194 RequestItems: {
195 [sourceTableId]: [
196 {
197 PutRequest: {
198 Item: {
199 pk: { S: "batch-write#1" },
200 sk: { S: "item" },
201 data: { S: "first item" },
202 },
203 },
204 },
205 {
206 PutRequest: {
207 Item: {
208 pk: { S: "batch-write#2" },
209 sk: { S: "item" },
210 data: { S: "second item" },
211 },
212 },
213 },
214 ],
215 },
216 }),
217 ).pipe(Effect.flatMap((r) => r.json));
218
219 expect(Object.keys((response as any).unprocessedItems)).toHaveLength(0);
220 }),
221 );
222 });
223
224 describe("BatchGetItem", () => {
225 test.provider("reads multiple items through the bound table", (_stack) =>
226 Effect.gen(function* () {
227 yield* send(
228 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/batch-write`), {
229 RequestItems: {
230 [sourceTableId]: [
231 {
232 PutRequest: {
233 Item: {
234 pk: { S: "batch-get#1" },
235 sk: { S: "item" },
236 data: { S: "first item" },
237 },
238 },
239 },
240 {
241 PutRequest: {
242 Item: {
243 pk: { S: "batch-get#2" },
244 sk: { S: "item" },
245 data: { S: "second item" },
246 },
247 },
248 },
249 ],
250 },
251 }),
252 );
253
254 const response = yield* send(
255 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/batch-get`), {
256 RequestItems: {
257 [sourceTableId]: {
258 Keys: [
259 { pk: { S: "batch-get#1" }, sk: { S: "item" } },
260 { pk: { S: "batch-get#2" }, sk: { S: "item" } },
261 ],
262 },
263 },
264 }),
265 ).pipe(Effect.flatMap((r) => r.json));
266
267 const items = Object.values((response as any).responses).flat();
268 expect(items).toHaveLength(2);
269 }),
270 );
271 });
272
273 describe("UpdateTimeToLive", () => {
274 test.provider("updates table ttl configuration", (_stack) =>
275 Effect.gen(function* () {
276 const response = yield* send(
277 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/update-ttl`), {
278 attributeName: "expiresAt",
279 enabled: true,
280 }),
281 ).pipe(Effect.flatMap((r) => r.json));
282
283 expect((response as any).timeToLiveSpecification).toEqual({
284 AttributeName: "expiresAt",
285 Enabled: true,
286 });
287 }),
288 );
289 });
290
291 describe("ExecuteStatement", () => {
292 test.provider("executes a PartiQL statement against the bound table", (_stack) =>
293 Effect.gen(function* () {
294 yield* send(
295 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/put`), {
296 pk: "statement#1",
297 sk: "item",
298 data: "statement data",
299 }),
300 );
301
302 const response = yield* send(
303 HttpClientRequest.bodyJsonUnsafe(
304 HttpClientRequest.post(`${baseUrl}/execute-statement`),
305 { pk: "statement#1", sk: "item" },
306 ),
307 ).pipe(Effect.flatMap((r) => r.json));
308
309 expect((response as any).items).toHaveLength(1);
310 expect((response as any).items[0].data.S).toBe("statement data");
311 }),
312 );
313 });
314
315 describe("BatchExecuteStatement", () => {
316 test.provider("executes PartiQL statements against the bound table", (_stack) =>
317 Effect.gen(function* () {
318 yield* send(
319 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/batch-write`), {
320 RequestItems: {
321 [sourceTableId]: [
322 {
323 PutRequest: {
324 Item: {
325 pk: { S: "batch-statement#1" },
326 sk: { S: "item" },
327 data: { S: "first item" },
328 },
329 },
330 },
331 {
332 PutRequest: {
333 Item: {
334 pk: { S: "batch-statement#2" },
335 sk: { S: "item" },
336 data: { S: "second item" },
337 },
338 },
339 },
340 ],
341 },
342 }),
343 );
344
345 const response = yield* fetchUntil(
346 send(
347 HttpClientRequest.bodyJsonUnsafe(
348 HttpClientRequest.post(`${baseUrl}/batch-execute-statement`),
349 {
350 first: { pk: "batch-statement#1", sk: "item" },
351 second: { pk: "batch-statement#2", sk: "item" },
352 },
353 ),
354 ).pipe(Effect.flatMap((r) => r.json)),
355 (body) => Array.isArray(body?.responses) && body.responses.length === 2,
356 );
357
358 expect((response as any).responses).toHaveLength(2);
359 }),
360 );
361 });
362
363 describe("ExecuteTransaction", () => {
364 test.provider("executes a PartiQL transaction against the table", (_stack) =>
365 Effect.gen(function* () {
366 yield* send(
367 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/put`), {
368 pk: "tx#1",
369 sk: "item1",
370 data: "first",
371 }),
372 );
373 yield* send(
374 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/put`), {
375 pk: "tx#1",
376 sk: "item2",
377 data: "second",
378 }),
379 );
380
381 const response = yield* fetchUntil(
382 send(HttpClientRequest.post(`${baseUrl}/execute-transaction`)).pipe(
383 Effect.flatMap((r) => r.json),
384 ),
385 (body) => Array.isArray(body?.responses) && body.responses.length === 2,
386 );
387
388 expect((response as any).responses).toHaveLength(2);
389 }),
390 );
391 });
392
393 describe("TransactWriteItems", () => {
394 test.provider("writes items transactionally through the bound table", (_stack) =>
395 Effect.gen(function* () {
396 const response = yield* send(
397 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/transact-write`), {
398 TransactItems: [
399 {
400 Put: {
401 Table: sourceTableId,
402 Item: {
403 pk: { S: "transact-write#1" },
404 sk: { S: "item" },
405 data: { S: "first item" },
406 },
407 },
408 },
409 {
410 Put: {
411 Table: sourceTableId,
412 Item: {
413 pk: { S: "transact-write#2" },
414 sk: { S: "item" },
415 data: { S: "second item" },
416 },
417 },
418 },
419 ],
420 }),
421 ).pipe(Effect.flatMap((r) => r.json));
422
423 expect((response as any).success).toBe(true);
424 }),
425 );
426 });
427
428 describe("TransactGetItems", () => {
429 test.provider("reads items transactionally through the bound table", (_stack) =>
430 Effect.gen(function* () {
431 yield* send(
432 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/transact-write`), {
433 TransactItems: [
434 {
435 Put: {
436 Table: sourceTableId,
437 Item: {
438 pk: { S: "transact-get#1" },
439 sk: { S: "item" },
440 data: { S: "first item" },
441 },
442 },
443 },
444 {
445 Put: {
446 Table: sourceTableId,
447 Item: {
448 pk: { S: "transact-get#2" },
449 sk: { S: "item" },
450 data: { S: "second item" },
451 },
452 },
453 },
454 ],
455 }),
456 );
457
458 const response = yield* fetchUntil(
459 send(
460 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/transact-get`), {
461 TransactItems: [
462 {
463 Get: {
464 Table: sourceTableId,
465 Key: { pk: { S: "transact-get#1" }, sk: { S: "item" } },
466 },
467 },
468 {
469 Get: {
470 Table: sourceTableId,
471 Key: { pk: { S: "transact-get#2" }, sk: { S: "item" } },
472 },
473 },
474 ],
475 }),
476 ).pipe(Effect.flatMap((r) => r.json)),
477 (body) => Array.isArray(body?.responses) && body.responses.length === 2,
478 );
479
480 expect((response as any).responses).toHaveLength(2);
481 }),
482 );
483 });
484
485 describe("UpdateItem", () => {
486 test.provider("updates an existing item", (_stack) =>
487 Effect.gen(function* () {
488 yield* send(
489 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/put`), {
490 pk: "update-test#1",
491 sk: "item",
492 data: "original",
493 }),
494 );
495
496 const response = yield* send(
497 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/update`), {
498 pk: "update-test#1",
499 sk: "item",
500 data: "updated",
501 }),
502 ).pipe(Effect.flatMap((r) => r.json));
503
504 expect(response).toHaveProperty("success", true);
505 expect((response as any).attributes.data.S).toBe("updated");
506
507 const getResponse = yield* HttpClient.get(
508 `${baseUrl}/get?pk=${encodeURIComponent("update-test#1")}&sk=item`,
509 ).pipe(Effect.flatMap((r) => r.json));
510
511 expect((getResponse as any).item.data.S).toBe("updated");
512 }),
513 );
514 });
515
516 describe("DeleteItem", () => {
517 test.provider("deletes an existing item", (_stack) =>
518 Effect.gen(function* () {
519 yield* send(
520 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/put`), {
521 pk: "delete-test#1",
522 sk: "item",
523 data: "to delete",
524 }),
525 );
526
527 const response = yield* send(
528 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.delete(`${baseUrl}/delete`), {
529 pk: "delete-test#1",
530 sk: "item",
531 }),
532 ).pipe(Effect.flatMap((r) => r.json));
533
534 expect(response).toHaveProperty("success", true);
535
536 const getResponse = yield* HttpClient.get(
537 `${baseUrl}/get?pk=${encodeURIComponent("delete-test#1")}&sk=item`,
538 ).pipe(Effect.flatMap((r) => r.json));
539
540 expect((getResponse as any).item).toBeUndefined();
541 }),
542 );
543 });
544
545 describe("Query", () => {
546 test.provider("queries items by partition key", (_stack) =>
547 Effect.gen(function* () {
548 yield* send(
549 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/put`), {
550 pk: "query-test#1",
551 sk: "item1",
552 data: "first",
553 }),
554 );
555 yield* send(
556 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/put`), {
557 pk: "query-test#1",
558 sk: "item2",
559 data: "second",
560 }),
561 );
562
563 // DynamoDB Query is eventually consistent by default, so a
564 // brand-new PutItem may not appear in the immediately-following
565 // Query. Poll briefly until both items are visible.
566 const response = yield* Effect.gen(function* () {
567 const r = yield* HttpClient.get(
568 `${baseUrl}/query?pk=${encodeURIComponent("query-test#1")}`,
569 ).pipe(Effect.flatMap((r) => r.json));
570 if ((r as any).count !== 2) {
571 return yield* Effect.fail(new QueryNotConsistent());
572 }
573 return r;
574 }).pipe(
575 Effect.retry({
576 while: (e) => e._tag === "QueryNotConsistent",
577 schedule: Schedule.max([Schedule.fixed("500 millis"), Schedule.recurs(20)]),
578 }),
579 );
580
581 expect((response as any).count).toBe(2);
582 expect((response as any).items.length).toBe(2);
583 }),
584 );
585
586 test.provider("queries the multi-attribute-key GSI by composite partition key", (_stack) =>
587 Effect.gen(function* () {
588 // Items carry natural attributes — the GSI indexes them without
589 // synthetic concatenated keys. The third item differs only in
590 // `subcategory`, so the composite partition key must exclude it.
591 const items = [
592 {
593 pk: "multi-query#1",
594 sk: "item1",
595 category: "electronics",
596 subcategory: "audio",
597 rank: "featured",
598 },
599 {
600 pk: "multi-query#2",
601 sk: "item2",
602 category: "electronics",
603 subcategory: "audio",
604 rank: "standard",
605 },
606 {
607 pk: "multi-query#3",
608 sk: "item3",
609 category: "electronics",
610 subcategory: "video",
611 rank: "featured",
612 },
613 ];
614 yield* Effect.forEach(items, (item) =>
615 send(HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/put`), item)),
616 );
617
618 // Both HASH attributes must be equality conditions; poll through
619 // the GSI's asynchronous propagation.
620 const byPartition = yield* Effect.gen(function* () {
621 const r = yield* HttpClient.get(
622 `${baseUrl}/query-multi?category=electronics&subcategory=audio`,
623 ).pipe(Effect.flatMap((r) => r.json));
624 if ((r as any).count !== 2) {
625 return yield* Effect.fail(new QueryNotConsistent());
626 }
627 return r;
628 }).pipe(
629 Effect.retry({
630 while: (e) => e._tag === "QueryNotConsistent",
631 schedule: Schedule.max([Schedule.fixed("500 millis"), Schedule.recurs(30)]),
632 }),
633 );
634 expect((byPartition as any).count).toBe(2);
635 expect(((byPartition as any).items as any[]).map((item) => item.pk.S).sort()).toEqual([
636 "multi-query#1",
637 "multi-query#2",
638 ]);
639
640 // Narrow by the first RANGE attribute (left-to-right).
641 const byRank = yield* HttpClient.get(
642 `${baseUrl}/query-multi?category=electronics&subcategory=audio&rank=featured`,
643 ).pipe(Effect.flatMap((r) => r.json));
644 expect((byRank as any).count).toBe(1);
645 expect((byRank as any).items[0].pk.S).toBe("multi-query#1");
646 }),
647 );
648 });
649
650 describe("ListTables", () => {
651 test.provider("lists the deployed table", (_stack) =>
652 Effect.gen(function* () {
653 // DescribeTable can momentarily return an empty body on a freshly
654 // provisioned table; poll until the table description is populated.
655 const described = yield* fetchUntil(
656 HttpClient.get(`${baseUrl}/describe-table`).pipe(Effect.flatMap((r) => r.json)),
657 (body) => Boolean(body?.table?.TableName),
658 );
659
660 // ListTables is eventually consistent and a new table can lag its
661 // appearance in the account-wide listing by a few seconds.
662 const response = yield* fetchUntil(
663 HttpClient.get(`${baseUrl}/list-tables`).pipe(Effect.flatMap((r) => r.json)),
664 (body) =>
665 Array.isArray(body?.tableNames) &&
666 body.tableNames.includes((described as any).table.TableName),
667 );
668
669 expect((response as any).tableNames).toContain((described as any).table.TableName);
670 }),
671 );
672 });
673
674 describe("ListTagsOfResource", () => {
675 test.provider("lists alchemy ownership tags for the table", (_stack) =>
676 Effect.gen(function* () {
677 const response = yield* HttpClient.get(`${baseUrl}/list-tags`).pipe(
678 Effect.flatMap((r) => r.json),
679 );
680
681 const keys = ((response as any).tags ?? []).map((tag: any) => tag.Key);
682 expect(keys).toContain("alchemy::stack");
683 expect(keys).toContain("alchemy::stage");
684 expect(keys).toContain("alchemy::id");
685 }),
686 );
687 });
688
689 describe("RestoreTableToPointInTime", () => {
690 test.provider(
691 "returns a structured error when point-in-time recovery is unavailable",
692 (_stack) =>
693 Effect.gen(function* () {
694 // The route always returns a structured `{ ok, ... }` body; a
695 // missing `ok` means the request hit a not-yet-ready instance, so
696 // poll until the binding produces its structured result.
697 const response = yield* fetchUntil(
698 send(
699 HttpClientRequest.bodyJsonUnsafe(
700 HttpClientRequest.post(`${baseUrl}/restore-table`),
701 {},
702 ),
703 ).pipe(Effect.flatMap((r) => r.json)),
704 (body) => typeof body?.ok === "boolean",
705 );
706
707 expect((response as any).ok).toBe(false);
708 expect([
709 "PointInTimeRecoveryUnavailableException",
710 "TableAlreadyExistsException",
711 ]).toContain((response as any).error);
712 }),
713 );
714 });
715
716 describe("Scan", () => {
717 test.provider("scans all items in the table", (_stack) =>
718 Effect.gen(function* () {
719 yield* send(
720 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/put`), {
721 pk: "scan-test#1",
722 sk: "item",
723 data: "scan data",
724 }),
725 );
726
727 const response = yield* HttpClient.get(`${baseUrl}/scan`).pipe(
728 Effect.flatMap((r) => r.json),
729 );
730
731 expect((response as any).count).toBeGreaterThanOrEqual(1);
732 expect((response as any).items.length).toBeGreaterThanOrEqual(1);
733 }),
734 );
735 });
736
737 describe("Backups", () => {
738 test.provider(
739 "creates, describes, lists, restore-conflicts, and deletes a backup",
740 (_stack) =>
741 Effect.gen(function* () {
742 const created = (yield* send(
743 HttpClientRequest.bodyJsonUnsafe(HttpClientRequest.post(`${baseUrl}/create-backup`), {
744 name: "bindings-test-backup",
745 }),
746 ).pipe(Effect.flatMap((r) => r.json))) as { backupArn?: string; status?: string };
747 expect(created.backupArn).toBeTruthy();
748 const backupArn = created.backupArn!;
749
750 yield* Effect.gen(function* () {
751 // Wait for the on-demand backup to become AVAILABLE (it is
752 // CREATING for a few seconds on an empty table).
753 const described = yield* fetchUntil<{ status?: string }>(
754 HttpClient.get(
755 `${baseUrl}/describe-backup?arn=${encodeURIComponent(backupArn)}`,
756 ).pipe(Effect.flatMap((r) => r.json)),
757 (body) => body?.status === "AVAILABLE",
758 );
759 expect(described.status).toBe("AVAILABLE");
760
761 const listed = (yield* HttpClient.get(`${baseUrl}/list-backups`).pipe(
762 Effect.flatMap((r) => r.json),
763 )) as { backupArns: string[] };
764 expect(listed.backupArns).toContain(backupArn);
765
766 // Restoring into the already-deployed target table conflicts —
767 // proves the RestoreTableFromBackup binding round-trips with a
768 // typed error and never creates an unmanaged table.
769 const restored = (yield* send(
770 HttpClientRequest.bodyJsonUnsafe(
771 HttpClientRequest.post(`${baseUrl}/restore-from-backup`),
772 { arn: backupArn },
773 ),
774 ).pipe(Effect.flatMap((r) => r.json))) as { ok: boolean; error?: string };
775 expect(restored.ok).toBe(false);
776 expect(restored.error).toBe("TableAlreadyExistsException");
777 }).pipe(
778 // Always delete the backup, even if an assertion above failed —
779 // zero orphans.
780 Effect.ensuring(
781 send(
782 HttpClientRequest.bodyJsonUnsafe(
783 HttpClientRequest.delete(`${baseUrl}/delete-backup`),
784 { arn: backupArn },
785 ),
786 ).pipe(Effect.ignore),
787 ),
788 );
789
790 // The delete in `ensuring` already ran; verify it took effect (the
791 // backup drops out of ListBackups, allowing for eventual
792 // consistency).
793 const afterDelete = yield* fetchUntil<{ backupArns: string[] }>(
794 HttpClient.get(`${baseUrl}/list-backups`).pipe(Effect.flatMap((r) => r.json)),
795 (body) => Array.isArray(body?.backupArns) && !body.backupArns.includes(backupArn),
796 );
797 expect(afterDelete.backupArns).not.toContain(backupArn);
798 }),
799 { timeout: 120_000 },
800 );
801 });
802
803 describe("DescribeContinuousBackups", () => {
804 test.provider("reads the continuous backups / PITR status", (_stack) =>
805 Effect.gen(function* () {
806 const response = (yield* HttpClient.get(`${baseUrl}/describe-continuous-backups`).pipe(
807 Effect.flatMap((r) => r.json),
808 )) as { continuousBackupsStatus?: string; pitrStatus?: string };
809
810 expect(response.continuousBackupsStatus).toBe("ENABLED");
811 // The fixture table does not enable PITR.
812 expect(response.pitrStatus).toBe("DISABLED");
813 }),
814 );
815 });
816
817 describe("Exports", () => {
818 test.provider(
819 "surfaces a typed error when PITR is unavailable and lists exports",
820 (_stack) =>
821 Effect.gen(function* () {
822 // PITR is disabled on the fixture table, so starting an export
823 // fails fast with the typed tag (and never writes to the bucket).
824 const exported = (yield* send(
825 HttpClientRequest.bodyJsonUnsafe(
826 HttpClientRequest.post(`${baseUrl}/export-table`),
827 {},
828 ),
829 ).pipe(Effect.flatMap((r) => r.json))) as { ok: boolean; error?: string };
830 expect(exported.ok).toBe(false);
831 expect(exported.error).toBe("PointInTimeRecoveryUnavailableException");
832
833 const listed = (yield* HttpClient.get(`${baseUrl}/list-exports`).pipe(
834 Effect.flatMap((r) => r.json),
835 )) as { exportArns: string[] };
836 expect(Array.isArray(listed.exportArns)).toBe(true);
837 expect(listed.exportArns).toHaveLength(0);
838 }),
839 { timeout: 120_000 },
840 );
841
842 test.provider("DescribeExport returns a typed not-found error", (_stack) =>
843 Effect.gen(function* () {
844 const described = (yield* HttpClient.get(`${baseUrl}/describe-table`).pipe(
845 Effect.flatMap((r) => r.json),
846 )) as { table: { TableArn: string } };
847 const exportArn = `${described.table.TableArn}/export/01700000000000-00000000`;
848
849 const response = (yield* HttpClient.get(
850 `${baseUrl}/describe-export?arn=${encodeURIComponent(exportArn)}`,
851 ).pipe(Effect.flatMap((r) => r.json))) as { ok: boolean; error?: string };
852 expect(response.ok).toBe(false);
853 expect(response.error).toBe("ExportNotFoundException");
854 }),
855 );
856 });
857 },
858);
859
860class QueryNotConsistent extends Data.TaggedError("QueryNotConsistent") {}
861
862// A request hit a Lambda instance that hadn't fully converged yet (cold
863// secondary instance, IAM propagation lag, control-plane eventual
864// consistency), so the bound call returned a not-yet-populated shape. Retry.
865class BindingNotConsistent extends Data.TaggedError("BindingNotConsistent") {}
866
867// Poll an HTTP-driven binding call until its parsed body satisfies `ready`.
868// Control-plane reads (ListTables/DescribeTable) and PartiQL transactions can
869// briefly observe stale/empty state right after the table is provisioned.
870const fetchUntil = <A>(
871 fetch: Effect.Effect<unknown, any, HttpClient.HttpClient>,
872 ready: (body: any) => boolean,
873) =>
874 fetch.pipe(
875 Effect.flatMap((body) =>
876 ready(body) ? Effect.succeed(body as A) : Effect.fail(new BindingNotConsistent()),
877 ),
878 Effect.retry({
879 while: (e) => e._tag === "BindingNotConsistent",
880 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(20)]),
881 }),
882 );
883
884// The same `*Http` bindings on a Cloudflare Worker host: the Worker gets an
885// IAM user, access key and least-privilege role, and signs each request with
886// the assumed role in the table's region.
887const workerHost = Test.make({
888 providers: Layer.mergeAll(Cloudflare.providers(), AWS.providers()),
889});
890
891class WorkerNotReady extends Data.TaggedError("WorkerNotReady")<{
892 readonly status: number;
893 readonly body: string;
894}> {}
895
896class TableStillExists extends Data.TaggedError("TableStillExists") {}
897
898// A fresh workers.dev URL, a fresh IAM access key and a fresh role trust
899// policy all take seconds to propagate: retry non-2xx answers for ~90s.
900const sendToWorker = (request: HttpClientRequest.HttpClientRequest) =>
901 HttpClient.execute(request).pipe(
902 Effect.flatMap((response) =>
903 response.status >= 200 && response.status < 300
904 ? Effect.succeed(response)
905 : response.text.pipe(
906 Effect.flatMap((body) =>
907 Effect.fail(new WorkerNotReady({ status: response.status, body })),
908 ),
909 ),
910 ),
911 Effect.tapError((error) =>
912 Effect.logInfo(
913 `DynamoDB Worker host: not ready (${error._tag === "WorkerNotReady" ? `${error.status} ${error.body.slice(0, 300)}` : error._tag})`,
914 ),
915 ),
916 Effect.retry({
917 schedule: Schedule.max([Schedule.spaced("3 seconds"), Schedule.recurs(30)]),
918 }),
919 );
920
921describe(
922 "DynamoDB Bindings on a Cloudflare Worker",
923 {
924 tags: [
925 "provider:aws",
926 "provider:aws:dynamodb",
927 "provider:aws:iam",
928 "provider:cloudflare",
929 "live",
930 ],
931 },
932 () => {
933 workerHost.test.provider(
934 "GetItemHttp, PutItemHttp, ScanHttp and BatchGetItemHttp sign with the Worker's assumed role",
935 (stack) =>
936 Effect.gen(function* () {
937 yield* stack.destroy();
938
939 const { url, tableName } = yield* stack.deploy(
940 Effect.gen(function* () {
941 const worker = yield* LinksWorker;
942 const table = yield* LinksTable;
943 return {
944 url: worker.url.as<string>(),
945 tableName: table.tableName,
946 };
947 }),
948 );
949 expect(url).toBeDefined();
950
951 const put = yield* sendToWorker(
952 HttpClientRequest.put(`${url}/links/alchemy`).pipe(
953 HttpClientRequest.bodyText("https://alchemy.run"),
954 ),
955 );
956 expect(yield* put.json).toEqual({
957 id: "alchemy",
958 url: "https://alchemy.run",
959 });
960
961 // Out of band: the item landed in the table.
962 const stored = yield* DynamoDBApi.getItem({
963 TableName: tableName,
964 Key: { id: { S: "alchemy" } },
965 ConsistentRead: true,
966 });
967 expect(stored.Item?.url?.S).toEqual("https://alchemy.run");
968
969 const got = yield* sendToWorker(HttpClientRequest.get(`${url}/links/alchemy`));
970 expect(yield* got.json).toEqual({
971 id: "alchemy",
972 url: "https://alchemy.run",
973 });
974
975 // A second item written out of band shows up in the Worker's scan.
976 yield* DynamoDBApi.putItem({
977 TableName: tableName,
978 Item: { id: { S: "effect" }, url: { S: "https://effect.website" } },
979 });
980 const scanned = yield* sendToWorker(HttpClientRequest.get(`${url}/links`));
981 expect(yield* scanned.json).toEqual({ ids: ["alchemy", "effect"] });
982
983 // Both items in one BatchGetItem round trip.
984 const batch = yield* sendToWorker(
985 HttpClientRequest.get(`${url}/links?ids=alchemy,effect`),
986 );
987 expect(yield* batch.json).toEqual({
988 urls: ["https://alchemy.run", "https://effect.website"],
989 });
990
991 yield* stack.destroy();
992
993 yield* DynamoDBApi.describeTable({ TableName: tableName }).pipe(
994 Effect.flatMap(() => Effect.fail(new TableStillExists())),
995 Effect.retry({
996 while: (e) => e._tag === "TableStillExists",
997 schedule: Schedule.max([Schedule.fixed("1 second"), Schedule.recurs(30)]),
998 }),
999 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
1000 );
1001 }),
1002 { timeout: 240_000 },
1003 );
1004 },
1005);