1import * as ce from "@distilled.cloud/aws/cost-explorer";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { createPhysicalName } from "../../PhysicalName.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
9import type { Providers } from "../Providers.ts";
10import { fetchCeTags, pinCe, syncCeTags, toResourceTags } from "./common.ts";
11
12/**
13 * A subscriber notified when a monitored anomaly crosses the subscription's
14 * threshold.
15 */
16export interface AnomalySubscriber {
17 /**
18 * The destination — an email address (for `EMAIL`) or an SNS topic ARN
19 * (for `SNS`).
20 */
21 address: string;
22 /**
23 * How the subscriber is notified. `IMMEDIATE` frequency requires `SNS`
24 * subscribers; `DAILY`/`WEEKLY` require `EMAIL`.
25 */
26 type: "EMAIL" | "SNS" | (string & {});
27}
28
29export interface AnomalySubscriptionProps {
30 /**
31 * Name of the subscription. If omitted, a unique name is generated from
32 * the app, stage, and logical ID. Renaming updates the subscription in
33 * place.
34 */
35 subscriptionName?: string;
36 /**
37 * ARNs of the anomaly monitors this subscription listens to. Pass
38 * `monitor.monitorArn` outputs from {@link AnomalyMonitor} resources.
39 */
40 monitorArnList: string[];
41 /**
42 * Who gets notified when an anomaly crosses the threshold.
43 */
44 subscribers: AnomalySubscriber[];
45 /**
46 * How often notifications are sent. `IMMEDIATE` requires SNS subscribers;
47 * `DAILY` and `WEEKLY` send email digests.
48 */
49 frequency: "DAILY" | "IMMEDIATE" | "WEEKLY" | (string & {});
50 /**
51 * An expression gating notifications on anomaly impact — e.g. only alert
52 * when the total absolute impact exceeds $100:
53 * `{ Dimensions: { Key: "ANOMALY_TOTAL_IMPACT_ABSOLUTE", MatchOptions: ["GREATER_THAN_OR_EQUAL"], Values: ["100"] } }`
54 * (raw Cost Explorer `Expression` shape).
55 */
56 thresholdExpression?: ce.Expression;
57 /**
58 * User-defined tags to apply to the subscription.
59 */
60 tags?: Record<string, string>;
61}
62
63export interface AnomalySubscription extends Resource<
64 "AWS.CostExplorer.AnomalySubscription",
65 AnomalySubscriptionProps,
66 {
67 /** ARN of the anomaly subscription. */
68 subscriptionArn: string;
69 /** Name of the anomaly subscription. */
70 subscriptionName: string;
71 /** Account ID the subscription belongs to. */
72 accountId: string | undefined;
73 /** Current tags on the subscription. */
74 tags: Record<string, string>;
75 },
76 never,
77 Providers
78> {}
79
80/**
81 * A Cost Explorer anomaly alert subscription. Attaches email or SNS
82 * subscribers to one or more {@link AnomalyMonitor}s with a notification
83 * frequency and an impact threshold.
84 *
85 * Cost Explorer is a global service — all calls are pinned to `us-east-1`
86 * regardless of the stack region. Every property is mutable in place.
87 *
88 * ### Creating Anomaly Subscriptions
89 * **Example:** Daily email digest for anomalies over $100
90 * ```typescript
91 * import * as CostExplorer from "alchemy/AWS/CostExplorer";
92 *
93 * const monitor = yield* CostExplorer.AnomalyMonitor("ServiceSpend", {
94 * monitorType: "DIMENSIONAL",
95 * monitorDimension: "SERVICE",
96 * });
97 *
98 * const subscription = yield* CostExplorer.AnomalySubscription("Alerts", {
99 * monitorArnList: [monitor.monitorArn],
100 * frequency: "DAILY",
101 * subscribers: [{ type: "EMAIL", address: "team@example.com" }],
102 * thresholdExpression: {
103 * Dimensions: {
104 * Key: "ANOMALY_TOTAL_IMPACT_ABSOLUTE",
105 * MatchOptions: ["GREATER_THAN_OR_EQUAL"],
106 * Values: ["100"],
107 * },
108 * },
109 * });
110 * ```
111 *
112 * **Example:** Immediate SNS notifications
113 * ```typescript
114 * const subscription = yield* CostExplorer.AnomalySubscription("PagerFeed", {
115 * monitorArnList: [monitor.monitorArn],
116 * frequency: "IMMEDIATE",
117 * subscribers: [{ type: "SNS", address: topic.topicArn }],
118 * });
119 * ```
120 *
121 * @resource
122 */
123export const AnomalySubscription = Resource<AnomalySubscription>(
124 "AWS.CostExplorer.AnomalySubscription",
125);
126
127const sameStringSets = (l: readonly string[], r: readonly string[]) =>
128 l.length === r.length &&
129 [...l].sort().join("\n") === [...r].sort().join("\n");
130
131export const AnomalySubscriptionProvider = () =>
132 Provider.effect(
133 AnomalySubscription,
134 Effect.gen(function* () {
135 const createName = Effect.fn(function* (
136 id: string,
137 props: { subscriptionName?: string | undefined },
138 ) {
139 return (
140 props.subscriptionName ??
141 (yield* createPhysicalName({ id, maxLength: 100 }))
142 );
143 });
144
145 const getByArn = (subscriptionArn: string) =>
146 pinCe(
147 ce.getAnomalySubscriptions({
148 SubscriptionArnList: [subscriptionArn],
149 }),
150 ).pipe(
151 Effect.map((r) => r.AnomalySubscriptions[0]),
152 Effect.catchTag("UnknownSubscriptionException", () =>
153 Effect.succeed(undefined),
154 ),
155 );
156
157 const findByName = (subscriptionName: string) =>
158 pinCe(
159 ce.getAnomalySubscriptions.items({}).pipe(
160 Stream.filter((s) => s.SubscriptionName === subscriptionName),
161 Stream.take(1),
162 Stream.runCollect,
163 ),
164 ).pipe(Effect.map((chunk) => Array.from(chunk)[0]));
165
166 const toAttrs = Effect.fn(function* (live: ce.AnomalySubscription) {
167 const subscriptionArn = live.SubscriptionArn!;
168 return {
169 subscriptionArn,
170 subscriptionName: live.SubscriptionName,
171 accountId: live.AccountId,
172 tags: yield* fetchCeTags(subscriptionArn),
173 };
174 });
175
176 const toSubscribers = (
177 subscribers: AnomalySubscriber[],
178 ): ce.Subscriber[] =>
179 subscribers.map((s) => ({ Address: s.address, Type: s.type }));
180
181 return AnomalySubscription.Provider.of({
182 stables: ["subscriptionArn", "accountId"],
183 list: () =>
184 Effect.gen(function* () {
185 const subscriptions = yield* pinCe(
186 ce.getAnomalySubscriptions.items({}).pipe(Stream.runCollect),
187 ).pipe(Effect.map((chunk) => Array.from(chunk)));
188 return yield* Effect.forEach(
189 subscriptions.filter((s) => s.SubscriptionArn !== undefined),
190 toAttrs,
191 { concurrency: 10 },
192 );
193 }),
194 read: Effect.fn(function* ({ id, olds, output }) {
195 const live = output?.subscriptionArn
196 ? yield* getByArn(output.subscriptionArn)
197 : yield* findByName(yield* createName(id, olds ?? {}));
198 if (live?.SubscriptionArn === undefined) return undefined;
199 const attrs = yield* toAttrs(live);
200 return (yield* hasAlchemyTags(id, attrs.tags))
201 ? attrs
202 : Unowned(attrs);
203 }),
204 reconcile: Effect.fn(function* ({ id, news, output, session }) {
205 const name = yield* createName(id, news);
206 const internalTags = yield* createInternalTags(id);
207 const desiredTags: Record<string, string> = {
208 ...news.tags,
209 ...internalTags,
210 };
211 const desiredSubscribers = toSubscribers(news.subscribers);
212
213 // OBSERVE — cloud state is authoritative.
214 const live = output?.subscriptionArn
215 ? yield* getByArn(output.subscriptionArn)
216 : yield* findByName(name);
217
218 let subscriptionArn: string;
219 if (live?.SubscriptionArn === undefined) {
220 // ENSURE — create if missing. Subscription names are unique per
221 // account; tolerate the AlreadyExists race by adopting the
222 // same-name subscription and converging it (every property is
223 // mutable via UpdateAnomalySubscription).
224 const created = yield* pinCe(
225 ce.createAnomalySubscription({
226 AnomalySubscription: {
227 SubscriptionName: name,
228 MonitorArnList: news.monitorArnList,
229 Subscribers: desiredSubscribers,
230 Frequency: news.frequency,
231 ThresholdExpression: news.thresholdExpression,
232 },
233 ResourceTags: toResourceTags(desiredTags),
234 }),
235 ).pipe(
236 Effect.catchTag("AnomalySubscriptionAlreadyExists", (error) =>
237 Effect.gen(function* () {
238 const existing = yield* findByName(name);
239 const existingArn = existing?.SubscriptionArn;
240 if (existingArn === undefined) {
241 return yield* Effect.fail(error);
242 }
243 return yield* pinCe(
244 ce.updateAnomalySubscription({
245 SubscriptionArn: existingArn,
246 SubscriptionName: name,
247 MonitorArnList: news.monitorArnList,
248 Subscribers: desiredSubscribers,
249 Frequency: news.frequency,
250 ThresholdExpression: news.thresholdExpression,
251 }),
252 );
253 }),
254 ),
255 );
256 subscriptionArn = created.SubscriptionArn;
257 } else {
258 // SYNC — diff observed against desired; every field is mutable.
259 subscriptionArn = live.SubscriptionArn;
260 const observed = live;
261 const needsUpdate =
262 observed.SubscriptionName !== name ||
263 observed.Frequency !== news.frequency ||
264 !sameStringSets(observed.MonitorArnList, news.monitorArnList) ||
265 !sameStringSets(
266 observed.Subscribers.map((s) => `${s.Type}:${s.Address}`),
267 desiredSubscribers.map((s) => `${s.Type}:${s.Address}`),
268 ) ||
269 (news.thresholdExpression !== undefined &&
270 JSON.stringify(observed.ThresholdExpression) !==
271 JSON.stringify(news.thresholdExpression));
272 if (needsUpdate) {
273 yield* pinCe(
274 ce.updateAnomalySubscription({
275 SubscriptionArn: subscriptionArn,
276 SubscriptionName: name,
277 MonitorArnList: news.monitorArnList,
278 Subscribers: desiredSubscribers,
279 Frequency: news.frequency,
280 ThresholdExpression: news.thresholdExpression,
281 }),
282 );
283 }
284 }
285
286 // SYNC TAGS — diff against observed cloud tags.
287 yield* syncCeTags(subscriptionArn, desiredTags);
288
289 yield* session.note(subscriptionArn);
290 const final = yield* getByArn(subscriptionArn);
291 return {
292 subscriptionArn,
293 subscriptionName: name,
294 accountId: final?.AccountId,
295 tags: desiredTags,
296 };
297 }),
298 delete: Effect.fn(function* ({ output }) {
299 yield* pinCe(
300 ce.deleteAnomalySubscription({
301 SubscriptionArn: output.subscriptionArn,
302 }),
303 ).pipe(
304 Effect.catchTag("UnknownSubscriptionException", () => Effect.void),
305 );
306 }),
307 });
308 }),
309 );