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 { tagRecord } from "../../Tags.ts";
9import { GcpEnvironment } from "../Environment.ts";
10import {
11 createInternalLabels,
12 diffLabels,
13 hasAlchemyLabels,
14 toLabels,
15} from "../Labels.ts";
16import type { Providers } from "../Providers.ts";
17import {
18 asCountString,
19 DEFAULT_MEMORY_BYTES,
20 DEFAULT_VCPU_COUNT,
21 defaultSubnet,
22 fieldMask,
23 fingerprint,
24 getCluster,
25 listAlchemyClusters,
26 locationParent,
27 MAX_CLUSTER_ID_LENGTH,
28 normalizeLocation,
29 parseName,
30 replaceOnIdentity,
31 retryCreateQuota,
32 ResourceNotResolved,
33 rfc1035,
34 toPhysicalId,
35 userLabels,
36 waitForOperation,
37 waitUntilExists,
38 waitUntilGone,
39 waitUntilReady,
40} from "./internal.ts";
41
42export type NetworkConfig = {
43 /** VPC subnet `projects/{project}/regions/{region}/subnetworks/{subnet}`. */
44 subnet?: string;
45};
46
47export type AccessConfig = {
48 /** VPC networks granted access. At least one subnet is required. */
49 networkConfigs?: NetworkConfig[];
50};
51
52export type GcpConfig = {
53 /** Access configuration for the Kafka cluster. */
54 accessConfig?: AccessConfig;
55 /**
56 * Cloud KMS key for encryption. Immutable.
57 * `projects/{project}/locations/{location}/keyRings/{keyRing}/cryptoKeys/{key}`.
58 */
59 kmsKey?: string;
60};
61
62export type CapacityConfig = {
63 /**
64 * vCPUs to provision. Minimum 3.
65 * @default 3
66 */
67 vcpuCount?: number | string;
68 /**
69 * Memory in bytes. Minimum 3221225472 (3 GiB). CPU:memory GiB ratio
70 * must be between 1:1 and 1:8.
71 * @default 3221225472
72 */
73 memoryBytes?: number | string;
74};
75
76export type RebalanceConfig = {
77 /**
78 * Rebalance mode (`NO_REBALANCE`, `AUTO_REBALANCE_ON_SCALE_UP`).
79 * @default "NO_REBALANCE"
80 */
81 mode?: kafka.RebalanceConfigModeEnum | (string & {});
82};
83
84export type TrustConfig = {
85 /** Certificate Authority Service CA pools. Maximum 10. */
86 casConfigs?: Array<{ caPool?: string }>;
87};
88
89export type TlsConfig = {
90 /** Kafka `ssl.principal.mapping.rules`. Empty string clears rules. */
91 sslPrincipalMappingRules?: string;
92 /** Broker truststore. Enables mTLS when set. */
93 trustConfig?: TrustConfig;
94};
95
96export type UpdateOptions = {
97 /**
98 * Allow an upscale that shrinks per-broker size below 90% of current.
99 * @default false
100 */
101 allowBrokerDownscaleOnClusterUpscale?: boolean;
102};
103
104export type ClusterProps = {
105 /**
106 * Cluster id (the `{cluster}` segment of
107 * `projects/{project}/locations/{location}/clusters/{cluster}`).
108 * Lowercase letters, digits, and hyphens; max 53 characters (longer
109 * values are truncated). If omitted, a unique RFC1035 name of at most
110 * 53 characters is generated. Immutable — changing it replaces the
111 * cluster.
112 */
113 clusterId?: string;
114 /**
115 * Region (`us-central1`, …). Immutable. `US-CENTRAL1` is accepted and
116 * normalized to `us-central1`.
117 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
118 */
119 location?: string;
120 /**
121 * Subnets for broker PSC endpoints. Each is
122 * `projects/{project}/regions/{region}/subnetworks/{subnet}`. Defaults
123 * to the project's `default` subnet in `location`. Immutable — changing
124 * them replaces the cluster.
125 */
126 subnets?: string[];
127 /**
128 * Full GCP config. `subnets` / `kmsKey` win when both are set.
129 */
130 gcpConfig?: GcpConfig;
131 /**
132 * Customer-managed KMS key. Immutable.
133 */
134 kmsKey?: string;
135 /**
136 * Capacity. Defaults to 3 vCPU / 3 GiB.
137 */
138 capacityConfig?: CapacityConfig;
139 /**
140 * Rebalance behavior.
141 */
142 rebalanceConfig?: RebalanceConfig;
143 /**
144 * TLS / mTLS configuration.
145 */
146 tlsConfig?: TlsConfig;
147 /**
148 * Options that control how updates are applied.
149 */
150 updateOptions?: UpdateOptions;
151 /**
152 * User labels. Alchemy ownership labels are merged in automatically.
153 */
154 labels?: Record<string, string>;
155};
156
157export type Cluster = Resource<
158 "GCP.ManagedKafka.Cluster",
159 ClusterProps,
160 {
161 /** Full resource name `projects/{project}/locations/{location}/clusters/{cluster}`. */
162 name: string;
163 /** Cluster id (last path segment). */
164 clusterId: string;
165 /** Project id. */
166 project: string;
167 /** Location id. */
168 location: string;
169 /** User labels (Alchemy ownership labels stripped). */
170 labels: Record<string, string>;
171 /** Server-reported state (`CREATING`, `ACTIVE`, …). */
172 state: string | undefined;
173 /** vCPU count currently provisioned. */
174 vcpuCount: string | undefined;
175 /** Memory in bytes currently provisioned. */
176 memoryBytes: string | undefined;
177 /** Subnets currently attached. */
178 subnets: string[];
179 /** CMEK key, if any. */
180 kmsKey: string | undefined;
181 /** Rebalance mode. */
182 rebalanceMode: string | undefined;
183 /** Kafka software version. */
184 kafkaVersion: string | undefined;
185 /** Broker details when requested with FULL view. */
186 brokerDetails: kafka.BrokerDetails[];
187 /** RFC3339 creation timestamp. */
188 createTime: string | undefined;
189 /** RFC3339 last-update timestamp. */
190 updateTime: string | undefined;
191 },
192 never,
193 Providers
194>;
195
196/**
197 * A Managed Service for Apache Kafka cluster.
198 *
199 * Changing `clusterId`, `location`, `subnets`, or `kmsKey` replaces the
200 * cluster. Capacity, rebalance mode, TLS config, and labels update in
201 * place. Provisioning typically takes more than a minute.
202 *
203 * ### Creating a Cluster
204 * **Example:** Generated name, 3 vCPU / 3 GiB
205 * ```typescript
206 * const cluster = yield* GCP.ManagedKafka.Cluster("Brokers", {});
207 * ```
208 *
209 * **Example:** Explicit id, subnet, and labels
210 * ```typescript
211 * const cluster = yield* GCP.ManagedKafka.Cluster("Brokers", {
212 * clusterId: "app-kafka",
213 * location: "us-central1",
214 * subnets: [
215 * "projects/my-project/regions/us-central1/subnetworks/default",
216 * ],
217 * capacityConfig: { vcpuCount: 3, memoryBytes: 3221225472 },
218 * labels: { env: "prod" },
219 * });
220 * ```
221 *
222 * @resource
223 * @category ManagedKafka
224 */
225export const Cluster = Resource<Cluster>("GCP.ManagedKafka.Cluster");
226
227export class ClusterNotResolved extends Data.TaggedError(
228 "GCP.ManagedKafka.ClusterNotResolved",
229)<{
230 name: string;
231}> {}
232
233const resourceName = (project: string, location: string, clusterId: string) =>
234 `projects/${project}/locations/${location}/clusters/${clusterId}`;
235
236const subnetsOf = (
237 cluster: kafka.Cluster | ClusterProps,
238 project: string,
239 location: string,
240): string[] => {
241 const fromConfig = (cluster.gcpConfig?.accessConfig?.networkConfigs ?? [])
242 .map((config) => config.subnet)
243 .filter((subnet): subnet is string => subnet !== undefined);
244 if ("subnets" in cluster && cluster.subnets && cluster.subnets.length > 0) {
245 return cluster.subnets;
246 }
247 if (fromConfig.length > 0) return fromConfig;
248 return [defaultSubnet(project, location)];
249};
250
251const kmsOf = (cluster: kafka.Cluster | ClusterProps) =>
252 ("kmsKey" in cluster ? cluster.kmsKey : undefined) ??
253 cluster.gcpConfig?.kmsKey;
254
255const toAttrs = (cluster: kafka.Cluster, project: string) => {
256 const name = cluster.name ?? "";
257 const parsed = parseName(name, "clusters");
258 return {
259 name,
260 clusterId: parsed.id,
261 project: parsed.project || project,
262 location: parsed.location,
263 labels: userLabels(cluster.labels),
264 state: cluster.state,
265 vcpuCount: cluster.capacityConfig?.vcpuCount,
266 memoryBytes: cluster.capacityConfig?.memoryBytes,
267 subnets: subnetsOf(cluster, parsed.project || project, parsed.location),
268 kmsKey: cluster.gcpConfig?.kmsKey,
269 rebalanceMode: cluster.rebalanceConfig?.mode,
270 kafkaVersion: cluster.kafkaVersion,
271 brokerDetails: cluster.brokerDetails ?? [],
272 createTime: cluster.createTime,
273 updateTime: cluster.updateTime,
274 };
275};
276
277const desiredCapacity = (news: ClusterProps): kafka.CapacityConfig => ({
278 vcpuCount: asCountString(news.capacityConfig?.vcpuCount, DEFAULT_VCPU_COUNT),
279 memoryBytes: asCountString(
280 news.capacityConfig?.memoryBytes,
281 DEFAULT_MEMORY_BYTES,
282 ),
283});
284
285const desiredGcpConfig = (
286 news: ClusterProps,
287 project: string,
288 location: string,
289): kafka.GcpConfig => ({
290 accessConfig: {
291 networkConfigs: subnetsOf(news, project, location).map((subnet) => ({
292 subnet,
293 })),
294 },
295 kmsKey: kmsOf(news),
296});
297
298export const ClusterProvider = () =>
299 Provider.succeed(Cluster, {
300 stables: ["name", "clusterId", "project", "location", "createTime"],
301
302 diff: Effect.fn(function* ({ news, olds, output }) {
303 if (!isResolved(news)) return undefined;
304 const env = yield* GcpEnvironment.current;
305 const location = normalizeLocation(
306 news.location ?? output?.location,
307 env.region,
308 );
309 const previousSubnets = fingerprint(
310 [...(olds?.subnets ?? output?.subnets ?? [])].sort(),
311 );
312 const nextSubnets = fingerprint(
313 [...subnetsOf(news, env.project, location)].sort(),
314 );
315 return replaceOnIdentity({
316 previousId: olds?.clusterId ?? output?.clusterId,
317 nextId: news.clusterId
318 ? rfc1035(news.clusterId, "kafka", MAX_CLUSTER_ID_LENGTH)
319 : (olds?.clusterId ?? output?.clusterId),
320 previousLocation: normalizeLocation(
321 olds?.location ?? output?.location,
322 env.region,
323 ),
324 nextLocation: location,
325 extra:
326 previousSubnets !== nextSubnets ||
327 (kmsOf(news) ?? "") !== (olds?.kmsKey ?? output?.kmsKey ?? ""),
328 });
329 }),
330
331 read: Effect.fn(function* ({ id, olds, output }) {
332 const env = yield* GcpEnvironment.current;
333 const clusterId = yield* toPhysicalId(
334 id,
335 olds?.clusterId,
336 output?.clusterId,
337 "kafka",
338 MAX_CLUSTER_ID_LENGTH,
339 );
340 const location = normalizeLocation(
341 olds?.location ?? output?.location,
342 env.region,
343 );
344 const name =
345 output?.name ?? resourceName(env.project, location, clusterId);
346 const existing = yield* getCluster(name);
347 if (existing === undefined) return undefined;
348 const attrs = toAttrs(existing, env.project);
349 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
350 ? attrs
351 : Unowned(attrs);
352 }),
353
354 list: () =>
355 Effect.gen(function* () {
356 const env = yield* GcpEnvironment.current;
357 const clusters = yield* listAlchemyClusters(env.project, env.region);
358 return clusters.map((cluster) => toAttrs(cluster, env.project));
359 }),
360
361 reconcile: Effect.fn(function* ({ id, news, output }) {
362 const env = yield* GcpEnvironment.current;
363 const clusterId = yield* toPhysicalId(
364 id,
365 news.clusterId,
366 output?.clusterId,
367 "kafka",
368 MAX_CLUSTER_ID_LENGTH,
369 );
370 const location = normalizeLocation(
371 news.location ?? output?.location,
372 env.region,
373 );
374 const name =
375 output?.name ?? resourceName(env.project, location, clusterId);
376 const desiredLabels = {
377 ...toLabels(news.labels),
378 ...(yield* createInternalLabels(id)),
379 };
380 const capacity = desiredCapacity(news);
381 const gcpConfig = desiredGcpConfig(news, env.project, location);
382 const rebalanceConfig = news.rebalanceConfig;
383
384 let current = yield* getCluster(name);
385
386 if (current === undefined) {
387 const created = yield* kafka
388 .createProjectsLocationsClusters({
389 parent: locationParent(env.project, location),
390 clusterId,
391 body: {
392 gcpConfig,
393 capacityConfig: capacity,
394 rebalanceConfig,
395 tlsConfig: news.tlsConfig,
396 updateOptions: news.updateOptions,
397 labels: desiredLabels,
398 },
399 })
400 .pipe(
401 retryCreateQuota,
402 Effect.catchTag("Conflict", () => Effect.succeed(undefined)),
403 );
404 if (created !== undefined) {
405 yield* waitForOperation(created);
406 }
407 current = yield* waitUntilExists(getCluster(name), name);
408 }
409
410 if (current === undefined) {
411 return yield* new ClusterNotResolved({ name });
412 }
413
414 current = yield* waitUntilReady(
415 getCluster(name),
416 name,
417 (cluster) => cluster.state,
418 );
419
420 const observedLabels = tagRecord(current.labels);
421 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
422 const labelsChanged = upsert.length > 0 || removed.length > 0;
423 const capacityChanged =
424 (current.capacityConfig?.vcpuCount ?? "") !==
425 (capacity.vcpuCount ?? "") ||
426 (current.capacityConfig?.memoryBytes ?? "") !==
427 (capacity.memoryBytes ?? "");
428 const rebalanceChanged =
429 (current.rebalanceConfig?.mode ?? "") !== (rebalanceConfig?.mode ?? "");
430 const tlsChanged =
431 fingerprint(current.tlsConfig) !== fingerprint(news.tlsConfig);
432 const updateOptionsChanged =
433 fingerprint(current.updateOptions) !== fingerprint(news.updateOptions);
434
435 if (
436 labelsChanged ||
437 capacityChanged ||
438 rebalanceChanged ||
439 tlsChanged ||
440 updateOptionsChanged
441 ) {
442 const op = yield* kafka.patchProjectsLocationsClusters({
443 name,
444 updateMask: fieldMask([
445 labelsChanged && "labels",
446 capacityChanged && "capacity_config",
447 rebalanceChanged && "rebalance_config",
448 tlsChanged && "tls_config",
449 updateOptionsChanged && "update_options",
450 ]),
451 body: {
452 labels: desiredLabels,
453 capacityConfig: capacity,
454 rebalanceConfig,
455 tlsConfig: news.tlsConfig,
456 updateOptions: news.updateOptions,
457 },
458 });
459 yield* waitForOperation(op);
460 current = yield* waitUntilReady(
461 getCluster(name),
462 name,
463 (cluster) => cluster.state,
464 );
465 }
466
467 if (current === undefined) {
468 return yield* new ResourceNotResolved({ name });
469 }
470 return toAttrs(current, env.project);
471 }),
472
473 delete: Effect.fn(function* ({ output }) {
474 const op = yield* kafka
475 .deleteProjectsLocationsClusters({ name: output.name })
476 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
477 if (op !== undefined) {
478 yield* waitForOperation(op, { notFoundOk: true });
479 }
480 yield* waitUntilGone(getCluster(output.name), output.name);
481 }),
482 });