1import * as CloudTrail from "@/AWS/CloudTrail";
2import * as Lambda from "@/AWS/Lambda";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Result from "effect/Result";
7import * as Schedule from "effect/Schedule";
8import { HttpServerRequest } from "effect/http/HttpServerRequest";
9import * as HttpServerResponse from "effect/http/HttpServerResponse";
10import path from "pathe";
11
12const main = path.resolve(import.meta.dirname, "lake-handler.ts");
13
14/**
15 * Probe helper: run the bound operation and report either the success
16 * projection or the typed error tag, so the test can assert that operations
17 * rejected by CloudTrail fail with a TYPED tag (never an untyped catch-all).
18 */
19const tagOr = <A, E extends { _tag: string }, R>(
20 effect: Effect.Effect<A, E, R>,
21 onSuccess: (value: A) => Record<string, unknown>,
22) =>
23 Effect.result(effect).pipe(
24 Effect.map((result) =>
25 Result.isSuccess(result)
26 ? onSuccess(result.success)
27 : { errorTag: result.failure._tag },
28 ),
29 );
30
31export class CloudTrailLakeTestFunction extends Lambda.Function<Lambda.Function>()(
32 "CloudTrailLakeTestFunction",
33) {}
34
35export default CloudTrailLakeTestFunction.make(
36 {
37 main,
38 functionUrl: true,
39 // A fresh store settles through CREATED/STARTING_INGESTION before the
40 // first query is accepted; the /query/run retry must fit the invocation.
41 timeout: Duration.seconds(60),
42 },
43 Effect.gen(function* () {
44 // GATED FIXTURE: deploying this requires a CloudTrail-Lake-onboarded
45 // account (Lake is closed to new customers — see the typed
46 // CloudTrailLakeOnboardingClosed probe in EventDataStore.test.ts).
47 const store = yield* CloudTrail.EventDataStore("BindingsLake", {
48 multiRegionEnabled: false,
49 retentionPeriod: "7 days",
50 terminationProtectionEnabled: false,
51 });
52
53 const startQuery = yield* CloudTrail.StartQuery(store);
54 const describeQuery = yield* CloudTrail.DescribeQuery(store);
55 const getQueryResults = yield* CloudTrail.GetQueryResults(store);
56 const cancelQuery = yield* CloudTrail.CancelQuery(store);
57 const listQueries = yield* CloudTrail.ListQueries(store);
58 const generateQuery = yield* CloudTrail.GenerateQuery(store);
59
60 return {
61 fetch: Effect.gen(function* () {
62 const request = yield* HttpServerRequest;
63 const url = new URL(request.originalUrl);
64 const pathname = url.pathname;
65 const param = (name: string) => url.searchParams.get(name) ?? "";
66
67 // Start a Lake query; the QueryStatement callback receives the bound
68 // store's ID for the FROM clause. Rides out the store's
69 // CREATED/STARTING_INGESTION settling window with a bounded retry.
70 if (request.method === "GET" && pathname === "/query/run") {
71 const result = yield* startQuery({
72 QueryStatement: (id) => `SELECT eventID FROM ${id} LIMIT 1`,
73 }).pipe(
74 Effect.retry({
75 while: (e): boolean =>
76 e._tag === "InactiveEventDataStoreException" ||
77 e._tag === "EventDataStoreNotFoundException" ||
78 e._tag === "MaxConcurrentQueriesException",
79 schedule: Schedule.spaced("2 seconds"),
80 times: 10,
81 }),
82 );
83 return yield* HttpServerResponse.json({ queryId: result.QueryId });
84 }
85
86 if (request.method === "GET" && pathname === "/query/describe") {
87 const result = yield* describeQuery({ QueryId: param("id") });
88 return yield* HttpServerResponse.json({
89 status: result.QueryStatus,
90 });
91 }
92
93 if (request.method === "GET" && pathname === "/query/results") {
94 const result = yield* getQueryResults({ QueryId: param("id") });
95 return yield* HttpServerResponse.json({
96 status: result.QueryStatus,
97 rows: (result.QueryResultRows ?? []).length,
98 });
99 }
100
101 // Cancelling an already-finished query fails with the typed
102 // InactiveQueryException — either outcome proves IAM + typed errors.
103 if (request.method === "POST" && pathname === "/query/cancel") {
104 return yield* HttpServerResponse.json(
105 yield* tagOr(cancelQuery({ QueryId: param("id") }), (result) => ({
106 status: result.QueryStatus,
107 })),
108 );
109 }
110
111 // The bound store is injected as the request's EventDataStore.
112 if (request.method === "GET" && pathname === "/query/list") {
113 const result = yield* listQueries({ MaxResults: 10 });
114 return yield* HttpServerResponse.json({
115 ids: (result.Queries ?? []).map((q) => q.QueryId),
116 });
117 }
118
119 // Natural-language SQL generation may be region/entitlement gated —
120 // a rejection must surface as a TYPED tag.
121 if (request.method === "POST" && pathname === "/query/generate") {
122 return yield* HttpServerResponse.json(
123 yield* tagOr(
124 generateQuery({
125 Prompt: "How many events were recorded in the last day?",
126 }),
127 (result) => ({ statement: result.QueryStatement }),
128 ),
129 );
130 }
131
132 return yield* HttpServerResponse.json(
133 { error: "Not found", method: request.method, pathname },
134 { status: 404 },
135 );
136 }).pipe(Effect.orDie),
137 };
138 }).pipe(
139 Effect.provide(
140 Layer.mergeAll(
141 CloudTrail.StartQueryHttp,
142 CloudTrail.DescribeQueryHttp,
143 CloudTrail.GetQueryResultsHttp,
144 CloudTrail.CancelQueryHttp,
145 CloudTrail.ListQueriesHttp,
146 CloudTrail.GenerateQueryHttp,
147 ),
148 ),
149 ),
150);