1import * as Lambda from "@/AWS/Lambda";
2import * as Logs from "@/AWS/Logs";
3import * as Clock from "effect/Clock";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Stream from "effect/Stream";
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, "sink-handler.ts");
13
14export const SINK_STREAM_NAME = "alchemy-test-log-event-sink-stream";
15
16export class LogEventSinkFunction extends Lambda.Function<Lambda.Function>()(
17 "LogEventSinkFunction",
18) {}
19
20export default LogEventSinkFunction.make(
21 {
22 main,
23 functionUrl: true,
24 timeout: Duration.seconds(30),
25 },
26 Effect.gen(function* () {
27 const logGroup = yield* Logs.LogGroup("SinkLogGroup", {
28 retention: "1 day",
29 });
30 yield* Logs.LogStream("SinkLogStream", {
31 logGroupName: logGroup.logGroupName,
32 logStreamName: SINK_STREAM_NAME,
33 });
34
35 const sink = yield* Logs.LogEventSink(logGroup, {
36 logStreamName: SINK_STREAM_NAME,
37 });
38 const LogGroupName = yield* logGroup.logGroupName;
39
40 return {
41 fetch: Effect.gen(function* () {
42 const request = yield* HttpServerRequest;
43 const url = new URL(request.originalUrl);
44 const pathname = url.pathname;
45
46 if (request.method === "GET" && pathname === "/ready") {
47 return yield* HttpServerResponse.json({
48 ok: true,
49 logGroupName: yield* LogGroupName,
50 logStreamName: SINK_STREAM_NAME,
51 });
52 }
53
54 if (request.method === "POST" && pathname === "/sink") {
55 const body = (yield* request.json) as { messages: string[] };
56 const start = yield* Clock.currentTimeMillis;
57 // PutLogEvents requires chronological order within a batch —
58 // ordering stays with the caller, so stamp increasing timestamps.
59 const events = body.messages.map((message, index) => ({
60 timestamp: start + index,
61 message,
62 }));
63
64 yield* Stream.fromIterable(events).pipe(Stream.run(sink));
65
66 return yield* HttpServerResponse.json({
67 ok: true,
68 count: body.messages.length,
69 });
70 }
71
72 if (request.method === "POST" && pathname === "/sink-with-rejected") {
73 const body = (yield* request.json) as {
74 valid: string[];
75 rejected: string;
76 };
77 const now = yield* Clock.currentTimeMillis;
78 // Valid events now, then one event 3 hours in the future — more
79 // than the 2-hour PutLogEvents window, so the API permanently
80 // rejects it (rejectedLogEventsInfo.tooNewLogEventStartIndex)
81 // while still ingesting the valid events. The sink must drop the
82 // rejected event (no retry) and complete successfully.
83 const events = [
84 ...body.valid.map((message, index) => ({
85 timestamp: now + index,
86 message,
87 })),
88 {
89 timestamp: now + 3 * 60 * 60 * 1000,
90 message: body.rejected,
91 },
92 ];
93
94 yield* Stream.fromIterable(events).pipe(Stream.run(sink));
95
96 return yield* HttpServerResponse.json({
97 ok: true,
98 count: events.length,
99 });
100 }
101
102 return yield* HttpServerResponse.json(
103 { error: "Not found", method: request.method, pathname },
104 { status: 404 },
105 );
106 }).pipe(Effect.orDie),
107 };
108 }).pipe(
109 Effect.provide(
110 Layer.provideMerge(Logs.LogEventSinkHttp, Logs.PutLogEventsHttp),
111 ),
112 ),
113);