1import * as oam from "@distilled.cloud/aws/oam";
2import * as Retry from "@distilled.cloud/aws/Retry";
3import * as Data from "effect/Data";
4import * as Effect from "effect/Effect";
5import * as Schedule from "effect/Schedule";
6import * as Stream from "effect/Stream";
7import { createInternalTags, diffTags } from "../../Tags.ts";
8
9const mutationSchedule = Schedule.max([
10 Schedule.spaced("1 second"),
11 Schedule.recurs(20),
12]);
13
14const isRetryableOamMutationError = (
15 error: unknown,
16 conflict: boolean,
17): boolean => {
18 if (!error || typeof error !== "object") return false;
19 const tag = (error as { _tag?: string })._tag;
20 return (
21 tag === "TooManyRequestsException" ||
22 tag === "ServiceQuotaExceededException" ||
23 tag === "InternalServiceFault" ||
24 (conflict && tag === "ConflictException")
25 );
26};
27
28/** One non-nested, bounded retry owner for OAM control-plane mutations. */
29export const retryOamMutation = <A, E, R>(
30 effect: Effect.Effect<A, E, R>,
31 options?: { conflict?: boolean },
32): Effect.Effect<A, E, Exclude<R, Retry.Retry>> =>
33 Retry.policy({
34 while: (error) =>
35 isRetryableOamMutationError(error, options?.conflict === true),
36 schedule: mutationSchedule,
37 })(effect);
38
39class SinkLinksStillAttached extends Data.TaggedError(
40 "SinkLinksStillAttached",
41)<{ readonly sinkArn: string; readonly count: number }> {}
42
43class OamResourceStillExists extends Data.TaggedError(
44 "OamResourceStillExists",
45)<{ readonly arn: string }> {}
46
47const dependencySchedule = Schedule.max([
48 Schedule.spaced("2 seconds"),
49 Schedule.recurs(10),
50]);
51
52const goneSchedule = Schedule.max([
53 Schedule.spaced("1 second"),
54 Schedule.recurs(10),
55]);
56
57/**
58 * Delete a sink only after attached links have drained, then observe absence.
59 */
60export const deleteSinkAndWait = Effect.fn(function* (sinkArn: string) {
61 yield* oam.listAttachedLinks.pages({ SinkIdentifier: sinkArn }).pipe(
62 Stream.runCollect,
63 Effect.map((pages) =>
64 Array.from(pages).reduce((count, page) => count + page.Items.length, 0),
65 ),
66 Effect.flatMap((count) =>
67 count === 0
68 ? Effect.void
69 : Effect.fail(new SinkLinksStillAttached({ sinkArn, count })),
70 ),
71 Effect.retry({
72 while: (error) => error._tag === "SinkLinksStillAttached",
73 schedule: dependencySchedule,
74 }),
75 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
76 );
77
78 yield* retryOamMutation(oam.deleteSink({ Identifier: sinkArn }), {
79 // A just-deleted link can remain visible to DeleteSink after the list has
80 // converged. Retry only this bounded dependency-violation window.
81 conflict: true,
82 }).pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void));
83
84 const observe = Retry.none(oam.getSink({ Identifier: sinkArn }));
85 yield* observe.pipe(
86 Effect.flatMap(() =>
87 Effect.fail(new OamResourceStillExists({ arn: sinkArn })),
88 ),
89 Effect.retry({
90 while: (error) =>
91 error._tag === "OamResourceStillExists" ||
92 error._tag === "TooManyRequestsException",
93 schedule: goneSchedule,
94 }),
95 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
96 );
97});
98
99/** Idempotently delete a link and observe absence before its sink can delete. */
100export const deleteLinkAndWait = Effect.fn(function* (linkArn: string) {
101 yield* retryOamMutation(oam.deleteLink({ Identifier: linkArn })).pipe(
102 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
103 );
104
105 const observe = Retry.none(oam.getLink({ Identifier: linkArn }));
106 yield* observe.pipe(
107 Effect.flatMap(() =>
108 Effect.fail(new OamResourceStillExists({ arn: linkArn })),
109 ),
110 Effect.retry({
111 while: (error) =>
112 error._tag === "OamResourceStillExists" ||
113 error._tag === "TooManyRequestsException",
114 schedule: goneSchedule,
115 }),
116 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
117 );
118});
119
120/**
121 * Read the observed tags on an OAM sink or link, tolerating a
122 * not-yet-visible resource (returns `{}`).
123 */
124export const readOamTags = Effect.fn(function* (resourceArn: string) {
125 return yield* oam.listTagsForResource({ ResourceArn: resourceArn }).pipe(
126 Effect.map((r) => (r.Tags ?? {}) as Record<string, string>),
127 Effect.catchTag(["ResourceNotFoundException", "ValidationException"], () =>
128 Effect.succeed({} as Record<string, string>),
129 ),
130 );
131});
132
133/**
134 * Converge the tags on an OAM sink or link to the desired user tags merged
135 * with the internal Alchemy ownership tags, diffing against OBSERVED cloud
136 * tags so adoption converges.
137 */
138export const syncOamTags = Effect.fn(function* (
139 resourceArn: string,
140 id: string,
141 userTags: Record<string, string> | undefined,
142) {
143 const internalTags = yield* createInternalTags(id);
144 const desired = { ...(userTags ?? {}), ...internalTags };
145 const observed = yield* readOamTags(resourceArn);
146 const { upsert, removed } = diffTags(observed, desired);
147 if (upsert.length > 0) {
148 yield* retryOamMutation(
149 oam.tagResource({
150 ResourceArn: resourceArn,
151 Tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
152 }),
153 );
154 }
155 if (removed.length > 0) {
156 yield* retryOamMutation(
157 oam.untagResource({ ResourceArn: resourceArn, TagKeys: removed }),
158 );
159 }
160});