1import * as AWS from "@/AWS";
2import * as Console from "effect/Console";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Stream from "effect/Stream";
6import { HttpServerRequest } from "effect/http/HttpServerRequest";
7import * as HttpServerResponse from "effect/http/HttpServerResponse";
8import * as Layer from "effect/Layer";
9import path from "pathe";
10
11const main = path.resolve(import.meta.dirname, "metric-sink-handler.ts");
12
13/**
14 * Namespace + metric the sink fixture publishes under. The test isolates
15 * runs from each other with a unique `Run` dimension value per invocation.
16 */
17export const SINK_NAMESPACE = "Alchemy/MetricSinkTest";
18export const SINK_METRIC_NAME = "MetricSinkTestMetric";
19
20export class MetricSinkFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
21 "MetricSinkFunction",
22) {}
23
24export const MetricSinkFunctionLive = MetricSinkFunction.make(
25 {
26 main,
27 functionUrl: true,
28 // Streaming >1000 datums makes 2 sequential PutMetricData calls; give
29 // cold starts + both calls comfortable headroom over Lambda's 3s default.
30 timeout: Duration.seconds(30),
31 },
32 Effect.gen(function* () {
33 const sink = yield* AWS.CloudWatch.MetricSink({
34 Namespace: SINK_NAMESPACE,
35 });
36
37 return {
38 fetch: Effect.gen(function* () {
39 const request = yield* HttpServerRequest;
40 const pathname = new URL(request.originalUrl).pathname;
41
42 if (request.method === "GET" && pathname === "/ready") {
43 return yield* HttpServerResponse.json({ ok: true });
44 }
45
46 if (request.method === "POST" && pathname === "/sink") {
47 const body = (yield* request.json) as {
48 runId: string;
49 count: number;
50 };
51
52 // Raw MetricDatum entries — the caller owns the shape; the sink
53 // only packs/batches. >1000 datums forces a 1000 + remainder split.
54 yield* Stream.fromIterable(
55 Array.from(
56 { length: body.count },
57 () =>
58 ({
59 MetricName: SINK_METRIC_NAME,
60 Dimensions: [{ Name: "Run", Value: body.runId }],
61 Value: 1,
62 Unit: "Count",
63 }) satisfies AWS.CloudWatch.MetricSinkDatum,
64 ),
65 ).pipe(Stream.run(sink));
66
67 return yield* HttpServerResponse.json({
68 ok: true,
69 count: body.count,
70 });
71 }
72
73 return yield* HttpServerResponse.json(
74 { error: "Not found", method: request.method, pathname },
75 { status: 404 },
76 );
77 }).pipe(
78 Effect.tapError(Console.log),
79 Effect.catch((error) =>
80 Effect.succeed(
81 HttpServerResponse.text(`Internal server error: ${String(error)}`, {
82 status: 500,
83 }),
84 ),
85 ),
86 ),
87 };
88 }).pipe(
89 Effect.provide(
90 Layer.provideMerge(
91 AWS.CloudWatch.MetricSinkHttp,
92 AWS.CloudWatch.PutMetricDataHttp,
93 ),
94 ),
95 ),
96);
97export default MetricSinkFunctionLive;