1import * as apim from "@distilled.cloud/gcp/apim_v1alpha";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import { GcpEnvironment } from "../Environment.ts";
9import type { Providers } from "../Providers.ts";
10import {
11 ResourceFailed,
12 ResourceNotReady,
13 ResourceNotResolved,
14 expandObservationSource,
15 hasAlchemyId,
16 listAtLocation,
17 collectPages,
18 normalizeLocation,
19 parentOf,
20 parseName,
21 replaceOnIdentity,
22 resourceName,
23 sameStringList,
24 toPhysicalId,
25 waitForOperation,
26 waitUntilExists,
27 waitUntilGone,
28 waitUntilReady,
29} from "./internal.ts";
30
31const COLLECTION = "observationJobs";
32const CREATED = new Set(["DISABLED", "ENABLED"]);
33const ENABLED = new Set(["ENABLED"]);
34const DISABLED = new Set(["DISABLED"]);
35
36export type ObservationJobProps = {
37 /**
38 * Observation job id (the `{observation_job}` segment of
39 * `projects/{project}/locations/{location}/observationJobs/{observation_job}`).
40 * Must be 4-63 lowercase letters, digits, or hyphens. If omitted, a
41 * unique `alch-` prefixed name is generated. Immutable — changing it
42 * replaces the job.
43 */
44 observationJobId?: string;
45 /**
46 * Region of the job (`us-central1`, …). Immutable — changing it
47 * replaces the job.
48 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
49 */
50 location?: string;
51 /**
52 * Observation source names
53 * (`projects/{project}/locations/{location}/observationSources/{source}`)
54 * or bare source ids in the same location. Sources must be the same
55 * kind. Immutable — changing them replaces the job.
56 */
57 sources?: string[];
58 /**
59 * When true, the job is enabled and collects observations. Synced in
60 * place via enable/disable. Jobs are created disabled.
61 * @default false
62 */
63 enabled?: boolean;
64};
65
66export type ObservationJob = Resource<
67 "GCP.Apim.ObservationJob",
68 ObservationJobProps,
69 {
70 /** Full resource name. */
71 name: string;
72 /** Observation job id (last path segment). */
73 observationJobId: string;
74 /** Project id. */
75 project: string;
76 /** Location id. */
77 location: string;
78 /** Observation source resource names. */
79 sources: string[];
80 /** Whether the job is enabled. */
81 enabled: boolean;
82 /** Server-reported state. */
83 state: string | undefined;
84 /** RFC3339 creation timestamp. */
85 createTime: string | undefined;
86 /** RFC3339 last-update timestamp. */
87 updateTime: string | undefined;
88 },
89 never,
90 Providers
91>;
92
93/**
94 * An API Observation job — configuration that collects data about APIs
95 * from attached observation sources. Jobs have no labels, description,
96 * or patch API; Alchemy stamps ownership into generated ids (`alch-`)
97 * so `list` / nuke can find them. Identity, location, and sources
98 * replace the resource. `enabled` is synced in place.
99 *
100 * ### Creating an Observation Job
101 * **Example:** Disabled job with a source
102 * ```typescript
103 * const job = yield* GCP.Apim.ObservationJob("Shadow", {
104 * sources: [source.name],
105 * });
106 * ```
107 *
108 * **Example:** Enable collection
109 * ```typescript
110 * const job = yield* GCP.Apim.ObservationJob("Shadow", {
111 * sources: [source.name],
112 * enabled: true,
113 * });
114 * ```
115 *
116 * @resource
117 * @category Apim
118 */
119export const ObservationJob = Resource<ObservationJob>(
120 "GCP.Apim.ObservationJob",
121);
122
123const toAttrs = (
124 location: string,
125 item: apim.ObservationJob,
126 project: string,
127) => {
128 const name = item.name ?? "";
129 const parsed = parseName(name, COLLECTION);
130 const state = item.state;
131 return {
132 name,
133 observationJobId: parsed.id,
134 project: parsed.project || project,
135 location: parsed.location || location,
136 sources: [...(item.sources ?? [])],
137 enabled: (state ?? "").toUpperCase() === "ENABLED",
138 state,
139 createTime: item.createTime,
140 updateTime: item.updateTime,
141 };
142};
143
144const getByName = (name: string) =>
145 name.length === 0
146 ? Effect.succeed(undefined)
147 : apim
148 .getProjectsLocationsObservationJobs({ name })
149 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
150
151const listOwned = (project: string, region: string) =>
152 listAtLocation(project, region, (parent) =>
153 collectPages(
154 apim.listProjectsLocationsObservationJobs.pages({
155 parent,
156 pageSize: 1000,
157 }),
158 (page) => page.observationJobs,
159 ),
160 ).pipe(
161 Effect.map((items) =>
162 items.filter((item) =>
163 hasAlchemyId(parseName(item.name ?? "", COLLECTION).id),
164 ),
165 ),
166 );
167
168const runAndWait = (operation: apim.Operation | undefined) =>
169 Effect.gen(function* () {
170 if (operation !== undefined) {
171 yield* waitForOperation(operation);
172 }
173 });
174
175export const ObservationJobProvider = () =>
176 Provider.succeed(ObservationJob, {
177 stables: ["name", "observationJobId", "project", "location", "createTime"],
178
179 diff: Effect.fn(function* ({ news, olds, output }) {
180 if (!isResolved(news)) return undefined;
181 const env = yield* GcpEnvironment.current;
182 const previousSources = olds?.sources ?? output?.sources;
183 return replaceOnIdentity({
184 previousId: olds?.observationJobId ?? output?.observationJobId,
185 nextId:
186 news.observationJobId ??
187 olds?.observationJobId ??
188 output?.observationJobId,
189 previousLocation: normalizeLocation(
190 olds?.location ?? output?.location ?? env.region,
191 ),
192 nextLocation: normalizeLocation(
193 news.location ?? olds?.location ?? output?.location ?? env.region,
194 ),
195 extra:
196 previousSources !== undefined &&
197 news.sources !== undefined &&
198 !sameStringList(previousSources, news.sources),
199 });
200 }),
201
202 read: Effect.fn(function* ({ id, olds, output }) {
203 const env = yield* GcpEnvironment.current;
204 const observationJobId = yield* toPhysicalId(
205 id,
206 olds?.observationJobId,
207 output?.observationJobId,
208 "job",
209 );
210 const location = normalizeLocation(
211 olds?.location ?? output?.location ?? env.region,
212 );
213 const name =
214 output?.name ??
215 resourceName(env.project, location, COLLECTION, observationJobId);
216 const existing = yield* getByName(name);
217 if (existing === undefined) return undefined;
218 const attrs = toAttrs(location, existing, env.project);
219 if (output !== undefined || hasAlchemyId(attrs.observationJobId)) {
220 return attrs;
221 }
222 return Unowned(attrs);
223 }),
224
225 list: () =>
226 Effect.gen(function* () {
227 const env = yield* GcpEnvironment.current;
228 const items = yield* listOwned(env.project, env.region);
229 return items.map((item) => toAttrs(env.region, item, env.project));
230 }),
231
232 reconcile: Effect.fn(function* ({ id, news, output }) {
233 const env = yield* GcpEnvironment.current;
234 const observationJobId = yield* toPhysicalId(
235 id,
236 news.observationJobId,
237 output?.observationJobId,
238 "job",
239 );
240 const location = normalizeLocation(
241 news.location ?? output?.location ?? env.region,
242 );
243 const name = resourceName(
244 env.project,
245 location,
246 COLLECTION,
247 observationJobId,
248 );
249 const sources = (news.sources ?? []).map((source) =>
250 expandObservationSource(source, env.project, location),
251 );
252 const wantEnabled = news.enabled === true;
253
254 let current = yield* getByName(output?.name ?? name);
255
256 if (current === undefined) {
257 const created = yield* apim
258 .createProjectsLocationsObservationJobs({
259 parent: parentOf(env.project, location),
260 observationJobId,
261 body: {
262 sources: sources.length > 0 ? sources : undefined,
263 },
264 })
265 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
266 if (created !== undefined) {
267 yield* waitForOperation(created);
268 }
269 current = yield* waitUntilExists(getByName(name), name);
270 }
271
272 if (current === undefined) {
273 return yield* new ResourceNotResolved({ name });
274 }
275
276 const resource = current.name ?? name;
277 current = yield* waitUntilReady(
278 getByName(resource),
279 resource,
280 (item) => item.state,
281 CREATED,
282 );
283
284 const state = (current.state ?? "").toUpperCase();
285 if (wantEnabled && state !== "ENABLED") {
286 const operation = yield* apim
287 .enableProjectsLocationsObservationJobs({
288 name: resource,
289 body: {},
290 })
291 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
292 yield* runAndWait(operation);
293 current = yield* waitUntilReady(
294 getByName(resource),
295 resource,
296 (item) => item.state,
297 ENABLED,
298 );
299 } else if (!wantEnabled && state === "ENABLED") {
300 const operation = yield* apim
301 .disableProjectsLocationsObservationJobs({
302 name: resource,
303 body: {},
304 })
305 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
306 yield* runAndWait(operation);
307 current = yield* waitUntilReady(
308 getByName(resource),
309 resource,
310 (item) => item.state,
311 DISABLED,
312 );
313 }
314
315 return toAttrs(location, current, env.project);
316 }),
317
318 delete: Effect.fn(function* ({ output }) {
319 const existing = yield* getByName(output.name);
320 const state = (existing?.state ?? "").toUpperCase();
321 if (
322 existing !== undefined &&
323 (state === "ENABLED" || state === "ENABLING")
324 ) {
325 const disable = yield* apim
326 .disableProjectsLocationsObservationJobs({
327 name: output.name,
328 body: {},
329 })
330 .pipe(
331 Effect.catchTag(["NotFound", "Conflict"], () =>
332 Effect.succeed(undefined),
333 ),
334 );
335 if (disable !== undefined) {
336 yield* waitForOperation(disable, { notFoundOk: true });
337 }
338 yield* waitUntilReady(
339 getByName(output.name),
340 output.name,
341 (item) => item.state,
342 DISABLED,
343 ).pipe(
344 Effect.catchIf(
345 (error) =>
346 error instanceof ResourceNotResolved ||
347 error instanceof ResourceNotReady ||
348 error instanceof ResourceFailed,
349 () => Effect.void,
350 ),
351 );
352 }
353
354 const operation = yield* apim
355 .deleteProjectsLocationsObservationJobs({ name: output.name })
356 .pipe(
357 Effect.retry({
358 while: (error) => error._tag === "Conflict",
359 times: 8,
360 schedule: Schedule.spaced("2 seconds"),
361 }),
362 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
363 );
364 if (operation !== undefined) {
365 yield* waitForOperation(operation, { notFoundOk: true });
366 }
367 yield* waitUntilGone(getByName(output.name), output.name);
368 }),
369 });