1import * as AWS from "@/AWS";
2import { MetricStream } from "@/AWS/CloudWatch";
3import * as Provider from "@/Provider";
4import * as Test from "@/Test/Alchemy";
5import { AWSEnvironment } from "@/AWS/Environment";
6import * as cloudwatch from "@distilled.cloud/aws/cloudwatch";
7import * as firehose from "@distilled.cloud/aws/firehose";
8import * as iam from "@distilled.cloud/aws/iam";
9import * as s3 from "@distilled.cloud/aws/s3";
10import { expect } from "alchemy-test";
11import * as Effect from "effect/Effect";
12import * as Schedule from "effect/Schedule";
13
14const { test } = Test.make({ providers: AWS.providers() });
15
16// Deterministic out-of-band names (no Date.now — stable across re-runs).
17const suffix = "alchemy-test-metricstream-list";
18const bucketName = `${suffix}-bucket`;
19const firehoseRoleName = `${suffix}-fh-role`;
20const firehoseName = `${suffix}-fh`;
21const streamRoleName = `${suffix}-ms-role`;
22
23const trustPolicy = (service: string) =>
24 JSON.stringify({
25 Version: "2012-10-17",
26 Statement: [
27 {
28 Effect: "Allow",
29 Principal: { Service: service },
30 Action: "sts:AssumeRole",
31 },
32 ],
33 });
34
35// `list()` enumerates every CloudWatch metric stream in the account/region via
36// the paginated `cloudwatch.listMetricStreams` op, re-reading each entry with
37// `getMetricStream` to produce the full Attributes shape. A metric stream
38// requires a live Firehose delivery stream + an assumable IAM role, so we stand
39// up the Firehose (S3-backed) and both roles out-of-band via distilled, deploy a
40// real MetricStream, resolve the provider via the typed `findProvider`, call
41// `list()`, and assert the deployed stream appears in the exhaustively
42// paginated result.
43//
44// Gated off by default (AWS_TEST_METRICSTREAM=1): the live setup provisions a
45// Firehose delivery stream (CREATING -> ACTIVE can take a couple of minutes) and
46// depends on fresh-IAM-role propagation, which is too slow/flaky for default CI.
47test.provider.skipIf(!process.env.AWS_TEST_METRICSTREAM)(
48 "list enumerates the deployed metric stream",
49 (stack) =>
50 Effect.gen(function* () {
51 const { region } = yield* AWSEnvironment.current;
52
53 // Idempotent cleanup: every delete tolerates ONLY its typed not-found
54 // tag, so a real failure (e.g. a dependency violation) surfaces and
55 // fails the test instead of silently orphaning the resource.
56 const emptyBucket = Effect.gen(function* () {
57 const listed = yield* s3.listObjectsV2({ Bucket: bucketName });
58 const objects = (listed.Contents ?? []).flatMap((o) =>
59 o.Key ? [{ Key: o.Key }] : [],
60 );
61 if (objects.length > 0) {
62 yield* s3.deleteObjects({
63 Bucket: bucketName,
64 Delete: { Objects: objects },
65 });
66 }
67 return listed.IsTruncated === true;
68 }).pipe(Effect.repeat({ until: (truncated) => !truncated, times: 8 }));
69
70 const cleanup = Effect.gen(function* () {
71 yield* firehose
72 .deleteDeliveryStream({
73 DeliveryStreamName: firehoseName,
74 AllowForceDelete: true,
75 })
76 .pipe(
77 Effect.retry({
78 while: (e) => e._tag === "ResourceInUseException",
79 schedule: Schedule.spaced("5 seconds"),
80 times: 8,
81 }),
82 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
83 );
84 yield* iam
85 .deleteRolePolicy({
86 RoleName: firehoseRoleName,
87 PolicyName: "s3",
88 })
89 .pipe(Effect.catchTag("NoSuchEntityException", () => Effect.void));
90 yield* iam
91 .deleteRole({ RoleName: firehoseRoleName })
92 .pipe(Effect.catchTag("NoSuchEntityException", () => Effect.void));
93 yield* iam
94 .deleteRolePolicy({
95 RoleName: streamRoleName,
96 PolicyName: "firehose",
97 })
98 .pipe(Effect.catchTag("NoSuchEntityException", () => Effect.void));
99 yield* iam
100 .deleteRole({ RoleName: streamRoleName })
101 .pipe(Effect.catchTag("NoSuchEntityException", () => Effect.void));
102 // The Firehose may have delivered objects into the bucket — empty it
103 // first, and retry once more if a late delivery races the delete.
104 yield* emptyBucket.pipe(
105 Effect.andThen(s3.deleteBucket({ Bucket: bucketName })),
106 Effect.retry({
107 while: (e) => e._tag === "BucketNotEmpty",
108 schedule: Schedule.spaced("2 seconds"),
109 times: 3,
110 }),
111 Effect.catchTag("NoSuchBucket", () => Effect.void),
112 );
113 });
114
115 yield* stack.destroy();
116 yield* cleanup;
117
118 yield* Effect.gen(function* () {
119 // S3 destination bucket for the Firehose.
120 yield* s3
121 .createBucket({
122 Bucket: bucketName,
123 ...(region === "us-east-1"
124 ? {}
125 : {
126 CreateBucketConfiguration: {
127 LocationConstraint: region,
128 },
129 }),
130 } as any)
131 .pipe(
132 Effect.catchTag("BucketAlreadyOwnedByYou", () => Effect.void),
133 Effect.catchTag("BucketAlreadyExists", () => Effect.void),
134 );
135
136 const bucketArn = `arn:aws:s3:::${bucketName}`;
137
138 // Role CloudWatch assumes to write metrics into the Firehose. Created
139 // first so it has the longest time to propagate before use.
140 const streamRole = yield* iam.createRole({
141 RoleName: streamRoleName,
142 AssumeRolePolicyDocument: trustPolicy(
143 "streams.metrics.cloudwatch.amazonaws.com",
144 ),
145 });
146 const streamRoleArn = streamRole.Role.Arn;
147
148 // Role the Firehose assumes to write to S3.
149 const firehoseRole = yield* iam.createRole({
150 RoleName: firehoseRoleName,
151 AssumeRolePolicyDocument: trustPolicy("firehose.amazonaws.com"),
152 });
153 const firehoseRoleArn = firehoseRole.Role.Arn;
154
155 yield* iam.putRolePolicy({
156 RoleName: firehoseRoleName,
157 PolicyName: "s3",
158 PolicyDocument: JSON.stringify({
159 Version: "2012-10-17",
160 Statement: [
161 {
162 Effect: "Allow",
163 Action: [
164 "s3:AbortMultipartUpload",
165 "s3:GetBucketLocation",
166 "s3:GetObject",
167 "s3:ListBucket",
168 "s3:ListBucketMultipartUploads",
169 "s3:PutObject",
170 ],
171 Resource: [bucketArn, `${bucketArn}/*`],
172 },
173 ],
174 }),
175 });
176
177 // Create the Firehose, retrying while IAM role propagation makes the
178 // delivery role temporarily un-assumable.
179 const created = yield* firehose
180 .createDeliveryStream({
181 DeliveryStreamName: firehoseName,
182 DeliveryStreamType: "DirectPut",
183 S3DestinationConfiguration: {
184 RoleARN: firehoseRoleArn,
185 BucketARN: bucketArn,
186 },
187 })
188 .pipe(
189 Effect.catchTag("ResourceInUseException", () =>
190 firehose
191 .describeDeliveryStream({ DeliveryStreamName: firehoseName })
192 .pipe(
193 Effect.map((d) => ({
194 DeliveryStreamARN:
195 d.DeliveryStreamDescription.DeliveryStreamARN,
196 })),
197 ),
198 ),
199 Effect.retry({
200 while: (e) => e._tag === "InvalidArgumentException",
201 schedule: Schedule.max([
202 Schedule.spaced("5 seconds"),
203 Schedule.recurs(8),
204 ]),
205 }),
206 );
207
208 const firehoseArn = created.DeliveryStreamARN!;
209
210 // Wait for the Firehose to become ACTIVE before CloudWatch will accept it.
211 yield* firehose
212 .describeDeliveryStream({ DeliveryStreamName: firehoseName })
213 .pipe(
214 Effect.map((d) => d.DeliveryStreamDescription.DeliveryStreamStatus),
215 Effect.tap((status) =>
216 status === "ACTIVE"
217 ? Effect.void
218 : Effect.fail("not-active" as const),
219 ),
220 Effect.retry({
221 while: (e) => e === "not-active",
222 schedule: Schedule.max([
223 Schedule.spaced("10 seconds"),
224 Schedule.recurs(10),
225 ]),
226 }),
227 Effect.catch(() => Effect.void),
228 );
229
230 yield* iam.putRolePolicy({
231 RoleName: streamRoleName,
232 PolicyName: "firehose",
233 PolicyDocument: JSON.stringify({
234 Version: "2012-10-17",
235 Statement: [
236 {
237 Effect: "Allow",
238 Action: ["firehose:PutRecord", "firehose:PutRecordBatch"],
239 Resource: [firehoseArn],
240 },
241 ],
242 }),
243 });
244
245 const deployed = yield* stack.deploy(
246 Effect.gen(function* () {
247 return yield* MetricStream("ListMetricStream", {
248 name: "alchemy-test-metricstream-list",
249 FirehoseArn: firehoseArn,
250 RoleArn: streamRoleArn,
251 OutputFormat: "json",
252 });
253 }),
254 );
255
256 const provider = yield* Provider.findProvider(MetricStream);
257 const all = yield* provider.list();
258
259 expect(
260 all.some((ms) => ms.metricStreamName === deployed.metricStreamName),
261 ).toBe(true);
262
263 yield* stack.destroy();
264
265 // Out-of-band assert-gone: getMetricStream returns the typed
266 // ResourceNotFoundException once the stream is deleted.
267 const gone = yield* cloudwatch
268 .getMetricStream({ Name: deployed.metricStreamName })
269 .pipe(
270 Effect.map(() => false),
271 Effect.catchTag("ResourceNotFoundException", () =>
272 Effect.succeed(true),
273 ),
274 );
275 expect(gone).toBe(true);
276 }).pipe(Effect.ensuring(cleanup.pipe(Effect.orDie)));
277 }),
278 {
279 tags: [
280 "provider:aws",
281 "provider:aws:cloudwatch",
282 "provider:aws:firehose",
283 "provider:aws:iam",
284 "provider:aws:s3",
285 "live",
286 ],
287 timeout: 240_000,
288 },
289);