1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import * as logs from "@distilled.cloud/aws/cloudwatch-logs";
4import { expect } from "alchemy-test";
5import * as Data from "effect/Data";
6import * as Effect from "effect/Effect";
7import * as Schedule from "effect/Schedule";
8import * as HttpBody from "effect/http/HttpBody";
9import * as HttpClient from "effect/http/HttpClient";
10import LogEventSinkFunctionLive, {
11 LogEventSinkFunction,
12} from "./sink-handler.ts";
13
14const { test } = Test.make({ providers: AWS.providers() });
15
16class FunctionNotReady extends Data.TaggedError("FunctionNotReady") {}
17
18class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
19 readonly status: number;
20}> {}
21
22class EventsNotVisible extends Data.TaggedError("EventsNotVisible")<{
23 readonly missing: readonly string[];
24}> {}
25
26/** Poll the fixture's /ready route until it serves the resolved log target. */
27const waitForReady = (url: string) =>
28 Effect.gen(function* () {
29 const response = yield* HttpClient.get(url);
30 if (response.status !== 200) {
31 return yield* Effect.fail(new FunctionNotReady());
32 }
33 // A fresh function URL can briefly 200 before the captured env (the log
34 // group name) has propagated — keep polling until it is a string.
35 const json = (yield* response.json) as any;
36 if (
37 typeof json?.logGroupName === "string" &&
38 typeof json?.logStreamName === "string"
39 ) {
40 return {
41 logGroupName: json.logGroupName as string,
42 logStreamName: json.logStreamName as string,
43 };
44 }
45 return yield* Effect.fail(new FunctionNotReady());
46 }).pipe(
47 Effect.retry({
48 while: (error) => error._tag === "FunctionNotReady",
49 schedule: Schedule.max([
50 Schedule.fixed("2 seconds"),
51 Schedule.recurs(75),
52 ]),
53 }),
54 );
55
56/** POST a JSON body to the fixture, retrying transient 5xx cold-start noise. */
57const postJson = (url: string, body: unknown) =>
58 Effect.gen(function* () {
59 const response = yield* HttpClient.post(url, {
60 body: yield* HttpBody.json(body),
61 });
62 if (response.status >= 500) {
63 return yield* Effect.fail(
64 new TransientUpstream({ status: response.status }),
65 );
66 }
67 return yield* response.json;
68 }).pipe(
69 Effect.retry({
70 while: (e) => e._tag === "TransientUpstream",
71 schedule: Schedule.max([
72 Schedule.exponential("1 second"),
73 Schedule.recurs(6),
74 ]),
75 }),
76 );
77
78/**
79 * Out-of-band read: poll GetLogEvents until every marker is visible in the
80 * stream (ingestion lags a few seconds), returning the observed messages.
81 */
82const waitForEvents = (
83 logGroupName: string,
84 logStreamName: string,
85 markers: readonly string[],
86) =>
87 Effect.gen(function* () {
88 const result = yield* logs.getLogEvents({
89 logGroupName,
90 logStreamName,
91 startFromHead: true,
92 });
93 const messages = (result.events ?? []).map((event) => event.message ?? "");
94 const missing = markers.filter(
95 (marker) => !messages.some((message) => message.includes(marker)),
96 );
97 if (missing.length > 0) {
98 return yield* Effect.fail(new EventsNotVisible({ missing }));
99 }
100 return messages;
101 }).pipe(
102 Effect.retry({
103 // Ride out both ingestion lag and the brief post-create window where
104 // the stream is not yet describable.
105 while: (e) =>
106 e._tag === "EventsNotVisible" || e._tag === "ResourceNotFoundException",
107 schedule: Schedule.max([
108 Schedule.fixed("2 seconds"),
109 Schedule.recurs(30),
110 ]),
111 }),
112 );
113
114test.provider(
115 "LogEventSink streams events through a deployed Lambda and drops rejected events",
116 (stack) =>
117 Effect.gen(function* () {
118 // Leading destroy reconciles away any partial deployment left behind by
119 // a previous crashed run (physical names are deterministic, so the
120 // re-deploy adopts and the trailing destroy cleans up).
121 yield* stack.destroy();
122
123 const fn = yield* stack.deploy(
124 LogEventSinkFunction.pipe(Effect.provide(LogEventSinkFunctionLive)),
125 );
126 const baseUrl = fn.functionUrl!.replace(/\/+$/, "");
127
128 const { logGroupName, logStreamName } = yield* waitForReady(
129 `${baseUrl}/ready`,
130 );
131
132 // 25 markers through the sink in one request-scoped drain.
133 const markers = Array.from(
134 { length: 25 },
135 (_, i) => `sink-${i}-${crypto.randomUUID()}`,
136 );
137 const response = yield* postJson(`${baseUrl}/sink`, {
138 messages: markers,
139 });
140 expect((response as any).ok).toBe(true);
141 expect((response as any).count).toBe(markers.length);
142
143 // Out-of-band: read the stream directly via distilled.
144 yield* waitForEvents(logGroupName, logStreamName, markers);
145
146 // Partial rejection: PutLogEvents has no transient per-event failure
147 // mode, so the partial-failure path the API supports is permanent
148 // rejection — an event >2h in the future is skipped by the API while
149 // the valid events in the same batch are still ingested. The sink must
150 // drop it (REJECT, no retry) and complete without error.
151 const validMarkers = Array.from(
152 { length: 3 },
153 (_, i) => `valid-${i}-${crypto.randomUUID()}`,
154 );
155 const rejectedMarker = `rejected-${crypto.randomUUID()}`;
156 const rejectedResponse = yield* postJson(
157 `${baseUrl}/sink-with-rejected`,
158 { valid: validMarkers, rejected: rejectedMarker },
159 );
160 expect((rejectedResponse as any).ok).toBe(true);
161
162 const messages = yield* waitForEvents(
163 logGroupName,
164 logStreamName,
165 validMarkers,
166 );
167 // The rejected (too-new) event was never ingested.
168 expect(messages.some((message) => message.includes(rejectedMarker))).toBe(
169 false,
170 );
171
172 yield* stack.destroy();
173
174 // Assert the sink's log group is gone after the final destroy.
175 const remaining = yield* logs.describeLogGroups({
176 logGroupNamePrefix: logGroupName,
177 limit: 1,
178 });
179 expect(
180 (remaining.logGroups ?? []).some(
181 (group) => group.logGroupName === logGroupName,
182 ),
183 ).toBe(false);
184 }),
185 {
186 tags: ["provider:aws", "provider:aws:lambda", "provider:aws:logs", "live"],
187 timeout: 240_000,
188 },
189);