1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import { describe, expect } from "alchemy-test";
4import * as Effect from "effect/Effect";
5import * as Schedule from "effect/Schedule";
6import * as HttpClient from "effect/http/HttpClient";
7import * as HttpClientRequest from "effect/http/HttpClientRequest";
8
9import TimestreamTestFunctionLive, { TimestreamTestFunction } from "./handler";
10
11const { test } = Test.make({ providers: AWS.providers() });
12
13// Amazon Timestream for LiveAnalytics is closed to new AWS customers; the
14// testing account receives the typed `TimestreamNotOnboarded` gate on every
15// operation (asserted ungated in Database.test.ts). The Lambda E2E — write
16// time-series points through the WriteRecords binding, then read them back
17// through the Query binding — is therefore gated behind AWS_TEST_TIMESTREAM=1
18// so an onboarded account can run it unchanged.
19describe(
20 "AWS.Timestream Bindings",
21 {
22 tags: [
23 "provider:aws",
24 "provider:aws:lambda",
25 "provider:aws:timestream",
26 "live",
27 ],
28 },
29 () => {
30 test.provider.skipIf(!process.env.AWS_TEST_TIMESTREAM)(
31 "Lambda writes records and queries them back",
32 (stack) =>
33 Effect.gen(function* () {
34 const { functionUrl } = yield* stack.deploy(
35 Effect.gen(function* () {
36 return yield* TimestreamTestFunction;
37 }).pipe(Effect.provide(TimestreamTestFunctionLive)),
38 );
39 expect(functionUrl).toBeTruthy();
40 const baseUrl = functionUrl!.replace(/\/+$/, "");
41
42 // Write a point through the WriteRecords binding. Retry through
43 // function-URL cold start / IAM propagation.
44 const writeResponse = yield* HttpClient.execute(
45 HttpClientRequest.post(`${baseUrl}/write`).pipe(
46 HttpClientRequest.bodyJsonUnsafe({
47 host: "web-1",
48 value: "42.0",
49 }),
50 ),
51 ).pipe(
52 Effect.flatMap((response) =>
53 response.status === 200
54 ? Effect.succeed(response)
55 : Effect.fail(new Error(`write not ready: ${response.status}`)),
56 ),
57 Effect.retry({
58 schedule: Schedule.max([
59 Schedule.exponential("1 second"),
60 Schedule.recurs(8),
61 ]),
62 }),
63 );
64 const written = (yield* writeResponse.json) as {
65 recordsIngested: { Total?: number } | undefined;
66 };
67 expect(written.recordsIngested?.Total).toBeGreaterThanOrEqual(1);
68
69 // Query the point back through the Query binding. Memory-store writes
70 // are queryable near-immediately; retry a few times regardless.
71 const rows = yield* HttpClient.get(`${baseUrl}/query`).pipe(
72 Effect.flatMap((response) =>
73 response.status === 200
74 ? response.json
75 : Effect.fail(new Error(`query failed: ${response.status}`)),
76 ),
77 Effect.map(
78 (body) =>
79 (
80 body as {
81 rows: Array<{ Data: Array<{ ScalarValue?: string }> }>;
82 }
83 ).rows,
84 ),
85 Effect.flatMap((rows) =>
86 Number(rows[0]?.Data[0]?.ScalarValue ?? "0") >= 1
87 ? Effect.succeed(rows)
88 : Effect.fail(new Error("no rows counted yet")),
89 ),
90 Effect.retry({
91 schedule: Schedule.max([
92 Schedule.spaced("2 seconds"),
93 Schedule.recurs(10),
94 ]),
95 }),
96 );
97 expect(Number(rows[0]!.Data[0]!.ScalarValue)).toBeGreaterThanOrEqual(
98 1,
99 );
100
101 // Validate the same SQL through the PrepareQuery binding.
102 const prepared = (yield* HttpClient.get(`${baseUrl}/prepare`).pipe(
103 Effect.flatMap((response) =>
104 response.status === 200
105 ? response.json
106 : Effect.fail(new Error(`prepare failed: ${response.status}`)),
107 ),
108 Effect.retry({
109 schedule: Schedule.max([
110 Schedule.spaced("2 seconds"),
111 Schedule.recurs(8),
112 ]),
113 }),
114 )) as { columns: Array<{ Name?: string }> };
115 expect(prepared.columns[0]?.Name).toBe("c");
116
117 yield* stack.destroy();
118 }),
119 { timeout: 600_000 },
120 );
121 },
122);