1import * as cloudasset from "@distilled.cloud/gcp/cloudasset_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
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 encodeDescription,
14 grantCloudAssetPublisher,
15 hasOwnershipMarker,
16 lastSegment,
17 listFeeds,
18 NO_OP_EXPRESSION,
19 ownedByAlchemy,
20 parentOf,
21 parseDescription,
22 replaceOn,
23 scopeParent,
24 sameJson,
25 sameStringList,
26 sameText,
27 toPhysicalId,
28 updateMaskOf,
29} from "./internal.ts";
30
31export type FeedContentType = cloudasset.FeedContentTypeEnum | (string & {});
32
33export type FeedCondition = {
34 /**
35 * CEL expression evaluated against a `TemporalAsset` named
36 * `temporal_asset`. Example: `temporal_asset.deleted == true`.
37 */
38 expression?: string;
39 /**
40 * Short title shown in UIs that edit the expression.
41 */
42 title?: string;
43 /**
44 * Longer description of the expression. Feeds have no labels field, so
45 * Alchemy ownership is stored in a `[alchemy …]` prefix here and
46 * stripped from attributes.
47 */
48 description?: string;
49 /**
50 * Source location of the expression for error reporting.
51 */
52 location?: string;
53};
54
55export type FeedProps = {
56 /**
57 * Feed id (the `{feed}` segment of
58 * `projects/{project}/feeds/{feed}`). If omitted, a unique RFC1035
59 * name is generated. Immutable — changing it replaces the feed.
60 */
61 feedId?: string;
62 /**
63 * Parent project, folder, or organization
64 * (`projects/{project}`, `folders/{folder}`,
65 * `organizations/{organization}`). Defaults to the current project.
66 * Immutable — changing it replaces the feed.
67 */
68 parent?: string;
69 /**
70 * Full resource names to subscribe to. Specify `assetNames` and/or
71 * `assetTypes`. Example:
72 * `//compute.googleapis.com/projects/p/zones/z/instances/i`.
73 */
74 assetNames?: string[];
75 /**
76 * Asset types to subscribe to. Example: `storage.googleapis.com/Bucket`.
77 */
78 assetTypes?: string[];
79 /**
80 * Asset content included in updates (`RESOURCE`, `IAM_POLICY`,
81 * `ORG_POLICY`, `ACCESS_POLICY`, `OS_INVENTORY`, `RELATIONSHIP`).
82 * Omitted returns name and type only.
83 */
84 contentType?: FeedContentType;
85 /**
86 * Relationship types to emit when `contentType` is `RELATIONSHIP`.
87 */
88 relationshipTypes?: string[];
89 /**
90 * Pub/Sub topic that receives asset updates
91 * (`projects/{project}/topics/{topic}`). Alchemy grants the Cloud
92 * Asset service agent `roles/pubsub.publisher` on the topic.
93 */
94 pubsubTopic: string;
95 /**
96 * CEL condition that filters published updates. When omitted, Alchemy
97 * still stamps ownership into a no-op `true` condition.
98 */
99 condition?: FeedCondition;
100};
101
102export type Feed = Resource<
103 "GCP.CloudAsset.Feed",
104 FeedProps,
105 {
106 /** Full resource name `projects/{project}/feeds/{feed}`. */
107 name: string;
108 /** Feed id (last path segment). */
109 feedId: string;
110 /** Parent project, folder, or organization. */
111 parent: string;
112 /** Project id used when the feed was reconciled. */
113 project: string;
114 /** Full resource names the feed watches. */
115 assetNames: string[];
116 /** Asset types the feed watches. */
117 assetTypes: string[];
118 /** Content type included in updates. */
119 contentType: string | undefined;
120 /** Relationship types when `contentType` is `RELATIONSHIP`. */
121 relationshipTypes: string[];
122 /** Destination Pub/Sub topic. */
123 pubsubTopic: string | undefined;
124 /** User condition with the Alchemy ownership prefix stripped. */
125 condition: FeedCondition | undefined;
126 },
127 never,
128 Providers
129>;
130
131/**
132 * A Cloud Asset Inventory feed that publishes asset updates to Pub/Sub.
133 *
134 * Feeds have no labels field — Alchemy stamps ownership into
135 * `condition.description` so `list` / nuke can find them. Feed id and
136 * parent are identity. Asset filters, content type, destination topic,
137 * and condition update in place.
138 *
139 * ### Creating a Feed
140 * **Example:** Watch storage buckets
141 * ```typescript
142 * const topic = yield* GCP.PubSub.Topic("AssetEvents", {});
143 * const feed = yield* GCP.CloudAsset.Feed("Buckets", {
144 * pubsubTopic: topic.name,
145 * assetTypes: ["storage.googleapis.com/Bucket"],
146 * contentType: "RESOURCE",
147 * });
148 * ```
149 *
150 * **Example:** Named feed with a deletion filter
151 * ```typescript
152 * const feed = yield* GCP.CloudAsset.Feed("Buckets", {
153 * feedId: "bucket-deletes",
154 * pubsubTopic: topic.name,
155 * assetTypes: ["storage.googleapis.com/Bucket"],
156 * contentType: "RESOURCE",
157 * condition: {
158 * expression: "temporal_asset.deleted == true",
159 * title: "deletes",
160 * description: "publish deletions only",
161 * },
162 * });
163 * ```
164 *
165 * ### Updating a Feed
166 * **Example:** Also watch Pub/Sub topics
167 * ```typescript
168 * // Same logical id, changed props: the feed is patched in place.
169 * const feed = yield* GCP.CloudAsset.Feed("Buckets", {
170 * pubsubTopic: topic.name,
171 * assetTypes: [
172 * "storage.googleapis.com/Bucket",
173 * "pubsub.googleapis.com/Topic",
174 * ],
175 * contentType: "RESOURCE",
176 * });
177 * ```
178 *
179 * @resource
180 * @category CloudAsset
181 */
182export const Feed = Resource<Feed>("GCP.CloudAsset.Feed");
183
184export class FeedNotResolved extends Data.TaggedError(
185 "GCP.CloudAsset.FeedNotResolved",
186)<{
187 name: string;
188}> {}
189
190const resourceName = (parent: string, feedId: string) =>
191 `${parent}/feeds/${feedId}`;
192
193const contentTypeOf = (value: string | undefined) =>
194 !value || value === "CONTENT_TYPE_UNSPECIFIED" ? undefined : value;
195
196const toUserCondition = (
197 condition: cloudasset.Expr | undefined,
198): FeedCondition | undefined => {
199 if (condition === undefined) return undefined;
200 const { description } = parseDescription(condition.description);
201 const isInternalOnly =
202 (condition.expression === undefined ||
203 condition.expression === NO_OP_EXPRESSION) &&
204 condition.title === undefined &&
205 condition.location === undefined &&
206 description === undefined;
207 if (isInternalOnly) return undefined;
208 return {
209 expression: condition.expression,
210 title: condition.title,
211 description,
212 location: condition.location,
213 };
214};
215
216const desiredCondition = (
217 ownership: Record<string, string>,
218 condition: FeedCondition | undefined,
219): cloudasset.Expr => ({
220 expression: condition?.expression ?? NO_OP_EXPRESSION,
221 title: condition?.title,
222 location: condition?.location,
223 description: encodeDescription(ownership, condition?.description),
224});
225
226const toAttrs = (feed: cloudasset.Feed, project: string) => {
227 const name = feed.name ?? "";
228 return {
229 name,
230 feedId: lastSegment(name),
231 parent: parentOf(name, "feeds"),
232 project,
233 assetNames: feed.assetNames ?? [],
234 assetTypes: feed.assetTypes ?? [],
235 contentType: contentTypeOf(feed.contentType),
236 relationshipTypes: feed.relationshipTypes ?? [],
237 pubsubTopic: feed.feedOutputConfig?.pubsubDestination?.topic,
238 condition: toUserCondition(feed.condition),
239 };
240};
241
242const getByName = (name: string) =>
243 name.length === 0
244 ? Effect.succeed(undefined)
245 : cloudasset
246 .getFeeds({ name })
247 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
248
249const observe = (project: string, feedId: string, outputName?: string) =>
250 Effect.gen(function* () {
251 const parent = yield* scopeParent(project);
252 const candidates = [outputName, resourceName(parent, feedId)].filter(
253 (name): name is string => name !== undefined && name.length > 0,
254 );
255 for (const name of candidates) {
256 const found = yield* getByName(name);
257 if (found !== undefined) return found;
258 }
259 const feeds = yield* listFeeds(parent);
260 return feeds.find((feed) => lastSegment(feed.name ?? "") === feedId);
261 });
262
263const toFeedBody = (
264 news: FeedProps,
265 condition: cloudasset.Expr,
266): cloudasset.Feed => ({
267 assetNames: news.assetNames,
268 assetTypes: news.assetTypes,
269 contentType: news.contentType,
270 relationshipTypes: news.relationshipTypes,
271 feedOutputConfig: {
272 pubsubDestination: { topic: news.pubsubTopic },
273 },
274 condition,
275});
276
277export const FeedProvider = () =>
278 Provider.succeed(Feed, {
279 stables: ["name", "feedId", "parent", "project"],
280
281 diff: Effect.fn(function* ({ news, olds, output }) {
282 if (!isResolved(news)) return undefined;
283 return (
284 replaceOn(olds?.feedId ?? output?.feedId, news.feedId) ??
285 replaceOn(olds?.parent, news.parent)
286 );
287 }),
288
289 read: Effect.fn(function* ({ id, olds, output }) {
290 const env = yield* GcpEnvironment.current;
291 const feedId = yield* toPhysicalId(id, olds?.feedId, output?.feedId);
292 const existing = yield* observe(env.project, feedId, output?.name);
293 if (existing === undefined) return undefined;
294 const attrs = toAttrs(existing, env.project);
295 return (yield* ownedByAlchemy(id, existing.condition?.description))
296 ? attrs
297 : Unowned(attrs);
298 }),
299
300 list: () =>
301 Effect.gen(function* () {
302 const env = yield* GcpEnvironment.current;
303 const parent = yield* scopeParent(env.project);
304 const feeds = yield* listFeeds(parent);
305 return feeds
306 .filter((feed) => hasOwnershipMarker(feed.condition?.description))
307 .map((feed) => toAttrs(feed, env.project));
308 }),
309
310 reconcile: Effect.fn(function* ({ id, news, output }) {
311 const env = yield* GcpEnvironment.current;
312 const feedId = yield* toPhysicalId(id, news.feedId, output?.feedId);
313 const parent = yield* scopeParent(env.project, news.parent);
314 const name = resourceName(parent, feedId);
315 const ownership = yield* createInternalLabels(id);
316 const condition = desiredCondition(ownership, news.condition);
317 const body = toFeedBody(news, condition);
318 yield* grantCloudAssetPublisher(env.project, news.pubsubTopic);
319
320 let current = yield* observe(env.project, feedId, output?.name);
321
322 if (current === undefined) {
323 const created = yield* cloudasset
324 .createFeeds({
325 parent,
326 body: { feedId, feed: body },
327 })
328 .pipe(
329 Effect.retry({
330 while: (error) => error._tag === "BadRequest",
331 times: 8,
332 schedule: Schedule.spaced("2 seconds"),
333 }),
334 Effect.catchTag("Conflict", () =>
335 observe(env.project, feedId, output?.name),
336 ),
337 );
338 current = created ?? undefined;
339 }
340
341 if (current === undefined) {
342 return yield* new FeedNotResolved({ name });
343 }
344
345 const currentName = current.name ?? name;
346 const namesChanged = !sameStringList(current.assetNames, news.assetNames);
347 const typesChanged = !sameStringList(current.assetTypes, news.assetTypes);
348 const contentChanged = !sameText(
349 contentTypeOf(current.contentType),
350 contentTypeOf(news.contentType),
351 );
352 const relationshipsChanged = !sameStringList(
353 current.relationshipTypes,
354 news.relationshipTypes,
355 );
356 const topicChanged = !sameText(
357 current.feedOutputConfig?.pubsubDestination?.topic,
358 news.pubsubTopic,
359 );
360 const conditionChanged = !sameJson(current.condition, condition);
361 const updateMask = updateMaskOf(
362 namesChanged ? "asset_names" : undefined,
363 typesChanged ? "asset_types" : undefined,
364 contentChanged ? "content_type" : undefined,
365 relationshipsChanged ? "relationship_types" : undefined,
366 topicChanged ? "feed_output_config" : undefined,
367 conditionChanged ? "condition" : undefined,
368 );
369
370 if (updateMask.length > 0) {
371 current = yield* cloudasset.patchFeeds({
372 name: currentName,
373 body: {
374 updateMask,
375 feed: { ...body, name: currentName },
376 },
377 });
378 }
379
380 return toAttrs(current, env.project);
381 }),
382
383 delete: Effect.fn(function* ({ output }) {
384 if (!output.name) return;
385 yield* cloudasset
386 .deleteFeeds({ name: output.name })
387 .pipe(Effect.catchTag(["NotFound", "BadRequest"], () => Effect.void));
388 }),
389 });