1import * as mc from "@distilled.cloud/gcp/migrationcenter_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 { tagRecord } from "../../Tags.ts";
10import { GcpEnvironment } from "../Environment.ts";
11import {
12 createInternalLabels,
13 diffLabels,
14 hasAlchemyLabels,
15 toLabels,
16} from "../Labels.ts";
17import type { Providers } from "../Providers.ts";
18import {
19 expandParent,
20 fieldMask,
21 hasAlchemyLabelMap,
22 locationParent,
23 normalizeLocation,
24 parseName,
25 replaceOnIdentity,
26 ResourceNotResolved,
27 toPhysicalId,
28 userLabels,
29 waitForOperation,
30 waitUntilExists,
31 waitUntilGone,
32} from "./internal.ts";
33
34export type ImportJobState = mc.ImportJobStateEnum | (string & {});
35
36export type ImportJobProps = {
37 /**
38 * Import job id (the `{importJob}` segment of
39 * `projects/{project}/locations/{location}/importJobs/{importJob}`). If
40 * omitted, a unique RFC1035 name is generated. Immutable — changing it
41 * replaces the job.
42 */
43 importJobId?: string;
44 /**
45 * Region (`us-central1`, …). Immutable — changing it replaces the job.
46 * `US-CENTRAL1` is accepted and normalized to `us-central1`.
47 * @default the stack's GCP region (`GCP.Region`, else the profile
48 * region, else `us-central1`)
49 */
50 location?: string;
51 /**
52 * Source that receives imported frames. Full name or source id
53 * (combined with `location`). Immutable — changing it replaces the job.
54 */
55 assetSource: string;
56 /**
57 * User-friendly display name. Maximum length is 256 characters.
58 */
59 displayName?: string;
60 /**
61 * User labels. Alchemy ownership labels are merged in automatically.
62 */
63 labels?: Record<string, string>;
64};
65
66export type ImportJob = Resource<
67 "GCP.MigrationCenter.ImportJob",
68 ImportJobProps,
69 {
70 /** Full resource name. */
71 name: string;
72 /** Import job id (last path segment). */
73 importJobId: string;
74 /** Project id. */
75 project: string;
76 /** Location id. */
77 location: string;
78 /** Source that receives imported frames. */
79 assetSource: string | undefined;
80 /** User-friendly display name. */
81 displayName: string | undefined;
82 /** User labels (Alchemy ownership labels stripped). */
83 labels: Record<string, string>;
84 /** Server-reported state. */
85 state: string | undefined;
86 /** RFC3339 creation timestamp. */
87 createTime: string | undefined;
88 /** RFC3339 last-update timestamp. */
89 updateTime: string | undefined;
90 /** RFC3339 completion timestamp. */
91 completeTime: string | undefined;
92 },
93 never,
94 Providers
95>;
96
97/**
98 * A Migration Center import job that ingests asset frames from uploaded
99 * payload files into a source.
100 *
101 * `importJobId`, `location`, and `assetSource` are immutable. Display
102 * name and labels update in place. Nested import data files are force-
103 * deleted with the job.
104 *
105 * ### Creating an Import Job
106 * **Example:** Bind to an upload source
107 * ```typescript
108 * const source = yield* GCP.MigrationCenter.Source("Inventory", {
109 * type: "SOURCE_TYPE_UPLOAD",
110 * });
111 * const job = yield* GCP.MigrationCenter.ImportJob("Rvtools", {
112 * assetSource: source.name,
113 * displayName: "rvtools-import",
114 * });
115 * ```
116 *
117 * @resource
118 * @category MigrationCenter
119 */
120export const ImportJob = Resource<ImportJob>("GCP.MigrationCenter.ImportJob");
121
122const resourceName = (project: string, location: string, importJobId: string) =>
123 `${locationParent(project, location)}/importJobs/${importJobId}`;
124
125const sourceOf = (value: string, project: string, location: string) =>
126 expandParent(value, project, location, "sources");
127
128const toAttrs = (item: mc.ImportJob, project: string, region: string) => {
129 const name = item.name ?? "";
130 const parsed = parseName(name, "importJobs", region);
131 return {
132 name,
133 importJobId: parsed.id,
134 project: parsed.project || project,
135 location: parsed.location,
136 assetSource: item.assetSource,
137 displayName: item.displayName,
138 labels: userLabels(item.labels),
139 state: item.state,
140 createTime: item.createTime,
141 updateTime: item.updateTime,
142 completeTime: item.completeTime,
143 };
144};
145
146const getByName = (name: string) =>
147 name.length === 0
148 ? Effect.succeed(undefined)
149 : mc
150 .getProjectsLocationsImportJobs({
151 name,
152 view: "IMPORT_JOB_VIEW_FULL",
153 })
154 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
155
156const listOwned = (project: string, region: string) =>
157 mc.listProjectsLocationsImportJobs
158 .pages({
159 parent: `projects/${project}/locations/-`,
160 pageSize: 1000,
161 view: "IMPORT_JOB_VIEW_BASIC",
162 })
163 .pipe(
164 Stream.flatMap((page) => Stream.fromIterable(page.importJobs ?? [])),
165 Stream.filter((item) => hasAlchemyLabelMap(item.labels)),
166 Stream.runCollect,
167 Effect.map((chunk) => Array.from(chunk)),
168 Effect.catchTag("NotFound", () =>
169 mc.listProjectsLocationsImportJobs
170 .pages({
171 parent: locationParent(project, region),
172 pageSize: 1000,
173 view: "IMPORT_JOB_VIEW_BASIC",
174 })
175 .pipe(
176 Stream.flatMap((page) =>
177 Stream.fromIterable(page.importJobs ?? []),
178 ),
179 Stream.filter((item) => hasAlchemyLabelMap(item.labels)),
180 Stream.runCollect,
181 Effect.map((chunk) => Array.from(chunk)),
182 Effect.catchTag("NotFound", () =>
183 Effect.succeed([] as mc.ImportJob[]),
184 ),
185 ),
186 ),
187 );
188
189export const ImportJobProvider = () =>
190 Provider.succeed(ImportJob, {
191 stables: ["name", "importJobId", "project", "location", "createTime"],
192
193 diff: Effect.fn(function* ({ news, olds, output }) {
194 if (!isResolved(news)) return undefined;
195 const env = yield* GcpEnvironment.current;
196 const previousSource = olds?.assetSource ?? output?.assetSource;
197 const nextSource = news.assetSource;
198 const sourceChanged =
199 previousSource !== undefined &&
200 nextSource !== undefined &&
201 previousSource !== nextSource &&
202 !previousSource.endsWith(`/${nextSource}`) &&
203 !nextSource.endsWith(`/${previousSource}`);
204 return replaceOnIdentity({
205 previousId: olds?.importJobId ?? output?.importJobId,
206 nextId: news.importJobId ?? olds?.importJobId ?? output?.importJobId,
207 previousLocation: normalizeLocation(
208 olds?.location ?? output?.location,
209 env.region,
210 ),
211 nextLocation: normalizeLocation(
212 news.location ?? olds?.location ?? output?.location,
213 env.region,
214 ),
215 extra: sourceChanged,
216 });
217 }),
218
219 read: Effect.fn(function* ({ id, olds, output }) {
220 const env = yield* GcpEnvironment.current;
221 const importJobId = yield* toPhysicalId(
222 id,
223 olds?.importJobId,
224 output?.importJobId,
225 "importjob",
226 );
227 const location = normalizeLocation(
228 olds?.location ?? output?.location,
229 env.region,
230 );
231 const name =
232 output?.name ?? resourceName(env.project, location, importJobId);
233 const existing = yield* getByName(name);
234 if (existing === undefined) return undefined;
235 const attrs = toAttrs(existing, env.project, env.region);
236 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
237 ? attrs
238 : Unowned(attrs);
239 }),
240
241 list: () =>
242 Effect.gen(function* () {
243 const env = yield* GcpEnvironment.current;
244 const items = yield* listOwned(env.project, env.region);
245 return items.map((item) => toAttrs(item, env.project, env.region));
246 }),
247
248 reconcile: Effect.fn(function* ({ id, news, output }) {
249 const env = yield* GcpEnvironment.current;
250 const importJobId = yield* toPhysicalId(
251 id,
252 news.importJobId,
253 output?.importJobId,
254 "importjob",
255 );
256 const location = normalizeLocation(
257 news.location ?? output?.location,
258 env.region,
259 );
260 const name = resourceName(env.project, location, importJobId);
261 const assetSource = sourceOf(news.assetSource, env.project, location);
262 const desiredLabels = {
263 ...toLabels(news.labels),
264 ...(yield* createInternalLabels(id)),
265 };
266 const displayName = news.displayName ?? importJobId;
267
268 let current = yield* getByName(output?.name ?? name);
269
270 if (current === undefined) {
271 const created = yield* mc
272 .createProjectsLocationsImportJobs({
273 parent: locationParent(env.project, location),
274 importJobId,
275 body: {
276 assetSource,
277 displayName,
278 labels: desiredLabels,
279 },
280 })
281 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
282 if (created !== undefined) {
283 yield* waitForOperation(created);
284 }
285 current = yield* waitUntilExists(getByName(name), name);
286 }
287
288 if (current === undefined) {
289 return yield* new ResourceNotResolved({ name });
290 }
291
292 const observedLabels = tagRecord(current.labels);
293 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
294 const labelsChanged = upsert.length > 0 || removed.length > 0;
295 const displayNameChanged = (current.displayName ?? "") !== displayName;
296 const mask = fieldMask([
297 labelsChanged && "labels",
298 displayNameChanged && "displayName",
299 ]);
300
301 if (mask.length > 0) {
302 const operation = yield* mc.patchProjectsLocationsImportJobs({
303 name: current.name ?? name,
304 updateMask: mask,
305 body: {
306 name: current.name ?? name,
307 displayName,
308 labels: desiredLabels,
309 },
310 });
311 yield* waitForOperation(operation);
312 current = yield* waitUntilExists(
313 getByName(current.name ?? name),
314 current.name ?? name,
315 );
316 }
317
318 return toAttrs(current, env.project, env.region);
319 }),
320
321 delete: Effect.fn(function* ({ output }) {
322 const operation = yield* mc
323 .deleteProjectsLocationsImportJobs({
324 name: output.name,
325 force: true,
326 })
327 .pipe(
328 Effect.retry({
329 while: (error) => error._tag === "Conflict",
330 times: 8,
331 schedule: Schedule.spaced("2 seconds"),
332 }),
333 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
334 );
335 if (operation !== undefined) {
336 yield* waitForOperation(operation, { notFoundOk: true });
337 }
338 yield* waitUntilGone(getByName(output.name), output.name);
339 }),
340 });