1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import { describe, expect } from "alchemy-test";
5import * as Data from "effect/Data";
6import * as Effect from "effect/Effect";
7import * as Schedule from "effect/Schedule";
8import * as HttpClient from "effect/http/HttpClient";
9import * as HttpClientRequest from "effect/http/HttpClientRequest";
10import LogsTestFunctionLive, { LogsTestFunction } from "./handler.ts";
11
12const testOptions = { providers: AWS.providers() };
13const { test, beforeAll, afterAll } = Test.make(testOptions);
14const sharedStack = Core.scratchStack(testOptions, "LogsBindings");
15
16// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
17// over 60s on a fresh deploy under parallel-suite load. Budget ~150s of
18// readiness polling.
19const readinessPolicy = Schedule.max([
20 Schedule.fixed("2 seconds"),
21 Schedule.recurs(75),
22]);
23
24let baseUrl: string;
25
26class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
27 readonly status: number;
28 readonly body: string;
29}> {}
30
31class MarkerNotVisible extends Data.TaggedError("MarkerNotVisible")<{
32 readonly marker: string;
33}> {}
34
35// The shared Lambda fixture occasionally answers a transient 5xx under
36// parallel load (cold re-init, IAM propagation). Retry 5xx only; a genuine
37// 4xx/assertion failure surfaces immediately.
38const send = (request: HttpClientRequest.HttpClientRequest) =>
39 HttpClient.execute(request).pipe(
40 Effect.flatMap((response) =>
41 response.status >= 500
42 ? response.text.pipe(
43 Effect.flatMap((body) =>
44 Effect.fail(
45 new TransientUpstream({ status: response.status, body }),
46 ),
47 ),
48 )
49 : Effect.succeed(response),
50 ),
51 Effect.retry({
52 while: (e) => e._tag === "TransientUpstream",
53 schedule: Schedule.max([
54 Schedule.exponential("500 millis"),
55 Schedule.recurs(6),
56 ]),
57 }),
58 );
59
60const putMarker = (marker: string) =>
61 send(HttpClientRequest.post(`${baseUrl}/put?message=${marker}`)).pipe(
62 Effect.flatMap((r) => r.json),
63 );
64
65// FilterLogEvents observes freshly-ingested events after a short delay —
66// poll bounded (~40s).
67const filterUntilVisible = (marker: string) =>
68 send(HttpClientRequest.get(`${baseUrl}/filter?pattern=${marker}`)).pipe(
69 Effect.flatMap((r) => r.json),
70 Effect.flatMap((body) => {
71 const messages = (body as { messages: (string | undefined)[] }).messages;
72 return messages.some((message) => message?.includes(marker))
73 ? Effect.succeed(messages)
74 : Effect.fail(new MarkerNotVisible({ marker }));
75 }),
76 Effect.retry({
77 while: (e) => e._tag === "MarkerNotVisible",
78 schedule: Schedule.max([
79 Schedule.fixed("2 seconds"),
80 Schedule.recurs(20),
81 ]),
82 }),
83 );
84
85describe(
86 "Logs Bindings",
87 {
88 tags: ["provider:aws", "provider:aws:lambda", "provider:aws:logs", "live"],
89 },
90 () => {
91 beforeAll(
92 Effect.gen(function* () {
93 yield* Effect.logInfo("Logs test setup: destroying previous resources");
94 yield* sharedStack.destroy();
95
96 yield* Effect.logInfo("Logs test setup: deploying fixture");
97 const { functionUrl } = yield* sharedStack.deploy(
98 Effect.gen(function* () {
99 return yield* LogsTestFunction;
100 }).pipe(Effect.provide(LogsTestFunctionLive)),
101 );
102
103 expect(functionUrl).toBeTruthy();
104 baseUrl = functionUrl!.replace(/\/+$/, "");
105 const readinessUrl = `${baseUrl}/get-events`;
106
107 yield* Effect.logInfo(
108 `Logs test setup: probing readiness at ${readinessUrl}`,
109 );
110 yield* HttpClient.get(readinessUrl).pipe(
111 Effect.flatMap((response) =>
112 response.status === 200
113 ? Effect.succeed(response)
114 : Effect.fail(
115 new Error(`Function not ready: ${response.status}`),
116 ),
117 ),
118 Effect.tapError((error) =>
119 Effect.logWarning(
120 `Logs test setup: fixture not ready yet (${String(error)})`,
121 ),
122 ),
123 Effect.retry({ schedule: readinessPolicy }),
124 );
125 }),
126 { timeout: 240_000 },
127 );
128
129 afterAll.skipIf(!!process.env.NO_DESTROY)(sharedStack.destroy(), {
130 timeout: 120_000,
131 });
132
133 describe("PutLogEvents", () => {
134 test.provider("writes a log event through the binding", (_stack) =>
135 Effect.gen(function* () {
136 const marker = `put-${crypto.randomUUID()}`;
137 const response = yield* putMarker(marker);
138 expect((response as any).ok).toBe(true);
139 expect((response as any).rejected).toBeNull();
140 }),
141 );
142 });
143
144 describe("FilterLogEvents", () => {
145 test.provider(
146 "round-trips a marker written by the fixture",
147 (_stack) =>
148 Effect.gen(function* () {
149 const marker = `filter-${crypto.randomUUID()}`;
150 yield* putMarker(marker);
151 const messages = yield* filterUntilVisible(marker);
152 expect(messages.some((m) => m?.includes(marker))).toBe(true);
153 }),
154 { timeout: 90_000 },
155 );
156 });
157
158 describe("GetLogEvents", () => {
159 test.provider(
160 "reads events back from the bound stream",
161 (_stack) =>
162 Effect.gen(function* () {
163 const marker = `get-${crypto.randomUUID()}`;
164 yield* putMarker(marker);
165
166 const messages = yield* send(
167 HttpClientRequest.get(`${baseUrl}/get-events`),
168 ).pipe(
169 Effect.flatMap((r) => r.json),
170 Effect.flatMap((body) => {
171 const found = (body as { messages: (string | undefined)[] })
172 .messages;
173 return found.some((message) => message?.includes(marker))
174 ? Effect.succeed(found)
175 : Effect.fail(new MarkerNotVisible({ marker }));
176 }),
177 Effect.retry({
178 while: (e) => e._tag === "MarkerNotVisible",
179 schedule: Schedule.max([
180 Schedule.fixed("2 seconds"),
181 Schedule.recurs(20),
182 ]),
183 }),
184 );
185 expect(messages.some((m) => m?.includes(marker))).toBe(true);
186 }),
187 { timeout: 90_000 },
188 );
189 });
190
191 describe("StartQuery", () => {
192 test.provider(
193 "starts an Insights query scoped to the bound group",
194 (_stack) =>
195 Effect.gen(function* () {
196 const response = yield* send(
197 HttpClientRequest.get(`${baseUrl}/query`),
198 ).pipe(Effect.flatMap((r) => r.json));
199 expect((response as any).queryId).toBeTruthy();
200 }),
201 { timeout: 90_000 },
202 );
203 });
204
205 describe("GetQueryResults", () => {
206 test.provider(
207 "polls the started query to completion",
208 (_stack) =>
209 Effect.gen(function* () {
210 const response = yield* send(
211 HttpClientRequest.get(`${baseUrl}/query`),
212 ).pipe(Effect.flatMap((r) => r.json));
213 expect((response as any).status).toBe("Complete");
214 expect((response as any).resultCount).toBeGreaterThanOrEqual(0);
215 }),
216 { timeout: 90_000 },
217 );
218 });
219
220 describe("StopQuery", () => {
221 test.provider(
222 "stops (or observes completion of) a started query",
223 (_stack) =>
224 Effect.gen(function* () {
225 const response = yield* send(
226 HttpClientRequest.get(`${baseUrl}/stop-query`),
227 ).pipe(Effect.flatMap((r) => r.json));
228 const body = response as {
229 ok: boolean;
230 stopped: boolean;
231 error: string | null;
232 };
233 expect(body.ok).toBe(true);
234 // Either the stop landed, or a benign completion/registration race
235 // surfaced as one of the two typed tags — anything else (e.g.
236 // AccessDenied) dies in the fixture and fails the request.
237 expect(
238 body.stopped === true ||
239 body.error === "InvalidParameterException" ||
240 body.error === "ResourceNotFoundException",
241 ).toBe(true);
242 }),
243 { timeout: 90_000 },
244 );
245 });
246
247 describe("GetLogRecord", () => {
248 test.provider(
249 "fetches the full record behind a query-result @ptr",
250 (_stack) =>
251 Effect.gen(function* () {
252 const marker = `record-${crypto.randomUUID()}`;
253 yield* putMarker(marker);
254 // Insights sees freshly-ingested events after a short delay —
255 // poll the query route until it returns rows (bounded, ~60s).
256 const ptr = yield* send(
257 HttpClientRequest.get(`${baseUrl}/query`),
258 ).pipe(
259 Effect.flatMap((r) => r.json),
260 Effect.flatMap((body) => {
261 const found = (body as { ptr: string | null }).ptr;
262 return found
263 ? Effect.succeed(found)
264 : Effect.fail(new MarkerNotVisible({ marker }));
265 }),
266 Effect.retry({
267 while: (e) => e._tag === "MarkerNotVisible",
268 schedule: Schedule.max([
269 Schedule.fixed("5 seconds"),
270 Schedule.recurs(10),
271 ]),
272 }),
273 );
274
275 const record = yield* send(
276 HttpClientRequest.get(
277 `${baseUrl}/record?ptr=${encodeURIComponent(ptr)}`,
278 ),
279 ).pipe(Effect.flatMap((r) => r.json));
280 expect((record as any).message).toBeTruthy();
281 }),
282 { timeout: 180_000 },
283 );
284 });
285
286 describe("GetLogGroupFields", () => {
287 test.provider(
288 "discovers fields present in the bound group",
289 (_stack) =>
290 Effect.gen(function* () {
291 const marker = `fields-${crypto.randomUUID()}`;
292 yield* putMarker(marker);
293 const response = yield* send(
294 HttpClientRequest.get(`${baseUrl}/fields`),
295 ).pipe(Effect.flatMap((r) => r.json));
296 expect(Array.isArray((response as any).fields)).toBe(true);
297 }),
298 { timeout: 90_000 },
299 );
300 });
301
302 describe("DescribeLogStreams", () => {
303 test.provider(
304 "lists the streams of the bound group",
305 (_stack) =>
306 Effect.gen(function* () {
307 const response = yield* send(
308 HttpClientRequest.get(`${baseUrl}/streams`),
309 ).pipe(Effect.flatMap((r) => r.json));
310 const streams = (response as { streams: string[] }).streams;
311 expect(streams).toContain("alchemy-test-bindings-stream");
312 }),
313 { timeout: 90_000 },
314 );
315 });
316
317 describe("CreateLogStream + DeleteLogStream", () => {
318 test.provider(
319 "creates and deletes a dynamic stream through the bindings",
320 (_stack) =>
321 Effect.gen(function* () {
322 const name = "alchemy-test-bindings-dynamic-stream";
323 const response = yield* send(
324 HttpClientRequest.post(
325 `${baseUrl}/stream-lifecycle?name=${name}`,
326 ),
327 ).pipe(Effect.flatMap((r) => r.json));
328 expect((response as any).seen).toBe(true);
329 expect((response as any).gone).toBe(true);
330 }),
331 { timeout: 90_000 },
332 );
333 });
334 },
335);