1import * as dm from "@distilled.cloud/gcp/datamigration_v1";
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 * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { GcpEnvironment } from "../Environment.ts";
10import { createInternalLabels } from "../Labels.ts";
11import type { Providers } from "../Providers.ts";
12import {
13 encodeOwnershipLine,
14 fieldMask,
15 fingerprint,
16 hasOwnershipMarker,
17 locationParent,
18 normalizeLocation,
19 ownedByAlchemy,
20 parseName,
21 parseOwnership,
22 replaceOnIdentity,
23 ResourceNotResolved,
24 toPhysicalId,
25 waitForOperation,
26 waitUntilExists,
27 waitUntilGone,
28} from "./internal.ts";
29
30export type DatabaseEngineInfo = dm.DatabaseEngineInfo;
31export type ConversionWorkspaceSourceProvider =
32 | dm.ConversionWorkspaceSourceProviderEnum
33 | (string & {});
34export type ConversionWorkspaceDestinationProvider =
35 | dm.ConversionWorkspaceDestinationProviderEnum
36 | (string & {});
37
38export type ConversionWorkspaceProps = {
39 /**
40 * Conversion workspace id (the `{conversionWorkspace}` segment of
41 * `projects/{project}/locations/{location}/conversionWorkspaces/{conversionWorkspace}`).
42 * If omitted, a unique RFC1035 name is generated. Immutable — changing
43 * it replaces the workspace.
44 */
45 conversionWorkspaceId?: string;
46 /**
47 * Region (`us-central1`, …). Immutable — changing it replaces the
48 * workspace. `US-CENTRAL1` is accepted and normalized to `us-central1`.
49 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
50 */
51 location?: string;
52 /**
53 * User-friendly display name. Maximum length is 63 characters including
54 * Alchemy's ownership marker. Conversion workspaces have no labels
55 * field, so ownership is stored in a `[alchemy …]` prefix and stripped
56 * from attributes.
57 */
58 displayName?: string;
59 /**
60 * Source engine and version. Immutable — changing it replaces the
61 * workspace.
62 */
63 source: DatabaseEngineInfo;
64 /**
65 * Destination engine and version. Immutable — changing it replaces the
66 * workspace.
67 */
68 destination: DatabaseEngineInfo;
69 /**
70 * Provider for the source database (`CLOUDSQL`, `RDS`, `AURORA`,
71 * `ALLOYDB`, `AZURE_DATABASE`). Immutable — changing it replaces the
72 * workspace.
73 */
74 sourceProvider?: ConversionWorkspaceSourceProvider;
75 /**
76 * Provider for the destination database. Immutable — changing it
77 * replaces the workspace.
78 */
79 destinationProvider?: ConversionWorkspaceDestinationProvider;
80 /**
81 * Database-pair specific workspace settings
82 * (`convert_foreign_key_to_interleave`, `skip_triggers`, …).
83 */
84 globalSettings?: Record<string, string>;
85};
86
87export type ConversionWorkspace = Resource<
88 "GCP.DataMigration.ConversionWorkspace",
89 ConversionWorkspaceProps,
90 {
91 /** Full resource name. */
92 name: string;
93 /** Conversion workspace id (last path segment). */
94 conversionWorkspaceId: string;
95 /** Project id. */
96 project: string;
97 /** Location id. */
98 location: string;
99 /** User display name with the Alchemy ownership prefix stripped. */
100 displayName: string | undefined;
101 /** Source engine details. */
102 source: DatabaseEngineInfo | undefined;
103 /** Destination engine details. */
104 destination: DatabaseEngineInfo | undefined;
105 /** Source database provider. */
106 sourceProvider: string | undefined;
107 /** Destination database provider. */
108 destinationProvider: string | undefined;
109 /** Workspace-level mapping-engine settings. */
110 globalSettings: Record<string, string>;
111 /** Whether the workspace has uncommitted changes. */
112 hasUncommittedChanges: boolean | undefined;
113 /** Latest commit id. */
114 latestCommitId: string | undefined;
115 /** RFC3339 latest commit timestamp. */
116 latestCommitTime: string | undefined;
117 /** RFC3339 creation timestamp. */
118 createTime: string | undefined;
119 /** RFC3339 last-update timestamp. */
120 updateTime: string | undefined;
121 },
122 never,
123 Providers
124>;
125
126/**
127 * A Database Migration Service conversion workspace: source and
128 * destination engines plus optional mapping-engine settings used by
129 * heterogeneous migrations.
130 *
131 * Conversion workspaces have no labels field — Alchemy stamps ownership
132 * into the display name so `list` / nuke can find them. `source`,
133 * `destination`, and their providers are replacement triggers. Display
134 * name and `globalSettings` update in place.
135 *
136 * ### Creating a Conversion Workspace
137 * **Example:** MySQL to PostgreSQL
138 * ```typescript
139 * const workspace = yield* GCP.DataMigration.ConversionWorkspace("MysqlToPg", {
140 * displayName: "mysql-to-pg",
141 * source: { engine: "MYSQL", version: "8.0" },
142 * destination: { engine: "POSTGRESQL", version: "14" },
143 * });
144 * ```
145 *
146 * **Example:** Oracle to PostgreSQL with settings
147 * ```typescript
148 * const workspace = yield* GCP.DataMigration.ConversionWorkspace("OraToPg", {
149 * source: { engine: "ORACLE", version: "19" },
150 * destination: { engine: "POSTGRESQL", version: "15" },
151 * globalSettings: { skip_triggers: "true" },
152 * });
153 * ```
154 *
155 * @resource
156 * @category DataMigration
157 */
158export const ConversionWorkspace = Resource<ConversionWorkspace>(
159 "GCP.DataMigration.ConversionWorkspace",
160);
161
162const resourceName = (
163 project: string,
164 location: string,
165 conversionWorkspaceId: string,
166) =>
167 `${locationParent(project, location)}/conversionWorkspaces/${conversionWorkspaceId}`;
168
169const settingsOf = (
170 value: Record<string, string | undefined> | null | undefined,
171): Record<string, string> => {
172 const next: Record<string, string> = {};
173 for (const [key, item] of Object.entries(value ?? {})) {
174 if (item !== undefined) next[key] = item;
175 }
176 return next;
177};
178
179const toAttrs = (workspace: dm.ConversionWorkspace, project: string) => {
180 const name = workspace.name ?? "";
181 const parsed = parseName(name, "conversionWorkspaces");
182 const ownership = parseOwnership(workspace.displayName);
183 return {
184 name,
185 conversionWorkspaceId: parsed.id,
186 project: parsed.project || project,
187 location: parsed.location,
188 displayName: ownership.text,
189 source: workspace.source,
190 destination: workspace.destination,
191 sourceProvider: workspace.sourceProvider,
192 destinationProvider: workspace.destinationProvider,
193 globalSettings: settingsOf(workspace.globalSettings),
194 hasUncommittedChanges: workspace.hasUncommittedChanges,
195 latestCommitId: workspace.latestCommitId,
196 latestCommitTime: workspace.latestCommitTime,
197 createTime: workspace.createTime,
198 updateTime: workspace.updateTime,
199 };
200};
201
202const getByName = (name: string) =>
203 name.length === 0
204 ? Effect.succeed(undefined)
205 : dm
206 .getProjectsLocationsConversionWorkspaces({ name })
207 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
208
209const listOwned = (project: string, region: string) =>
210 dm.listProjectsLocationsConversionWorkspaces
211 .pages({
212 parent: `projects/${project}/locations/-`,
213 pageSize: 1000,
214 })
215 .pipe(
216 Stream.flatMap((page) =>
217 Stream.fromIterable(page.conversionWorkspaces ?? []),
218 ),
219 Stream.filter((item) => hasOwnershipMarker(item.displayName)),
220 Stream.runCollect,
221 Effect.map((chunk) => Array.from(chunk)),
222 Effect.catchTag("NotFound", () =>
223 dm.listProjectsLocationsConversionWorkspaces
224 .pages({
225 parent: locationParent(project, region),
226 pageSize: 1000,
227 })
228 .pipe(
229 Stream.flatMap((page) =>
230 Stream.fromIterable(page.conversionWorkspaces ?? []),
231 ),
232 Stream.filter((item) => hasOwnershipMarker(item.displayName)),
233 Stream.runCollect,
234 Effect.map((chunk) => Array.from(chunk)),
235 Effect.catchTag("NotFound", () =>
236 Effect.succeed([] as dm.ConversionWorkspace[]),
237 ),
238 ),
239 ),
240 );
241
242const engineOf = (value: DatabaseEngineInfo | undefined) =>
243 fingerprint({
244 engine: value?.engine,
245 version: value?.version,
246 });
247
248export const ConversionWorkspaceProvider = () =>
249 Provider.succeed(ConversionWorkspace, {
250 stables: [
251 "name",
252 "conversionWorkspaceId",
253 "project",
254 "location",
255 "createTime",
256 ],
257
258 diff: Effect.fn(function* ({ news, olds, output }) {
259 if (!isResolved(news)) return undefined;
260 const env = yield* GcpEnvironment.current;
261 return replaceOnIdentity({
262 previousId:
263 olds?.conversionWorkspaceId ?? output?.conversionWorkspaceId,
264 nextId:
265 news.conversionWorkspaceId ??
266 olds?.conversionWorkspaceId ??
267 output?.conversionWorkspaceId,
268 previousLocation: normalizeLocation(
269 olds?.location ?? output?.location,
270 env.region,
271 ),
272 nextLocation: normalizeLocation(
273 news.location ?? olds?.location ?? output?.location,
274 env.region,
275 ),
276 extra:
277 engineOf(news.source) !== engineOf(olds?.source ?? output?.source) ||
278 engineOf(news.destination) !==
279 engineOf(olds?.destination ?? output?.destination) ||
280 (news.sourceProvider ??
281 olds?.sourceProvider ??
282 output?.sourceProvider) !==
283 (olds?.sourceProvider ?? output?.sourceProvider) ||
284 (news.destinationProvider ??
285 olds?.destinationProvider ??
286 output?.destinationProvider) !==
287 (olds?.destinationProvider ?? output?.destinationProvider),
288 });
289 }),
290
291 read: Effect.fn(function* ({ id, olds, output }) {
292 const env = yield* GcpEnvironment.current;
293 const conversionWorkspaceId = yield* toPhysicalId(
294 id,
295 olds?.conversionWorkspaceId,
296 output?.conversionWorkspaceId,
297 "workspace",
298 );
299 const location = normalizeLocation(
300 olds?.location ?? output?.location,
301 env.region,
302 );
303 const name =
304 output?.name ??
305 resourceName(env.project, location, conversionWorkspaceId);
306 const existing = yield* getByName(name);
307 if (existing === undefined) return undefined;
308 const attrs = toAttrs(existing, env.project);
309 return (yield* ownedByAlchemy(id, existing.displayName))
310 ? attrs
311 : Unowned(attrs);
312 }),
313
314 list: () =>
315 Effect.gen(function* () {
316 const env = yield* GcpEnvironment.current;
317 const items = yield* listOwned(env.project, env.region);
318 return items.map((item) => toAttrs(item, env.project));
319 }),
320
321 reconcile: Effect.fn(function* ({ id, news, output }) {
322 const env = yield* GcpEnvironment.current;
323 const conversionWorkspaceId = yield* toPhysicalId(
324 id,
325 news.conversionWorkspaceId,
326 output?.conversionWorkspaceId,
327 "workspace",
328 );
329 const location = normalizeLocation(
330 news.location ?? output?.location,
331 env.region,
332 );
333 const name = resourceName(env.project, location, conversionWorkspaceId);
334 const ownership = yield* createInternalLabels(id);
335 const displayName = encodeOwnershipLine(ownership, news.displayName);
336 const globalSettings = news.globalSettings;
337
338 let current = yield* getByName(output?.name ?? name);
339
340 if (current === undefined) {
341 const created = yield* dm
342 .createProjectsLocationsConversionWorkspaces({
343 parent: locationParent(env.project, location),
344 conversionWorkspaceId,
345 body: {
346 displayName,
347 source: news.source,
348 destination: news.destination,
349 sourceProvider: news.sourceProvider,
350 destinationProvider: news.destinationProvider,
351 globalSettings,
352 },
353 })
354 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
355 if (created !== undefined) {
356 yield* waitForOperation(created);
357 }
358 current = yield* waitUntilExists(getByName(name), name);
359 }
360
361 if (current === undefined) {
362 return yield* new ResourceNotResolved({ name });
363 }
364
365 const observedDisplay = current.displayName ?? "";
366 const displayNameChanged = observedDisplay !== displayName;
367 const settingsChanged =
368 fingerprint(settingsOf(current.globalSettings)) !==
369 fingerprint(globalSettings);
370 const mask = fieldMask([
371 displayNameChanged && "displayName",
372 settingsChanged && "globalSettings",
373 ]);
374
375 if (mask.length > 0) {
376 const operation = yield* dm.patchProjectsLocationsConversionWorkspaces({
377 name: current.name ?? name,
378 updateMask: mask,
379 body: {
380 name: current.name ?? name,
381 displayName,
382 globalSettings,
383 },
384 });
385 yield* waitForOperation(operation);
386 current = yield* waitUntilExists(
387 getByName(current.name ?? name),
388 current.name ?? name,
389 );
390 }
391
392 return toAttrs(current, env.project);
393 }),
394
395 delete: Effect.fn(function* ({ output }) {
396 const operation = yield* dm
397 .deleteProjectsLocationsConversionWorkspaces({
398 name: output.name,
399 force: true,
400 })
401 .pipe(
402 Effect.retry({
403 while: (error) => error._tag === "Conflict",
404 times: 8,
405 schedule: Schedule.spaced("2 seconds"),
406 }),
407 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
408 );
409 if (operation !== undefined) {
410 yield* waitForOperation(operation, { notFoundOk: true });
411 }
412 yield* waitUntilGone(getByName(output.name), output.name);
413 }),
414 });