1import * as dms from "@distilled.cloud/aws/database-migration-service";
2import * as Effect from "effect/Effect";
3import * as Redacted from "effect/Redacted";
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
13/**
14 * Direction of a DMS endpoint relative to the replication:
15 * `source` reads from the database, `target` writes to it.
16 */
17export type EndpointType = "source" | "target";
18
19export interface EndpointProps {
20 /**
21 * Database-migration endpoint identifier. Must be lowercase, 1-255
22 * characters, begin with a letter, and contain only letters, digits, and
23 * hyphens (no two consecutive hyphens, no trailing hyphen). If omitted, a
24 * deterministic physical name is generated. Changing it replaces the
25 * endpoint.
26 */
27 endpointIdentifier?: string;
28 /**
29 * Whether this endpoint is the migration `source` or `target`.
30 */
31 endpointType: EndpointType;
32 /**
33 * Database engine, e.g. `"mysql"`, `"postgres"`, `"aurora"`, `"s3"`,
34 * `"kinesis"`, `"docdb"`. Determines which `*Settings` block applies.
35 */
36 engineName: string;
37 /**
38 * User name used to connect to the endpoint database.
39 */
40 username?: string;
41 /**
42 * Password used to connect to the endpoint database. Marked sensitive.
43 */
44 password?: Redacted.Redacted<string>;
45 /**
46 * Host name of the endpoint database server.
47 */
48 serverName?: string;
49 /**
50 * Port the endpoint database listens on.
51 */
52 port?: number;
53 /**
54 * Name of the endpoint database.
55 */
56 databaseName?: string;
57 /**
58 * Additional attributes accepted by DMS as a semicolon-separated string.
59 */
60 extraConnectionAttributes?: string;
61 /**
62 * Customer-managed KMS key for connection-secret encryption. Changing the
63 * key replaces the endpoint.
64 * @default AWS-owned DMS key
65 */
66 kmsKeyId?: string;
67 /**
68 * ARN of a certificate DMS uses for SSL connections to the endpoint.
69 */
70 certificateArn?: string;
71 /**
72 * SSL mode for the connection.
73 * @default "none"
74 */
75 sslMode?: dms.DmsSslModeValue;
76 /**
77 * IAM role ARN DMS assumes to access the endpoint (required for S3,
78 * DynamoDB, Kinesis, and other AWS-service targets).
79 */
80 serviceAccessRoleArn?: string;
81 /**
82 * Externally-defined table definition JSON (S3/DynamoDB targets).
83 */
84 externalTableDefinition?: string;
85 /** Settings for an Amazon S3 endpoint. */
86 s3Settings?: dms.S3Settings;
87 /** Settings for an Amazon DynamoDB target endpoint. */
88 dynamoDbSettings?: dms.DynamoDbSettings;
89 /** Settings for an Amazon Kinesis Data Streams target endpoint. */
90 kinesisSettings?: dms.KinesisSettings;
91 /** Settings for an Apache Kafka target endpoint. */
92 kafkaSettings?: dms.KafkaSettings;
93 /** Settings for an OpenSearch/Elasticsearch target endpoint. */
94 elasticsearchSettings?: dms.ElasticsearchSettings;
95 /** Settings for an Amazon Neptune target endpoint. */
96 neptuneSettings?: dms.NeptuneSettings;
97 /** Settings for an Amazon Redshift target endpoint. */
98 redshiftSettings?: dms.RedshiftSettings;
99 /** Settings for a PostgreSQL endpoint. */
100 postgreSQLSettings?: dms.PostgreSQLSettings;
101 /** Settings for a MySQL endpoint. */
102 mySQLSettings?: dms.MySQLSettings;
103 /** Settings for an Oracle endpoint. */
104 oracleSettings?: dms.OracleSettings;
105 /** Settings for a Microsoft SQL Server endpoint. */
106 microsoftSQLServerSettings?: dms.MicrosoftSQLServerSettings;
107 /** Settings for a MongoDB endpoint. */
108 mongoDbSettings?: dms.MongoDbSettings;
109 /** Settings for an Amazon DocumentDB endpoint. */
110 docDbSettings?: dms.DocDbSettings;
111 /** Settings for a Redis target endpoint. */
112 redisSettings?: dms.RedisSettings;
113 /**
114 * User-defined tags for the endpoint.
115 */
116 tags?: Record<string, string>;
117}
118
119export interface Endpoint extends Resource<
120 "AWS.DMS.Endpoint",
121 EndpointProps,
122 {
123 /** The endpoint identifier (unique per account/region). */
124 endpointIdentifier: string;
125 /** The ARN of the endpoint. */
126 endpointArn: string;
127 /** Whether the endpoint is a `source` or `target`. */
128 endpointType: string;
129 /** The database engine of the endpoint, e.g. `postgres`, `mysql`. */
130 engineName: string;
131 /** The current status of the endpoint, e.g. `active`. */
132 status: string | undefined;
133 /** The tags attached to the endpoint. */
134 tags: Record<string, string>;
135 },
136 never,
137 Providers
138> {}
139
140/**
141 * An AWS Database Migration Service (DMS) endpoint — the source or target
142 * database of a replication. Endpoints are metadata-only (they store
143 * connection information, not data), so they are free and fast to create.
144 * ### Creating Endpoints
145 * **Example:** MySQL Source Endpoint
146 * ```typescript
147 * const source = yield* Endpoint("Source", {
148 * endpointType: "source",
149 * engineName: "mysql",
150 * serverName: "source-db.example.com",
151 * port: 3306,
152 * username: "admin",
153 * password: Redacted.make("super-secret"),
154 * databaseName: "app",
155 * });
156 * ```
157 *
158 * **Example:** S3 Target Endpoint
159 * ```typescript
160 * const target = yield* Endpoint("Target", {
161 * endpointType: "target",
162 * engineName: "s3",
163 * serviceAccessRoleArn: role.roleArn,
164 * s3Settings: {
165 * BucketName: bucket.bucketName,
166 * ServiceAccessRoleArn: role.roleArn,
167 * },
168 * });
169 * ```
170 *
171 * @resource
172 */
173export const Endpoint = Resource<Endpoint>("AWS.DMS.Endpoint");
174
175const toTagRecord = (
176 tags: Array<{ Key?: string; Value?: string }> | undefined,
177): Record<string, string> =>
178 Object.fromEntries(
179 (tags ?? [])
180 .filter(
181 (tag): tag is { Key: string; Value: string } =>
182 typeof tag.Key === "string" && typeof tag.Value === "string",
183 )
184 .map((tag) => [tag.Key, tag.Value]),
185 );
186
187export const EndpointProvider = () =>
188 Provider.effect(
189 Endpoint,
190 Effect.gen(function* () {
191 const toName = (id: string, props: EndpointProps) =>
192 props.endpointIdentifier
193 ? Effect.succeed(props.endpointIdentifier)
194 : createPhysicalName({ id, maxLength: 200, lowercase: true });
195
196 // DMS has no get-by-identifier; describeEndpoints with an `endpoint-id`
197 // filter returns the single matching endpoint (or ResourceNotFoundFault).
198 const findEndpoint = Effect.fn(function* (identifier: string) {
199 const response = yield* dms
200 .describeEndpoints({
201 Filters: [{ Name: "endpoint-id", Values: [identifier] }],
202 })
203 .pipe(
204 Effect.catchTag("ResourceNotFoundFault", () =>
205 Effect.succeed(undefined),
206 ),
207 );
208 return response?.Endpoints?.[0];
209 });
210
211 const readTags = Effect.fn(function* (arn: string) {
212 const response = yield* dms
213 .listTagsForResource({ ResourceArn: arn })
214 .pipe(Effect.catch(() => Effect.succeed(undefined)));
215 return toTagRecord(response?.TagList);
216 });
217
218 const toAttrs = Effect.fn(function* (endpoint: dms.Endpoint) {
219 if (!endpoint.EndpointIdentifier || !endpoint.EndpointArn) {
220 return yield* Effect.fail(
221 new Error(
222 `DMS endpoint '${endpoint.EndpointIdentifier}' is missing its identifier or ARN`,
223 ),
224 );
225 }
226 return {
227 endpointIdentifier: endpoint.EndpointIdentifier,
228 endpointArn: endpoint.EndpointArn,
229 endpointType: endpoint.EndpointType ?? "source",
230 engineName: endpoint.EngineName ?? "",
231 status: endpoint.Status,
232 tags: yield* readTags(endpoint.EndpointArn),
233 };
234 });
235
236 // Settings blocks and secret material shared by create + modify.
237 const buildSettings = (props: EndpointProps) => ({
238 Username: props.username,
239 Password: props.password,
240 ServerName: props.serverName,
241 Port: props.port,
242 DatabaseName: props.databaseName,
243 ExtraConnectionAttributes: props.extraConnectionAttributes,
244 CertificateArn: props.certificateArn,
245 SslMode: props.sslMode,
246 ServiceAccessRoleArn: props.serviceAccessRoleArn,
247 ExternalTableDefinition: props.externalTableDefinition,
248 S3Settings: props.s3Settings,
249 DynamoDbSettings: props.dynamoDbSettings,
250 KinesisSettings: props.kinesisSettings,
251 KafkaSettings: props.kafkaSettings,
252 ElasticsearchSettings: props.elasticsearchSettings,
253 NeptuneSettings: props.neptuneSettings,
254 RedshiftSettings: props.redshiftSettings,
255 PostgreSQLSettings: props.postgreSQLSettings,
256 MySQLSettings: props.mySQLSettings,
257 OracleSettings: props.oracleSettings,
258 MicrosoftSQLServerSettings: props.microsoftSQLServerSettings,
259 MongoDbSettings: props.mongoDbSettings,
260 DocDbSettings: props.docDbSettings,
261 RedisSettings: props.redisSettings,
262 });
263
264 return {
265 stables: ["endpointIdentifier", "endpointArn"],
266
267 diff: Effect.fn(function* ({ id, olds, news }) {
268 if (!isResolved(news)) return undefined;
269 if (
270 (yield* toName(id, olds ?? ({} as EndpointProps))) !==
271 (yield* toName(id, news))
272 ) {
273 return { action: "replace" } as const;
274 }
275 // KMS key is create-only.
276 if ((news.kmsKeyId ?? undefined) !== (olds?.kmsKeyId ?? undefined)) {
277 return { action: "replace" } as const;
278 }
279 }),
280
281 read: Effect.fn(function* ({ id, olds, output }) {
282 const name =
283 output?.endpointIdentifier ??
284 (yield* toName(id, olds ?? ({} as EndpointProps)));
285 const endpoint = yield* findEndpoint(name);
286 if (!endpoint?.EndpointArn) return undefined;
287 const attrs = yield* toAttrs(endpoint);
288 return (yield* hasAlchemyTags(id, attrs.tags))
289 ? attrs
290 : Unowned(attrs);
291 }),
292
293 reconcile: Effect.fn(function* ({ id, news, output, session }) {
294 const name = output?.endpointIdentifier ?? (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* findEndpoint(name);
300
301 // 2. Ensure — create if missing; tolerate AlreadyExists as a race.
302 if (observed === undefined) {
303 observed = yield* dms
304 .createEndpoint({
305 EndpointIdentifier: name,
306 EndpointType: news.endpointType,
307 EngineName: news.engineName,
308 KmsKeyId: news.kmsKeyId,
309 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
310 Key,
311 Value,
312 })),
313 ...buildSettings(news),
314 })
315 .pipe(
316 Effect.map((r) => r.Endpoint),
317 Effect.catchTag("ResourceAlreadyExistsFault", () =>
318 findEndpoint(name),
319 ),
320 );
321 } else {
322 // 3. Sync — the endpoint exists; push desired configuration. DMS
323 // modifyEndpoint is a full upsert of the connection settings,
324 // so send them all (cheap, no delta computation needed).
325 observed = yield* dms
326 .modifyEndpoint({
327 EndpointArn: observed.EndpointArn!,
328 EndpointIdentifier: name,
329 EndpointType: news.endpointType,
330 EngineName: news.engineName,
331 ...buildSettings(news),
332 })
333 .pipe(Effect.map((r) => r.Endpoint));
334 }
335
336 if (!observed?.EndpointArn) {
337 return yield* Effect.fail(
338 new Error(`DMS endpoint '${name}' has no ARN after reconcile`),
339 );
340 }
341
342 // 3b. Sync tags — diff against OBSERVED cloud tags.
343 const arn = observed.EndpointArn;
344 const observedTags = yield* readTags(arn);
345 const { upsert, removed } = diffTags(observedTags, desiredTags);
346 if (upsert.length > 0) {
347 yield* dms.addTagsToResource({ ResourceArn: arn, Tags: upsert });
348 }
349 if (removed.length > 0) {
350 yield* dms.removeTagsFromResource({
351 ResourceArn: arn,
352 TagKeys: removed,
353 });
354 }
355
356 yield* session.note(name);
357 return yield* toAttrs(observed);
358 }),
359
360 delete: Effect.fn(function* ({ output }) {
361 yield* dms
362 .deleteEndpoint({ EndpointArn: output.endpointArn })
363 .pipe(Effect.catchTag("ResourceNotFoundFault", () => Effect.void));
364 }),
365
366 list: () =>
367 dms.describeEndpoints.pages({}).pipe(
368 Stream.runCollect,
369 Effect.map((chunk) =>
370 Array.from(chunk).flatMap((page) =>
371 (page.Endpoints ?? []).filter(
372 (endpoint) =>
373 endpoint.EndpointIdentifier !== undefined &&
374 endpoint.EndpointArn !== undefined,
375 ),
376 ),
377 ),
378 Effect.flatMap(
379 Effect.forEach((endpoint) => toAttrs(endpoint), {
380 concurrency: 4,
381 }),
382 ),
383 ),
384 };
385 }),
386 );