1import * as apigee from "@distilled.cloud/gcp/apigee_v1";
2import { createHash } from "node:crypto";
3import * as Data from "effect/Data";
4import * as Effect from "effect/Effect";
5import * as Schedule from "effect/Schedule";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import { createPhysicalName } from "../../PhysicalName.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { unzipFiles, zipFiles } from "../../Util/zip.ts";
12import { GcpEnvironment } from "../Environment.ts";
13import type { Providers } from "../Providers.ts";
14import {
15 createInternalLabels,
16 encodeOwnership,
17 hasAlchemyLabels,
18 hasOwnershipMarker,
19 lastSegment,
20 orgIdOf,
21 orgParent,
22 parseOwnership,
23 xmlEscape,
24 xmlUnescape,
25} from "./ownership.ts";
26
27const MAX_ID_LENGTH = 255;
28
29export type SharedflowProps = {
30 /**
31 * Shared flow id (the `{sharedflow}` segment of
32 * `organizations/{org}/sharedflows/{sharedflow}`). If omitted, a unique
33 * name is generated from the stack, stage, and logical id. Restrict
34 * characters to `A-Za-z0-9._-`. Immutable — changing it replaces the
35 * shared flow.
36 */
37 sharedflowId?: string;
38 /**
39 * Apigee organization id. Defaults to the stack GCP project. Immutable —
40 * changing it replaces the shared flow.
41 */
42 organization?: string;
43 /**
44 * Space id to associate with this shared flow. Only applied on create;
45 * subsequent space changes call `move`.
46 */
47 space?: string;
48 /**
49 * Base64-encoded ZIP shared-flow bundle. If omitted, Alchemy uploads a
50 * minimal empty shared flow. Shared flows have no labels field, so
51 * Alchemy ownership is stored in the bundle `Description` for `list` /
52 * nuke. Uploading a new bundle (or changing `description`) creates a
53 * new revision.
54 */
55 bundle?: string;
56 /**
57 * Human-readable description stored in the shared-flow bundle. Alchemy
58 * ownership (`alchemy-stack` / `alchemy-stage` / `alchemy-id`) is
59 * prefixed and stripped from attributes.
60 */
61 description?: string;
62};
63
64export type Sharedflow = Resource<
65 "GCP.Apigee.Sharedflow",
66 SharedflowProps,
67 {
68 /** Full resource name `organizations/{org}/sharedflows/{sharedflow}`. */
69 name: string;
70 /** Shared flow id (last path segment). */
71 sharedflowId: string;
72 /** Apigee organization id. */
73 organization: string;
74 /** Space id, if the shared flow is associated with a space. */
75 space: string | undefined;
76 /** Latest revision id. */
77 latestRevisionId: string | undefined;
78 /** Known revision ids. */
79 revision: string[];
80 /** User description with the Alchemy ownership prefix stripped. */
81 description: string | undefined;
82 /** Created-at timestamp in milliseconds since epoch. */
83 createdAt: string | undefined;
84 /** Last-modified timestamp in milliseconds since epoch. */
85 lastModifiedAt: string | undefined;
86 },
87 never,
88 Providers
89>;
90
91/**
92 * An Apigee shared flow — a reusable sequence of policies that API
93 * proxies (or other shared flows) invoke with a FlowCallout.
94 *
95 * Shared flows have no labels field, so Alchemy stamps ownership into
96 * the bundle `Description` for `list` / nuke. `sharedflowId` and
97 * `organization` are identity — changing either replaces the shared
98 * flow. Uploading a bundle or changing `description` creates a new
99 * revision. `space` is moved in place.
100 *
101 * ### Creating a Shared Flow
102 * **Example:** Generated empty bundle
103 * ```typescript
104 * const flow = yield* GCP.Apigee.Sharedflow("Traffic", {
105 * description: "rate limit and spike arrest",
106 * });
107 * ```
108 *
109 * **Example:** Explicit id and space
110 * ```typescript
111 * const flow = yield* GCP.Apigee.Sharedflow("Traffic", {
112 * sharedflowId: "traffic-management",
113 * space: team.spaceId,
114 * description: "rate limit and spike arrest",
115 * });
116 * ```
117 *
118 * ### Updating a Shared Flow
119 * **Example:** New description (creates a revision)
120 * ```typescript
121 * const flow = yield* GCP.Apigee.Sharedflow("Traffic", {
122 * description: "rate limit only",
123 * });
124 * ```
125 *
126 * @resource
127 * @category Apigee
128 */
129export const Sharedflow = Resource<Sharedflow>("GCP.Apigee.Sharedflow");
130
131export class SharedflowNotResolved extends Data.TaggedError(
132 "GCP.Apigee.SharedflowNotResolved",
133)<{
134 name: string;
135}> {}
136
137const sanitizeId = (name: string): string => {
138 let next = name.replace(/[^A-Za-z0-9._-]/g, "-").replace(/^-+/, "");
139 if (!/^[A-Za-z]/.test(next)) {
140 next = `s${next}`;
141 }
142 next = next.slice(0, MAX_ID_LENGTH).replace(/[^A-Za-z0-9]+$/g, "");
143 return next.length > 0 ? next : "sharedflow";
144};
145
146const resourceName = (org: string, sharedflowId: string) =>
147 `${orgParent(org)}/sharedflows/${sharedflowId}`;
148
149const sharedflowIdOf = (flow: apigee.GoogleCloudApigeeV1SharedFlow) => {
150 const raw = flow.name ?? "";
151 return raw.includes("/") ? lastSegment(raw) : raw;
152};
153
154const toId = (
155 id: string,
156 sharedflowId: string | undefined,
157 existing?: string,
158) =>
159 Effect.gen(function* () {
160 if (sharedflowId !== undefined) return sanitizeId(sharedflowId);
161 if (existing !== undefined) return existing;
162 return sanitizeId(
163 yield* createPhysicalName({
164 id,
165 maxLength: MAX_ID_LENGTH,
166 lowercase: true,
167 }),
168 );
169 });
170
171const sha12 = (value: string) =>
172 Effect.sync(() =>
173 createHash("sha256").update(value).digest("hex").slice(0, 12),
174 );
175
176const bundleXml = (sharedflowId: string, description: string) =>
177 `<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
178<SharedFlowBundle name="${xmlEscape(sharedflowId)}">
179 <ConfigurationVersion majorVersion="4" minorVersion="0"/>
180 <Description>${xmlEscape(description)}</Description>
181 <DisplayName>${xmlEscape(sharedflowId)}</DisplayName>
182 <SharedFlows>
183 <SharedFlow>default</SharedFlow>
184 </SharedFlows>
185</SharedFlowBundle>
186`;
187
188const defaultFlowXml = `<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
189<SharedFlow name="default">
190</SharedFlow>
191`;
192
193const toBase64 = (bytes: Uint8Array) =>
194 Effect.sync(() => Buffer.from(bytes).toString("base64"));
195
196const makeEmptyBundle = (sharedflowId: string, description: string) =>
197 zipFiles([
198 {
199 path: `sharedflowbundle/${sharedflowId}.xml`,
200 content: bundleXml(sharedflowId, description),
201 },
202 {
203 path: "sharedflowbundle/sharedflows/default.xml",
204 content: defaultFlowXml,
205 },
206 ]).pipe(Effect.flatMap(toBase64));
207
208const bundleXmlPath = (paths: ReadonlyArray<string>) =>
209 paths.find((path) => /^sharedflowbundle\/[^/]+\.xml$/i.test(path));
210
211const stampBundle = (base64: string, description: string) =>
212 Effect.gen(function* () {
213 const bytes = yield* Effect.sync(() => Buffer.from(base64, "base64"));
214 const entries = yield* unzipFiles(bytes);
215 const xmlPath = bundleXmlPath(Object.keys(entries));
216 if (xmlPath !== undefined) {
217 const xml = yield* Effect.sync(() =>
218 Buffer.from(entries[xmlPath]!).toString("utf8"),
219 );
220 const escaped = `<Description>${xmlEscape(description)}</Description>`;
221 const next = xml.includes("<Description>")
222 ? xml.replace(/<Description>[\s\S]*?<\/Description>/, escaped)
223 : xml.replace(
224 /<\/SharedFlowBundle>/,
225 ` ${escaped}\n</SharedFlowBundle>`,
226 );
227 entries[xmlPath] = yield* Effect.sync(() => Buffer.from(next, "utf8"));
228 }
229 // zipFiles sorts entries and stamps a fixed timestamp, so restamping the
230 // bundle is byte-identical for identical content.
231 const archive = yield* zipFiles(
232 Object.entries(entries).map(([path, content]) => ({ path, content })),
233 );
234 return yield* toBase64(archive);
235 });
236
237const desiredBundle = (
238 sharedflowId: string,
239 description: string,
240 bundle: string | undefined,
241) =>
242 bundle !== undefined
243 ? stampBundle(bundle, description)
244 : makeEmptyBundle(sharedflowId, description);
245
246const toAttrs = (
247 flow: apigee.GoogleCloudApigeeV1SharedFlow,
248 org: string,
249 description: string | undefined,
250) => {
251 const sharedflowId = sharedflowIdOf(flow);
252 const parsed = parseOwnership(description);
253 return {
254 name: resourceName(org, sharedflowId),
255 sharedflowId,
256 organization: org,
257 space: flow.space,
258 latestRevisionId: flow.latestRevisionId,
259 revision: [...(flow.revision ?? [])],
260 description: parsed.text,
261 createdAt: flow.metaData?.createdAt,
262 lastModifiedAt: flow.metaData?.lastModifiedAt,
263 };
264};
265
266const getByName = (name: string) =>
267 apigee
268 .getOrganizationsSharedflows({ name })
269 .pipe(
270 Effect.catchTag(["NotFound", "ApigeeResourceNotFound"], () =>
271 Effect.succeed(undefined),
272 ),
273 );
274
275const getRevision = (name: string) =>
276 apigee
277 .getOrganizationsSharedflowsRevisions({ name, format: "bundle" })
278 .pipe(
279 Effect.catchTag(["NotFound", "ApigeeResourceNotFound"], () =>
280 Effect.succeed(undefined),
281 ),
282 );
283
284const descriptionFromBundle = (body: apigee.GoogleApiHttpBody | undefined) =>
285 Effect.gen(function* () {
286 const data = body?.data;
287 if (data === undefined || data.length === 0) return undefined;
288 const bytes = yield* Effect.sync(() => Buffer.from(data, "base64"));
289 const entries = yield* unzipFiles(bytes);
290 const xmlPath = bundleXmlPath(Object.keys(entries));
291 if (xmlPath === undefined) return undefined;
292 const xml = yield* Effect.sync(() =>
293 Buffer.from(entries[xmlPath]!).toString("utf8"),
294 );
295 const match = xml.match(/<Description>([\s\S]*?)<\/Description>/);
296 if (match?.[1] === undefined) return undefined;
297 return xmlUnescape(match[1]);
298 });
299
300const revisionName = (org: string, sharedflowId: string, revisionId: string) =>
301 `${resourceName(org, sharedflowId)}/revisions/${revisionId}`;
302
303const observedDescription = (
304 org: string,
305 flow: apigee.GoogleCloudApigeeV1SharedFlow,
306) =>
307 Effect.gen(function* () {
308 const sharedflowId = sharedflowIdOf(flow);
309 const latest = flow.latestRevisionId;
310 if (!sharedflowId || !latest) return undefined;
311 const revision = yield* getRevision(
312 revisionName(org, sharedflowId, latest),
313 );
314 return yield* descriptionFromBundle(revision);
315 });
316
317const importBundle = (input: {
318 org: string;
319 sharedflowId: string;
320 space?: string;
321 data: string;
322}) =>
323 apigee.createOrganizationsSharedflows({
324 parent: orgParent(input.org),
325 action: "import",
326 name: input.sharedflowId,
327 space: input.space,
328 body: {
329 contentType: "application/octet-stream",
330 data: input.data,
331 },
332 });
333
334const waitUntilExists = (name: string) =>
335 getByName(name).pipe(
336 Effect.flatMap((flow) =>
337 flow
338 ? Effect.succeed(flow)
339 : Effect.fail(new SharedflowNotResolved({ name })),
340 ),
341 Effect.retry({
342 while: (error) => error._tag === "GCP.Apigee.SharedflowNotResolved",
343 times: 8,
344 schedule: Schedule.exponential("250 millis"),
345 }),
346 );
347
348export const SharedflowProvider = () =>
349 Provider.succeed(Sharedflow, {
350 stables: ["name", "sharedflowId", "organization", "createdAt"],
351
352 diff: Effect.fn(function* ({ news, olds, output }) {
353 if (!isResolved(news)) return undefined;
354 const previousId = olds?.sharedflowId ?? output?.sharedflowId;
355 const idChanged =
356 previousId !== undefined &&
357 news.sharedflowId !== undefined &&
358 sanitizeId(news.sharedflowId) !== previousId;
359 const previousOrg = olds?.organization ?? output?.organization;
360 const orgChanged =
361 previousOrg !== undefined &&
362 news.organization !== undefined &&
363 orgIdOf(news.organization, previousOrg) !== previousOrg;
364 if (idChanged || orgChanged) {
365 return { action: "replace" as const, deleteFirst: false };
366 }
367 return undefined;
368 }),
369
370 read: Effect.fn(function* ({ id, olds, output }) {
371 const env = yield* GcpEnvironment.current;
372 const org = orgIdOf(
373 olds?.organization ?? output?.organization,
374 env.project,
375 );
376 const sharedflowId = yield* toId(
377 id,
378 olds?.sharedflowId,
379 output?.sharedflowId,
380 );
381 const name = output?.name ?? resourceName(org, sharedflowId);
382 const existing = yield* getByName(name);
383 if (existing === undefined) return undefined;
384 const description = yield* observedDescription(org, existing);
385 const attrs = toAttrs(existing, org, description);
386 const { labels } = parseOwnership(description);
387 return (yield* hasAlchemyLabels(id, labels)) ? attrs : Unowned(attrs);
388 }),
389
390 list: () =>
391 Effect.gen(function* () {
392 const env = yield* GcpEnvironment.current;
393 const org = env.project;
394 const page = yield* apigee
395 .listOrganizationsSharedflows({
396 parent: orgParent(org),
397 includeRevisions: true,
398 includeMetaData: true,
399 })
400 .pipe(
401 Effect.catchTag(["NotFound", "ApigeeResourceNotFound"], () =>
402 Effect.succeed({ sharedFlows: [] }),
403 ),
404 );
405 const owned = yield* Effect.forEach(
406 page.sharedFlows ?? [],
407 (flow) =>
408 Effect.gen(function* () {
409 const description = yield* observedDescription(org, flow);
410 if (!hasOwnershipMarker(description)) return undefined;
411 return toAttrs(flow, org, description);
412 }),
413 { concurrency: 8 },
414 );
415 return owned.filter((flow) => flow !== undefined);
416 }),
417
418 reconcile: Effect.fn(function* ({ id, news, output }) {
419 const env = yield* GcpEnvironment.current;
420 const org = orgIdOf(
421 news.organization ?? output?.organization,
422 env.project,
423 );
424 const sharedflowId = yield* toId(
425 id,
426 news.sharedflowId,
427 output?.sharedflowId,
428 );
429 const name = resourceName(org, sharedflowId);
430 const ownership = yield* createInternalLabels(id);
431 const extra =
432 news.bundle !== undefined
433 ? { bundle: yield* sha12(news.bundle) }
434 : undefined;
435 const desiredDescription = encodeOwnership(
436 ownership,
437 news.description,
438 extra,
439 );
440 const data = yield* desiredBundle(
441 sharedflowId,
442 desiredDescription,
443 news.bundle,
444 );
445
446 let current = yield* getByName(output?.name ?? name);
447
448 if (current === undefined) {
449 yield* importBundle({
450 org,
451 sharedflowId,
452 space: news.space,
453 data,
454 }).pipe(Effect.catchTag("Conflict", () => Effect.void));
455 current = yield* waitUntilExists(name);
456 }
457
458 if (current === undefined) {
459 return yield* new SharedflowNotResolved({ name });
460 }
461
462 const observed = yield* observedDescription(org, current);
463 if ((observed ?? "") !== desiredDescription) {
464 yield* importBundle({
465 org,
466 sharedflowId,
467 data,
468 });
469 current = yield* waitUntilExists(name);
470 }
471
472 const desiredSpace = news.space ?? "";
473 const observedSpace = current.space ?? "";
474 if (desiredSpace !== observedSpace) {
475 current = yield* apigee.moveOrganizationsSharedflows({
476 name,
477 body: { space: news.space },
478 });
479 }
480
481 const description = yield* observedDescription(org, current);
482 return toAttrs(current, org, description);
483 }),
484
485 delete: Effect.fn(function* ({ output }) {
486 yield* apigee
487 .deleteOrganizationsSharedflows({ name: output.name })
488 .pipe(
489 Effect.catchTag(
490 ["NotFound", "ApigeeResourceNotFound"],
491 () => Effect.void,
492 ),
493 );
494 }),
495 });