1import * as AWS from "@/AWS";
2import * as Test from "@/Test/Alchemy";
3import * as cloudwatch from "@distilled.cloud/aws/cloudwatch";
4import * as lambda from "@distilled.cloud/aws/lambda";
5import { expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import * as HttpBody from "effect/http/HttpBody";
10import * as HttpClient from "effect/http/HttpClient";
11import {
12 MetricSinkFunction,
13 MetricSinkFunctionLive,
14 SINK_METRIC_NAME,
15 SINK_NAMESPACE,
16} from "./metric-sink-handler";
17
18const { test } = Test.make({ providers: AWS.providers() });
19
20class FunctionNotReady extends Data.TaggedError("FunctionNotReady") {}
21
22class MetricsNotVisible extends Data.TaggedError("MetricsNotVisible")<{
23 readonly observed: number;
24 readonly expected: number;
25}> {}
26
27const waitForFunctionReady = (url: string) =>
28 HttpClient.get(url).pipe(
29 Effect.flatMap((response) =>
30 response.status === 200
31 ? Effect.void
32 : Effect.fail(new FunctionNotReady()),
33 ),
34 Effect.retry({
35 while: (error) => error._tag === "FunctionNotReady",
36 schedule: Schedule.max([
37 Schedule.fixed("2 seconds"),
38 Schedule.recurs(75),
39 ]),
40 }),
41 );
42
43/**
44 * Out-of-band verification: poll GetMetricStatistics until the summed
45 * `Sum` across datapoints for the run's unique dimension reaches the
46 * published total. Standard-resolution data is typically queryable within
47 * seconds to ~2 minutes of PutMetricData; the poll is bounded at ~3 min.
48 */
49const waitForMetricSum = Effect.fn(function* (
50 runId: string,
51 expected: number,
52 startTime: Date,
53) {
54 return yield* Effect.gen(function* () {
55 const stats = yield* cloudwatch.getMetricStatistics({
56 Namespace: SINK_NAMESPACE,
57 MetricName: SINK_METRIC_NAME,
58 Dimensions: [{ Name: "Run", Value: runId }],
59 StartTime: startTime,
60 EndTime: new Date(Date.now() + 60_000),
61 Period: 60,
62 Statistics: ["Sum", "SampleCount"],
63 });
64 const observed = (stats.Datapoints ?? []).reduce(
65 (total, point) => total + (point.Sum ?? 0),
66 0,
67 );
68 if (observed < expected) {
69 return yield* Effect.fail(new MetricsNotVisible({ observed, expected }));
70 }
71 return observed;
72 }).pipe(
73 Effect.retry({
74 while: (e) => e._tag === "MetricsNotVisible",
75 schedule: Schedule.max([
76 Schedule.fixed("5 seconds"),
77 Schedule.recurs(36),
78 ]),
79 }),
80 );
81});
82
83test.provider(
84 "MetricSink batches >1000 datums through a deployed Lambda",
85 (stack) =>
86 Effect.gen(function* () {
87 // Leading destroy reconciles away any prior partial/crashed deployment.
88 yield* stack.destroy();
89
90 const fn = yield* stack.deploy(
91 MetricSinkFunction.pipe(Effect.provide(MetricSinkFunctionLive)),
92 );
93 const baseUrl = fn.functionUrl!.replace(/\/+$/, "");
94
95 yield* waitForFunctionReady(`${baseUrl}/ready`);
96
97 // 1500 datums > the PutMetricData limit of 1000 per call, so the
98 // batched sink must split the stream into 2 sequential API calls
99 // (1000 + 500). PutMetricData is all-or-nothing, so Sum === 1500
100 // proves both calls landed.
101 const runId = crypto.randomUUID();
102 const count = 1500;
103 const startTime = new Date(Date.now() - 5 * 60_000);
104
105 const response = yield* HttpClient.post(`${baseUrl}/sink`, {
106 body: yield* HttpBody.json({ runId, count }),
107 }).pipe(
108 Effect.flatMap((result) =>
109 result.status === 200
110 ? Effect.succeed(result)
111 : Effect.fail("not ready"),
112 ),
113 // Fresh function URLs can serve transient non-200s while IAM and
114 // DNS propagate — retry only that window.
115 Effect.retry({
116 while: (error) => error === "not ready",
117 schedule: Schedule.max([
118 Schedule.fixed("2 seconds"),
119 Schedule.recurs(30),
120 ]),
121 }),
122 Effect.flatMap((result) => result.json),
123 );
124
125 expect((response as any).ok).toBe(true);
126 expect((response as any).count).toBe(count);
127
128 const observed = yield* waitForMetricSum(runId, count, startTime);
129 expect(observed).toBe(count);
130
131 yield* stack.destroy();
132
133 // Out-of-band assert-gone: the deployed Lambda no longer exists after
134 // the final destroy (custom metrics themselves are not deletable and
135 // age out on their own).
136 const gone = yield* lambda
137 .getFunction({ FunctionName: fn.functionName })
138 .pipe(
139 Effect.map(() => false),
140 Effect.catchTag("ResourceNotFoundException", () =>
141 Effect.succeed(true),
142 ),
143 );
144 expect(gone).toBe(true);
145 }),
146 // Deploy (~60-120s) + readiness poll (bounded ~150s) + metric visibility
147 // poll (bounded ~180s) + destroy. All waits are bounded.
148 {
149 tags: [
150 "provider:aws",
151 "provider:aws:cloudwatch",
152 "provider:aws:lambda",
153 "live",
154 ],
155 timeout: 420_000,
156 },
157);