1import * as dms from "@distilled.cloud/aws/database-migration-service";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
11import type { Providers } from "../Providers.ts";
12
13export interface ReplicationInstanceProps {
14 /**
15 * Replication-instance identifier. Must be lowercase, 1-63 characters,
16 * begin with a letter, and contain only letters, digits, and hyphens. If
17 * omitted, a deterministic physical name is generated. Changing it replaces
18 * the instance.
19 */
20 replicationInstanceIdentifier?: string;
21 /**
22 * Compute and memory class, e.g. `"dms.t3.micro"`, `"dms.c5.large"`.
23 */
24 replicationInstanceClass: string;
25 /**
26 * Storage in gigabytes allocated to the replication instance.
27 * @default service-chosen for the instance class
28 */
29 allocatedStorage?: number;
30 /**
31 * VPC security group IDs controlling network access to the instance.
32 */
33 vpcSecurityGroupIds?: string[];
34 /**
35 * Availability Zone the instance is created in. Changing it replaces the
36 * instance.
37 */
38 availabilityZone?: string;
39 /**
40 * Subnet group the instance is placed in. Changing it replaces the
41 * instance.
42 */
43 replicationSubnetGroupIdentifier?: string;
44 /**
45 * Weekly maintenance window in `ddd:hh24:mi-ddd:hh24:mi` (UTC) format.
46 */
47 preferredMaintenanceWindow?: string;
48 /**
49 * Whether to provision the instance across multiple Availability Zones.
50 * @default false
51 */
52 multiAZ?: boolean;
53 /**
54 * DMS engine version.
55 * @default latest
56 */
57 engineVersion?: string;
58 /**
59 * Whether minor engine upgrades are applied automatically during the
60 * maintenance window.
61 * @default true
62 */
63 autoMinorVersionUpgrade?: boolean;
64 /**
65 * Customer-managed KMS key for storage encryption. Changing it replaces the
66 * instance.
67 * @default AWS-owned DMS key
68 */
69 kmsKeyId?: string;
70 /**
71 * Whether the instance gets a public IP address. Changing it replaces the
72 * instance.
73 * @default true
74 */
75 publiclyAccessible?: boolean;
76 /**
77 * Network type: `"IPV4"`, `"IPV6"`, or `"DUAL"`. Changing it replaces the
78 * instance.
79 */
80 networkType?: string;
81 /**
82 * User-defined tags for the instance.
83 */
84 tags?: Record<string, string>;
85}
86
87export interface ReplicationInstance extends Resource<
88 "AWS.DMS.ReplicationInstance",
89 ReplicationInstanceProps,
90 {
91 /** The replication instance identifier (unique per account/region). */
92 replicationInstanceIdentifier: string;
93 /** The ARN of the replication instance. */
94 replicationInstanceArn: string;
95 /** The compute class of the instance, e.g. `dms.t3.micro`. */
96 replicationInstanceClass: string;
97 /** The current status of the instance, e.g. `available`. */
98 status: string | undefined;
99 /** The DMS engine version running on the instance. */
100 engineVersion: string | undefined;
101 /** The private IP addresses of the instance. */
102 privateIpAddresses: string[];
103 /** The public IP addresses of the instance (when publicly accessible). */
104 publicIpAddresses: string[];
105 /** The tags attached to the replication instance. */
106 tags: Record<string, string>;
107 },
108 never,
109 Providers
110> {}
111
112/**
113 * A DMS replication instance — the managed compute that runs migration and
114 * replication tasks. Provisioning takes several minutes and the instance is
115 * billed hourly while it exists, so create it only when a migration is
116 * running and destroy it promptly.
117 * ### Creating a Replication Instance
118 * **Example:** Small Instance in a Subnet Group
119 * ```typescript
120 * const instance = yield* ReplicationInstance("Migration", {
121 * replicationInstanceClass: "dms.t3.micro",
122 * allocatedStorage: 50,
123 * replicationSubnetGroupIdentifier: subnetGroup.replicationSubnetGroupIdentifier,
124 * publiclyAccessible: false,
125 * });
126 * ```
127 *
128 * @resource
129 */
130export const ReplicationInstance = Resource<ReplicationInstance>(
131 "AWS.DMS.ReplicationInstance",
132);
133
134const toTagRecord = (
135 tags: Array<{ Key?: string; Value?: string }> | undefined,
136): Record<string, string> =>
137 Object.fromEntries(
138 (tags ?? [])
139 .filter(
140 (tag): tag is { Key: string; Value: string } =>
141 typeof tag.Key === "string" && typeof tag.Value === "string",
142 )
143 .map((tag) => [tag.Key, tag.Value]),
144 );
145
146export const ReplicationInstanceProvider = () =>
147 Provider.effect(
148 ReplicationInstance,
149 Effect.gen(function* () {
150 const toName = (id: string, props: ReplicationInstanceProps) =>
151 props.replicationInstanceIdentifier
152 ? Effect.succeed(props.replicationInstanceIdentifier)
153 : createPhysicalName({ id, maxLength: 60, lowercase: true });
154
155 const findInstance = Effect.fn(function* (identifier: string) {
156 const response = yield* dms
157 .describeReplicationInstances({
158 Filters: [
159 { Name: "replication-instance-id", Values: [identifier] },
160 ],
161 })
162 .pipe(
163 Effect.catchTag("ResourceNotFoundFault", () =>
164 Effect.succeed(undefined),
165 ),
166 );
167 return response?.ReplicationInstances?.[0];
168 });
169
170 const readTags = Effect.fn(function* (arn: string) {
171 const response = yield* dms
172 .listTagsForResource({ ResourceArn: arn })
173 .pipe(Effect.catch(() => Effect.succeed(undefined)));
174 return toTagRecord(response?.TagList);
175 });
176
177 // Provisioning and modifications both surface as a transitional status;
178 // budget ~15 minutes (15s x 60) for the instance to become available.
179 const waitForInstance = Effect.fn(function* (identifier: string) {
180 const readinessPolicy = Schedule.max([
181 Schedule.fixed("15 seconds"),
182 Schedule.recurs(60),
183 ]);
184 return yield* findInstance(identifier).pipe(
185 Effect.flatMap((instance) => {
186 if (!instance?.ReplicationInstanceArn) {
187 return Effect.fail(
188 new Error(`DMS replication instance '${identifier}' not found`),
189 );
190 }
191 if (instance.ReplicationInstanceStatus !== "available") {
192 return Effect.fail(
193 new Error(
194 `DMS replication instance '${identifier}' not available (status: ${instance.ReplicationInstanceStatus})`,
195 ),
196 );
197 }
198 return Effect.succeed(instance);
199 }),
200 Effect.retry({ schedule: readinessPolicy }),
201 );
202 });
203
204 const toAttrs = Effect.fn(function* (instance: dms.ReplicationInstance) {
205 const identifier = instance.ReplicationInstanceIdentifier;
206 if (!identifier || !instance.ReplicationInstanceArn) {
207 return yield* Effect.fail(
208 new Error(
209 `DMS replication instance '${identifier}' is missing its identifier or ARN`,
210 ),
211 );
212 }
213 return {
214 replicationInstanceIdentifier: identifier,
215 replicationInstanceArn: instance.ReplicationInstanceArn,
216 replicationInstanceClass: instance.ReplicationInstanceClass ?? "",
217 status: instance.ReplicationInstanceStatus,
218 engineVersion: instance.EngineVersion,
219 privateIpAddresses: [
220 ...(instance.ReplicationInstancePrivateIpAddresses ?? []),
221 ],
222 publicIpAddresses: [
223 ...(instance.ReplicationInstancePublicIpAddresses ?? []),
224 ],
225 tags: yield* readTags(instance.ReplicationInstanceArn),
226 };
227 });
228
229 return {
230 stables: ["replicationInstanceIdentifier", "replicationInstanceArn"],
231
232 diff: Effect.fn(function* ({ id, olds, news }) {
233 if (!isResolved(news)) return undefined;
234 if (
235 (yield* toName(
236 id,
237 olds ??
238 ({
239 replicationInstanceClass: "",
240 } as ReplicationInstanceProps),
241 )) !== (yield* toName(id, news))
242 ) {
243 return { action: "replace" } as const;
244 }
245 // Create-only properties force a replacement.
246 if (
247 (news.publiclyAccessible ?? undefined) !==
248 (olds?.publiclyAccessible ?? undefined)
249 ) {
250 return { action: "replace" } as const;
251 }
252 if ((news.kmsKeyId ?? undefined) !== (olds?.kmsKeyId ?? undefined)) {
253 return { action: "replace" } as const;
254 }
255 if (
256 (news.availabilityZone ?? undefined) !==
257 (olds?.availabilityZone ?? undefined)
258 ) {
259 return { action: "replace" } as const;
260 }
261 if (
262 (news.replicationSubnetGroupIdentifier ?? undefined) !==
263 (olds?.replicationSubnetGroupIdentifier ?? undefined)
264 ) {
265 return { action: "replace" } as const;
266 }
267 if (
268 (news.networkType ?? undefined) !== (olds?.networkType ?? undefined)
269 ) {
270 return { action: "replace" } as const;
271 }
272 }),
273
274 read: Effect.fn(function* ({ id, olds, output }) {
275 const name =
276 output?.replicationInstanceIdentifier ??
277 (yield* toName(
278 id,
279 olds ??
280 ({
281 replicationInstanceClass: "",
282 } as ReplicationInstanceProps),
283 ));
284 const instance = yield* findInstance(name);
285 if (!instance?.ReplicationInstanceArn) return undefined;
286 const attrs = yield* toAttrs(instance);
287 return (yield* hasAlchemyTags(id, attrs.tags))
288 ? attrs
289 : Unowned(attrs);
290 }),
291
292 reconcile: Effect.fn(function* ({ id, news, output, session }) {
293 const name =
294 output?.replicationInstanceIdentifier ?? (yield* toName(id, news));
295 const internalTags = yield* createInternalTags(id);
296 const desiredTags = { ...internalTags, ...news.tags };
297
298 // 1. Observe — cloud state is authoritative.
299 let observed = yield* findInstance(name);
300
301 // 2. Ensure — create if missing; tolerate AlreadyExists as a race.
302 if (observed === undefined) {
303 yield* dms
304 .createReplicationInstance({
305 ReplicationInstanceIdentifier: name,
306 ReplicationInstanceClass: news.replicationInstanceClass,
307 AllocatedStorage: news.allocatedStorage,
308 VpcSecurityGroupIds: news.vpcSecurityGroupIds,
309 AvailabilityZone: news.availabilityZone,
310 ReplicationSubnetGroupIdentifier:
311 news.replicationSubnetGroupIdentifier,
312 PreferredMaintenanceWindow: news.preferredMaintenanceWindow,
313 MultiAZ: news.multiAZ,
314 EngineVersion: news.engineVersion,
315 AutoMinorVersionUpgrade: news.autoMinorVersionUpgrade,
316 KmsKeyId: news.kmsKeyId,
317 PubliclyAccessible: news.publiclyAccessible,
318 NetworkType: news.networkType,
319 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
320 Key,
321 Value,
322 })),
323 })
324 .pipe(
325 Effect.catchTag(
326 "ResourceAlreadyExistsFault",
327 () => Effect.void,
328 ),
329 );
330 }
331
332 // Wait for availability so sync/modify calls don't hit
333 // InvalidResourceStateFault.
334 observed = yield* waitForInstance(name);
335
336 // 3. Sync — compute the modify delta from OBSERVED state. Immutable
337 // properties are handled by diff (replacement), so only mutable
338 // aspects are compared here.
339 const modify: dms.ModifyReplicationInstanceMessage = {
340 ReplicationInstanceArn: observed.ReplicationInstanceArn!,
341 ApplyImmediately: true,
342 };
343 let dirty = false;
344 if (
345 news.replicationInstanceClass !== observed.ReplicationInstanceClass
346 ) {
347 modify.ReplicationInstanceClass = news.replicationInstanceClass;
348 dirty = true;
349 }
350 if (
351 news.allocatedStorage !== undefined &&
352 news.allocatedStorage !== observed.AllocatedStorage
353 ) {
354 modify.AllocatedStorage = news.allocatedStorage;
355 dirty = true;
356 }
357 if (news.multiAZ !== undefined && news.multiAZ !== observed.MultiAZ) {
358 modify.MultiAZ = news.multiAZ;
359 dirty = true;
360 }
361 if (
362 news.engineVersion !== undefined &&
363 news.engineVersion !== observed.EngineVersion
364 ) {
365 modify.EngineVersion = news.engineVersion;
366 modify.AllowMajorVersionUpgrade = true;
367 dirty = true;
368 }
369 if (
370 news.preferredMaintenanceWindow !== undefined &&
371 news.preferredMaintenanceWindow !==
372 observed.PreferredMaintenanceWindow
373 ) {
374 modify.PreferredMaintenanceWindow = news.preferredMaintenanceWindow;
375 dirty = true;
376 }
377
378 if (dirty) {
379 yield* dms.modifyReplicationInstance(modify);
380 observed = yield* waitForInstance(name);
381 }
382
383 // 3b. Sync tags — diff against OBSERVED cloud tags.
384 const arn = observed.ReplicationInstanceArn!;
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 // Deletion of an instance mid-create/modify is rejected with
403 // InvalidResourceStateFault — deletion is already impossible until it
404 // settles, so treat that (and NotFound) as success/in-progress.
405 yield* dms
406 .deleteReplicationInstance({
407 ReplicationInstanceArn: output.replicationInstanceArn,
408 })
409 .pipe(
410 Effect.catchTag("ResourceNotFoundFault", () => Effect.void),
411 Effect.catchTag("InvalidResourceStateFault", () => Effect.void),
412 );
413 }),
414
415 list: () =>
416 dms.describeReplicationInstances.pages({}).pipe(
417 Stream.runCollect,
418 Effect.map((chunk) =>
419 Array.from(chunk).flatMap((page) =>
420 (page.ReplicationInstances ?? []).filter(
421 (instance) =>
422 instance.ReplicationInstanceIdentifier !== undefined &&
423 instance.ReplicationInstanceArn !== undefined,
424 ),
425 ),
426 ),
427 Effect.flatMap(
428 Effect.forEach((instance) => toAttrs(instance), {
429 concurrency: 4,
430 }),
431 ),
432 ),
433 };
434 }),
435 );