1import * as finspace from "@distilled.cloud/aws/finspace";
2import * as Data from "effect/Data";
3import type * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Schedule from "effect/Schedule";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { createInternalTags } from "../../Tags.ts";
11import { toWireSeconds } from "../../Util/Duration.ts";
12import type { Providers } from "../Providers.ts";
13
14export type KxClusterStatus = finspace.KxClusterStatus;
15export type KxClusterType = finspace.KxClusterType;
16export type KxAzMode = finspace.KxAzMode;
17export type VpcConfiguration = finspace.VpcConfiguration;
18export type CapacityConfiguration = finspace.CapacityConfiguration;
19export type CodeConfiguration = finspace.CodeConfiguration;
20export type KxCommandLineArgument = finspace.KxCommandLineArgument;
21export type KxDatabaseConfiguration = finspace.KxDatabaseConfiguration;
22export type KxCacheStorageConfiguration = finspace.KxCacheStorageConfiguration;
23export type KxSavedownStorageConfiguration =
24 finspace.KxSavedownStorageConfiguration;
25export type KxScalingGroupConfiguration = finspace.KxScalingGroupConfiguration;
26export type TickerplantLogConfiguration = finspace.TickerplantLogConfiguration;
27
28/**
29 * Auto-scaling policy for a dedicated {@link KxCluster}.
30 */
31export interface AutoScalingConfiguration {
32 /** Lowest number of nodes to scale in to. */
33 minNodeCount?: number;
34 /** Highest number of nodes to scale out to. */
35 maxNodeCount?: number;
36 /** The metric the auto-scaling policy tracks. */
37 autoScalingMetric?: finspace.AutoScalingMetric;
38 /** The desired value of the chosen metric. */
39 metricTarget?: number;
40 /**
41 * The cooldown after a scale-in event before another scaling event —
42 * e.g. `"5 minutes"`. Sent to AWS as whole seconds.
43 */
44 scaleInCooldown?: Duration.Input;
45 /**
46 * The cooldown after a scale-out event before another scaling event —
47 * e.g. `"5 minutes"`. Sent to AWS as whole seconds.
48 */
49 scaleOutCooldown?: Duration.Input;
50}
51
52/** Convert the alchemy-facing auto-scaling config to the wire shape (seconds). */
53const toWireAutoScaling = (
54 config: AutoScalingConfiguration,
55): finspace.AutoScalingConfiguration => ({
56 minNodeCount: config.minNodeCount,
57 maxNodeCount: config.maxNodeCount,
58 autoScalingMetric: config.autoScalingMetric,
59 metricTarget: config.metricTarget,
60 scaleInCooldownSeconds: toWireSeconds(config.scaleInCooldown),
61 scaleOutCooldownSeconds: toWireSeconds(config.scaleOutCooldown),
62});
63
64export interface KxClusterProps {
65 /**
66 * Identifier of the kdb environment the cluster runs in. Changing it
67 * replaces the cluster.
68 */
69 environmentId: string;
70 /**
71 * Name of the kdb cluster. Changing it replaces the cluster.
72 * @default ${app}-${id}-${stage}-${suffix}
73 */
74 clusterName?: string;
75 /**
76 * The type of cluster — `HDB`, `RDB`, `GATEWAY`, `GP` or `TICKERPLANT`.
77 * Changing it replaces the cluster.
78 */
79 clusterType: KxClusterType;
80 /**
81 * The kdb release the cluster runs (e.g. `"1.0"`). Changing it replaces
82 * the cluster.
83 */
84 releaseLabel: string;
85 /**
86 * VPC the cluster network interfaces are placed in. Changing it replaces
87 * the cluster.
88 */
89 vpcConfiguration: VpcConfiguration;
90 /**
91 * Availability-zone placement mode — `SINGLE` or `MULTI`. Changing it
92 * replaces the cluster.
93 */
94 azMode: KxAzMode;
95 /**
96 * The availability zone id to place the cluster in (required when
97 * `azMode` is `SINGLE`). Changing it replaces the cluster.
98 */
99 availabilityZoneId?: string;
100 /**
101 * Dedicated node capacity for the cluster. Mutually exclusive with
102 * `scalingGroupConfiguration`. Changing it replaces the cluster.
103 */
104 capacityConfiguration?: CapacityConfiguration;
105 /**
106 * Placement on a shared scaling group instead of dedicated capacity.
107 * Changing it replaces the cluster.
108 */
109 scalingGroupConfiguration?: KxScalingGroupConfiguration;
110 /**
111 * Auto-scaling policy for a dedicated cluster. Changing it replaces the
112 * cluster.
113 */
114 autoScalingConfiguration?: AutoScalingConfiguration;
115 /**
116 * Temporary savedown storage for an `RDB` cluster. Changing it replaces
117 * the cluster.
118 */
119 savedownStorageConfiguration?: KxSavedownStorageConfiguration;
120 /**
121 * Databases to mount on the cluster, with optional cache configuration.
122 */
123 databases?: KxDatabaseConfiguration[];
124 /**
125 * Cache storage (e.g. `CACHE_1000`) sized for the mounted databases.
126 * Changing it replaces the cluster.
127 */
128 cacheStorageConfigurations?: KxCacheStorageConfiguration[];
129 /**
130 * Tickerplant log volumes for a `TICKERPLANT` cluster. Changing it
131 * replaces the cluster.
132 */
133 tickerplantLogConfiguration?: TickerplantLogConfiguration;
134 /**
135 * A description of the cluster.
136 */
137 description?: string;
138 /**
139 * The S3 location of the q code to deploy on the cluster.
140 */
141 code?: CodeConfiguration;
142 /**
143 * Path (relative to the code root) of the q script run at cluster start.
144 */
145 initializationScript?: string;
146 /**
147 * Command-line arguments passed to the kdb process.
148 */
149 commandLineArguments?: KxCommandLineArgument[];
150 /**
151 * ARN of the IAM execution role the cluster assumes to access AWS
152 * resources.
153 */
154 executionRole?: string;
155 /**
156 * Tags to associate with the cluster (applied at creation).
157 */
158 tags?: Record<string, string>;
159}
160
161export interface KxCluster extends Resource<
162 "AWS.FinSpace.KxCluster",
163 KxClusterProps,
164 {
165 /**
166 * Identifier of the kdb environment the cluster runs in.
167 */
168 environmentId: string;
169 /**
170 * The cluster's name.
171 */
172 clusterName: string;
173 /**
174 * Current lifecycle status of the cluster.
175 */
176 status: KxClusterStatus | undefined;
177 /**
178 * The cluster's type.
179 */
180 clusterType: KxClusterType | undefined;
181 /**
182 * The kdb release the cluster runs.
183 */
184 releaseLabel: string | undefined;
185 /**
186 * Availability-zone placement mode.
187 */
188 azMode: KxAzMode | undefined;
189 },
190 never,
191 Providers
192> {}
193
194/**
195 * A kdb cluster inside an Amazon FinSpace Managed kdb environment — the
196 * compute that mounts kdb databases and serves q queries.
197 *
198 * :::caution
199 * Cluster provisioning is slow (tens of minutes) and bills per node-hour
200 * while it exists. Live lifecycle tests are gated behind
201 * `AWS_TEST_FINSPACE=1`.
202 * :::
203 * ### Creating kdb Clusters
204 * **Example:** HDB Cluster on Dedicated Capacity
205 * ```typescript
206 * import * as AWS from "alchemy/AWS";
207 *
208 * const cluster = yield* AWS.FinSpace.KxCluster("Hdb", {
209 * environmentId: env.environmentId,
210 * clusterType: "HDB",
211 * releaseLabel: "1.0",
212 * azMode: "SINGLE",
213 * availabilityZoneId: "use1-az1",
214 * capacityConfiguration: { nodeType: "kx.s.large", nodeCount: 1 },
215 * vpcConfiguration: {
216 * vpcId: vpc.vpcId,
217 * securityGroupIds: [sg.securityGroupId],
218 * subnetIds: [subnet.subnetId],
219 * ipAddressType: "IP_V4",
220 * },
221 * databases: [{ databaseName: db.databaseName }],
222 * });
223 * ```
224 *
225 * @resource
226 */
227export const KxCluster = Resource<KxCluster>("AWS.FinSpace.KxCluster");
228
229const createClusterName = (
230 id: string,
231 props: { clusterName?: string | undefined },
232) =>
233 props.clusterName
234 ? Effect.succeed(props.clusterName)
235 : createPhysicalName({ id, maxLength: 63 });
236
237const isGone = (status: KxClusterStatus | undefined) =>
238 status === "DELETED" || status === "DELETING";
239
240const readCluster = Effect.fn(function* (
241 environmentId: string,
242 clusterName: string,
243) {
244 const response = yield* finspace
245 .getKxCluster({ environmentId, clusterName })
246 .pipe(
247 Effect.catchTag("ResourceNotFoundException", () =>
248 Effect.succeed(undefined),
249 ),
250 );
251 if (!response || isGone(response.status)) return undefined;
252 return response;
253});
254
255const toAttributes = (
256 cluster: finspace.GetKxClusterResponse,
257 environmentId: string,
258 clusterName: string,
259): KxCluster["Attributes"] => ({
260 environmentId,
261 clusterName: cluster.clusterName ?? clusterName,
262 status: cluster.status,
263 clusterType: cluster.clusterType,
264 releaseLabel: cluster.releaseLabel,
265 azMode: cluster.azMode,
266});
267
268/**
269 * A cluster still transitioning toward the awaited status — retried by the
270 * bounded schedule in {@link waitForClusterStatus}.
271 */
272class KxClusterNotReady extends Data.TaggedError("KxClusterNotReady")<{
273 readonly clusterName: string;
274 readonly status: string | undefined;
275}> {}
276
277/**
278 * A cluster whose asynchronous provisioning converged to a terminal failure
279 * status (`CREATE_FAILED` / `DELETE_FAILED`).
280 */
281export class KxClusterProvisioningFailed extends Data.TaggedError(
282 "KxClusterProvisioningFailed",
283)<{
284 readonly clusterName: string;
285 readonly status: string | undefined;
286 readonly statusReason: string | undefined;
287}> {}
288
289// Explicitly-typed retry wrapper — an inline `Effect.retry` in provider
290// lifecycle code leaks `Retry.Return`'s conditional type into declaration
291// emit and widens the provider layer to `unknown` for every consumer of
292// `AWS.providers()`.
293const retryWhileNotReady = <A, E extends { readonly _tag: string }, R>(
294 self: Effect.Effect<A, E, R>,
295): Effect.Effect<A, E, R> =>
296 Effect.retry(self, {
297 while: (e) => e._tag === "KxClusterNotReady",
298 // Cluster provisioning is slow (tens of minutes); poll every 20s up to
299 // ~40 min.
300 schedule: Schedule.max([
301 Schedule.spaced("20 seconds"),
302 Schedule.recurs(120),
303 ]),
304 });
305
306const waitForClusterStatus = (
307 environmentId: string,
308 clusterName: string,
309 target: "RUNNING" | "DELETED",
310) =>
311 retryWhileNotReady(
312 Effect.gen(function* () {
313 const response = yield* finspace
314 .getKxCluster({ environmentId, clusterName })
315 .pipe(
316 Effect.catchTag("ResourceNotFoundException", () =>
317 Effect.succeed(undefined),
318 ),
319 );
320 const status = response?.status;
321 if (target === "DELETED") {
322 if (response === undefined || status === "DELETED") return;
323 if (status === "DELETE_FAILED") {
324 return yield* Effect.fail(
325 new KxClusterProvisioningFailed({
326 clusterName,
327 status,
328 statusReason: response.statusReason,
329 }),
330 );
331 }
332 return yield* Effect.fail(
333 new KxClusterNotReady({ clusterName, status }),
334 );
335 }
336 if (status === "RUNNING") return;
337 if (status === "CREATE_FAILED") {
338 return yield* Effect.fail(
339 new KxClusterProvisioningFailed({
340 clusterName,
341 status,
342 statusReason: response?.statusReason,
343 }),
344 );
345 }
346 // response === undefined right after create is eventual consistency —
347 // keep polling on the bounded schedule.
348 return yield* Effect.fail(new KxClusterNotReady({ clusterName, status }));
349 }),
350 );
351
352const sameJson = (a: unknown, b: unknown) =>
353 JSON.stringify(a ?? null) === JSON.stringify(b ?? null);
354
355export const KxClusterProvider = () =>
356 Provider.effect(
357 KxCluster,
358 Effect.gen(function* () {
359 return {
360 stables: ["environmentId", "clusterName"],
361 // Clusters are keyed by their parent kdb environment — there is no
362 // account-wide enumeration without an environment id.
363 list: () => Effect.succeed([]),
364 read: Effect.fn(function* ({ id, olds, output }) {
365 const environmentId = output?.environmentId ?? olds?.environmentId;
366 if (environmentId === undefined) return undefined;
367 const clusterName =
368 output?.clusterName ?? (yield* createClusterName(id, olds ?? {}));
369 const cluster = yield* readCluster(environmentId, clusterName);
370 if (!cluster) return undefined;
371 // Clusters have no tag-read surface (no ARN in Get/List responses)
372 // — ownership cannot be verified, so a name match is owned.
373 return toAttributes(cluster, environmentId, clusterName);
374 }),
375 diff: Effect.fn(function* ({ id, news, olds }) {
376 if (!isResolved(news)) return;
377 if (olds === undefined) return;
378 const oldName = yield* createClusterName(id, olds);
379 const newName = yield* createClusterName(id, news);
380 // Everything except code/databases/description is fixed at
381 // creation.
382 if (
383 olds.environmentId !== news.environmentId ||
384 oldName !== newName ||
385 olds.clusterType !== news.clusterType ||
386 olds.releaseLabel !== news.releaseLabel ||
387 olds.azMode !== news.azMode ||
388 olds.availabilityZoneId !== news.availabilityZoneId ||
389 !sameJson(olds.vpcConfiguration, news.vpcConfiguration) ||
390 !sameJson(olds.capacityConfiguration, news.capacityConfiguration) ||
391 !sameJson(
392 olds.scalingGroupConfiguration,
393 news.scalingGroupConfiguration,
394 ) ||
395 !sameJson(
396 olds.autoScalingConfiguration &&
397 toWireAutoScaling(olds.autoScalingConfiguration),
398 news.autoScalingConfiguration &&
399 toWireAutoScaling(news.autoScalingConfiguration),
400 ) ||
401 !sameJson(
402 olds.savedownStorageConfiguration,
403 news.savedownStorageConfiguration,
404 ) ||
405 !sameJson(
406 olds.cacheStorageConfigurations,
407 news.cacheStorageConfigurations,
408 ) ||
409 !sameJson(
410 olds.tickerplantLogConfiguration,
411 news.tickerplantLogConfiguration,
412 ) ||
413 olds.executionRole !== news.executionRole
414 ) {
415 return { action: "replace" } as const;
416 }
417 }),
418 reconcile: Effect.fn(function* ({ id, news, output, session }) {
419 if (!news) {
420 return yield* Effect.fail(
421 new Error("FinSpace KxCluster requires props"),
422 );
423 }
424 const environmentId = news.environmentId;
425 const clusterName =
426 output?.clusterName ?? (yield* createClusterName(id, news));
427 const internalTags = yield* createInternalTags(id);
428 // One idempotency token per reconcile.
429 const clientToken = yield* Effect.sync(() => crypto.randomUUID());
430
431 // Observe — cloud state is authoritative.
432 let cluster = yield* readCluster(environmentId, clusterName);
433
434 // Ensure — create if missing (tolerating a Conflict race), then
435 // wait for RUNNING.
436 if (cluster === undefined) {
437 yield* finspace
438 .createKxCluster({
439 clientToken,
440 environmentId,
441 clusterName,
442 clusterType: news.clusterType,
443 releaseLabel: news.releaseLabel,
444 vpcConfiguration: news.vpcConfiguration,
445 azMode: news.azMode,
446 availabilityZoneId: news.availabilityZoneId,
447 capacityConfiguration: news.capacityConfiguration,
448 scalingGroupConfiguration: news.scalingGroupConfiguration,
449 autoScalingConfiguration:
450 news.autoScalingConfiguration &&
451 toWireAutoScaling(news.autoScalingConfiguration),
452 savedownStorageConfiguration: news.savedownStorageConfiguration,
453 databases: news.databases,
454 cacheStorageConfigurations: news.cacheStorageConfigurations,
455 tickerplantLogConfiguration: news.tickerplantLogConfiguration,
456 clusterDescription: news.description,
457 code: news.code,
458 initializationScript: news.initializationScript,
459 commandLineArguments: news.commandLineArguments,
460 executionRole: news.executionRole,
461 tags: { ...internalTags, ...news.tags },
462 })
463 .pipe(
464 Effect.catchTag("ConflictException", () =>
465 Effect.succeed(undefined),
466 ),
467 );
468 yield* session.note(`Creating kdb cluster ${clusterName}...`);
469 yield* waitForClusterStatus(environmentId, clusterName, "RUNNING");
470 cluster = yield* readCluster(environmentId, clusterName);
471 if (cluster === undefined) {
472 return yield* Effect.fail(
473 new Error(`failed to read created kdb cluster ${clusterName}`),
474 );
475 }
476 }
477
478 // Sync code configuration — only when the caller manages code and
479 // the observed deployment drifted.
480 if (
481 news.code !== undefined &&
482 (!sameJson(news.code, cluster.code) ||
483 (news.initializationScript ?? "") !==
484 (cluster.initializationScript ?? "") ||
485 !sameJson(
486 news.commandLineArguments,
487 cluster.commandLineArguments,
488 ))
489 ) {
490 yield* finspace.updateKxClusterCodeConfiguration({
491 environmentId,
492 clusterName,
493 clientToken,
494 code: news.code,
495 initializationScript: news.initializationScript,
496 commandLineArguments: news.commandLineArguments,
497 });
498 yield* session.note(`Updated code on kdb cluster ${clusterName}`);
499 yield* waitForClusterStatus(environmentId, clusterName, "RUNNING");
500 }
501
502 // Sync mounted databases — only when the caller manages them and
503 // the observed set drifted.
504 if (
505 news.databases !== undefined &&
506 !sameJson(news.databases, cluster.databases)
507 ) {
508 yield* finspace.updateKxClusterDatabases({
509 environmentId,
510 clusterName,
511 clientToken,
512 databases: news.databases,
513 });
514 yield* session.note(
515 `Updated databases on kdb cluster ${clusterName}`,
516 );
517 yield* waitForClusterStatus(environmentId, clusterName, "RUNNING");
518 }
519
520 yield* session.note(clusterName);
521
522 const final = yield* readCluster(environmentId, clusterName);
523 if (!final) {
524 return yield* Effect.fail(
525 new Error(`failed to read reconciled kdb cluster ${clusterName}`),
526 );
527 }
528 return toAttributes(final, environmentId, clusterName);
529 }),
530 delete: Effect.fn(function* ({ output }) {
531 yield* finspace
532 .deleteKxCluster({
533 environmentId: output.environmentId,
534 clusterName: output.clusterName,
535 })
536 .pipe(
537 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
538 );
539 yield* waitForClusterStatus(
540 output.environmentId,
541 output.clusterName,
542 "DELETED",
543 );
544 }),
545 };
546 }),
547 );