1import * as dms from "@distilled.cloud/aws/database-migration-service";
2import * as iam from "@distilled.cloud/aws/iam";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import { createPhysicalName } from "../../PhysicalName.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
12import { AWSEnvironment } from "../Environment.ts";
13import type { Providers } from "../Providers.ts";
14
15export interface ReplicationSubnetGroupProps {
16 /**
17 * Subnet-group identifier. Must be lowercase, 1-255 characters, and cannot
18 * be the reserved value `default`. If omitted, a deterministic physical
19 * name is generated. Changing it replaces the subnet group.
20 */
21 replicationSubnetGroupIdentifier?: string;
22 /**
23 * Human-readable description of the subnet group.
24 */
25 description?: string;
26 /**
27 * VPC subnet IDs the replication instance can be placed in. Must span at
28 * least two Availability Zones.
29 */
30 subnetIds: string[];
31 /**
32 * User-defined tags for the subnet group.
33 */
34 tags?: Record<string, string>;
35}
36
37export interface ReplicationSubnetGroup extends Resource<
38 "AWS.DMS.ReplicationSubnetGroup",
39 ReplicationSubnetGroupProps,
40 {
41 /** The subnet group identifier (unique per account/region). */
42 replicationSubnetGroupIdentifier: string;
43 /** The ARN of the replication subnet group. */
44 replicationSubnetGroupArn: string;
45 /** The VPC the subnets belong to. */
46 vpcId: string | undefined;
47 /** The IDs of the subnets in the group. */
48 subnetIds: string[];
49 /** The current status of the subnet group, e.g. `Complete`. */
50 status: string | undefined;
51 /** The tags attached to the subnet group. */
52 tags: Record<string, string>;
53 },
54 never,
55 Providers
56> {}
57
58/**
59 * A DMS replication subnet group — the set of VPC subnets a replication
60 * instance can be launched into. Must cover at least two Availability Zones.
61 * Free and fast to create.
62 * ### Creating a Subnet Group
63 * **Example:** Two-AZ Subnet Group
64 * ```typescript
65 * const subnetGroup = yield* ReplicationSubnetGroup("Migration", {
66 * description: "DMS replication subnets",
67 * subnetIds: [subnetA.subnetId, subnetB.subnetId],
68 * });
69 * ```
70 *
71 * @resource
72 */
73export const ReplicationSubnetGroup = Resource<ReplicationSubnetGroup>(
74 "AWS.DMS.ReplicationSubnetGroup",
75);
76
77const toTagRecord = (
78 tags: Array<{ Key?: string; Value?: string }> | undefined,
79): Record<string, string> =>
80 Object.fromEntries(
81 (tags ?? [])
82 .filter(
83 (tag): tag is { Key: string; Value: string } =>
84 typeof tag.Key === "string" && typeof tag.Value === "string",
85 )
86 .map((tag) => [tag.Key, tag.Value]),
87 );
88
89const sameStringSet = (
90 a: readonly string[] | undefined,
91 b: readonly string[] | undefined,
92): boolean => {
93 const left = [...(a ?? [])].sort();
94 const right = [...(b ?? [])].sort();
95 return left.length === right.length && left.every((v, i) => v === right[i]);
96};
97
98const DMS_VPC_ROLE_NAME = "dms-vpc-role";
99const DMS_VPC_POLICY_ARN =
100 "arn:aws:iam::aws:policy/service-role/AmazonDMSVPCManagementRole";
101const DMS_VPC_ROLE_OWNER_TAG = {
102 Key: "alchemy::managed-by",
103 Value: "AWS.DMS.ReplicationSubnetGroup",
104} as const;
105
106export const ReplicationSubnetGroupProvider = () =>
107 Provider.effect(
108 ReplicationSubnetGroup,
109 Effect.gen(function* () {
110 const toName = (id: string, props: ReplicationSubnetGroupProps) =>
111 props.replicationSubnetGroupIdentifier
112 ? Effect.succeed(props.replicationSubnetGroupIdentifier)
113 : createPhysicalName({ id, maxLength: 200, lowercase: true });
114
115 // Subnet-group ARNs deterministically embed the identifier, so we build
116 // them rather than reading one back (describe returns no ARN).
117 const toArn = Effect.fn(function* (identifier: string) {
118 const { accountId, region } = yield* AWSEnvironment.current;
119 return `arn:aws:dms:${region}:${accountId}:subgrp:${identifier}`;
120 });
121
122 const findGroup = Effect.fn(function* (identifier: string) {
123 const response = yield* dms
124 .describeReplicationSubnetGroups({
125 Filters: [
126 { Name: "replication-subnet-group-id", Values: [identifier] },
127 ],
128 })
129 .pipe(
130 Effect.catchTag("ResourceNotFoundFault", () =>
131 Effect.succeed(undefined),
132 ),
133 );
134 return response?.ReplicationSubnetGroups?.[0];
135 });
136
137 // DMS requires the account-wide `dms-vpc-role` (trusting
138 // dms.amazonaws.com with the AmazonDMSVPCManagementRole managed policy)
139 // before any VPC-touching operation. The console creates it implicitly;
140 // API callers must ensure it. Tag only roles created by this provider so
141 // deletion can distinguish our helper from pre-existing account
142 // infrastructure with the same AWS-mandated name.
143 const ensureDmsVpcRole = Effect.gen(function* () {
144 yield* iam
145 .createRole({
146 RoleName: DMS_VPC_ROLE_NAME,
147 AssumeRolePolicyDocument: JSON.stringify({
148 Version: "2012-10-17",
149 Statement: [
150 {
151 Effect: "Allow",
152 Principal: { Service: "dms.amazonaws.com" },
153 Action: "sts:AssumeRole",
154 },
155 ],
156 }),
157 Tags: [DMS_VPC_ROLE_OWNER_TAG],
158 })
159 .pipe(
160 Effect.catchTag("EntityAlreadyExistsException", () => Effect.void),
161 );
162 // Idempotent — attaching an already-attached managed policy is a no-op.
163 yield* iam.attachRolePolicy({
164 RoleName: DMS_VPC_ROLE_NAME,
165 PolicyArn: DMS_VPC_POLICY_ARN,
166 });
167 }).pipe(
168 // A concurrent last-subnet-group delete can remove the shared helper
169 // between create/get and policy attachment. Re-observe and ensure it
170 // again instead of surfacing a cross-test race.
171 Effect.retry({
172 while: (error) =>
173 error._tag === "NoSuchEntityException" ||
174 error._tag === "ConcurrentModificationException" ||
175 error._tag === "LimitExceededException" ||
176 error._tag === "ServiceFailureException",
177 schedule: Schedule.max([
178 Schedule.fixed("1 second"),
179 Schedule.recurs(10),
180 ]),
181 }),
182 );
183
184 const deleteOwnedDmsVpcRoleIfUnused = Effect.gen(function* () {
185 // Bound the whole observe/detach/delete loop to 30 seconds. Every pass
186 // rechecks DMS first, so a concurrent subnet-group create protects the
187 // role and takes responsibility for deleting it later.
188 for (let attempt = 0; attempt < 30; attempt++) {
189 const groups = yield* dms.describeReplicationSubnetGroups({});
190 if ((groups.ReplicationSubnetGroups ?? []).length > 0) return;
191
192 const tags = yield* iam
193 .listRoleTags({ RoleName: DMS_VPC_ROLE_NAME })
194 .pipe(
195 Effect.catchTag("NoSuchEntityException", () =>
196 Effect.succeed(undefined),
197 ),
198 );
199 if (tags === undefined) return;
200 const owned = (tags.Tags ?? []).some(
201 (tag) =>
202 tag.Key === DMS_VPC_ROLE_OWNER_TAG.Key &&
203 tag.Value === DMS_VPC_ROLE_OWNER_TAG.Value,
204 );
205 if (!owned) return;
206
207 const deleted = yield* Effect.gen(function* () {
208 yield* iam
209 .detachRolePolicy({
210 RoleName: DMS_VPC_ROLE_NAME,
211 PolicyArn: DMS_VPC_POLICY_ARN,
212 })
213 .pipe(
214 Effect.catchTag("NoSuchEntityException", () => Effect.void),
215 );
216 yield* iam
217 .deleteRole({ RoleName: DMS_VPC_ROLE_NAME })
218 .pipe(
219 Effect.catchTag("NoSuchEntityException", () => Effect.void),
220 );
221 return true;
222 }).pipe(
223 Effect.catchTag(
224 [
225 "ConcurrentModificationException",
226 "DeleteConflictException",
227 "LimitExceededException",
228 "ServiceFailureException",
229 ],
230 () => Effect.succeed(false),
231 ),
232 );
233 if (!deleted) {
234 yield* Effect.sleep("1 second");
235 continue;
236 }
237
238 const remaining = yield* iam
239 .getRole({ RoleName: DMS_VPC_ROLE_NAME })
240 .pipe(
241 Effect.catchTag("NoSuchEntityException", () =>
242 Effect.succeed(undefined),
243 ),
244 );
245 if (remaining === undefined) return;
246 yield* Effect.sleep("1 second");
247 }
248 return yield* Effect.die(
249 new Error(
250 `IAM role ${DMS_VPC_ROLE_NAME} remained observable 30 seconds after the last DMS subnet group was deleted`,
251 ),
252 );
253 });
254
255 const readTags = Effect.fn(function* (arn: string) {
256 const response = yield* dms
257 .listTagsForResource({ ResourceArn: arn })
258 .pipe(Effect.catch(() => Effect.succeed(undefined)));
259 return toTagRecord(response?.TagList);
260 });
261
262 const toAttrs = Effect.fn(function* (group: dms.ReplicationSubnetGroup) {
263 const identifier = group.ReplicationSubnetGroupIdentifier;
264 if (!identifier) {
265 return yield* Effect.fail(
266 new Error("DMS replication subnet group is missing its identifier"),
267 );
268 }
269 const arn = yield* toArn(identifier);
270 return {
271 replicationSubnetGroupIdentifier: identifier,
272 replicationSubnetGroupArn: arn,
273 vpcId: group.VpcId,
274 subnetIds: (group.Subnets ?? [])
275 .map((s) => s.SubnetIdentifier)
276 .filter((s): s is string => typeof s === "string"),
277 status: group.SubnetGroupStatus,
278 tags: yield* readTags(arn),
279 };
280 });
281
282 return {
283 stables: [
284 "replicationSubnetGroupIdentifier",
285 "replicationSubnetGroupArn",
286 ],
287
288 diff: Effect.fn(function* ({ id, olds, news }) {
289 if (!isResolved(news)) return undefined;
290 if (
291 (yield* toName(
292 id,
293 olds ?? ({ subnetIds: [] } as ReplicationSubnetGroupProps),
294 )) !== (yield* toName(id, news))
295 ) {
296 return { action: "replace" } as const;
297 }
298 }),
299
300 read: Effect.fn(function* ({ id, olds, output }) {
301 const name =
302 output?.replicationSubnetGroupIdentifier ??
303 (yield* toName(
304 id,
305 olds ?? ({ subnetIds: [] } as ReplicationSubnetGroupProps),
306 ));
307 const group = yield* findGroup(name);
308 if (!group?.ReplicationSubnetGroupIdentifier) return undefined;
309 const attrs = yield* toAttrs(group);
310 return (yield* hasAlchemyTags(id, attrs.tags))
311 ? attrs
312 : Unowned(attrs);
313 }),
314
315 reconcile: Effect.fn(function* ({ id, news, output, session }) {
316 const name =
317 output?.replicationSubnetGroupIdentifier ??
318 (yield* toName(id, news));
319 const internalTags = yield* createInternalTags(id);
320 const desiredTags = { ...internalTags, ...news.tags };
321 const description = news.description ?? "Managed by Alchemy";
322
323 // 1. Observe — cloud state is authoritative.
324 let observed = yield* findGroup(name);
325
326 // 2. Ensure — create if missing; tolerate AlreadyExists as a race.
327 if (observed === undefined) {
328 yield* ensureDmsVpcRole;
329 yield* dms
330 .createReplicationSubnetGroup({
331 ReplicationSubnetGroupIdentifier: name,
332 ReplicationSubnetGroupDescription: description,
333 SubnetIds: news.subnetIds,
334 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
335 Key,
336 Value,
337 })),
338 })
339 .pipe(
340 // A freshly-created dms-vpc-role takes a few seconds to
341 // propagate to DMS, which surfaces as AccessDeniedFault
342 // ("The IAM Role ... is not configured properly").
343 Effect.catchTag("AccessDeniedFault", (error) =>
344 ensureDmsVpcRole.pipe(Effect.andThen(Effect.fail(error))),
345 ),
346 Effect.retry({
347 while: (e) => e._tag === "AccessDeniedFault",
348 schedule: Schedule.spaced("5 seconds"),
349 times: 10,
350 }),
351 Effect.catchTag(
352 "ResourceAlreadyExistsFault",
353 () => Effect.void,
354 ),
355 );
356 observed = yield* findGroup(name);
357 } else {
358 // 3. Sync — modify description / subnet set in place when they drift.
359 const observedSubnets = (observed.Subnets ?? [])
360 .map((s) => s.SubnetIdentifier)
361 .filter((s): s is string => typeof s === "string");
362 if (
363 description !== observed.ReplicationSubnetGroupDescription ||
364 !sameStringSet(news.subnetIds, observedSubnets)
365 ) {
366 yield* dms.modifyReplicationSubnetGroup({
367 ReplicationSubnetGroupIdentifier: name,
368 ReplicationSubnetGroupDescription: description,
369 SubnetIds: news.subnetIds,
370 });
371 observed = yield* findGroup(name);
372 }
373 }
374
375 if (!observed?.ReplicationSubnetGroupIdentifier) {
376 return yield* Effect.fail(
377 new Error(
378 `DMS replication subnet group '${name}' not found after reconcile`,
379 ),
380 );
381 }
382
383 // 3b. Sync tags — diff against OBSERVED cloud tags.
384 const arn = yield* toArn(name);
385 const observedTags = yield* readTags(arn);
386 const { upsert, removed } = diffTags(observedTags, desiredTags);
387 if (upsert.length > 0) {
388 yield* dms.addTagsToResource({ ResourceArn: arn, Tags: upsert });
389 }
390 if (removed.length > 0) {
391 yield* dms.removeTagsFromResource({
392 ResourceArn: arn,
393 TagKeys: removed,
394 });
395 }
396
397 yield* session.note(name);
398 return yield* toAttrs(observed);
399 }),
400
401 delete: Effect.fn(function* ({ output }) {
402 yield* dms
403 .deleteReplicationSubnetGroup({
404 ReplicationSubnetGroupIdentifier:
405 output.replicationSubnetGroupIdentifier,
406 })
407 .pipe(Effect.catchTag("ResourceNotFoundFault", () => Effect.void));
408
409 // Observe actual absence before discarding state or considering the
410 // shared helper unused. DMS deletion is eventually consistent.
411 for (let attempt = 0; attempt < 30; attempt++) {
412 if (
413 (yield* findGroup(output.replicationSubnetGroupIdentifier)) ===
414 undefined
415 ) {
416 yield* deleteOwnedDmsVpcRoleIfUnused;
417 return;
418 }
419 yield* Effect.sleep("1 second");
420 }
421 yield* Effect.die(
422 new Error(
423 `DMS replication subnet group ${output.replicationSubnetGroupIdentifier} remained observable 30 seconds after delete`,
424 ),
425 );
426 }),
427
428 list: () =>
429 dms.describeReplicationSubnetGroups.pages({}).pipe(
430 Stream.runCollect,
431 Effect.map((chunk) =>
432 Array.from(chunk).flatMap((page) =>
433 (page.ReplicationSubnetGroups ?? []).filter(
434 (group) =>
435 group.ReplicationSubnetGroupIdentifier !== undefined,
436 ),
437 ),
438 ),
439 Effect.flatMap(
440 Effect.forEach((group) => toAttrs(group), { concurrency: 4 }),
441 ),
442 ),
443 };
444 }),
445 );