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 fieldMask,
14 fingerprint,
15 getConnector,
16 listAlchemyConnectClusters,
17 normalizeLocation,
18 parseName,
19 replaceOnIdentity,
20 retryTransient,
21 rfc1035,
22 stringMapOf,
23 toPhysicalId,
24 waitUntilGone,
25 waitUntilReady,
26} from "./internal.ts";
27
28export type TaskRetryPolicy = {
29 /** Minimum backoff before retrying a failed task (e.g. `"60s"`). */
30 minimumBackoff?: string;
31 /** Maximum backoff before retrying a failed task (e.g. `"43200s"`). */
32 maximumBackoff?: string;
33 /**
34 * Disable task retry.
35 * @default false
36 */
37 taskRetryDisabled?: boolean;
38};
39
40export type ConnectClustersConnectorProps = {
41 /**
42 * Parent Connect cluster. Full name
43 * `projects/{project}/locations/{location}/connectClusters/{id}` or
44 * the Connect cluster id (combined with `location`). Immutable —
45 * changing it replaces the connector.
46 */
47 connectCluster: string;
48 /**
49 * Region used when `connectCluster` is a bare id.
50 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
51 */
52 location?: string;
53 /**
54 * Connector id. If omitted, a unique RFC1035 name is generated.
55 * Immutable — changing it replaces the connector.
56 */
57 connectorId?: string;
58 /**
59 * Connector config (`connector.class`, `tasks.max`, converters, …).
60 */
61 configs?: Record<string, string>;
62 /**
63 * Task restart policy.
64 */
65 taskRestartPolicy?: TaskRetryPolicy;
66};
67
68export type ConnectClustersConnector = Resource<
69 "GCP.ManagedKafka.ConnectClustersConnector",
70 ConnectClustersConnectorProps,
71 {
72 /** Full resource name `.../connectClusters/{id}/connectors/{connector}`. */
73 name: string;
74 /** Connector id. */
75 connectorId: string;
76 /** Parent Connect cluster resource name. */
77 connectCluster: string;
78 /** Project id. */
79 project: string;
80 /** Location id. */
81 location: string;
82 /** Connector configs. */
83 configs: Record<string, string>;
84 /** Server-reported state. */
85 state: string | undefined;
86 /** Task restart policy currently applied. */
87 taskRestartPolicy: TaskRetryPolicy | undefined;
88 },
89 never,
90 Providers
91>;
92
93/**
94 * A Kafka Connect connector on a Managed Kafka Connect cluster.
95 *
96 * Changing `connectorId`, `connectCluster`, or `location` replaces the
97 * connector. Configs and the task restart policy update in place.
98 *
99 * ### Creating a Connector
100 * **Example:** Pub/Sub sink
101 * ```typescript
102 * const connector = yield* GCP.ManagedKafka.ConnectClustersConnector(
103 * "Sink",
104 * {
105 * connectCluster: connect.name,
106 * configs: {
107 * "connector.class":
108 * "com.google.pubsub.kafka.sink.CloudPubSubSinkConnector",
109 * "tasks.max": "1",
110 * topics: "orders",
111 * "cps.project": "my-project",
112 * "cps.topic": "orders-sink",
113 * },
114 * },
115 * );
116 * ```
117 *
118 * @resource
119 * @category ManagedKafka
120 */
121export const ConnectClustersConnector = Resource<ConnectClustersConnector>(
122 "GCP.ManagedKafka.ConnectClustersConnector",
123);
124
125export class ConnectClustersConnectorNotResolved extends Data.TaggedError(
126 "GCP.ManagedKafka.ConnectClustersConnectorNotResolved",
127)<{
128 name: string;
129}> {}
130
131const connectClusterOf = (value: string, project: string, location: string) =>
132 expandParent(value, project, location, "connectClusters");
133
134const resourceName = (connectCluster: string, connectorId: string) =>
135 `${connectCluster}/connectors/${connectorId}`;
136
137const policyOf = (
138 policy: kafka.TaskRetryPolicy | TaskRetryPolicy | undefined,
139): TaskRetryPolicy | undefined => {
140 if (policy === undefined) return undefined;
141 return {
142 minimumBackoff: policy.minimumBackoff,
143 maximumBackoff: policy.maximumBackoff,
144 taskRetryDisabled: policy.taskRetryDisabled === true,
145 };
146};
147
148const toAttrs = (connector: kafka.Connector, project: string) => {
149 const name = connector.name ?? "";
150 const parsed = parseName(name, "connectors");
151 return {
152 name,
153 connectorId: parsed.id,
154 connectCluster: parsed.parent,
155 project: parsed.project || project,
156 location: parsed.location,
157 configs: stringMapOf(connector.configs),
158 state: connector.state,
159 taskRestartPolicy: policyOf(connector.taskRestartPolicy),
160 };
161};
162
163export const ConnectClustersConnectorProvider = () =>
164 Provider.succeed(ConnectClustersConnector, {
165 stables: ["name", "connectorId", "connectCluster", "project", "location"],
166
167 diff: Effect.fn(function* ({ news, olds, output }) {
168 if (!isResolved(news)) return undefined;
169 const env = yield* GcpEnvironment.current;
170 const location = normalizeLocation(
171 news.location ?? output?.location,
172 env.region,
173 );
174 return replaceOnIdentity({
175 previousId: olds?.connectorId ?? output?.connectorId,
176 nextId: news.connectorId
177 ? rfc1035(news.connectorId, "connector")
178 : (olds?.connectorId ?? output?.connectorId),
179 previousParent: olds?.connectCluster ?? output?.connectCluster,
180 nextParent: connectClusterOf(
181 news.connectCluster,
182 env.project,
183 location,
184 ),
185 });
186 }),
187
188 read: Effect.fn(function* ({ id, olds, output }) {
189 const env = yield* GcpEnvironment.current;
190 const location = normalizeLocation(
191 olds?.location ?? output?.location,
192 env.region,
193 );
194 const connectorId = yield* toPhysicalId(
195 id,
196 olds?.connectorId,
197 output?.connectorId,
198 "connector",
199 );
200 const connectCluster =
201 olds?.connectCluster !== undefined
202 ? connectClusterOf(olds.connectCluster, env.project, location)
203 : (output?.connectCluster ?? "");
204 const name =
205 output?.name ??
206 (connectCluster.length > 0
207 ? resourceName(connectCluster, connectorId)
208 : "");
209 const existing = yield* getConnector(name);
210 if (existing === undefined) return undefined;
211 const attrs = toAttrs(existing, env.project);
212 return output !== undefined || olds !== undefined
213 ? attrs
214 : Unowned(attrs);
215 }),
216
217 list: () =>
218 Effect.gen(function* () {
219 const env = yield* GcpEnvironment.current;
220 const clusters = yield* listAlchemyConnectClusters(
221 env.project,
222 env.region,
223 );
224 const connectors = yield* Effect.forEach(
225 clusters.filter((cluster) => (cluster.name ?? "").length > 0),
226 (cluster: kafka.ConnectCluster) =>
227 collectPages(
228 kafka.listProjectsLocationsConnectClustersConnectors.pages({
229 parent: cluster.name!,
230 pageSize: 1000,
231 }),
232 (page) => page.connectors,
233 ).pipe(
234 Effect.catchTag("NotFound", () =>
235 Effect.succeed([] as kafka.Connector[]),
236 ),
237 ),
238 { concurrency: 4 },
239 );
240 return connectors
241 .flat()
242 .map((connector) => toAttrs(connector, env.project));
243 }),
244
245 reconcile: Effect.fn(function* ({ id, news, output }) {
246 const env = yield* GcpEnvironment.current;
247 const location = normalizeLocation(
248 news.location ?? output?.location,
249 env.region,
250 );
251 const connectCluster = connectClusterOf(
252 news.connectCluster,
253 env.project,
254 location,
255 );
256 const connectorId = yield* toPhysicalId(
257 id,
258 news.connectorId,
259 output?.connectorId,
260 "connector",
261 );
262 const name = output?.name ?? resourceName(connectCluster, connectorId);
263 const configs = news.configs;
264 const taskRestartPolicy = news.taskRestartPolicy;
265
266 let current = yield* getConnector(name);
267
268 if (current === undefined) {
269 const created = yield* retryTransient(
270 kafka
271 .createProjectsLocationsConnectClustersConnectors({
272 parent: connectCluster,
273 connectorId,
274 body: {
275 configs,
276 taskRestartPolicy,
277 },
278 })
279 .pipe(Effect.catchTag("Conflict", () => getConnector(name))),
280 );
281 current = created ?? undefined;
282 }
283
284 if (current === undefined) {
285 return yield* new ConnectClustersConnectorNotResolved({ name });
286 }
287
288 const configsChanged =
289 fingerprint(stringMapOf(current.configs)) !==
290 fingerprint(stringMapOf(configs));
291 const policyChanged =
292 fingerprint(policyOf(current.taskRestartPolicy)) !==
293 fingerprint(policyOf(taskRestartPolicy));
294
295 if (configsChanged || policyChanged) {
296 current = yield* kafka.patchProjectsLocationsConnectClustersConnectors({
297 name,
298 updateMask: fieldMask([
299 configsChanged && "configs",
300 policyChanged && "task_restart_policy",
301 ]),
302 body: {
303 configs,
304 taskRestartPolicy,
305 },
306 });
307 }
308
309 current = yield* waitUntilReady(
310 getConnector(current.name ?? name),
311 current.name ?? name,
312 (connector) => connector.state,
313 );
314 return toAttrs(current, env.project);
315 }),
316
317 delete: Effect.fn(function* ({ output }) {
318 yield* kafka
319 .deleteProjectsLocationsConnectClustersConnectors({
320 name: output.name,
321 })
322 .pipe(Effect.catchTag("NotFound", () => Effect.void));
323 yield* waitUntilGone(getConnector(output.name), output.name);
324 }),
325 });