1import * as Lambda from "@/AWS/Lambda";
2import * as Logs from "@/AWS/Logs";
3import * as Clock from "effect/Clock";
4import * as Data from "effect/Data";
5import * as Duration from "effect/Duration";
6import * as Effect from "effect/Effect";
7import * as Layer from "effect/Layer";
8import * as Schedule from "effect/Schedule";
9import { HttpServerRequest } from "effect/http/HttpServerRequest";
10import * as HttpServerResponse from "effect/http/HttpServerResponse";
11import path from "pathe";
12
13const main = path.resolve(import.meta.dirname, "handler.ts");
14
15export class LogsTestFunction extends Lambda.Function<Lambda.Function>()(
16 "LogsTestFunction",
17) {}
18
19class QueryNotComplete extends Data.TaggedError("QueryNotComplete")<{
20 readonly status: string | undefined;
21}> {}
22
23export default LogsTestFunction.make(
24 {
25 main,
26 functionUrl: true,
27 timeout: Duration.seconds(30),
28 },
29 Effect.gen(function* () {
30 const logGroup = yield* Logs.LogGroup("BindingsLogGroup", {
31 retention: "1 day",
32 });
33 const logStream = yield* Logs.LogStream("BindingsLogStream", {
34 logGroupName: logGroup.logGroupName,
35 logStreamName: "alchemy-test-bindings-stream",
36 });
37
38 const putLogEvents = yield* Logs.PutLogEvents(logGroup);
39 const filterLogEvents = yield* Logs.FilterLogEvents(logGroup);
40 const getLogEvents = yield* Logs.GetLogEvents(logGroup);
41 const startQuery = yield* Logs.StartQuery(logGroup);
42 const getQueryResults = yield* Logs.GetQueryResults(logGroup);
43 const stopQuery = yield* Logs.StopQuery(logGroup);
44 const getLogRecord = yield* Logs.GetLogRecord(logGroup);
45 const getLogGroupFields = yield* Logs.GetLogGroupFields(logGroup);
46 const describeLogStreams = yield* Logs.DescribeLogStreams(logGroup);
47 const createLogStream = yield* Logs.CreateLogStream(logGroup);
48 const deleteLogStream = yield* Logs.DeleteLogStream(logGroup);
49
50 const LogStreamName = yield* logStream.logStreamName;
51
52 return {
53 fetch: Effect.gen(function* () {
54 const request = yield* HttpServerRequest;
55 const url = new URL(request.originalUrl);
56 const pathname = url.pathname;
57 const logStreamName = yield* LogStreamName;
58
59 if (request.method === "POST" && pathname === "/put") {
60 const message = url.searchParams.get("message") ?? "no-message";
61 const timestamp = yield* Clock.currentTimeMillis;
62 const result = yield* putLogEvents({
63 logStreamName,
64 logEvents: [{ timestamp, message }],
65 });
66 return yield* HttpServerResponse.json({
67 ok: true,
68 rejected: result.rejectedLogEventsInfo ?? null,
69 });
70 }
71
72 if (request.method === "GET" && pathname === "/filter") {
73 const pattern = url.searchParams.get("pattern") ?? "";
74 const result = yield* filterLogEvents({
75 filterPattern: `"${pattern}"`,
76 });
77 return yield* HttpServerResponse.json({
78 messages: (result.events ?? []).map((event) => event.message),
79 });
80 }
81
82 if (request.method === "GET" && pathname === "/get-events") {
83 const result = yield* getLogEvents({
84 logStreamName,
85 startFromHead: true,
86 });
87 return yield* HttpServerResponse.json({
88 messages: (result.events ?? []).map((event) => event.message),
89 });
90 }
91
92 if (request.method === "GET" && pathname === "/query") {
93 const now = yield* Clock.currentTimeMillis;
94 const { queryId } = yield* startQuery({
95 queryString: "fields @timestamp, @message | limit 10",
96 startTime: Math.floor(now / 1000) - 3600,
97 endTime: Math.floor(now / 1000) + 60,
98 });
99 if (!queryId) {
100 return yield* HttpServerResponse.json(
101 { error: "no queryId" },
102 { status: 500 },
103 );
104 }
105 // Poll bounded (~20s) — well inside the 30s function timeout.
106 const results = yield* getQueryResults({ queryId }).pipe(
107 Effect.flatMap((response) =>
108 response.status === "Complete"
109 ? Effect.succeed(response)
110 : Effect.fail(
111 new QueryNotComplete({ status: response.status }),
112 ),
113 ),
114 Effect.retry({
115 while: (error) => error._tag === "QueryNotComplete",
116 schedule: Schedule.max([
117 Schedule.fixed("2 seconds"),
118 Schedule.recurs(10),
119 ]),
120 }),
121 );
122 // Surface the first row's @ptr so the test can exercise GetLogRecord.
123 const ptr =
124 (results.results ?? [])
125 .flat()
126 .find((field) => field.field === "@ptr")?.value ?? null;
127 return yield* HttpServerResponse.json({
128 queryId,
129 status: results.status,
130 resultCount: (results.results ?? []).length,
131 ptr,
132 });
133 }
134
135 if (request.method === "GET" && pathname === "/record") {
136 const ptr = url.searchParams.get("ptr");
137 if (!ptr) {
138 return yield* HttpServerResponse.json(
139 { error: "missing ptr" },
140 { status: 400 },
141 );
142 }
143 const { logRecord } = yield* getLogRecord({ logRecordPointer: ptr });
144 return yield* HttpServerResponse.json({
145 message: logRecord?.["@message"] ?? null,
146 });
147 }
148
149 if (request.method === "GET" && pathname === "/stop-query") {
150 const now = yield* Clock.currentTimeMillis;
151 const { queryId } = yield* startQuery({
152 queryString: "fields @timestamp, @message | limit 10000",
153 startTime: Math.floor(now / 1000) - 86_400,
154 endTime: Math.floor(now / 1000) + 60,
155 });
156 if (!queryId) {
157 return yield* HttpServerResponse.json(
158 { error: "no queryId" },
159 { status: 500 },
160 );
161 }
162 // Give the query registration a beat to propagate before stopping.
163 yield* Effect.sleep("500 millis");
164 const outcome = yield* stopQuery({ queryId }).pipe(
165 Effect.map((response) => ({
166 stopped: response.success ?? true,
167 error: null as string | null,
168 })),
169 // Benign races: the query can complete before the stop lands
170 // (InvalidParameterException) or the stop can outrun the query's
171 // registration (ResourceNotFoundException). Both still prove the
172 // binding round-tripped with valid credentials.
173 Effect.catchTag(
174 ["InvalidParameterException", "ResourceNotFoundException"],
175 (error) => Effect.succeed({ stopped: false, error: error._tag }),
176 ),
177 );
178 return yield* HttpServerResponse.json({ ok: true, ...outcome });
179 }
180
181 if (request.method === "GET" && pathname === "/fields") {
182 const { logGroupFields } = yield* getLogGroupFields();
183 return yield* HttpServerResponse.json({
184 fields: (logGroupFields ?? []).map((field) => field.name),
185 });
186 }
187
188 if (request.method === "GET" && pathname === "/streams") {
189 const { logStreams } = yield* describeLogStreams({ limit: 10 });
190 return yield* HttpServerResponse.json({
191 streams: (logStreams ?? []).map((stream) => stream.logStreamName),
192 });
193 }
194
195 if (request.method === "POST" && pathname === "/stream-lifecycle") {
196 const name = url.searchParams.get("name");
197 if (!name) {
198 return yield* HttpServerResponse.json(
199 { error: "missing name" },
200 { status: 400 },
201 );
202 }
203 yield* createLogStream({ logStreamName: name }).pipe(
204 Effect.catchTag(
205 "ResourceAlreadyExistsException",
206 () => Effect.void,
207 ),
208 );
209 const { logStreams } = yield* describeLogStreams({
210 logStreamNamePrefix: name,
211 });
212 const seen = (logStreams ?? []).some(
213 (stream) => stream.logStreamName === name,
214 );
215 yield* deleteLogStream({ logStreamName: name }).pipe(
216 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
217 );
218 const after = yield* describeLogStreams({
219 logStreamNamePrefix: name,
220 });
221 const gone = !(after.logStreams ?? []).some(
222 (stream) => stream.logStreamName === name,
223 );
224 return yield* HttpServerResponse.json({ seen, gone });
225 }
226
227 return yield* HttpServerResponse.json(
228 { error: "Not found", method: request.method, pathname },
229 { status: 404 },
230 );
231 }).pipe(Effect.orDie),
232 };
233 }).pipe(
234 Effect.provide(
235 Layer.mergeAll(
236 Logs.PutLogEventsHttp,
237 Logs.FilterLogEventsHttp,
238 Logs.GetLogEventsHttp,
239 Logs.StartQueryHttp,
240 Logs.GetQueryResultsHttp,
241 Logs.StopQueryHttp,
242 Logs.GetLogRecordHttp,
243 Logs.GetLogGroupFieldsHttp,
244 Logs.DescribeLogStreamsHttp,
245 Logs.CreateLogStreamHttp,
246 Logs.DeleteLogStreamHttp,
247 ),
248 ),
249 ),
250);