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 conversionWorkspaceOf,
14 encodeOwnershipLine,
15 fingerprint,
16 hasOwnershipMarker,
17 listConversionWorkspaces,
18 normalizeLocation,
19 ownedByAlchemy,
20 parseName,
21 parseOwnership,
22 replaceOnIdentity,
23 ResourceNotResolved,
24 toPhysicalId,
25 waitUntilExists,
26 waitUntilGone,
27} from "./internal.ts";
28
29export type MappingRuleRuleScope = dm.MappingRuleRuleScopeEnum | (string & {});
30export type MappingRuleState = dm.MappingRuleStateEnum | (string & {});
31export type MappingRuleFilter = dm.MappingRuleFilter;
32export type FilterTableColumns = dm.FilterTableColumns;
33export type SinglePackageChange = dm.SinglePackageChange;
34export type SingleEntityRename = dm.SingleEntityRename;
35export type SetTablePrimaryKey = dm.SetTablePrimaryKey;
36export type MultiEntityRename = dm.MultiEntityRename;
37export type EntityMove = dm.EntityMove;
38export type SourceSqlChange = dm.SourceSqlChange;
39export type MultiColumnDatatypeChange = dm.MultiColumnDatatypeChange;
40export type ConvertRowIdToColumn = dm.ConvertRowIdToColumn;
41export type SingleColumnChange = dm.SingleColumnChange;
42export type ConditionalColumnSetValue = dm.ConditionalColumnSetValue;
43
44export type ConversionWorkspacesMappingRuleProps = {
45 /**
46 * Parent conversion workspace. Full name
47 * `projects/{project}/locations/{location}/conversionWorkspaces/{conversionWorkspace}`
48 * or the workspace id (combined with `location`). Immutable — changing
49 * it replaces the rule.
50 */
51 conversionWorkspace: string;
52 /**
53 * Region used when `conversionWorkspace` is a bare id.
54 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
55 */
56 location?: string;
57 /**
58 * Mapping rule id (the `{mappingRule}` segment). If omitted, a unique
59 * RFC1035 name is generated. Immutable — changing it replaces the rule.
60 */
61 mappingRuleId?: string;
62 /**
63 * Rule scope (`DATABASE_ENTITY_TYPE_SCHEMA`,
64 * `DATABASE_ENTITY_TYPE_TABLE`, …). The API has no patch method —
65 * changing the rule body replaces the rule.
66 */
67 ruleScope: MappingRuleRuleScope;
68 /**
69 * Entities the rule applies to.
70 */
71 filter?: MappingRuleFilter;
72 /**
73 * Application order. Lower values run first.
74 * @default "1"
75 */
76 ruleOrder?: string;
77 /**
78 * User-friendly display name. Mapping rules have no labels field, so
79 * ownership is stored in a `[alchemy …]` prefix and stripped from
80 * attributes. The API has no patch method — changing display name
81 * replaces the rule.
82 */
83 displayName?: string;
84 /**
85 * Mapping rule state (`ENABLED` or `DISABLED`).
86 * @default "ENABLED"
87 */
88 state?: MappingRuleState;
89 /** Rename a single entity. */
90 singleEntityRename?: SingleEntityRename;
91 /** Rename multiple entities. */
92 multiEntityRename?: MultiEntityRename;
93 /** Move entities into another schema. */
94 entityMove?: EntityMove;
95 /** Change SQL for a function or procedure. */
96 sourceSqlChange?: SourceSqlChange;
97 /** Change a single package. */
98 singlePackageChange?: SinglePackageChange;
99 /** Set a table primary key. */
100 setTablePrimaryKey?: SetTablePrimaryKey;
101 /** Filter table columns. */
102 filterTableColumns?: FilterTableColumns;
103 /** Change multiple column data types. */
104 multiColumnDataTypeChange?: MultiColumnDatatypeChange;
105 /** Convert ROWID to a column. */
106 convertRowidColumn?: ConvertRowIdToColumn;
107 /** Change a single column. */
108 singleColumnChange?: SingleColumnChange;
109 /** Conditionally transform column values. */
110 conditionalColumnSetValue?: ConditionalColumnSetValue;
111};
112
113export type ConversionWorkspacesMappingRule = Resource<
114 "GCP.DataMigration.ConversionWorkspacesMappingRule",
115 ConversionWorkspacesMappingRuleProps,
116 {
117 /** Full resource name. */
118 name: string;
119 /** Mapping rule id (last path segment). */
120 mappingRuleId: string;
121 /** Parent conversion workspace resource name. */
122 conversionWorkspace: string;
123 /** Project id. */
124 project: string;
125 /** Location id. */
126 location: string;
127 /** Rule scope. */
128 ruleScope: string | undefined;
129 /** Entity filter. */
130 filter: MappingRuleFilter | undefined;
131 /** Application order. */
132 ruleOrder: string | undefined;
133 /** User display name with the Alchemy ownership prefix stripped. */
134 displayName: string | undefined;
135 /** Mapping rule state. */
136 state: string | undefined;
137 /** Single-entity rename. */
138 singleEntityRename: SingleEntityRename | undefined;
139 /** Multi-entity rename. */
140 multiEntityRename: MultiEntityRename | undefined;
141 /** Entity move. */
142 entityMove: EntityMove | undefined;
143 /** Source SQL change. */
144 sourceSqlChange: SourceSqlChange | undefined;
145 /** Package change. */
146 singlePackageChange: SinglePackageChange | undefined;
147 /** Primary key change. */
148 setTablePrimaryKey: SetTablePrimaryKey | undefined;
149 /** Table column filter. */
150 filterTableColumns: FilterTableColumns | undefined;
151 /** Multi-column data type change. */
152 multiColumnDataTypeChange: MultiColumnDatatypeChange | undefined;
153 /** ROWID-to-column conversion. */
154 convertRowidColumn: ConvertRowIdToColumn | undefined;
155 /** Single-column change. */
156 singleColumnChange: SingleColumnChange | undefined;
157 /** Conditional column value transform. */
158 conditionalColumnSetValue: ConditionalColumnSetValue | undefined;
159 /** Revision id. */
160 revisionId: string | undefined;
161 /** RFC3339 revision timestamp. */
162 revisionCreateTime: string | undefined;
163 },
164 never,
165 Providers
166>;
167
168/**
169 * A mapping rule attached to a Database Migration Service conversion
170 * workspace. Rules transform source schema entities into the destination
171 * schema (rename, move, type change, …).
172 *
173 * Mapping rules have no labels field and no patch method — Alchemy stamps
174 * ownership into the display name, and any change to the rule body
175 * replaces the rule.
176 *
177 * ### Creating a Mapping Rule
178 * **Example:** Rename a schema
179 * ```typescript
180 * const workspace = yield* GCP.DataMigration.ConversionWorkspace("MysqlToPg", {
181 * source: { engine: "MYSQL", version: "8.0" },
182 * destination: { engine: "POSTGRESQL", version: "14" },
183 * });
184 * const rule = yield* GCP.DataMigration.ConversionWorkspacesMappingRule("Rename", {
185 * conversionWorkspace: workspace.name,
186 * ruleScope: "DATABASE_ENTITY_TYPE_SCHEMA",
187 * filter: { entities: ["src_schema"] },
188 * singleEntityRename: { newName: "dst_schema" },
189 * });
190 * ```
191 *
192 * @resource
193 * @category DataMigration
194 */
195export const ConversionWorkspacesMappingRule =
196 Resource<ConversionWorkspacesMappingRule>(
197 "GCP.DataMigration.ConversionWorkspacesMappingRule",
198 );
199
200const DEFAULT_ORDER = "1";
201
202const resourceName = (conversionWorkspace: string, mappingRuleId: string) =>
203 `${conversionWorkspace}/mappingRules/${mappingRuleId}`;
204
205// The API omits `name` from mapping rules (create and get alike), so callers
206// that know the rule's name pass it in.
207const toAttrs = (item: dm.MappingRule, project: string, knownName?: string) => {
208 const name = item.name || knownName || "";
209 const parsed = parseName(name, "mappingRules");
210 const ownership = parseOwnership(item.displayName);
211 return {
212 name,
213 mappingRuleId: parsed.id,
214 conversionWorkspace: parsed.parent,
215 project: parsed.project || project,
216 location: parsed.location,
217 ruleScope: item.ruleScope,
218 filter: item.filter,
219 ruleOrder: item.ruleOrder,
220 displayName: ownership.text,
221 state: item.state,
222 singleEntityRename: item.singleEntityRename,
223 multiEntityRename: item.multiEntityRename,
224 entityMove: item.entityMove,
225 sourceSqlChange: item.sourceSqlChange,
226 singlePackageChange: item.singlePackageChange,
227 setTablePrimaryKey: item.setTablePrimaryKey,
228 filterTableColumns: item.filterTableColumns,
229 multiColumnDataTypeChange: item.multiColumnDataTypeChange,
230 convertRowidColumn: item.convertRowidColumn,
231 singleColumnChange: item.singleColumnChange,
232 conditionalColumnSetValue: item.conditionalColumnSetValue,
233 revisionId: item.revisionId,
234 revisionCreateTime: item.revisionCreateTime,
235 };
236};
237
238const ruleBody = (item: {
239 ruleScope?: string;
240 filter?: MappingRuleFilter;
241 ruleOrder?: string;
242 displayName?: string;
243 state?: string;
244 singleEntityRename?: SingleEntityRename;
245 multiEntityRename?: MultiEntityRename;
246 entityMove?: EntityMove;
247 sourceSqlChange?: SourceSqlChange;
248 singlePackageChange?: SinglePackageChange;
249 setTablePrimaryKey?: SetTablePrimaryKey;
250 filterTableColumns?: FilterTableColumns;
251 multiColumnDataTypeChange?: MultiColumnDatatypeChange;
252 convertRowidColumn?: ConvertRowIdToColumn;
253 singleColumnChange?: SingleColumnChange;
254 conditionalColumnSetValue?: ConditionalColumnSetValue;
255}) =>
256 fingerprint({
257 ruleScope: item.ruleScope,
258 filter: item.filter ?? {},
259 ruleOrder: item.ruleOrder ?? DEFAULT_ORDER,
260 displayName: item.displayName,
261 state: item.state,
262 singleEntityRename: item.singleEntityRename,
263 multiEntityRename: item.multiEntityRename,
264 entityMove: item.entityMove,
265 sourceSqlChange: item.sourceSqlChange,
266 singlePackageChange: item.singlePackageChange,
267 setTablePrimaryKey: item.setTablePrimaryKey,
268 filterTableColumns: item.filterTableColumns,
269 multiColumnDataTypeChange: item.multiColumnDataTypeChange,
270 convertRowidColumn: item.convertRowidColumn,
271 singleColumnChange: item.singleColumnChange,
272 conditionalColumnSetValue: item.conditionalColumnSetValue,
273 });
274
275// A name with an empty segment means the parent workspace never resolved
276// (e.g. its create failed), so there is no rule to observe or delete.
277const isWellFormed = (name: string) => name.length > 0 && !name.includes("//");
278
279const getByName = (name: string) =>
280 !isWellFormed(name)
281 ? Effect.succeed(undefined)
282 : dm.getProjectsLocationsConversionWorkspacesMappingRules({ name }).pipe(
283 // Deleted rules stay readable as a `DELETED` revision.
284 Effect.map((rule) => (rule.state === "DELETED" ? undefined : rule)),
285 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
286 );
287
288const listRules = (parent: string) =>
289 dm.listProjectsLocationsConversionWorkspacesMappingRules
290 .pages({ parent, pageSize: 1000 })
291 .pipe(
292 Stream.flatMap((page) => Stream.fromIterable(page.mappingRules ?? [])),
293 Stream.runCollect,
294 Effect.map((chunk) => Array.from(chunk)),
295 Effect.catchTag("NotFound", () =>
296 Effect.succeed([] as dm.MappingRule[]),
297 ),
298 );
299
300const listOwned = (project: string, region: string) =>
301 Effect.gen(function* () {
302 const workspaces = yield* listConversionWorkspaces(project, region);
303 const groups = yield* Effect.forEach(
304 workspaces.filter((item) => (item.name ?? "").length > 0),
305 (workspace) => listRules(workspace.name!),
306 { concurrency: 4 },
307 );
308 return groups.flat().filter((item) => hasOwnershipMarker(item.displayName));
309 });
310
311export const ConversionWorkspacesMappingRuleProvider = () =>
312 Provider.succeed(ConversionWorkspacesMappingRule, {
313 stables: [
314 "name",
315 "mappingRuleId",
316 "conversionWorkspace",
317 "project",
318 "location",
319 ],
320
321 diff: Effect.fn(function* ({ news, olds, output }) {
322 if (!isResolved(news)) return undefined;
323 const env = yield* GcpEnvironment.current;
324 const previous = {
325 ruleScope: olds?.ruleScope ?? output?.ruleScope,
326 filter: olds?.filter ?? output?.filter,
327 ruleOrder: olds?.ruleOrder ?? output?.ruleOrder,
328 displayName: olds?.displayName ?? output?.displayName,
329 state: olds?.state ?? output?.state,
330 singleEntityRename:
331 olds?.singleEntityRename ?? output?.singleEntityRename,
332 multiEntityRename: olds?.multiEntityRename ?? output?.multiEntityRename,
333 entityMove: olds?.entityMove ?? output?.entityMove,
334 sourceSqlChange: olds?.sourceSqlChange ?? output?.sourceSqlChange,
335 singlePackageChange:
336 olds?.singlePackageChange ?? output?.singlePackageChange,
337 setTablePrimaryKey:
338 olds?.setTablePrimaryKey ?? output?.setTablePrimaryKey,
339 filterTableColumns:
340 olds?.filterTableColumns ?? output?.filterTableColumns,
341 multiColumnDataTypeChange:
342 olds?.multiColumnDataTypeChange ?? output?.multiColumnDataTypeChange,
343 convertRowidColumn:
344 olds?.convertRowidColumn ?? output?.convertRowidColumn,
345 singleColumnChange:
346 olds?.singleColumnChange ?? output?.singleColumnChange,
347 conditionalColumnSetValue:
348 olds?.conditionalColumnSetValue ?? output?.conditionalColumnSetValue,
349 };
350 return replaceOnIdentity({
351 previousId: olds?.mappingRuleId ?? output?.mappingRuleId,
352 nextId:
353 news.mappingRuleId ?? olds?.mappingRuleId ?? output?.mappingRuleId,
354 previousLocation: normalizeLocation(
355 olds?.location ?? output?.location,
356 env.region,
357 ),
358 nextLocation: normalizeLocation(
359 news.location ?? olds?.location ?? output?.location,
360 env.region,
361 ),
362 previousParent:
363 olds?.conversionWorkspace ?? output?.conversionWorkspace,
364 nextParent: news.conversionWorkspace,
365 extra:
366 (olds !== undefined || output !== undefined) &&
367 ruleBody(news) !== ruleBody(previous),
368 });
369 }),
370
371 read: Effect.fn(function* ({ id, olds, output }) {
372 const env = yield* GcpEnvironment.current;
373 const location = normalizeLocation(
374 olds?.location ?? output?.location,
375 env.region,
376 );
377 const conversionWorkspace = conversionWorkspaceOf(
378 olds?.conversionWorkspace ?? output?.conversionWorkspace ?? "",
379 env.project,
380 location,
381 );
382 const mappingRuleId = yield* toPhysicalId(
383 id,
384 olds?.mappingRuleId,
385 output?.mappingRuleId,
386 "rule",
387 );
388 const name =
389 output?.name ??
390 (conversionWorkspace.length > 0
391 ? resourceName(conversionWorkspace, mappingRuleId)
392 : "");
393 const existing = yield* getByName(name);
394 if (existing === undefined) return undefined;
395 const attrs = toAttrs(existing, env.project, name);
396 return (yield* ownedByAlchemy(id, existing.displayName))
397 ? attrs
398 : Unowned(attrs);
399 }),
400
401 list: () =>
402 Effect.gen(function* () {
403 const env = yield* GcpEnvironment.current;
404 const items = yield* listOwned(env.project, env.region);
405 return items.map((item) => toAttrs(item, env.project));
406 }),
407
408 reconcile: Effect.fn(function* ({ id, news, output }) {
409 const env = yield* GcpEnvironment.current;
410 const location = normalizeLocation(
411 news.location ?? output?.location,
412 env.region,
413 );
414 const conversionWorkspace = conversionWorkspaceOf(
415 news.conversionWorkspace,
416 env.project,
417 location,
418 );
419 const mappingRuleId = yield* toPhysicalId(
420 id,
421 news.mappingRuleId,
422 output?.mappingRuleId,
423 "rule",
424 );
425 const name = resourceName(conversionWorkspace, mappingRuleId);
426 const ownership = yield* createInternalLabels(id);
427 const displayName = encodeOwnershipLine(ownership, news.displayName);
428 const ruleOrder = news.ruleOrder ?? DEFAULT_ORDER;
429 const filter = news.filter ?? {};
430
431 let current = yield* getByName(output?.name ?? name);
432
433 if (current === undefined) {
434 const created = yield* dm
435 .createProjectsLocationsConversionWorkspacesMappingRules({
436 parent: conversionWorkspace,
437 mappingRuleId,
438 body: {
439 displayName,
440 ruleScope: news.ruleScope,
441 filter,
442 ruleOrder,
443 state: news.state ?? "ENABLED",
444 singleEntityRename: news.singleEntityRename,
445 multiEntityRename: news.multiEntityRename,
446 entityMove: news.entityMove,
447 sourceSqlChange: news.sourceSqlChange,
448 singlePackageChange: news.singlePackageChange,
449 setTablePrimaryKey: news.setTablePrimaryKey,
450 filterTableColumns: news.filterTableColumns,
451 multiColumnDataTypeChange: news.multiColumnDataTypeChange,
452 convertRowidColumn: news.convertRowidColumn,
453 singleColumnChange: news.singleColumnChange,
454 conditionalColumnSetValue: news.conditionalColumnSetValue,
455 },
456 })
457 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
458 if (created !== undefined) {
459 current = created;
460 } else {
461 current = yield* waitUntilExists(getByName(name), name);
462 }
463 }
464
465 if (current === undefined) {
466 return yield* new ResourceNotResolved({ name });
467 }
468
469 return toAttrs(current, env.project, output?.name ?? name);
470 }),
471
472 delete: Effect.fn(function* ({ output }) {
473 if (!isWellFormed(output.name)) return;
474 const deleted = yield* dm
475 .deleteProjectsLocationsConversionWorkspacesMappingRules({
476 name: output.name,
477 })
478 .pipe(
479 Effect.as(true),
480 Effect.retry({
481 while: (error) => error._tag === "Conflict",
482 times: 8,
483 schedule: Schedule.spaced("2 seconds"),
484 }),
485 // The API answers NotFound for rules it still serves on GET (they
486 // only go away with their workspace), so there is nothing to wait on.
487 Effect.catchTag("NotFound", () => Effect.succeed(false)),
488 );
489 if (deleted) {
490 yield* waitUntilGone(getByName(output.name), output.name);
491 }
492 }),
493 });