1import type * as Effect from "effect/Effect";
2import type * as Stream from "effect/Stream";
3import {
4 consumeBusEvents,
5 type EventRecord,
6 type EventRouteProps,
7} from "../EventBridge/EventSource.ts";
8
9/**
10 * The `detail` payload Cost Anomaly Detection delivers to EventBridge when
11 * a monitor detects a cost anomaly. Fields mirror the `Anomaly` shape the
12 * `GetAnomalies` API returns; the schema grows over time so unknown fields
13 * are preserved.
14 */
15export interface AnomalyEventDetail {
16 /** Id of the detected anomaly (usable with `ProvideAnomalyFeedback`). */
17 anomalyId?: string;
18 /** First day the anomaly was observed (ISO timestamp). */
19 anomalyStartDate?: string;
20 /** Last day the anomaly was observed (ISO timestamp). */
21 anomalyEndDate?: string;
22 /** How anomalous the spend is. */
23 anomalyScore?: {
24 currentScore?: number;
25 maxScore?: number;
26 };
27 /** The dollar impact of the anomaly. */
28 impact?: {
29 maxImpact?: number;
30 totalActualSpend?: number;
31 totalExpectedSpend?: number;
32 totalImpact?: number;
33 totalImpactPercentage?: number;
34 };
35 /** The services / accounts / usage types driving the anomaly. */
36 rootCauses?: {
37 linkedAccount?: string;
38 linkedAccountName?: string;
39 region?: string;
40 service?: string;
41 usageType?: string;
42 impact?: { contribution?: number };
43 }[];
44 /** ARN of the monitor that detected the anomaly. */
45 monitorArn?: string;
46 /** Name of the monitor that detected the anomaly. */
47 monitorName?: string;
48 /** The monitored dimension's value (e.g. the service name). */
49 dimensionValue?: string;
50 /** 12-digit id of the account the anomaly was detected in. */
51 accountId?: string;
52 /** Name of the account the anomaly was detected in. */
53 accountName?: string;
54 /** Console deep link to the anomaly. */
55 anomalyDetailsLink?: string;
56 /** Additional event fields (the schema grows over time). */
57 [key: string]: unknown;
58}
59
60/** A Cost Anomaly Detection EventBridge event delivered to the handler. */
61export type AnomalyEvent = EventRecord<AnomalyEventDetail>;
62
63export interface AnomalyEventSourceProps extends EventRouteProps {
64 /**
65 * Logical id for the backing EventBridge rule.
66 * @default "CostAnomalyEvents"
67 */
68 id?: string;
69 /**
70 * Restrict to anomalies detected by specific monitors (matched against
71 * the event's top-level `resources`, which carries the monitor's ARN).
72 */
73 monitorArns?: readonly string[];
74}
75
76/**
77 * Event source connecting Cost Anomaly Detection to the hosting compute.
78 * When an {@link AnomalyMonitor} detects a cost anomaly, Cost Explorer
79 * publishes an `Anomaly Detected` event (source `aws.ce`, best-effort
80 * delivery) to the account's default EventBridge bus; this subscribes the
81 * host Function to those events so it can page, open tickets, or trigger
82 * automated cost-control actions the moment unusual spend appears.
83 *
84 * Cost Anomaly Detection publishes to EventBridge automatically — no
85 * additional resource is created besides the EventBridge rule targeting
86 * the host. Events are delivered to the Cost Anomaly Detection home region
87 * (`us-east-1`), so deploy the consuming stack there. Provide the
88 * host-specific implementation layer (e.g. `AWS.Lambda.EventSource`) on
89 * the Function effect.
90 *
91 * ### Consuming Anomaly Events
92 * **Example:** React To Detected Cost Anomalies
93 * ```typescript
94 * import * as AWS from "alchemy/AWS";
95 * import * as Stream from "effect/Stream";
96 *
97 * export default AlertFunction.make(
98 * { main: import.meta.url },
99 * Effect.gen(function* () {
100 * yield* AWS.CostExplorer.consumeAnomalyEvents({}, (events) =>
101 * Stream.runForEach(events, (event) =>
102 * Effect.log(
103 * `anomaly ${event.detail.anomalyId}: $${event.detail.impact?.totalImpact} on ${event.detail.dimensionValue}`,
104 * ),
105 * ),
106 * );
107 * return {};
108 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
109 * );
110 * ```
111 *
112 * **Example:** Only Anomalies From One Monitor
113 * ```typescript
114 * yield* AWS.CostExplorer.consumeAnomalyEvents(
115 * { monitorArns: ["arn:aws:ce::123456789012:anomalymonitor/abcd..."] },
116 * (events) =>
117 * Stream.runForEach(events, (event) => Effect.log(event.detail)),
118 * );
119 * ```
120 */
121export const consumeAnomalyEvents = <StreamReq = never, Req = never>(
122 props: AnomalyEventSourceProps,
123 process: (
124 events: Stream.Stream<AnomalyEvent, never, StreamReq>,
125 ) => Effect.Effect<void, never, Req>,
126) =>
127 consumeBusEvents(
128 props.id ?? "CostAnomalyEvents",
129 {
130 source: ["aws.ce"],
131 "detail-type": ["Anomaly Detected"],
132 ...(props.monitorArns !== undefined
133 ? { resources: [...props.monitorArns] }
134 : {}),
135 },
136 { description: props.description, state: props.state },
137 process,
138 );