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 {
11 createInternalLabels,
12 diffLabels,
13 hasAlchemyLabels,
14 toLabels,
15} from "../Labels.ts";
16import type { Providers } from "../Providers.ts";
17import {
18 collectPages,
19 connectionProfileOf,
20 emptyMessage,
21 fieldMask,
22 fingerprint,
23 hasAlchemyLabelMap,
24 lastSegment,
25 listAtLocation,
26 locationParent,
27 normalizeLocation,
28 parseName,
29 replaceOnIdentity,
30 ResourceNotResolved,
31 settleOperation,
32 toPhysicalId,
33 userLabels,
34 waitUntilExists,
35 waitUntilGone,
36} from "./internal.ts";
37
38export type StreamState = ds.StreamStateEnum | (string & {});
39export type SourceConfig = ds.SourceConfig;
40export type DestinationConfig = ds.DestinationConfig;
41export type BackfillAllStrategy = ds.BackfillAllStrategy;
42export type BigQueryDestinationConfig = ds.BigQueryDestinationConfig;
43export type GcsDestinationConfig = ds.GcsDestinationConfig;
44export type MysqlSourceConfig = ds.MysqlSourceConfig;
45export type PostgresqlSourceConfig = ds.PostgresqlSourceConfig;
46export type OracleSourceConfig = ds.OracleSourceConfig;
47export type SqlServerSourceConfig = ds.SqlServerSourceConfig;
48export type MongodbSourceConfig = ds.MongodbSourceConfig;
49export type SpannerSourceConfig = ds.SpannerSourceConfig;
50export type SalesforceSourceConfig = ds.SalesforceSourceConfig;
51export type SalesforceMarketingCloudSourceConfig =
52 ds.SalesforceMarketingCloudSourceConfig;
53export type ServiceNowSourceConfig = ds.ServiceNowSourceConfig;
54export type RuleSet = ds.RuleSet;
55export type DatastreamError = ds.Datastream_Error;
56
57export type StreamProps = {
58 /**
59 * Stream id (the `{stream}` segment of
60 * `projects/{project}/locations/{location}/streams/{stream}`). If
61 * omitted, a unique RFC1035 name is generated. Immutable — changing it
62 * replaces the stream.
63 */
64 streamId?: string;
65 /**
66 * Region (`us-central1`, …). Immutable — changing it replaces the
67 * stream. `US-CENTRAL1` is accepted and normalized to `us-central1`.
68 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
69 */
70 location?: string;
71 /**
72 * User-friendly display name.
73 */
74 displayName?: string;
75 /**
76 * User labels. Alchemy ownership labels are merged in automatically.
77 */
78 labels?: Record<string, string>;
79 /**
80 * Source connection profile and engine-specific include/exclude
81 * objects. `sourceConnectionProfile` may be a full name or id
82 * (combined with `location`). Changing the source profile replaces the
83 * stream; other source config updates in place.
84 */
85 sourceConfig: SourceConfig;
86 /**
87 * Destination connection profile and BigQuery / GCS writer config.
88 * `destinationConnectionProfile` may be a full name or id (combined
89 * with `location`). Changing the destination profile replaces the
90 * stream; other destination config updates in place.
91 */
92 destinationConfig: DestinationConfig;
93 /**
94 * Automatically backfill objects included in the source config.
95 * Mutually exclusive with `backfillNone`.
96 */
97 backfillAll?: BackfillAllStrategy;
98 /**
99 * Do not automatically backfill any objects. Mutually exclusive with
100 * `backfillAll`.
101 */
102 backfillNone?: Record<string, never>;
103 /**
104 * Customer-managed Cloud KMS key used to encrypt stream data.
105 * Immutable — changing it replaces the stream.
106 */
107 customerManagedEncryptionKey?: string;
108 /**
109 * Customization rule sets applied to stream objects.
110 */
111 ruleSets?: ReadonlyArray<RuleSet>;
112 /**
113 * Desired stream state (`NOT_STARTED`, `RUNNING`, `PAUSED`, …).
114 */
115 state?: StreamState;
116 /**
117 * Create or update without validating connectivity.
118 */
119 force?: boolean;
120 /**
121 * Validate without creating or updating resources.
122 */
123 validateOnly?: boolean;
124};
125
126export type Stream = Resource<
127 "GCP.Datastream.Stream",
128 StreamProps,
129 {
130 /** Full resource name. */
131 name: string;
132 /** Stream id (last path segment). */
133 streamId: string;
134 /** Project id. */
135 project: string;
136 /** Location id. */
137 location: string;
138 /** User-friendly display name. */
139 displayName: string | undefined;
140 /** User labels (Alchemy ownership labels stripped). */
141 labels: Record<string, string>;
142 /** Source connection profile and engine config. */
143 sourceConfig: SourceConfig | undefined;
144 /** Destination connection profile and writer config. */
145 destinationConfig: DestinationConfig | undefined;
146 /** Backfill-all strategy. */
147 backfillAll: BackfillAllStrategy | undefined;
148 /** Backfill-none strategy. */
149 backfillNone: Record<string, never> | undefined;
150 /** Customer-managed encryption key. */
151 customerManagedEncryptionKey: string | undefined;
152 /** Customization rule sets. */
153 ruleSets: ReadonlyArray<RuleSet> | undefined;
154 /** Server-reported state. */
155 state: string | undefined;
156 /** Errors on the stream. */
157 errors: ReadonlyArray<DatastreamError> | undefined;
158 /** RFC3339 last recovery timestamp. */
159 lastRecoveryTime: string | undefined;
160 /** Whether the stream satisfies physical zone isolation. */
161 satisfiesPzi: boolean | undefined;
162 /** Whether the stream satisfies physical zone separation. */
163 satisfiesPzs: boolean | undefined;
164 /** RFC3339 creation timestamp. */
165 createTime: string | undefined;
166 /** RFC3339 last-update timestamp. */
167 updateTime: string | undefined;
168 },
169 never,
170 Providers
171>;
172
173/**
174 * A Datastream stream: continuously replicates data from a source
175 * connection profile to a BigQuery or Cloud Storage destination.
176 *
177 * `streamId`, `location`, `customerManagedEncryptionKey`, and the source
178 * / destination connection profile identities are replacement triggers.
179 * Display name, labels, include/exclude objects, backfill strategy,
180 * rule sets, and desired `state` update in place.
181 *
182 * ### Creating a Stream
183 * **Example:** MySQL to BigQuery
184 * ```typescript
185 * const source = yield* GCP.Datastream.ConnectionProfile("MysqlSrc", {
186 * mysqlProfile: {
187 * hostname: "10.0.0.8",
188 * username: "datastream",
189 * password: process.env.MYSQL_PASSWORD,
190 * },
191 * force: true,
192 * });
193 * const dest = yield* GCP.Datastream.ConnectionProfile("BqDest", {
194 * bigqueryProfile: {},
195 * });
196 * const stream = yield* GCP.Datastream.Stream("MysqlToBq", {
197 * sourceConfig: {
198 * sourceConnectionProfile: source.name,
199 * mysqlSourceConfig: {
200 * includeObjects: { mysqlDatabases: [{ database: "app" }] },
201 * },
202 * },
203 * destinationConfig: {
204 * destinationConnectionProfile: dest.name,
205 * bigqueryDestinationConfig: {
206 * singleTargetDataset: { datasetId: "app_replica" },
207 * },
208 * },
209 * backfillAll: {},
210 * force: true,
211 * });
212 * ```
213 *
214 * @resource
215 * @category Datastream
216 */
217export const Stream = Resource<Stream>("GCP.Datastream.Stream");
218
219const resourceName = (project: string, location: string, streamId: string) =>
220 `${locationParent(project, location)}/streams/${streamId}`;
221
222const expandSource = (
223 config: SourceConfig,
224 project: string,
225 location: string,
226): SourceConfig => ({
227 ...config,
228 sourceConnectionProfile: config.sourceConnectionProfile
229 ? connectionProfileOf(config.sourceConnectionProfile, project, location)
230 : undefined,
231});
232
233const expandDestination = (
234 config: DestinationConfig,
235 project: string,
236 location: string,
237): DestinationConfig => ({
238 ...config,
239 destinationConnectionProfile: config.destinationConnectionProfile
240 ? connectionProfileOf(
241 config.destinationConnectionProfile,
242 project,
243 location,
244 )
245 : undefined,
246});
247
248const profileIdOf = (value: string | undefined) => lastSegment(value ?? "");
249
250const toAttrs = (item: ds.Stream, project: string, region: string) => {
251 const name = item.name ?? "";
252 const parsed = parseName(name, "streams", region);
253 return {
254 name,
255 streamId: parsed.id,
256 project: parsed.project || project,
257 location: parsed.location,
258 displayName: item.displayName,
259 labels: userLabels(item.labels),
260 sourceConfig: item.sourceConfig,
261 destinationConfig: item.destinationConfig,
262 backfillAll: item.backfillAll,
263 backfillNone: emptyMessage(item.backfillNone),
264 customerManagedEncryptionKey: item.customerManagedEncryptionKey,
265 ruleSets: item.ruleSets,
266 state: item.state,
267 errors: item.errors,
268 lastRecoveryTime: item.lastRecoveryTime,
269 satisfiesPzi: item.satisfiesPzi,
270 satisfiesPzs: item.satisfiesPzs,
271 createTime: item.createTime,
272 updateTime: item.updateTime,
273 };
274};
275
276const getByName = (name: string) =>
277 name.length === 0
278 ? Effect.succeed(undefined)
279 : ds
280 .getProjectsLocationsStreams({ name })
281 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
282
283const listOwned = (project: string, region: string) =>
284 listAtLocation(project, region, (parent) =>
285 collectPages(
286 ds.listProjectsLocationsStreams.pages({
287 parent,
288 pageSize: 1000,
289 }),
290 (page) => page.streams,
291 ).pipe(
292 Effect.map((items) =>
293 items.filter((item) => hasAlchemyLabelMap(item.labels)),
294 ),
295 ),
296 );
297
298const sourceConfigWithoutProfile = (config: SourceConfig | undefined) => {
299 if (config === undefined) return undefined;
300 const { sourceConnectionProfile: _ignored, ...rest } = config;
301 return rest;
302};
303
304const destinationConfigWithoutProfile = (
305 config: DestinationConfig | undefined,
306) => {
307 if (config === undefined) return undefined;
308 const { destinationConnectionProfile: _ignored, ...rest } = config;
309 return rest;
310};
311
312export const StreamProvider = () =>
313 Provider.succeed(Stream, {
314 stables: ["name", "streamId", "project", "location", "createTime"],
315
316 diff: Effect.fn(function* ({ news, olds, output }) {
317 if (!isResolved(news)) return undefined;
318 const env = yield* GcpEnvironment.current;
319 const previousSource = profileIdOf(
320 olds?.sourceConfig?.sourceConnectionProfile ??
321 output?.sourceConfig?.sourceConnectionProfile,
322 );
323 const nextSource = profileIdOf(
324 news.sourceConfig.sourceConnectionProfile ??
325 olds?.sourceConfig?.sourceConnectionProfile ??
326 output?.sourceConfig?.sourceConnectionProfile,
327 );
328 const previousDest = profileIdOf(
329 olds?.destinationConfig?.destinationConnectionProfile ??
330 output?.destinationConfig?.destinationConnectionProfile,
331 );
332 const nextDest = profileIdOf(
333 news.destinationConfig.destinationConnectionProfile ??
334 olds?.destinationConfig?.destinationConnectionProfile ??
335 output?.destinationConfig?.destinationConnectionProfile,
336 );
337 const previousCmek =
338 olds?.customerManagedEncryptionKey ??
339 output?.customerManagedEncryptionKey;
340 const nextCmek = news.customerManagedEncryptionKey ?? previousCmek;
341 return replaceOnIdentity({
342 previousId: olds?.streamId ?? output?.streamId,
343 nextId: news.streamId ?? olds?.streamId ?? output?.streamId,
344 previousLocation: normalizeLocation(
345 olds?.location ?? output?.location,
346 env.region,
347 ),
348 nextLocation: normalizeLocation(
349 news.location ?? olds?.location ?? output?.location,
350 env.region,
351 ),
352 extra:
353 (previousSource.length > 0 &&
354 nextSource.length > 0 &&
355 previousSource !== nextSource) ||
356 (previousDest.length > 0 &&
357 nextDest.length > 0 &&
358 previousDest !== nextDest) ||
359 (previousCmek !== undefined &&
360 nextCmek !== undefined &&
361 previousCmek !== nextCmek) ||
362 (previousCmek === undefined && nextCmek !== undefined),
363 });
364 }),
365
366 read: Effect.fn(function* ({ id, olds, output }) {
367 const env = yield* GcpEnvironment.current;
368 const streamId = yield* toPhysicalId(
369 id,
370 olds?.streamId,
371 output?.streamId,
372 "stream",
373 );
374 const location = normalizeLocation(
375 olds?.location ?? output?.location,
376 env.region,
377 );
378 const name =
379 output?.name ?? resourceName(env.project, location, streamId);
380 const existing = yield* getByName(name);
381 if (existing === undefined) return undefined;
382 const attrs = toAttrs(existing, env.project, env.region);
383 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
384 ? attrs
385 : Unowned(attrs);
386 }),
387
388 list: () =>
389 Effect.gen(function* () {
390 const env = yield* GcpEnvironment.current;
391 const items = yield* listOwned(env.project, env.region);
392 return items.map((item) => toAttrs(item, env.project, env.region));
393 }),
394
395 reconcile: Effect.fn(function* ({ id, news, output }) {
396 const env = yield* GcpEnvironment.current;
397 const streamId = yield* toPhysicalId(
398 id,
399 news.streamId,
400 output?.streamId,
401 "stream",
402 );
403 const location = normalizeLocation(
404 news.location ?? output?.location,
405 env.region,
406 );
407 const name = resourceName(env.project, location, streamId);
408 const desiredLabels = {
409 ...toLabels(news.labels),
410 ...(yield* createInternalLabels(id)),
411 };
412 const displayName = news.displayName ?? streamId;
413 const sourceConfig = expandSource(
414 news.sourceConfig,
415 env.project,
416 location,
417 );
418 const destinationConfig = expandDestination(
419 news.destinationConfig,
420 env.project,
421 location,
422 );
423
424 let current = yield* getByName(output?.name ?? name);
425
426 const backfillAll =
427 news.backfillAll !== undefined
428 ? news.backfillAll
429 : news.backfillNone !== undefined
430 ? undefined
431 : current?.backfillAll;
432 const backfillNone =
433 news.backfillNone !== undefined
434 ? {}
435 : news.backfillAll !== undefined
436 ? undefined
437 : (emptyMessage(current?.backfillNone) ??
438 (current?.backfillAll === undefined ? {} : undefined));
439 const body: ds.Stream = {
440 displayName,
441 labels: desiredLabels,
442 sourceConfig,
443 destinationConfig,
444 backfillAll,
445 backfillNone: backfillAll === undefined ? backfillNone : undefined,
446 customerManagedEncryptionKey: news.customerManagedEncryptionKey,
447 ruleSets: news.ruleSets ? [...news.ruleSets] : undefined,
448 state: news.state,
449 };
450
451 if (current === undefined) {
452 const created = yield* ds
453 .createProjectsLocationsStreams({
454 parent: locationParent(env.project, location),
455 streamId,
456 force: news.force,
457 validateOnly: news.validateOnly,
458 body,
459 })
460 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
461 yield* settleOperation(created);
462 current = yield* waitUntilExists(getByName(name), name);
463 }
464
465 if (current === undefined) {
466 return yield* new ResourceNotResolved({ name });
467 }
468
469 const observedLabels = tagRecord(current.labels);
470 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
471 const labelsChanged = upsert.length > 0 || removed.length > 0;
472 const displayNameChanged = (current.displayName ?? "") !== displayName;
473 const sourceChanged =
474 fingerprint(sourceConfigWithoutProfile(current.sourceConfig)) !==
475 fingerprint(sourceConfigWithoutProfile(sourceConfig));
476 const destinationChanged =
477 fingerprint(
478 destinationConfigWithoutProfile(current.destinationConfig),
479 ) !== fingerprint(destinationConfigWithoutProfile(destinationConfig));
480 const backfillChanged =
481 fingerprint({
482 backfillAll: current.backfillAll,
483 backfillNone: emptyMessage(current.backfillNone),
484 }) !==
485 fingerprint({
486 backfillAll,
487 backfillNone: backfillAll === undefined ? backfillNone : undefined,
488 });
489 const ruleSetsChanged =
490 fingerprint(current.ruleSets) !== fingerprint(news.ruleSets);
491 const stateChanged =
492 news.state !== undefined && (current.state ?? "") !== news.state;
493 const mask = fieldMask([
494 labelsChanged && "labels",
495 displayNameChanged && "displayName",
496 sourceChanged && "sourceConfig",
497 destinationChanged && "destinationConfig",
498 backfillChanged &&
499 (backfillAll !== undefined ? "backfillAll" : "backfillNone"),
500 ruleSetsChanged && "ruleSets",
501 stateChanged && "state",
502 ]);
503
504 if (mask.length > 0) {
505 const patch: ds.Stream = { name: current.name ?? name };
506 if (labelsChanged) patch.labels = desiredLabels;
507 if (displayNameChanged) patch.displayName = displayName;
508 if (sourceChanged) patch.sourceConfig = sourceConfig;
509 if (destinationChanged) patch.destinationConfig = destinationConfig;
510 if (backfillChanged) {
511 patch.backfillAll = backfillAll;
512 patch.backfillNone =
513 backfillAll === undefined ? backfillNone : undefined;
514 }
515 if (ruleSetsChanged) {
516 patch.ruleSets = news.ruleSets ? [...news.ruleSets] : undefined;
517 }
518 if (stateChanged) patch.state = news.state;
519 const operation = yield* ds.patchProjectsLocationsStreams({
520 name: current.name ?? name,
521 updateMask: mask,
522 force: news.force,
523 validateOnly: news.validateOnly,
524 body: patch,
525 });
526 yield* settleOperation(operation);
527 current = yield* waitUntilExists(
528 getByName(current.name ?? name),
529 current.name ?? name,
530 );
531 }
532
533 return toAttrs(current, env.project, env.region);
534 }),
535
536 delete: Effect.fn(function* ({ output }) {
537 const operation = yield* ds
538 .deleteProjectsLocationsStreams({
539 name: output.name,
540 })
541 .pipe(
542 Effect.retry({
543 while: (error) => error._tag === "Conflict",
544 times: 8,
545 schedule: Schedule.spaced("2 seconds"),
546 }),
547 Effect.catchTag(["NotFound", "Conflict"], () =>
548 Effect.succeed(undefined),
549 ),
550 );
551 // Stream deletes take a few minutes.
552 yield* settleOperation(operation, { notFoundOk: true });
553 yield* waitUntilGone(getByName(output.name), output.name, {
554 times: 15,
555 interval: "4 seconds",
556 });
557 }),
558 });