1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import { expect } from "alchemy-test";
4import * as Data from "effect/Data";
5import * as Effect from "effect/Effect";
6import * as Schedule from "effect/Schedule";
7import * as HttpClient from "effect/http/HttpClient";
8import * as HttpClientRequest from "effect/http/HttpClientRequest";
9
10import CloudTrailLakeTestFunctionLive, {
11 CloudTrailLakeTestFunction,
12} from "./lake-handler";
13
14const { test } = Test.make({ providers: AWS.providers() });
15
16class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
17 readonly status: number;
18 readonly body: string;
19}> {}
20
21// Retry transient 5xx from the Lambda fixture (cold re-init, IAM propagation
22// on the freshly attached cloudtrail policy, the store's settling window
23// exhausting the handler's bounded retry). Genuine 4xx return immediately.
24const send = (request: HttpClientRequest.HttpClientRequest) =>
25 HttpClient.execute(request).pipe(
26 Effect.flatMap((response) =>
27 response.status >= 500
28 ? response.text.pipe(
29 Effect.flatMap((body) =>
30 Effect.fail(
31 new TransientUpstream({ status: response.status, body }),
32 ),
33 ),
34 )
35 : Effect.succeed(response),
36 ),
37 Effect.retry({
38 while: (e) => e._tag === "TransientUpstream",
39 schedule: Schedule.max([
40 Schedule.exponential("1 second"),
41 Schedule.recurs(8),
42 ]),
43 }),
44 );
45
46const getJson = (url: string) =>
47 send(HttpClientRequest.get(url)).pipe(Effect.flatMap((r) => r.json));
48
49class QueryStillRunning extends Data.TaggedError("QueryStillRunning") {}
50
51// GATED: CloudTrail Lake is closed to new customers (see the typed
52// CloudTrailLakeOnboardingClosed probe in EventDataStore.test.ts) — deploying
53// the fixture's EventDataStore only works on an account that was onboarded
54// to Lake before the closure. Set AWS_TEST_CLOUDTRAIL_LAKE=1 to run it.
55test.provider.skipIf(!process.env.AWS_TEST_CLOUDTRAIL_LAKE)(
56 "Lake query bindings: start, describe, results, list, cancel, generate",
57 (stack) =>
58 Effect.gen(function* () {
59 yield* stack.destroy();
60
61 const { functionUrl } = yield* stack.deploy(
62 Effect.gen(function* () {
63 return yield* CloudTrailLakeTestFunction;
64 }).pipe(Effect.provide(CloudTrailLakeTestFunctionLive)),
65 );
66 expect(functionUrl).toBeTruthy();
67 const baseUrl = functionUrl!.replace(/\/+$/, "");
68
69 // Readiness: fresh function URLs take a few seconds to serve 200s.
70 yield* HttpClient.get(`${baseUrl}/query/list`).pipe(
71 Effect.flatMap((response) =>
72 response.status === 200
73 ? Effect.succeed(response)
74 : Effect.fail(new Error(`Function not ready: ${response.status}`)),
75 ),
76 Effect.retry({
77 schedule: Schedule.max([
78 Schedule.fixed("2 seconds"),
79 Schedule.recurs(60),
80 ]),
81 }),
82 );
83
84 // StartQuery — the fixture threads the store's ID into the FROM
85 // clause and rides out the store's settling window.
86 const started = (yield* getJson(`${baseUrl}/query/run`)) as any;
87 expect(started.queryId).toBeTruthy();
88
89 // DescribeQuery — poll to a terminal status (an empty store finishes
90 // fast; FAILED still proves IAM + typed wiring).
91 const terminal = yield* getJson(
92 `${baseUrl}/query/describe?id=${started.queryId}`,
93 ).pipe(
94 Effect.flatMap((body: any) =>
95 body.status === "QUEUED" || body.status === "RUNNING"
96 ? Effect.fail(new QueryStillRunning())
97 : Effect.succeed(body),
98 ),
99 Effect.retry({
100 while: (e) => e._tag === "QueryStillRunning",
101 schedule: Schedule.max([
102 Schedule.fixed("3 seconds"),
103 Schedule.recurs(20),
104 ]),
105 }),
106 );
107 expect(["FINISHED", "FAILED", "CANCELLED", "TIMED_OUT"]).toContain(
108 (terminal as any).status,
109 );
110
111 // GetQueryResults — an empty store yields zero rows.
112 const results = (yield* getJson(
113 `${baseUrl}/query/results?id=${started.queryId}`,
114 )) as any;
115 expect(results.rows).toBeGreaterThanOrEqual(0);
116
117 // ListQueries — the injected EventDataStore scopes the listing.
118 const list = (yield* getJson(`${baseUrl}/query/list`)) as any;
119 expect(list.ids).toContain(started.queryId);
120
121 // CancelQuery — the query already finished, so CloudTrail rejects
122 // with the typed InactiveQueryException (a fresh cancel racing the
123 // query instead returns CANCELLED); both prove the wiring.
124 const cancelled = (yield* send(
125 HttpClientRequest.post(`${baseUrl}/query/cancel?id=${started.queryId}`),
126 ).pipe(Effect.flatMap((r) => r.json))) as any;
127 if (cancelled.errorTag !== undefined) {
128 expect(cancelled.errorTag.length).toBeGreaterThan(0);
129 } else {
130 expect(cancelled.status).toBeTruthy();
131 }
132
133 // GenerateQuery — natural-language SQL generation; success or a
134 // typed rejection both prove the wiring.
135 const generated = (yield* send(
136 HttpClientRequest.post(`${baseUrl}/query/generate`),
137 ).pipe(Effect.flatMap((r) => r.json))) as any;
138 if (generated.errorTag !== undefined) {
139 expect(generated.errorTag.length).toBeGreaterThan(0);
140 } else {
141 expect(generated.statement).toBeTruthy();
142 }
143
144 yield* stack.destroy();
145 }),
146 {
147 tags: [
148 "provider:aws",
149 "provider:aws:cloudtrail",
150 "provider:aws:lambda",
151 "live",
152 ],
153 timeout: 600_000,
154 },
155);