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 expandParent,
23 fieldMask,
24 fingerprint,
25 getConnectCluster,
26 listAlchemyConnectClusters,
27 locationParent,
28 normalizeLocation,
29 parseName,
30 replaceOnIdentity,
31 retryCreateQuota,
32 ResourceNotResolved,
33 rfc1035,
34 stringMapOf,
35 MAX_CLUSTER_ID_LENGTH,
36 toPhysicalId,
37 userLabels,
38 waitForOperation,
39 waitUntilExists,
40 waitUntilGone,
41 waitUntilReady,
42} from "./internal.ts";
43
44export type ConnectNetworkConfig = {
45 /**
46 * Primary VPC subnet for Connect workers. Must be RFC1918 and at least
47 * `/22`. Defaults to the project's `default` subnet.
48 */
49 primarySubnet?: string;
50 /** Additional subnets. Deprecated — Connect can reach any endpoint. */
51 additionalSubnets?: string[];
52 /** Extra DNS domain names visible to workers. */
53 dnsDomainNames?: string[];
54};
55
56export type ConnectAccessConfig = {
57 /** VPC networks granted access. At least one is required. */
58 networkConfigs?: ConnectNetworkConfig[];
59};
60
61export type ConnectGcpConfig = {
62 /** Access configuration. */
63 accessConfig?: ConnectAccessConfig;
64 /** Secret Manager versions loaded into workers. Maximum 32. */
65 secretPaths?: string[];
66};
67
68export type ConnectCapacityConfig = {
69 /**
70 * vCPUs to provision. Minimum 3.
71 * @default 3
72 */
73 vcpuCount?: number | string;
74 /**
75 * Memory in bytes. Minimum 3221225472 (3 GiB).
76 * @default 3221225472
77 */
78 memoryBytes?: number | string;
79};
80
81export type ConnectClusterProps = {
82 /**
83 * Connect cluster id. Lowercase letters, digits, and hyphens; max 53
84 * characters (longer values are truncated). If omitted, a unique RFC1035
85 * name of at most 53 characters is generated. Immutable — changing it
86 * replaces the cluster.
87 */
88 connectClusterId?: string;
89 /**
90 * Region (`us-central1`, …). Immutable.
91 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
92 */
93 location?: string;
94 /**
95 * Kafka cluster this Connect cluster attaches to
96 * (`projects/{project}/locations/{location}/clusters/{cluster}`).
97 * Immutable — changing it replaces the Connect cluster.
98 */
99 kafkaCluster: string;
100 /**
101 * Primary subnet for workers. Defaults to the project's `default`
102 * subnet in `location`. Immutable.
103 */
104 primarySubnet?: string;
105 /** Additional subnets. */
106 additionalSubnets?: string[];
107 /** Extra DNS domain names. */
108 dnsDomainNames?: string[];
109 /** Full GCP config. Flattened fields win when both are set. */
110 gcpConfig?: ConnectGcpConfig;
111 /** Capacity. Defaults to 3 vCPU / 3 GiB. */
112 capacityConfig?: ConnectCapacityConfig;
113 /** Worker config overrides. Currently unused by the API. */
114 config?: Record<string, string>;
115 /** Secret Manager versions loaded into workers. */
116 secretPaths?: string[];
117 /**
118 * User labels. Alchemy ownership labels are merged in automatically.
119 */
120 labels?: Record<string, string>;
121};
122
123export type ConnectCluster = Resource<
124 "GCP.ManagedKafka.ConnectCluster",
125 ConnectClusterProps,
126 {
127 /** Full resource name `.../connectClusters/{connectCluster}`. */
128 name: string;
129 /** Connect cluster id. */
130 connectClusterId: string;
131 /** Project id. */
132 project: string;
133 /** Location id. */
134 location: string;
135 /** Attached Kafka cluster. */
136 kafkaCluster: string;
137 /** User labels (Alchemy ownership labels stripped). */
138 labels: Record<string, string>;
139 /** Server-reported state. */
140 state: string | undefined;
141 /** vCPU count. */
142 vcpuCount: string | undefined;
143 /** Memory in bytes. */
144 memoryBytes: string | undefined;
145 /** Primary subnet. */
146 primarySubnet: string | undefined;
147 /** Worker config. */
148 config: Record<string, string>;
149 /** RFC3339 creation timestamp. */
150 createTime: string | undefined;
151 /** RFC3339 last-update timestamp. */
152 updateTime: string | undefined;
153 },
154 never,
155 Providers
156>;
157
158/**
159 * A Managed Service for Apache Kafka Connect cluster.
160 *
161 * Changing `connectClusterId`, `location`, `kafkaCluster`, or the primary
162 * subnet replaces the cluster. Capacity, labels, and worker config update
163 * in place. Provisioning typically takes more than a minute.
164 *
165 * ### Creating a Connect Cluster
166 * **Example:** Attach to a Kafka cluster
167 * ```typescript
168 * const connect = yield* GCP.ManagedKafka.ConnectCluster("Connect", {
169 * kafkaCluster: cluster.name,
170 * });
171 * ```
172 *
173 * **Example:** Explicit id and labels
174 * ```typescript
175 * const connect = yield* GCP.ManagedKafka.ConnectCluster("Connect", {
176 * connectClusterId: "app-connect",
177 * kafkaCluster: cluster.name,
178 * labels: { env: "prod" },
179 * });
180 * ```
181 *
182 * @resource
183 * @category ManagedKafka
184 */
185export const ConnectCluster = Resource<ConnectCluster>(
186 "GCP.ManagedKafka.ConnectCluster",
187);
188
189export class ConnectClusterNotResolved extends Data.TaggedError(
190 "GCP.ManagedKafka.ConnectClusterNotResolved",
191)<{
192 name: string;
193}> {}
194
195const resourceName = (
196 project: string,
197 location: string,
198 connectClusterId: string,
199) =>
200 `projects/${project}/locations/${location}/connectClusters/${connectClusterId}`;
201
202const kafkaClusterOf = (value: string, project: string, location: string) =>
203 expandParent(value, project, location, "clusters");
204
205const primarySubnetOf = (
206 news: ConnectClusterProps,
207 project: string,
208 location: string,
209) =>
210 news.primarySubnet ??
211 news.gcpConfig?.accessConfig?.networkConfigs?.[0]?.primarySubnet ??
212 defaultSubnet(project, location);
213
214const desiredGcpConfig = (
215 news: ConnectClusterProps,
216 project: string,
217 location: string,
218): kafka.ConnectGcpConfig => ({
219 accessConfig: {
220 networkConfigs: [
221 {
222 primarySubnet: primarySubnetOf(news, project, location),
223 additionalSubnets:
224 news.additionalSubnets ??
225 news.gcpConfig?.accessConfig?.networkConfigs?.[0]?.additionalSubnets,
226 dnsDomainNames:
227 news.dnsDomainNames ??
228 news.gcpConfig?.accessConfig?.networkConfigs?.[0]?.dnsDomainNames,
229 },
230 ],
231 },
232 secretPaths: news.secretPaths ?? news.gcpConfig?.secretPaths,
233});
234
235const toAttrs = (cluster: kafka.ConnectCluster, project: string) => {
236 const name = cluster.name ?? "";
237 const parsed = parseName(name, "connectClusters");
238 return {
239 name,
240 connectClusterId: parsed.id,
241 project: parsed.project || project,
242 location: parsed.location,
243 kafkaCluster: cluster.kafkaCluster ?? "",
244 labels: userLabels(cluster.labels),
245 state: cluster.state,
246 vcpuCount: cluster.capacityConfig?.vcpuCount,
247 memoryBytes: cluster.capacityConfig?.memoryBytes,
248 primarySubnet:
249 cluster.gcpConfig?.accessConfig?.networkConfigs?.[0]?.primarySubnet,
250 config: stringMapOf(cluster.config),
251 createTime: cluster.createTime,
252 updateTime: cluster.updateTime,
253 };
254};
255
256export const ConnectClusterProvider = () =>
257 Provider.succeed(ConnectCluster, {
258 stables: [
259 "name",
260 "connectClusterId",
261 "project",
262 "location",
263 "kafkaCluster",
264 "createTime",
265 ],
266
267 diff: Effect.fn(function* ({ news, olds, output }) {
268 if (!isResolved(news)) return undefined;
269 const env = yield* GcpEnvironment.current;
270 const location = normalizeLocation(
271 news.location ?? output?.location,
272 env.region,
273 );
274 const previousKafka = olds?.kafkaCluster ?? output?.kafkaCluster ?? "";
275 const nextKafka = kafkaClusterOf(
276 news.kafkaCluster,
277 env.project,
278 location,
279 );
280 const previousSubnet = olds?.primarySubnet ?? output?.primarySubnet ?? "";
281 const nextSubnet = primarySubnetOf(news, env.project, location);
282 return replaceOnIdentity({
283 previousId: olds?.connectClusterId ?? output?.connectClusterId,
284 nextId: news.connectClusterId
285 ? rfc1035(news.connectClusterId, "connect", MAX_CLUSTER_ID_LENGTH)
286 : (olds?.connectClusterId ?? output?.connectClusterId),
287 previousLocation: normalizeLocation(
288 olds?.location ?? output?.location,
289 env.region,
290 ),
291 nextLocation: location,
292 extra: previousKafka !== nextKafka || previousSubnet !== nextSubnet,
293 });
294 }),
295
296 read: Effect.fn(function* ({ id, olds, output }) {
297 const env = yield* GcpEnvironment.current;
298 const connectClusterId = yield* toPhysicalId(
299 id,
300 olds?.connectClusterId,
301 output?.connectClusterId,
302 "connect",
303 MAX_CLUSTER_ID_LENGTH,
304 );
305 const location = normalizeLocation(
306 olds?.location ?? output?.location,
307 env.region,
308 );
309 const name =
310 output?.name ?? resourceName(env.project, location, connectClusterId);
311 const existing = yield* getConnectCluster(name);
312 if (existing === undefined) return undefined;
313 const attrs = toAttrs(existing, env.project);
314 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
315 ? attrs
316 : Unowned(attrs);
317 }),
318
319 list: () =>
320 Effect.gen(function* () {
321 const env = yield* GcpEnvironment.current;
322 const clusters = yield* listAlchemyConnectClusters(
323 env.project,
324 env.region,
325 );
326 return clusters.map((cluster) => toAttrs(cluster, env.project));
327 }),
328
329 reconcile: Effect.fn(function* ({ id, news, output }) {
330 const env = yield* GcpEnvironment.current;
331 const connectClusterId = yield* toPhysicalId(
332 id,
333 news.connectClusterId,
334 output?.connectClusterId,
335 "connect",
336 MAX_CLUSTER_ID_LENGTH,
337 );
338 const location = normalizeLocation(
339 news.location ?? output?.location,
340 env.region,
341 );
342 const name =
343 output?.name ?? resourceName(env.project, location, connectClusterId);
344 const kafkaCluster = kafkaClusterOf(
345 news.kafkaCluster,
346 env.project,
347 location,
348 );
349 const desiredLabels = {
350 ...toLabels(news.labels),
351 ...(yield* createInternalLabels(id)),
352 };
353 const capacity = {
354 vcpuCount: asCountString(
355 news.capacityConfig?.vcpuCount,
356 DEFAULT_VCPU_COUNT,
357 ),
358 memoryBytes: asCountString(
359 news.capacityConfig?.memoryBytes,
360 DEFAULT_MEMORY_BYTES,
361 ),
362 };
363 const gcpConfig = desiredGcpConfig(news, env.project, location);
364
365 let current = yield* getConnectCluster(name);
366
367 if (current === undefined) {
368 const created = yield* kafka
369 .createProjectsLocationsConnectClusters({
370 parent: locationParent(env.project, location),
371 connectClusterId,
372 body: {
373 kafkaCluster,
374 gcpConfig,
375 capacityConfig: capacity,
376 config: news.config,
377 labels: desiredLabels,
378 },
379 })
380 .pipe(
381 retryCreateQuota,
382 Effect.catchTag("Conflict", () => Effect.succeed(undefined)),
383 );
384 if (created !== undefined) {
385 yield* waitForOperation(created);
386 }
387 current = yield* waitUntilExists(getConnectCluster(name), name);
388 }
389
390 if (current === undefined) {
391 return yield* new ConnectClusterNotResolved({ name });
392 }
393
394 current = yield* waitUntilReady(
395 getConnectCluster(name),
396 name,
397 (cluster) => cluster.state,
398 );
399
400 const observedLabels = tagRecord(current.labels);
401 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
402 const labelsChanged = upsert.length > 0 || removed.length > 0;
403 const capacityChanged =
404 (current.capacityConfig?.vcpuCount ?? "") !==
405 (capacity.vcpuCount ?? "") ||
406 (current.capacityConfig?.memoryBytes ?? "") !==
407 (capacity.memoryBytes ?? "");
408 const configChanged =
409 fingerprint(stringMapOf(current.config)) !==
410 fingerprint(stringMapOf(news.config));
411 const secretsChanged =
412 fingerprint(current.gcpConfig?.secretPaths ?? []) !==
413 fingerprint(gcpConfig.secretPaths ?? []);
414
415 if (labelsChanged || capacityChanged || configChanged || secretsChanged) {
416 const op = yield* kafka.patchProjectsLocationsConnectClusters({
417 name,
418 updateMask: fieldMask([
419 labelsChanged && "labels",
420 capacityChanged && "capacity_config",
421 configChanged && "config",
422 secretsChanged && "gcp_config.secret_paths",
423 ]),
424 body: {
425 labels: desiredLabels,
426 capacityConfig: capacity,
427 config: news.config,
428 gcpConfig,
429 },
430 });
431 yield* waitForOperation(op);
432 current = yield* waitUntilReady(
433 getConnectCluster(name),
434 name,
435 (cluster) => cluster.state,
436 );
437 }
438
439 if (current === undefined) {
440 return yield* new ResourceNotResolved({ name });
441 }
442 return toAttrs(current, env.project);
443 }),
444
445 delete: Effect.fn(function* ({ output }) {
446 const op = yield* kafka
447 .deleteProjectsLocationsConnectClusters({ name: output.name })
448 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
449 if (op !== undefined) {
450 yield* waitForOperation(op, { notFoundOk: true });
451 }
452 yield* waitUntilGone(getConnectCluster(output.name), output.name);
453 }),
454 });