1import * as ds from "@distilled.cloud/gcp/datastream_v1";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
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 { createInternalLabels, hasAlchemyLabels, toLabels } from "../Labels.ts";
11import type { Providers } from "../Providers.ts";
12import {
13 collectPages,
14 fingerprint,
15 hasAlchemyLabelMap,
16 listAtLocation,
17 locationParent,
18 networkOf,
19 normalizeLocation,
20 parseName,
21 replaceOnIdentity,
22 ResourceNotResolved,
23 settleOperation,
24 toPhysicalId,
25 userLabels,
26 waitUntilExists,
27 waitUntilGone,
28} from "./internal.ts";
29
30export type PrivateConnectionState =
31 | ds.PrivateConnectionStateEnum
32 | (string & {});
33export type VpcPeeringConfig = ds.VpcPeeringConfig;
34export type PscInterfaceConfig = ds.PscInterfaceConfig;
35type DatastreamError = ds.Datastream_Error;
36
37export type PrivateConnectionProps = {
38 /**
39 * Private connection id (the `{privateConnection}` segment of
40 * `projects/{project}/locations/{location}/privateConnections/{privateConnection}`).
41 * If omitted, a unique RFC1035 name is generated. Immutable — changing
42 * it replaces the connection.
43 */
44 privateConnectionId?: string;
45 /**
46 * Region (`us-central1`, …). Immutable — changing it replaces the
47 * connection. `US-CENTRAL1` is accepted and normalized to `us-central1`.
48 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
49 */
50 location?: string;
51 /**
52 * User-friendly display name. The API has no patch method — changing
53 * display name replaces the connection.
54 */
55 displayName?: string;
56 /**
57 * User labels. Alchemy ownership labels are merged in automatically.
58 * The API has no patch method — changing labels replaces the
59 * connection.
60 */
61 labels?: Record<string, string>;
62 /**
63 * VPC peering configuration. Mutually exclusive with
64 * `pscInterfaceConfig`. Immutable — changing it replaces the
65 * connection. `vpc` may be a full resource name or a bare network id.
66 */
67 vpcPeeringConfig?: {
68 vpc: string;
69 subnet: string;
70 };
71 /**
72 * PSC interface configuration. Mutually exclusive with
73 * `vpcPeeringConfig`. Immutable — changing it replaces the connection.
74 */
75 pscInterfaceConfig?: PscInterfaceConfig;
76 /**
77 * Skip validations on create.
78 */
79 force?: boolean;
80 /**
81 * PSC interface only: resolve the tenant project without creating.
82 */
83 validateOnly?: boolean;
84};
85
86export type PrivateConnection = Resource<
87 "GCP.Datastream.PrivateConnection",
88 PrivateConnectionProps,
89 {
90 /** Full resource name. */
91 name: string;
92 /** Private connection id (last path segment). */
93 privateConnectionId: string;
94 /** Project id. */
95 project: string;
96 /** Location id. */
97 location: string;
98 /** User-friendly display name. */
99 displayName: string | undefined;
100 /** User labels (Alchemy ownership labels stripped). */
101 labels: Record<string, string>;
102 /** VPC peering configuration. */
103 vpcPeeringConfig: VpcPeeringConfig | undefined;
104 /** PSC interface configuration. */
105 pscInterfaceConfig: PscInterfaceConfig | undefined;
106 /** Server-reported state. */
107 state: string | undefined;
108 /** Failure details when `state` is `FAILED`. */
109 error: DatastreamError | undefined;
110 /** Whether the connection satisfies physical zone isolation. */
111 satisfiesPzi: boolean | undefined;
112 /** Whether the connection satisfies physical zone separation. */
113 satisfiesPzs: boolean | undefined;
114 /** RFC3339 creation timestamp. */
115 createTime: string | undefined;
116 /** RFC3339 last-update timestamp. */
117 updateTime: string | undefined;
118 },
119 never,
120 Providers
121>;
122
123/**
124 * A Datastream private connection: VPC peering or a PSC interface
125 * between Datastream and a consumer network.
126 *
127 * The API has no patch method. `privateConnectionId`, `location`,
128 * `vpcPeeringConfig`, `pscInterfaceConfig`, display name, and labels are
129 * replacement triggers. Delete always force-deletes child routes.
130 *
131 * ### Creating a Private Connection
132 * **Example:** VPC peering
133 * ```typescript
134 * const network = yield* GCP.Compute.Network("DsVpc", {
135 * autoCreateSubnetworks: false,
136 * });
137 * const peering = yield* GCP.Datastream.PrivateConnection("DsPeer", {
138 * displayName: "ds-peer",
139 * vpcPeeringConfig: {
140 * vpc: network.networkName,
141 * subnet: "10.9.0.0/29",
142 * },
143 * labels: { env: "test" },
144 * });
145 * ```
146 *
147 * @resource
148 * @category Datastream
149 */
150export const PrivateConnection = Resource<PrivateConnection>(
151 "GCP.Datastream.PrivateConnection",
152);
153
154const resourceName = (
155 project: string,
156 location: string,
157 privateConnectionId: string,
158) =>
159 `${locationParent(project, location)}/privateConnections/${privateConnectionId}`;
160
161const peeringOf = (
162 value:
163 | PrivateConnectionProps["vpcPeeringConfig"]
164 | VpcPeeringConfig
165 | undefined,
166 project: string,
167) =>
168 value?.vpc === undefined && value?.subnet === undefined
169 ? undefined
170 : {
171 vpc: value?.vpc ? networkOf(value.vpc, project) : undefined,
172 subnet: value?.subnet,
173 };
174
175const kindOf = (value: {
176 vpcPeeringConfig?: unknown;
177 pscInterfaceConfig?: unknown;
178}) => (value.vpcPeeringConfig ? "vpc" : value.pscInterfaceConfig ? "psc" : "");
179
180export const toPrivateConnectionAttrs = (
181 item: ds.PrivateConnection,
182 project: string,
183 region: string,
184) => {
185 const name = item.name ?? "";
186 const parsed = parseName(name, "privateConnections", region);
187 return {
188 name,
189 privateConnectionId: parsed.id,
190 project: parsed.project || project,
191 location: parsed.location,
192 displayName: item.displayName,
193 labels: userLabels(item.labels),
194 vpcPeeringConfig: item.vpcPeeringConfig,
195 pscInterfaceConfig: item.pscInterfaceConfig,
196 state: item.state,
197 error: item.error,
198 satisfiesPzi: item.satisfiesPzi,
199 satisfiesPzs: item.satisfiesPzs,
200 createTime: item.createTime,
201 updateTime: item.updateTime,
202 };
203};
204
205export const getPrivateConnectionByName = (name: string) =>
206 name.length === 0
207 ? Effect.succeed(undefined)
208 : ds
209 .getProjectsLocationsPrivateConnections({ name })
210 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
211
212export const listPrivateConnections = (project: string, region: string) =>
213 listAtLocation(project, region, (parent) =>
214 collectPages(
215 ds.listProjectsLocationsPrivateConnections.pages({
216 parent,
217 pageSize: 1000,
218 }),
219 (page) => page.privateConnections,
220 ),
221 );
222
223const listOwned = (project: string, region: string) =>
224 listPrivateConnections(project, region).pipe(
225 Effect.map((items) =>
226 items.filter((item) => hasAlchemyLabelMap(item.labels)),
227 ),
228 );
229
230export const PrivateConnectionProvider = () =>
231 Provider.succeed(PrivateConnection, {
232 stables: [
233 "name",
234 "privateConnectionId",
235 "project",
236 "location",
237 "createTime",
238 ],
239
240 diff: Effect.fn(function* ({ news, olds, output }) {
241 if (!isResolved(news)) return undefined;
242 const env = yield* GcpEnvironment.current;
243 const previousKind = kindOf({
244 vpcPeeringConfig: olds?.vpcPeeringConfig ?? output?.vpcPeeringConfig,
245 pscInterfaceConfig:
246 olds?.pscInterfaceConfig ?? output?.pscInterfaceConfig,
247 });
248 const nextKind = kindOf(news) || previousKind;
249 const previousPeering = fingerprint(
250 olds?.vpcPeeringConfig ?? output?.vpcPeeringConfig,
251 );
252 const nextPeering = fingerprint(news.vpcPeeringConfig);
253 const previousPsc = fingerprint(
254 olds?.pscInterfaceConfig ?? output?.pscInterfaceConfig,
255 );
256 const nextPsc = fingerprint(news.pscInterfaceConfig);
257 const previousDisplay = olds?.displayName ?? output?.displayName;
258 const nextDisplay = news.displayName ?? previousDisplay;
259 const previousLabels = fingerprint(olds?.labels ?? output?.labels);
260 const nextLabels = fingerprint(
261 news.labels ?? olds?.labels ?? output?.labels,
262 );
263 return replaceOnIdentity({
264 previousId: olds?.privateConnectionId ?? output?.privateConnectionId,
265 nextId:
266 news.privateConnectionId ??
267 olds?.privateConnectionId ??
268 output?.privateConnectionId,
269 previousLocation: normalizeLocation(
270 olds?.location ?? output?.location,
271 env.region,
272 ),
273 nextLocation: normalizeLocation(
274 news.location ?? olds?.location ?? output?.location,
275 env.region,
276 ),
277 extra:
278 (olds !== undefined || output !== undefined) &&
279 (previousKind !== nextKind ||
280 (news.vpcPeeringConfig !== undefined &&
281 previousPeering !== nextPeering) ||
282 (news.pscInterfaceConfig !== undefined &&
283 previousPsc !== nextPsc) ||
284 previousDisplay !== nextDisplay ||
285 previousLabels !== nextLabels),
286 });
287 }),
288
289 read: Effect.fn(function* ({ id, olds, output }) {
290 const env = yield* GcpEnvironment.current;
291 const privateConnectionId = yield* toPhysicalId(
292 id,
293 olds?.privateConnectionId,
294 output?.privateConnectionId,
295 "pconn",
296 );
297 const location = normalizeLocation(
298 olds?.location ?? output?.location,
299 env.region,
300 );
301 const name =
302 output?.name ??
303 resourceName(env.project, location, privateConnectionId);
304 const existing = yield* getPrivateConnectionByName(name);
305 if (existing === undefined) return undefined;
306 const attrs = toPrivateConnectionAttrs(existing, env.project, env.region);
307 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
308 ? attrs
309 : Unowned(attrs);
310 }),
311
312 list: () =>
313 Effect.gen(function* () {
314 const env = yield* GcpEnvironment.current;
315 const items = yield* listOwned(env.project, env.region);
316 return items.map((item) =>
317 toPrivateConnectionAttrs(item, env.project, env.region),
318 );
319 }),
320
321 reconcile: Effect.fn(function* ({ id, news, output }) {
322 const env = yield* GcpEnvironment.current;
323 const privateConnectionId = yield* toPhysicalId(
324 id,
325 news.privateConnectionId,
326 output?.privateConnectionId,
327 "pconn",
328 );
329 const location = normalizeLocation(
330 news.location ?? output?.location,
331 env.region,
332 );
333 const name = resourceName(env.project, location, privateConnectionId);
334 const desiredLabels = {
335 ...toLabels(news.labels),
336 ...(yield* createInternalLabels(id)),
337 };
338 const displayName = news.displayName ?? privateConnectionId;
339 const vpcPeeringConfig = peeringOf(news.vpcPeeringConfig, env.project);
340
341 let current = yield* getPrivateConnectionByName(output?.name ?? name);
342
343 if (current === undefined) {
344 const created = yield* ds
345 .createProjectsLocationsPrivateConnections({
346 parent: locationParent(env.project, location),
347 privateConnectionId,
348 force: news.force,
349 validateOnly: news.validateOnly,
350 body: {
351 displayName,
352 labels: desiredLabels,
353 vpcPeeringConfig,
354 pscInterfaceConfig: news.pscInterfaceConfig,
355 },
356 })
357 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
358 // VPC peering takes 5-10 minutes.
359 yield* settleOperation(created);
360 current = yield* waitUntilExists(
361 getPrivateConnectionByName(name),
362 name,
363 );
364 }
365
366 if (current === undefined) {
367 return yield* new ResourceNotResolved({ name });
368 }
369
370 return toPrivateConnectionAttrs(current, env.project, env.region);
371 }),
372
373 delete: Effect.fn(function* ({ output }) {
374 const operation = yield* ds
375 .deleteProjectsLocationsPrivateConnections({
376 name: output.name,
377 force: true,
378 })
379 .pipe(
380 Effect.retry({
381 while: (error) => error._tag === "Conflict",
382 times: 8,
383 schedule: Schedule.spaced("2 seconds"),
384 }),
385 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
386 );
387 yield* settleOperation(operation, { notFoundOk: true });
388 yield* waitUntilGone(
389 getPrivateConnectionByName(output.name),
390 output.name,
391 { times: 8, interval: "4 seconds" },
392 );
393 }),
394 });