1import * as cloudwatch from "@distilled.cloud/aws/cloudwatch";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import type { Input } from "../../Input.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { hasAlchemyTags } from "../../Tags.ts";
10import { AWSEnvironment, type AccountID } from "../Environment.ts";
11import type { Providers } from "../Providers.ts";
12import type { RegionID } from "../Region.ts";
13import {
14 createName,
15 readResourceTags,
16 retryConcurrent,
17 updateResourceTags,
18} from "./common.ts";
19
20export type MetricStreamName = string;
21export type MetricStreamArn =
22 `arn:aws:cloudwatch:${RegionID}:${AccountID}:metric-stream/${string}`;
23
24export interface MetricStreamProps extends Omit<
25 cloudwatch.PutMetricStreamInput,
26 "Name" | "Tags"
27> {
28 /**
29 * Name of the metric stream. If omitted, a unique name is generated.
30 */
31 name?: MetricStreamName;
32 /**
33 * Whether the stream should be running after deployment.
34 * @default true
35 */
36 enabled?: boolean;
37 /**
38 * Optional tags to apply to the metric stream.
39 */
40 tags?: Record<string, string>;
41}
42
43export interface MetricStream extends Resource<
44 "AWS.CloudWatch.MetricStream",
45 MetricStreamProps,
46 {
47 /** Physical name of the metric stream. */
48 metricStreamName: MetricStreamName;
49 /** ARN of the metric stream. */
50 metricStreamArn: MetricStreamArn;
51 /** Current state of the stream (`running` or `stopped`). */
52 state: string | undefined;
53 /** The full metric stream description as last read from CloudWatch. */
54 metricStream: cloudwatch.GetMetricStreamOutput;
55 /** Tags on the metric stream, including the internal Alchemy ownership tags. */
56 tags: Record<string, string>;
57 },
58 never,
59 Providers
60> {}
61
62/**
63 * A CloudWatch metric stream — continuously exports CloudWatch metrics to
64 * a Kinesis Data Firehose delivery stream (and on to S3, Datadog, etc.).
65 * ### Creating Metric Streams
66 * **Example:** Firehose Delivery Stream
67 * ```typescript
68 * const stream = yield* MetricStream("MetricsExport", {
69 * FirehoseArn: "arn:aws:firehose:us-east-1:123456789012:deliverystream/example",
70 * RoleArn: "arn:aws:iam::123456789012:role/example",
71 * OutputFormat: "json",
72 * });
73 * ```
74 *
75 * **Example:** Stream Only Selected Namespaces
76 * ```typescript
77 * const stream = yield* MetricStream("LambdaMetricsExport", {
78 * FirehoseArn: firehose.deliveryStreamArn,
79 * RoleArn: role.roleArn,
80 * OutputFormat: "json",
81 * IncludeFilters: [{ Namespace: "AWS/Lambda" }],
82 * });
83 * ```
84 *
85 * ### Reading Metric Streams at Runtime
86 * **Example:** Read the Stream's State from a Function
87 * ```typescript
88 * // init — bind the stream to the function (see GetMetricStream)
89 * const getMetricStream = yield* AWS.CloudWatch.GetMetricStream(stream);
90 *
91 * // runtime
92 * const result = yield* getMetricStream();
93 * const state = result.State; // "running" | "stopped"
94 * ```
95 *
96 * @resource
97 */
98export const MetricStream = Resource<MetricStream>(
99 "AWS.CloudWatch.MetricStream",
100);
101
102export const MetricStreamProvider = () =>
103 Provider.effect(
104 MetricStream,
105 Effect.gen(function* () {
106 const createMetricStreamName = (
107 id: string,
108 props: { name?: string } = {},
109 ) => createName(id, props.name, 255);
110
111 const metricStreamArn = (name: string) =>
112 AWSEnvironment.current.pipe(
113 Effect.map(
114 (env) =>
115 `arn:aws:cloudwatch:${env.region}:${env.accountId}:metric-stream/${name}` as MetricStreamArn,
116 ),
117 );
118
119 const readMetricStream = Effect.fn(function* (name: string) {
120 const output = yield* cloudwatch
121 .getMetricStream({
122 Name: name,
123 })
124 .pipe(
125 Effect.catchTag("InvalidParameterValueException", () =>
126 Effect.succeed(undefined),
127 ),
128 Effect.catchTag("ResourceNotFoundException", () =>
129 Effect.succeed(undefined),
130 ),
131 );
132
133 if (!output?.Name || !output.Arn) {
134 return undefined;
135 }
136
137 const tags = yield* readResourceTags(output.Arn).pipe(
138 Effect.catchTag("ResourceNotFoundException", () =>
139 Effect.succeed({}),
140 ),
141 Effect.catchTag("InvalidParameterValueException", () =>
142 Effect.succeed({}),
143 ),
144 );
145
146 return {
147 metricStreamName: output.Name,
148 metricStreamArn: output.Arn as MetricStreamArn,
149 state: output.State,
150 metricStream: output,
151 tags,
152 };
153 });
154
155 const syncMetricStreamState = Effect.fn(function* ({
156 name,
157 enabled,
158 }: {
159 name: string;
160 enabled: boolean | undefined;
161 }) {
162 if (enabled === false) {
163 yield* retryConcurrent(
164 cloudwatch.stopMetricStreams({
165 Names: [name],
166 }),
167 );
168 return;
169 }
170
171 yield* retryConcurrent(
172 cloudwatch.startMetricStreams({
173 Names: [name],
174 }),
175 );
176 });
177
178 return {
179 stables: ["metricStreamName", "metricStreamArn"],
180 diff: Effect.fn(function* ({
181 id,
182 olds = {},
183 news = {} as Input<MetricStreamProps>,
184 }) {
185 if (!isResolved(news)) return undefined;
186 const oldName = yield* createMetricStreamName(id, olds);
187 const newName = yield* createMetricStreamName(id, news);
188
189 if (oldName !== newName) {
190 return { action: "replace" } as const;
191 }
192 }),
193 list: () =>
194 Effect.gen(function* () {
195 // Enumerate every metric stream in the account/region. The list
196 // op only returns summary entries, so re-read each by name to
197 // produce the full Attributes shape `read` returns.
198 const entries = yield* cloudwatch.listMetricStreams.pages({}).pipe(
199 Stream.runCollect,
200 Effect.map((chunk) =>
201 Array.from(chunk).flatMap((page) => page.Entries ?? []),
202 ),
203 );
204
205 const results = yield* Effect.forEach(
206 entries,
207 (entry) =>
208 entry.Name
209 ? readMetricStream(entry.Name)
210 : Effect.succeed(undefined),
211 { concurrency: 10 },
212 );
213
214 return results.filter(
215 (state): state is NonNullable<typeof state> =>
216 state !== undefined,
217 );
218 }),
219 read: Effect.fn(function* ({ id, olds, output }) {
220 const name =
221 output?.metricStreamName ??
222 (yield* createMetricStreamName(id, olds ?? {}));
223 const state = yield* readMetricStream(name);
224 if (!state) return undefined;
225 return (yield* hasAlchemyTags(id, state.tags))
226 ? state
227 : Unowned(state);
228 }),
229 reconcile: Effect.fn(function* ({ id, news, olds, output, session }) {
230 // Observe — pin the physical name from `output` if present;
231 // otherwise derive from desired props. Read existing so we have
232 // a baseline for tag-diffing on adoption.
233 const name =
234 output?.metricStreamName ??
235 (yield* createMetricStreamName(id, news));
236 const existing = yield* readMetricStream(name);
237
238 // Ensure — `putMetricStream` is an upsert; we send the full
239 // desired config every reconcile.
240 yield* retryConcurrent(
241 cloudwatch.putMetricStream({
242 ...news,
243 Name: name,
244 }),
245 );
246
247 // Sync running state — `enabled` drives start/stop independently
248 // of the put call.
249 yield* syncMetricStreamState({
250 name,
251 enabled: news.enabled,
252 });
253
254 // Sync tags — diff against `olds.tags` when we have prior state,
255 // otherwise fall back to what we observed (adoption path).
256 const tags = yield* updateResourceTags({
257 id,
258 resourceArn: yield* metricStreamArn(name),
259 olds: olds?.tags ?? existing?.tags,
260 news: news.tags,
261 });
262
263 yield* session.note(yield* metricStreamArn(name));
264
265 const state = yield* readMetricStream(name);
266 if (!state) {
267 return yield* Effect.fail(
268 new Error(`failed to read reconciled metric stream '${name}'`),
269 );
270 }
271
272 return {
273 ...state,
274 tags,
275 };
276 }),
277 delete: Effect.fn(function* ({ output }) {
278 const existing = yield* readMetricStream(output.metricStreamName);
279 if (!existing) {
280 return;
281 }
282
283 yield* retryConcurrent(
284 cloudwatch.deleteMetricStream({
285 Name: output.metricStreamName,
286 }),
287 ).pipe(
288 Effect.catchTag(
289 "InvalidParameterValueException",
290 () => Effect.void,
291 ),
292 );
293 }),
294 };
295 }),
296 );