1import * as redshiftserverless from "@distilled.cloud/aws/redshift-serverless";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import { createPhysicalName } from "../../PhysicalName.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
12import type { Providers } from "../Providers.ts";
13import { readTags, syncTags, toWireTags } from "./internal.ts";
14
15export interface WorkgroupProps {
16 /**
17 * Name of the workgroup. Must be 3-64 characters, lowercase letters,
18 * numbers, and hyphens. If omitted, a deterministic physical name is
19 * generated. Changing the name replaces the workgroup.
20 */
21 workgroupName?: string;
22 /**
23 * Name of the namespace this workgroup provides compute for. Changing the
24 * namespace replaces the workgroup.
25 */
26 namespaceName: string;
27 /**
28 * Base compute capacity in Redshift Processing Units (RPUs), 8-1024 in
29 * multiples of 8. 8 RPU is the cheapest configuration.
30 * @default 8
31 */
32 baseCapacity?: number;
33 /**
34 * Maximum RPUs the workgroup can burst to. Omit to disable the ceiling.
35 */
36 maxCapacity?: number;
37 /**
38 * Route all traffic through the VPC (enhanced VPC routing) instead of the
39 * internet.
40 * @default false
41 */
42 enhancedVpcRouting?: boolean;
43 /**
44 * Whether the workgroup is reachable from the public internet.
45 * @default false
46 */
47 publiclyAccessible?: boolean;
48 /**
49 * VPC subnet IDs the workgroup runs in. Must span at least three
50 * Availability Zones. Defaults to the account's default-VPC subnets.
51 */
52 subnetIds?: string[];
53 /**
54 * VPC security group IDs attached to the workgroup.
55 */
56 securityGroupIds?: string[];
57 /**
58 * Port the workgroup listens on.
59 * @default 5439
60 */
61 port?: number;
62 /**
63 * Redshift configuration parameters (e.g. `max_query_execution_time`,
64 * `enable_user_activity_logging`) as key/value pairs.
65 */
66 configParameters?: Record<string, string>;
67 /**
68 * User-defined tags for the workgroup.
69 */
70 tags?: Record<string, string>;
71}
72
73export interface Workgroup extends Resource<
74 "AWS.RedshiftServerless.Workgroup",
75 WorkgroupProps,
76 {
77 /**
78 * Name of the workgroup.
79 */
80 workgroupName: string;
81 /**
82 * ARN of the workgroup.
83 */
84 workgroupArn: string;
85 /**
86 * Unique ID of the workgroup.
87 */
88 workgroupId: string;
89 /**
90 * Name of the namespace the workgroup computes against.
91 */
92 namespaceName: string;
93 /**
94 * Current workgroup status (e.g. `"AVAILABLE"`).
95 */
96 status: string;
97 /**
98 * DNS address of the workgroup endpoint (pgwire host).
99 */
100 endpointAddress: string | undefined;
101 /**
102 * Port of the workgroup endpoint (5439 by default).
103 */
104 endpointPort: number | undefined;
105 /**
106 * Whether the endpoint is reachable from the public internet.
107 */
108 publiclyAccessible: boolean | undefined;
109 },
110 never,
111 Providers
112> {}
113
114/**
115 * An Amazon Redshift Serverless workgroup — the compute half of a serverless
116 * data warehouse.
117 *
118 * A workgroup provides on-demand compute (measured in RPUs) against a
119 * {@link Namespace}'s data. Creating a workgroup is asynchronous and takes
120 * roughly 2-5 minutes; the provider waits (bounded) for it to become
121 * `AVAILABLE`. Because a running workgroup bills against its RPU floor, tear
122 * it down promptly when you are done.
123 *
124 * ### Creating a Workgroup
125 * **Example:** Minimal (Cheapest) Workgroup
126 * ```typescript
127 * const namespace = yield* RedshiftServerless.Namespace("Analytics", {
128 * adminUsername: "admin",
129 * manageAdminPassword: true,
130 * });
131 * const workgroup = yield* RedshiftServerless.Workgroup("AnalyticsWg", {
132 * namespaceName: namespace.namespaceName,
133 * baseCapacity: 8,
134 * });
135 * // workgroup.endpointAddress -> "<wg>.<account>.<region>.redshift-serverless.amazonaws.com"
136 * ```
137 *
138 * ### Networking
139 * **Example:** Publicly Accessible with Explicit Subnets
140 * ```typescript
141 * const workgroup = yield* RedshiftServerless.Workgroup("AnalyticsWg", {
142 * namespaceName: namespace.namespaceName,
143 * baseCapacity: 8,
144 * publiclyAccessible: true,
145 * subnetIds: [subnetA.subnetId, subnetB.subnetId, subnetC.subnetId],
146 * securityGroupIds: [securityGroup.groupId],
147 * enhancedVpcRouting: false,
148 * });
149 * ```
150 *
151 * @resource
152 */
153export const Workgroup = Resource<Workgroup>(
154 "AWS.RedshiftServerless.Workgroup",
155);
156
157class WorkgroupNotSettled extends Data.TaggedError("WorkgroupNotSettled")<{
158 readonly workgroupName: string;
159 readonly status: string;
160}> {}
161
162const statusIs = (status: string | undefined, expected: string): boolean =>
163 status?.toUpperCase() === expected;
164
165const sameStringSet = (a: string[], b: string[]): boolean => {
166 const as = [...a].sort();
167 const bs = [...b].sort();
168 return as.length === bs.length && as.every((v, i) => v === bs[i]);
169};
170
171const toConfigList = (
172 config: Record<string, string> | undefined,
173): redshiftserverless.ConfigParameter[] | undefined =>
174 config === undefined
175 ? undefined
176 : Object.entries(config).map(([parameterKey, parameterValue]) => ({
177 parameterKey,
178 parameterValue,
179 }));
180
181export const WorkgroupProvider = () =>
182 Provider.effect(
183 Workgroup,
184 Effect.gen(function* () {
185 const toName = (id: string, props: Partial<WorkgroupProps>) =>
186 props.workgroupName
187 ? Effect.succeed(props.workgroupName)
188 : createPhysicalName({ id, maxLength: 64 });
189
190 const readWorkgroup = Effect.fn(function* (name: string) {
191 const response = yield* redshiftserverless
192 .getWorkgroup({ workgroupName: name })
193 .pipe(
194 Effect.catchTag("ResourceNotFoundException", () =>
195 Effect.succeed(undefined),
196 ),
197 );
198 return response?.workgroup;
199 });
200
201 // Workgroup create/update converge through CREATING/MODIFYING. Creation
202 // takes ~2-5 min; budget ~8 min (96 * 5s).
203 const waitForAvailable = Effect.fn(function* (name: string) {
204 return yield* readWorkgroup(name).pipe(
205 Effect.flatMap((wg) =>
206 wg !== undefined && !statusIs(wg.status, "AVAILABLE")
207 ? Effect.fail(
208 new WorkgroupNotSettled({
209 workgroupName: name,
210 status: wg.status ?? "UNKNOWN",
211 }),
212 )
213 : Effect.succeed(wg),
214 ),
215 Effect.retry({
216 while: (e) => e instanceof WorkgroupNotSettled,
217 schedule: Schedule.max([
218 Schedule.fixed("5 seconds"),
219 Schedule.recurs(96),
220 ]),
221 }),
222 );
223 });
224
225 const waitUntilGone = Effect.fn(function* (name: string) {
226 yield* readWorkgroup(name).pipe(
227 Effect.flatMap((wg) =>
228 wg === undefined
229 ? Effect.void
230 : Effect.fail(
231 new WorkgroupNotSettled({
232 workgroupName: name,
233 status: wg.status ?? "DELETING",
234 }),
235 ),
236 ),
237 Effect.retry({
238 while: (e) => e instanceof WorkgroupNotSettled,
239 schedule: Schedule.max([
240 Schedule.fixed("5 seconds"),
241 Schedule.recurs(96),
242 ]),
243 }),
244 );
245 });
246
247 const toAttrs = (wg: redshiftserverless.Workgroup) => ({
248 workgroupName: wg.workgroupName!,
249 workgroupArn: wg.workgroupArn!,
250 workgroupId: wg.workgroupId!,
251 namespaceName: wg.namespaceName!,
252 status: wg.status ?? "UNKNOWN",
253 endpointAddress: wg.endpoint?.address,
254 endpointPort: wg.endpoint?.port,
255 publiclyAccessible: wg.publiclyAccessible,
256 });
257
258 // Redshift Serverless rejects multi-parameter updates: "You can't
259 // update multiple parameters in one request." Apply each drifted field
260 // as its own updateWorkgroup call, waiting for AVAILABLE between them.
261 const applyUpdate = Effect.fn(function* (
262 name: string,
263 patch: Omit<redshiftserverless.UpdateWorkgroupRequest, "workgroupName">,
264 ) {
265 yield* redshiftserverless.updateWorkgroup({
266 workgroupName: name,
267 ...patch,
268 });
269 return yield* waitForAvailable(name);
270 });
271
272 return {
273 stables: ["workgroupName", "workgroupArn", "workgroupId"],
274
275 diff: Effect.fn(function* ({ id, olds, news }) {
276 if (!isResolved(news)) return undefined;
277 if (
278 (yield* toName(id, olds ?? {})) !== (yield* toName(id, news ?? {}))
279 ) {
280 return { action: "replace" } as const;
281 }
282 // A workgroup can't be reassigned to another namespace.
283 if (
284 (news?.namespaceName ?? undefined) !==
285 (olds?.namespaceName ?? undefined)
286 ) {
287 return { action: "replace" } as const;
288 }
289 }),
290
291 read: Effect.fn(function* ({ id, olds, output }) {
292 const name = output?.workgroupName ?? (yield* toName(id, olds ?? {}));
293 const wg = yield* readWorkgroup(name);
294 if (wg === undefined) return undefined;
295 const attrs = toAttrs(wg);
296 const tags = yield* readTags(attrs.workgroupArn);
297 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
298 }),
299
300 reconcile: Effect.fn(function* ({ id, news, output, session }) {
301 const name = output?.workgroupName ?? (yield* toName(id, news));
302 const internalTags = yield* createInternalTags(id);
303 const desiredTags = { ...internalTags, ...news.tags };
304 const baseCapacity = news.baseCapacity ?? 8;
305
306 // 1. Observe.
307 let observed = yield* readWorkgroup(name);
308
309 // 2. Ensure — create if missing; tolerate a concurrent create.
310 if (observed === undefined) {
311 const created = yield* redshiftserverless
312 .createWorkgroup({
313 workgroupName: name,
314 namespaceName: news.namespaceName,
315 baseCapacity,
316 maxCapacity: news.maxCapacity,
317 enhancedVpcRouting: news.enhancedVpcRouting,
318 publiclyAccessible: news.publiclyAccessible,
319 subnetIds: news.subnetIds,
320 securityGroupIds: news.securityGroupIds,
321 port: news.port,
322 configParameters: toConfigList(news.configParameters),
323 tags: toWireTags(desiredTags),
324 })
325 .pipe(
326 Effect.map((r) => r.workgroup),
327 Effect.catchTag("ConflictException", () =>
328 redshiftserverless
329 .getWorkgroup({ workgroupName: name })
330 .pipe(Effect.map((r) => r.workgroup)),
331 ),
332 );
333 observed = created;
334 }
335
336 const settled = yield* waitForAvailable(name);
337 if (settled !== undefined) observed = settled;
338 if (observed === undefined) {
339 return yield* Effect.fail(
340 new Error(
341 `Redshift workgroup '${name}' disappeared while reconciling`,
342 ),
343 );
344 }
345
346 // 3. Sync — one updateWorkgroup per drifted field (multi-field
347 // updates are rejected). Only mutate fields the user specified.
348 if (
349 news.baseCapacity !== undefined &&
350 observed.baseCapacity !== baseCapacity
351 ) {
352 observed = (yield* applyUpdate(name, { baseCapacity })) ?? observed;
353 }
354 if (
355 news.enhancedVpcRouting !== undefined &&
356 observed.enhancedVpcRouting !== news.enhancedVpcRouting
357 ) {
358 observed =
359 (yield* applyUpdate(name, {
360 enhancedVpcRouting: news.enhancedVpcRouting,
361 })) ?? observed;
362 }
363 if (
364 news.publiclyAccessible !== undefined &&
365 observed.publiclyAccessible !== news.publiclyAccessible
366 ) {
367 observed =
368 (yield* applyUpdate(name, {
369 publiclyAccessible: news.publiclyAccessible,
370 })) ?? observed;
371 }
372 if (news.port !== undefined && observed.port !== news.port) {
373 observed =
374 (yield* applyUpdate(name, { port: news.port })) ?? observed;
375 }
376 if (
377 news.securityGroupIds !== undefined &&
378 !sameStringSet(
379 observed.securityGroupIds ?? [],
380 news.securityGroupIds,
381 )
382 ) {
383 observed =
384 (yield* applyUpdate(name, {
385 securityGroupIds: news.securityGroupIds,
386 })) ?? observed;
387 }
388
389 // 3b. Sync tags — diff against OBSERVED cloud tags.
390 yield* syncTags(observed.workgroupArn!, desiredTags);
391
392 // 4. Return fresh attributes.
393 yield* session.note(name);
394 return toAttrs(observed);
395 }),
396
397 delete: Effect.fn(function* ({ output }) {
398 yield* redshiftserverless
399 .deleteWorkgroup({ workgroupName: output.workgroupName })
400 .pipe(
401 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
402 // A concurrent modification may still be settling — retry.
403 Effect.retry({
404 while: (e) => e._tag === "ConflictException",
405 schedule: Schedule.max([
406 Schedule.fixed("5 seconds"),
407 Schedule.recurs(24),
408 ]),
409 }),
410 Effect.catchTag("ConflictException", () => Effect.void),
411 );
412 yield* waitUntilGone(output.workgroupName);
413 }),
414
415 list: () =>
416 redshiftserverless.listWorkgroups.pages({}).pipe(
417 Stream.runCollect,
418 Effect.map((chunk) =>
419 Array.from(chunk)
420 .flatMap((page) => page.workgroups ?? [])
421 .flatMap((wg) =>
422 wg.workgroupName !== undefined &&
423 wg.workgroupArn !== undefined &&
424 wg.workgroupId !== undefined &&
425 wg.namespaceName !== undefined
426 ? [
427 {
428 workgroupName: wg.workgroupName,
429 workgroupArn: wg.workgroupArn,
430 workgroupId: wg.workgroupId,
431 namespaceName: wg.namespaceName,
432 status: wg.status ?? "UNKNOWN",
433 endpointAddress: wg.endpoint?.address,
434 endpointPort: wg.endpoint?.port,
435 publiclyAccessible: wg.publiclyAccessible,
436 },
437 ]
438 : [],
439 ),
440 ),
441 ),
442 };
443 }),
444 );