1import * as oam from "@distilled.cloud/aws/oam";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { hasAlchemyTags } from "../../Tags.ts";
10import type { Providers } from "../Providers.ts";
11import {
12 deleteSinkAndWait,
13 readOamTags,
14 retryOamMutation,
15 syncOamTags,
16} from "./internal.ts";
17
18export interface SinkProps {
19 /**
20 * Name of the sink. If omitted, a unique name is generated from the app,
21 * stage and logical ID. Changing the name replaces the sink.
22 */
23 sinkName?: string;
24
25 /**
26 * The IAM resource policy that grants source accounts permission to link
27 * to this sink (`oam:CreateLink` / `oam:UpdateLink`). Provided as a JSON
28 * string or a plain policy document object.
29 *
30 * OAM has no delete-sink-policy API, so removing this prop leaves the
31 * last applied policy in place.
32 */
33 policy?: string | Record<string, any>;
34
35 /**
36 * User tags to attach to the sink. Merged with internal Alchemy tags.
37 */
38 tags?: Record<string, string>;
39}
40
41export interface Sink extends Resource<
42 "AWS.OAM.Sink",
43 SinkProps,
44 {
45 /** The name of the sink. */
46 sinkName: string;
47 /** The ARN of the sink (used as the `sinkIdentifier` of a `Link`). */
48 sinkArn: string;
49 /** The random ID string that AWS generated as part of the sink ARN. */
50 sinkId: string;
51 },
52 never,
53 Providers
54> {}
55
56/**
57 * A CloudWatch cross-account observability **sink** — the attachment point
58 * in a monitoring account that source accounts link to in order to share
59 * metrics, logs, traces, and Application Signals data.
60 *
61 * Each account can contain **one sink per region**. After creating a sink,
62 * attach a sink policy (the `policy` prop) that authorizes source accounts
63 * (or an entire organization) to create links to it.
64 *
65 * ### Creating a Sink
66 * **Example:** Basic Sink
67 * ```typescript
68 * import * as OAM from "alchemy/AWS/OAM";
69 *
70 * const sink = yield* OAM.Sink("MonitoringSink");
71 * ```
72 *
73 * **Example:** Sink with a policy authorizing source accounts
74 * ```typescript
75 * const sink = yield* OAM.Sink("MonitoringSink", {
76 * policy: {
77 * Version: "2012-10-17",
78 * Statement: [
79 * {
80 * Effect: "Allow",
81 * Principal: { AWS: ["111122223333"] },
82 * Action: ["oam:CreateLink", "oam:UpdateLink"],
83 * Resource: "*",
84 * Condition: {
85 * "ForAllValues:StringEquals": {
86 * "oam:ResourceTypes": [
87 * "AWS::CloudWatch::Metric",
88 * "AWS::Logs::LogGroup",
89 * ],
90 * },
91 * },
92 * },
93 * ],
94 * },
95 * });
96 * ```
97 *
98 * **Example:** Authorize an entire organization
99 * ```typescript
100 * const sink = yield* OAM.Sink("OrgSink", {
101 * policy: {
102 * Version: "2012-10-17",
103 * Statement: [
104 * {
105 * Effect: "Allow",
106 * Principal: "*",
107 * Action: ["oam:CreateLink", "oam:UpdateLink"],
108 * Resource: "*",
109 * Condition: {
110 * "ForAnyValue:StringEquals": { "aws:PrincipalOrgID": "o-xxxxxxxxxx" },
111 * },
112 * },
113 * ],
114 * },
115 * });
116 * ```
117 *
118 * @resource
119 */
120export const Sink = Resource<Sink>("AWS.OAM.Sink");
121
122/** Normalize a `policy` prop to a JSON string for `putSinkPolicy`. */
123const toPolicyString = (policy: string | Record<string, any>): string =>
124 typeof policy === "string" ? policy : JSON.stringify(policy);
125
126/**
127 * Semantic policy equality — AWS normalizes documents (e.g. account-ID
128 * principals become `arn:aws:iam::…:root`), so a raw string compare would
129 * re-put an unchanged policy forever. We compare parsed JSON; a normalized
130 * difference still re-puts, which is idempotent and harmless.
131 */
132const samePolicy = (a: string | undefined, b: string | undefined): boolean => {
133 if (a === undefined || b === undefined) return a === b;
134 try {
135 return JSON.stringify(JSON.parse(a)) === JSON.stringify(JSON.parse(b));
136 } catch {
137 return a === b;
138 }
139};
140
141export const SinkProvider = () =>
142 Provider.effect(
143 Sink,
144 Effect.gen(function* () {
145 // `props` may be undefined at runtime — all SinkProps fields are
146 // optional, so callers can omit the props object entirely.
147 const createName = Effect.fn(function* (
148 id: string,
149 props: SinkProps | undefined,
150 ) {
151 return props?.sinkName ?? (yield* createPhysicalName({ id }));
152 });
153
154 // Find a sink by name via the (small — quota is 1/region) list API.
155 const findByName = Effect.fn(function* (sinkName: string) {
156 const pages = yield* oam.listSinks.pages({}).pipe(Stream.runCollect);
157 return Array.from(pages)
158 .flatMap((page) => page.Items)
159 .find((item) => item.Name === sinkName);
160 });
161
162 return Sink.Provider.of({
163 stables: ["sinkName", "sinkArn", "sinkId"],
164 list: () =>
165 oam.listSinks.pages({}).pipe(
166 Stream.runCollect,
167 Effect.map((chunk) =>
168 Array.from(chunk).flatMap((page) =>
169 page.Items.filter(
170 (item) =>
171 item.Name != null && item.Arn != null && item.Id != null,
172 ).map((item) => ({
173 sinkName: item.Name!,
174 sinkArn: item.Arn!,
175 sinkId: item.Id!,
176 })),
177 ),
178 ),
179 ),
180 read: Effect.fn(function* ({ id, olds, output }) {
181 const sinkName =
182 output?.sinkName ?? (yield* createName(id, olds ?? {}));
183 const found = output?.sinkArn
184 ? yield* oam
185 .getSink({ Identifier: output.sinkArn })
186 .pipe(
187 Effect.catchTag("ResourceNotFoundException", () =>
188 Effect.succeed(undefined),
189 ),
190 )
191 : yield* findByName(sinkName);
192 if (!found?.Arn) return undefined;
193 const attrs = {
194 sinkName: found.Name ?? sinkName,
195 sinkArn: found.Arn,
196 sinkId: found.Id!,
197 };
198 const tags = yield* readOamTags(found.Arn);
199 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
200 }),
201 diff: Effect.fn(function* ({ id, news, olds }) {
202 if (!isResolved(news)) return undefined;
203 const oldName = yield* createName(id, olds);
204 const newName = yield* createName(id, news);
205 if (oldName !== newName) return { action: "replace" } as const;
206 return undefined;
207 }),
208 reconcile: Effect.fn(function* ({ id, news, output, session }) {
209 const sinkName = output?.sinkName ?? (yield* createName(id, news));
210
211 // OBSERVE — cloud state is authoritative; `output` is only an
212 // identifier cache. A cached ARN that no longer resolves falls
213 // through to "missing".
214 let live = output?.sinkArn
215 ? yield* oam
216 .getSink({ Identifier: output.sinkArn })
217 .pipe(
218 Effect.catchTag("ResourceNotFoundException", () =>
219 Effect.succeed(undefined),
220 ),
221 )
222 : yield* findByName(sinkName);
223
224 // ENSURE — create when missing; a ConflictException means a peer
225 // created the same-named sink concurrently, so re-observe.
226 if (live?.Arn == null) {
227 live = yield* retryOamMutation(
228 oam.createSink({ Name: sinkName, Tags: news?.tags }),
229 ).pipe(
230 Effect.catchTag("ConflictException", () => findByName(sinkName)),
231 );
232 }
233 const sinkArn = live!.Arn!;
234 const sinkId = live!.Id!;
235
236 // SYNC policy — diff the OBSERVED sink policy against the desired
237 // document and put only on change. There is no delete-policy API,
238 // so an absent `policy` prop leaves the current policy untouched.
239 const desiredPolicy = news?.policy;
240 if (desiredPolicy !== undefined) {
241 const desired = toPolicyString(desiredPolicy);
242 const observed = yield* oam
243 .getSinkPolicy({ SinkIdentifier: sinkArn })
244 .pipe(
245 Effect.map((r) => r.Policy),
246 Effect.catchTag("ResourceNotFoundException", () =>
247 Effect.succeed(undefined),
248 ),
249 );
250 if (!samePolicy(observed, desired)) {
251 yield* retryOamMutation(
252 oam.putSinkPolicy({
253 SinkIdentifier: sinkArn,
254 Policy: desired,
255 }),
256 );
257 }
258 }
259
260 // SYNC tags — against observed cloud tags (adoption-safe).
261 yield* syncOamTags(sinkArn, id, news?.tags);
262
263 yield* session.note(sinkArn);
264 return { sinkName, sinkArn, sinkId };
265 }),
266 delete: Effect.fn(function* ({ output }) {
267 yield* deleteSinkAndWait(output.sinkArn);
268 }),
269 });
270 }),
271 );