1import * as kafka from "@distilled.cloud/gcp/managedkafka_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
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 collectPages,
12 expandParent,
13 fingerprint,
14 getAcl,
15 listAlchemyClusters,
16 normalizeLocation,
17 parseName,
18 replaceOnIdentity,
19 retryTransient,
20 toPhysicalId,
21} from "./internal.ts";
22
23export type AclEntry = {
24 /**
25 * Operation (`ALL`, `READ`, `WRITE`, `CREATE`, `DELETE`, `ALTER`,
26 * `DESCRIBE`, `CLUSTER_ACTION`, `DESCRIBE_CONFIGS`, `ALTER_CONFIGS`,
27 * `IDEMPOTENT_WRITE`).
28 */
29 operation?: string;
30 /**
31 * Host. Must be `*` for Managed Service for Apache Kafka.
32 * @default "*"
33 */
34 host?: string;
35 /**
36 * Permission type (`ALLOW`, `DENY`).
37 */
38 permissionType?: string;
39 /**
40 * Principal (`User:sa@project.iam.gserviceaccount.com` or `User:*`).
41 */
42 principal?: string;
43};
44
45export type ClustersAclProps = {
46 /**
47 * Parent cluster. Full name
48 * `projects/{project}/locations/{location}/clusters/{cluster}` or the
49 * cluster id (combined with `location`). Immutable — changing it
50 * replaces the ACL.
51 */
52 cluster: string;
53 /**
54 * Region used when `cluster` is a bare id.
55 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
56 */
57 location?: string;
58 /**
59 * ACL id. Defines the Resource Pattern:
60 * `cluster`, `topic/{name}`, `consumerGroup/{name}`,
61 * `transactionalId/{name}`, `topicPrefixed/{name}`, `allTopics`,
62 * `allConsumerGroups`, `allTransactionalIds`. If omitted, a topic
63 * pattern is generated from the logical id. Immutable — changing it
64 * replaces the ACL.
65 */
66 aclId?: string;
67 /**
68 * ACL entries for the resource pattern. Maximum 100.
69 */
70 aclEntries: AclEntry[];
71};
72
73export type ClustersAcl = Resource<
74 "GCP.ManagedKafka.ClustersAcl",
75 ClustersAclProps,
76 {
77 /** Full resource name `.../clusters/{cluster}/acls/{acl_id}`. */
78 name: string;
79 /** ACL id (resource pattern). */
80 aclId: string;
81 /** Parent cluster resource name. */
82 cluster: string;
83 /** Project id. */
84 project: string;
85 /** Location id. */
86 location: string;
87 /** ACL entries currently applied. */
88 aclEntries: AclEntry[];
89 /** Pattern type (`LITERAL`, `PREFIXED`). */
90 patternType: string | undefined;
91 /** Resource name derived from the ACL id. */
92 resourceName: string | undefined;
93 /** Resource type (`CLUSTER`, `TOPIC`, `GROUP`, `TRANSACTIONAL_ID`). */
94 resourceType: string | undefined;
95 /** Etag for concurrency control. */
96 etag: string | undefined;
97 },
98 never,
99 Providers
100>;
101
102/**
103 * An ACL for a Kafka resource pattern on a Managed Kafka cluster.
104 *
105 * The ACL id encodes the resource pattern and is immutable. Entries
106 * update in place.
107 *
108 * ### Creating an ACL
109 * **Example:** Allow all users to describe a topic
110 * ```typescript
111 * const acl = yield* GCP.ManagedKafka.ClustersAcl("OrdersAcl", {
112 * cluster: cluster.name,
113 * aclId: "topic/orders",
114 * aclEntries: [
115 * {
116 * principal: "User:*",
117 * operation: "DESCRIBE",
118 * permissionType: "ALLOW",
119 * host: "*",
120 * },
121 * ],
122 * });
123 * ```
124 *
125 * **Example:** Cluster-wide ACL
126 * ```typescript
127 * const acl = yield* GCP.ManagedKafka.ClustersAcl("ClusterAcl", {
128 * cluster: cluster.name,
129 * aclId: "cluster",
130 * aclEntries: [
131 * {
132 * principal: "User:client@my-project.iam.gserviceaccount.com",
133 * operation: "CLUSTER_ACTION",
134 * permissionType: "ALLOW",
135 * host: "*",
136 * },
137 * ],
138 * });
139 * ```
140 *
141 * @resource
142 * @category ManagedKafka
143 */
144export const ClustersAcl = Resource<ClustersAcl>(
145 "GCP.ManagedKafka.ClustersAcl",
146);
147
148export class ClustersAclNotResolved extends Data.TaggedError(
149 "GCP.ManagedKafka.ClustersAclNotResolved",
150)<{
151 name: string;
152}> {}
153
154const clusterOf = (cluster: string, project: string, location: string) =>
155 expandParent(cluster, project, location, "clusters");
156
157const resourceName = (cluster: string, aclId: string) =>
158 `${cluster}/acls/${aclId}`;
159
160const entriesOf = (
161 entries: readonly kafka.AclEntry[] | undefined,
162): AclEntry[] =>
163 (entries ?? []).map((entry) => ({
164 operation: entry.operation,
165 host: entry.host ?? "*",
166 permissionType: entry.permissionType,
167 principal: entry.principal,
168 }));
169
170const entriesKey = (entries: readonly AclEntry[] | undefined) =>
171 fingerprint(
172 entriesOf(entries).map((entry) => ({
173 operation: (entry.operation ?? "").toUpperCase(),
174 host: entry.host ?? "*",
175 permissionType: (entry.permissionType ?? "").toUpperCase(),
176 principal: entry.principal ?? "",
177 })),
178 );
179
180const toAttrs = (acl: kafka.Acl, project: string) => {
181 const name = acl.name ?? "";
182 const parsed = parseName(name, "acls");
183 return {
184 name,
185 aclId: parsed.id,
186 cluster: parsed.parent,
187 project: parsed.project || project,
188 location: parsed.location,
189 aclEntries: entriesOf(acl.aclEntries),
190 patternType: acl.patternType,
191 resourceName: acl.resourceName,
192 resourceType: acl.resourceType,
193 etag: acl.etag,
194 };
195};
196
197const toAclId = (
198 id: string,
199 explicit: string | undefined,
200 existing: string | undefined,
201) =>
202 Effect.gen(function* () {
203 if (explicit !== undefined) return explicit;
204 if (existing !== undefined) return existing;
205 const topicId = yield* toPhysicalId(id, undefined, undefined, "topic");
206 return `topic/${topicId}`;
207 });
208
209export const ClustersAclProvider = () =>
210 Provider.succeed(ClustersAcl, {
211 stables: [
212 "name",
213 "aclId",
214 "cluster",
215 "project",
216 "location",
217 "patternType",
218 "resourceName",
219 "resourceType",
220 ],
221
222 diff: Effect.fn(function* ({ news, olds, output }) {
223 if (!isResolved(news)) return undefined;
224 const env = yield* GcpEnvironment.current;
225 const location = normalizeLocation(
226 news.location ?? output?.location,
227 env.region,
228 );
229 return replaceOnIdentity({
230 previousId: olds?.aclId ?? output?.aclId,
231 nextId: news.aclId ?? olds?.aclId ?? output?.aclId,
232 previousParent: olds?.cluster ?? output?.cluster,
233 nextParent: clusterOf(news.cluster, env.project, location),
234 });
235 }),
236
237 read: Effect.fn(function* ({ id, olds, output }) {
238 const env = yield* GcpEnvironment.current;
239 const location = normalizeLocation(
240 olds?.location ?? output?.location,
241 env.region,
242 );
243 const aclId = yield* toAclId(id, olds?.aclId, output?.aclId);
244 const cluster =
245 olds?.cluster !== undefined
246 ? clusterOf(olds.cluster, env.project, location)
247 : (output?.cluster ?? "");
248 const name =
249 output?.name ??
250 (cluster.length > 0 ? resourceName(cluster, aclId) : "");
251 const existing = yield* getAcl(name);
252 if (existing === undefined) return undefined;
253 const attrs = toAttrs(existing, env.project);
254 return output !== undefined || olds !== undefined
255 ? attrs
256 : Unowned(attrs);
257 }),
258
259 list: () =>
260 Effect.gen(function* () {
261 const env = yield* GcpEnvironment.current;
262 const clusters = yield* listAlchemyClusters(env.project, env.region);
263 const acls = yield* Effect.forEach(
264 clusters.filter((cluster) => (cluster.name ?? "").length > 0),
265 (cluster: kafka.Cluster) =>
266 collectPages(
267 kafka.listProjectsLocationsClustersAcls.pages({
268 parent: cluster.name!,
269 pageSize: 1000,
270 }),
271 (page) => page.acls,
272 ).pipe(
273 Effect.catchTag("NotFound", () =>
274 Effect.succeed([] as kafka.Acl[]),
275 ),
276 ),
277 { concurrency: 4 },
278 );
279 return acls.flat().map((acl) => toAttrs(acl, env.project));
280 }),
281
282 reconcile: Effect.fn(function* ({ id, news, output }) {
283 const env = yield* GcpEnvironment.current;
284 const location = normalizeLocation(
285 news.location ?? output?.location,
286 env.region,
287 );
288 const cluster = clusterOf(news.cluster, env.project, location);
289 const aclId = yield* toAclId(id, news.aclId, output?.aclId);
290 const name = output?.name ?? resourceName(cluster, aclId);
291 const aclEntries = news.aclEntries.map((entry) => ({
292 operation: entry.operation,
293 host: entry.host ?? "*",
294 permissionType: entry.permissionType,
295 principal: entry.principal,
296 }));
297
298 let current = yield* getAcl(name);
299
300 if (current === undefined) {
301 const created = yield* retryTransient(
302 kafka
303 .createProjectsLocationsClustersAcls({
304 parent: cluster,
305 aclId,
306 body: { aclEntries },
307 })
308 .pipe(Effect.catchTag("Conflict", () => getAcl(name))),
309 );
310 current = created ?? undefined;
311 }
312
313 if (current === undefined) {
314 return yield* new ClustersAclNotResolved({ name });
315 }
316
317 if (entriesKey(current.aclEntries) !== entriesKey(aclEntries)) {
318 current = yield* kafka.patchProjectsLocationsClustersAcls({
319 name,
320 updateMask: "acl_entries",
321 body: {
322 aclEntries,
323 etag: current.etag,
324 },
325 });
326 }
327
328 return toAttrs(current, env.project);
329 }),
330
331 delete: Effect.fn(function* ({ output }) {
332 yield* kafka
333 .deleteProjectsLocationsClustersAcls({ name: output.name })
334 .pipe(
335 Effect.catchTag(
336 ["NotFound", "AclClusterNotFound"],
337 () => Effect.void,
338 ),
339 );
340 }),
341 });