1import * as influxdb from "@distilled.cloud/aws/timestream-influxdb";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Redacted from "effect/Redacted";
5import * as EffectStream from "effect/Stream";
6import * as Schedule from "effect/Schedule";
7import { Unowned } from "../../AdoptPolicy.ts";
8import { isResolved } from "../../Diff.ts";
9import { createPhysicalName } from "../../PhysicalName.ts";
10import * as Provider from "../../Provider.ts";
11import { Resource } from "../../Resource.ts";
12import {
13 createInternalTags,
14 diffTags,
15 hasAlchemyTags,
16 type Tags,
17} from "../../Tags.ts";
18import type { Providers } from "../Providers.ts";
19
20export type DbInstanceStatus = influxdb.Status;
21
22export interface DbInstanceProps {
23 /**
24 * Name that uniquely identifies the DB instance. Must be 3–40 characters,
25 * start with a letter, and contain only lowercase letters, numbers, and
26 * hyphens.
27 * @default ${app}-${stage}-${id}
28 */
29 name?: string;
30 /**
31 * The compute instance type to run InfluxDB on (e.g. `db.influx.medium`).
32 */
33 dbInstanceType: influxdb.DbInstanceType;
34 /**
35 * The amount of storage to allocate, in GiB.
36 */
37 allocatedStorage: number;
38 /**
39 * The IDs of the VPC subnets the DB instance is attached to.
40 */
41 vpcSubnetIds: string[];
42 /**
43 * The VPC security group IDs to associate with the DB instance.
44 */
45 vpcSecurityGroupIds: string[];
46 /**
47 * The initial InfluxDB admin password. Stored in AWS Secrets Manager by
48 * Timestream; supplied only at creation. Pass a redacted value, e.g.
49 * `Redacted.make("super-secret-password")`.
50 */
51 password: Redacted.Redacted<string>;
52 /**
53 * The initial InfluxDB admin username.
54 * @default "admin"
55 */
56 username?: string;
57 /**
58 * The name of the initial InfluxDB organization.
59 */
60 organization?: string;
61 /**
62 * The name of the initial InfluxDB bucket.
63 */
64 bucket?: string;
65 /**
66 * The storage tier (IOPS profile) for the DB instance.
67 */
68 dbStorageType?: influxdb.DbStorageType;
69 /**
70 * Whether the DB instance has a public IP and is reachable from the internet.
71 * @default false
72 */
73 publiclyAccessible?: boolean;
74 /**
75 * Single- or multi-AZ deployment.
76 * @default "SINGLE_AZ"
77 */
78 deploymentType?: influxdb.DeploymentType;
79 /**
80 * The identifier of an InfluxDB parameter group to associate.
81 */
82 dbParameterGroupIdentifier?: string;
83 /**
84 * Log delivery (S3) configuration for InfluxDB engine logs.
85 */
86 logDeliveryConfiguration?: influxdb.LogDeliveryConfiguration;
87 /**
88 * The port InfluxDB listens on.
89 * @default 8086
90 */
91 port?: number;
92 /**
93 * The network protocol (IPv4 or dual-stack).
94 */
95 networkType?: influxdb.NetworkType;
96 /**
97 * Tags to associate with the DB instance.
98 */
99 tags?: Record<string, string>;
100}
101
102export interface DbInstance extends Resource<
103 "AWS.Timestream.DbInstance",
104 DbInstanceProps,
105 {
106 /**
107 * Service-assigned unique identifier for the DB instance.
108 */
109 id: string;
110 /**
111 * The DB instance's name.
112 */
113 name: string;
114 /**
115 * ARN of the DB instance.
116 */
117 arn: string;
118 /**
119 * Current lifecycle status of the DB instance.
120 */
121 status: DbInstanceStatus | undefined;
122 /**
123 * The connection endpoint (host) for the DB instance.
124 */
125 endpoint: string | undefined;
126 /**
127 * The port InfluxDB is listening on.
128 */
129 port: number | undefined;
130 /**
131 * ARN of the Secrets Manager secret holding the InfluxDB auth parameters.
132 */
133 influxAuthParametersSecretArn: string | undefined;
134 /**
135 * The compute instance type backing the DB instance.
136 */
137 dbInstanceType: influxdb.DbInstanceType | undefined;
138 /**
139 * The allocated storage, in GiB.
140 */
141 allocatedStorage: number | undefined;
142 /**
143 * The subnets the DB instance is attached to.
144 */
145 vpcSubnetIds: string[];
146 /**
147 * Current tags reported for the DB instance.
148 */
149 tags: Record<string, string>;
150 },
151 never,
152 Providers
153> {}
154
155/**
156 * An Amazon Timestream for InfluxDB DB instance — a managed, single-tenant
157 * InfluxDB engine for time-series workloads.
158 *
159 * `DbInstance` owns the instance's lifecycle and its mutable configuration
160 * (instance type, storage, port, parameter group, log delivery, deployment
161 * type, and tags). Networking (subnets, security groups), the initial
162 * credentials, organization, and bucket are fixed at creation.
163 *
164 * :::caution
165 * Provisioning a DB instance takes ~15–20 minutes and incurs EC2-backed cost.
166 * :::
167 * ### Creating DB Instances
168 * **Example:** Basic InfluxDB Instance
169 * ```typescript
170 * import * as Timestream from "alchemy/AWS/Timestream";
171 *
172 * const influx = yield* Timestream.DbInstance("Influx", {
173 * name: "my-influx",
174 * dbInstanceType: "db.influx.medium",
175 * allocatedStorage: 20,
176 * vpcSubnetIds: [subnetA.subnetId, subnetB.subnetId],
177 * vpcSecurityGroupIds: [securityGroup.groupId],
178 * password: Redacted.make("super-secret-password"),
179 * });
180 * ```
181 *
182 * @resource
183 */
184export const DbInstance = Resource<DbInstance>("AWS.Timestream.DbInstance");
185
186const createInstanceName = (id: string, props: { name?: string | undefined }) =>
187 Effect.gen(function* () {
188 if (props.name) {
189 return props.name;
190 }
191 return yield* createPhysicalName({ id, maxLength: 40 });
192 });
193
194const toAttrs = (
195 instance: influxdb.GetDbInstanceOutput,
196 tags: Record<string, string>,
197): DbInstance["Attributes"] => ({
198 id: instance.id,
199 name: instance.name,
200 arn: instance.arn,
201 status: instance.status,
202 endpoint: instance.endpoint,
203 port: instance.port,
204 influxAuthParametersSecretArn: instance.influxAuthParametersSecretArn,
205 dbInstanceType: instance.dbInstanceType,
206 allocatedStorage: instance.allocatedStorage,
207 vpcSubnetIds: instance.vpcSubnetIds,
208 tags,
209});
210
211const fetchTags = Effect.fn(function* (arn: string) {
212 const response = yield* influxdb
213 .listTagsForResource({ resourceArn: arn })
214 .pipe(
215 Effect.catchTag("ResourceNotFoundException", () =>
216 Effect.succeed(undefined),
217 ),
218 );
219 const tags = response?.tags ?? {};
220 return Object.fromEntries(
221 Object.entries(tags).filter(
222 (entry): entry is [string, string] => typeof entry[1] === "string",
223 ),
224 );
225});
226
227const readInstanceById = Effect.fn(function* (identifier: string) {
228 const instance = yield* influxdb
229 .getDbInstance({ identifier })
230 .pipe(
231 Effect.catchTag("ResourceNotFoundException", () =>
232 Effect.succeed(undefined),
233 ),
234 );
235 if (!instance) return undefined;
236 return toAttrs(instance, yield* fetchTags(instance.arn));
237});
238
239const findInstanceByName = Effect.fn(function* (name: string) {
240 const summaries = yield* influxdb.listDbInstances.pages({}).pipe(
241 EffectStream.runCollect,
242 Effect.map((chunk) =>
243 Array.from(chunk).flatMap((page) => page.items ?? []),
244 ),
245 );
246 const match = summaries.find((summary) => summary.name === name);
247 if (!match) return undefined;
248 return yield* readInstanceById(match.id);
249});
250
251/**
252 * A DB instance still transitioning toward the awaited status — retried by
253 * {@link waitForStatus}'s bounded schedule.
254 */
255class DbInstanceNotReady extends Data.TaggedError("DbInstanceNotReady")<{
256 readonly identifier: string;
257 readonly status: string | undefined;
258}> {}
259
260/**
261 * A DB instance whose asynchronous provisioning converged to a terminal
262 * failure status (`FAILED` / `REBOOT_FAILED`).
263 */
264export class DbInstanceProvisioningFailed extends Data.TaggedError(
265 "DbInstanceProvisioningFailed",
266)<{
267 readonly identifier: string;
268 readonly status: string;
269}> {}
270
271// Explicitly-typed retry wrapper — an inline `Effect.retry` in provider
272// lifecycle code leaks `Retry.Return`'s conditional type into declaration
273// emit and widens the provider layer to `unknown` for every consumer of
274// `AWS.providers()`.
275const retryWhileNotReady = <A, E extends { readonly _tag: string }, R>(
276 self: Effect.Effect<A, E, R>,
277): Effect.Effect<A, E, R> =>
278 Effect.retry(self, {
279 // ParseError: a CREATING instance can transiently decode incompletely
280 // while fields are still being populated — treat it as not-ready.
281 while: (e) => e._tag === "DbInstanceNotReady" || e._tag === "ParseError",
282 // Provisioning is slow (~15–20 min); poll every 20s up to ~30 min.
283 schedule: Schedule.max([
284 Schedule.spaced("20 seconds"),
285 Schedule.recurs(90),
286 ]),
287 });
288
289const waitForStatus = (identifier: string, target: "AVAILABLE" | "DELETED") =>
290 retryWhileNotReady(
291 Effect.gen(function* () {
292 const instance = yield* influxdb
293 .getDbInstance({ identifier })
294 .pipe(
295 Effect.catchTag("ResourceNotFoundException", () =>
296 Effect.succeed(undefined),
297 ),
298 );
299 if (target === "DELETED") {
300 if (instance === undefined) return;
301 return yield* Effect.fail(
302 new DbInstanceNotReady({ identifier, status: instance.status }),
303 );
304 }
305 if (instance?.status === "AVAILABLE") return;
306 if (
307 instance?.status === "FAILED" ||
308 instance?.status === "REBOOT_FAILED"
309 ) {
310 return yield* Effect.fail(
311 new DbInstanceProvisioningFailed({
312 identifier,
313 status: instance.status,
314 }),
315 );
316 }
317 return yield* Effect.fail(
318 new DbInstanceNotReady({ identifier, status: instance?.status }),
319 );
320 }),
321 );
322
323export const DbInstanceProvider = () =>
324 Provider.effect(
325 DbInstance,
326 Effect.gen(function* () {
327 return {
328 stables: ["id", "arn", "name"],
329 list: () =>
330 Effect.gen(function* () {
331 const summaries = yield* influxdb.listDbInstances.pages({}).pipe(
332 EffectStream.runCollect,
333 Effect.map((chunk) =>
334 Array.from(chunk).flatMap((page) => page.items ?? []),
335 ),
336 );
337 const hydrated = yield* Effect.forEach(
338 summaries,
339 (summary) => readInstanceById(summary.id),
340 { concurrency: 5 },
341 );
342 return hydrated.filter(
343 (attrs): attrs is DbInstance["Attributes"] => attrs !== undefined,
344 );
345 }),
346 read: Effect.fn(function* ({ id, olds, output }) {
347 const state = output?.id
348 ? yield* readInstanceById(output.id)
349 : yield* findInstanceByName(
350 yield* createInstanceName(id, olds ?? {}),
351 );
352 if (!state) return undefined;
353 return (yield* hasAlchemyTags(id, state.tags as Tags))
354 ? state
355 : Unowned(state);
356 }),
357 diff: Effect.fn(function* ({ id, news, olds = {} }) {
358 if (!isResolved(news)) return;
359 const oldName = yield* createInstanceName(id, olds);
360 const newName = yield* createInstanceName(id, news);
361 // Name and the networking/credential aspects are immutable — changing
362 // any of them requires a replacement.
363 const arraysDiffer = (a?: string[], b?: string[]) =>
364 JSON.stringify([...(a ?? [])].sort()) !==
365 JSON.stringify([...(b ?? [])].sort());
366 if (
367 oldName !== newName ||
368 arraysDiffer(olds.vpcSubnetIds, news.vpcSubnetIds) ||
369 arraysDiffer(olds.vpcSecurityGroupIds, news.vpcSecurityGroupIds) ||
370 olds.username !== news.username ||
371 olds.organization !== news.organization ||
372 olds.bucket !== news.bucket ||
373 olds.publiclyAccessible !== news.publiclyAccessible ||
374 olds.networkType !== news.networkType
375 ) {
376 return { action: "replace" } as const;
377 }
378 }),
379 reconcile: Effect.fn(function* ({ id, news, output, session }) {
380 if (!news) {
381 return yield* Effect.fail(
382 new Error("Timestream DbInstance requires props"),
383 );
384 }
385 const name = output?.name ?? (yield* createInstanceName(id, news));
386 const internalTags = yield* createInternalTags(id);
387 const desiredTags = { ...internalTags, ...news.tags };
388
389 // Observe — prefer the cached id; fall back to a name lookup so a
390 // create whose state failed to persist is adopted rather than
391 // duplicated.
392 let state = output?.id
393 ? yield* readInstanceById(output.id)
394 : yield* findInstanceByName(name);
395
396 // Ensure — create if missing, then wait for AVAILABLE.
397 if (state === undefined) {
398 const created = yield* influxdb.createDbInstance({
399 name,
400 dbInstanceType: news.dbInstanceType,
401 allocatedStorage: news.allocatedStorage,
402 vpcSubnetIds: news.vpcSubnetIds,
403 vpcSecurityGroupIds: news.vpcSecurityGroupIds,
404 password: Redacted.value(news.password),
405 username: news.username,
406 organization: news.organization,
407 bucket: news.bucket,
408 dbStorageType: news.dbStorageType,
409 publiclyAccessible: news.publiclyAccessible,
410 deploymentType: news.deploymentType,
411 dbParameterGroupIdentifier: news.dbParameterGroupIdentifier,
412 logDeliveryConfiguration: news.logDeliveryConfiguration,
413 port: news.port,
414 networkType: news.networkType,
415 tags: desiredTags,
416 });
417 yield* session.note(
418 `Creating DB instance ${name} (${created.id})...`,
419 );
420 yield* waitForStatus(created.id, "AVAILABLE");
421 state = yield* readInstanceById(created.id);
422 if (state === undefined) {
423 return yield* Effect.fail(
424 new Error(`failed to read created DB instance ${name}`),
425 );
426 }
427 }
428
429 // Sync mutable settings via UpdateDbInstance. It is a partial upsert;
430 // only send it when a mutable aspect drifted from observed state.
431 const needsUpdate =
432 (news.dbInstanceType !== undefined &&
433 news.dbInstanceType !== state.dbInstanceType) ||
434 (news.allocatedStorage !== undefined &&
435 news.allocatedStorage !== state.allocatedStorage) ||
436 (news.port !== undefined && news.port !== state.port) ||
437 news.dbParameterGroupIdentifier !== undefined ||
438 news.logDeliveryConfiguration !== undefined ||
439 news.deploymentType !== undefined ||
440 news.dbStorageType !== undefined;
441 if (needsUpdate) {
442 yield* influxdb.updateDbInstance({
443 identifier: state.id,
444 dbInstanceType: news.dbInstanceType,
445 allocatedStorage: news.allocatedStorage,
446 port: news.port,
447 dbParameterGroupIdentifier: news.dbParameterGroupIdentifier,
448 logDeliveryConfiguration: news.logDeliveryConfiguration,
449 deploymentType: news.deploymentType,
450 dbStorageType: news.dbStorageType,
451 });
452 yield* waitForStatus(state.id, "AVAILABLE");
453 yield* session.note(`Updated DB instance ${name}`);
454 }
455
456 // Sync tags — diff against observed cloud tags.
457 const { removed, upsert } = diffTags(state.tags, desiredTags);
458 if (removed.length > 0) {
459 yield* influxdb.untagResource({
460 resourceArn: state.arn,
461 tagKeys: removed,
462 });
463 }
464 if (upsert.length > 0) {
465 yield* influxdb.tagResource({
466 resourceArn: state.arn,
467 tags: Object.fromEntries(
468 upsert.map(({ Key, Value }) => [Key, Value]),
469 ),
470 });
471 }
472
473 yield* session.note(state.arn);
474
475 const final = yield* readInstanceById(state.id);
476 if (!final) {
477 return yield* Effect.fail(
478 new Error(`failed to read reconciled DB instance ${name}`),
479 );
480 }
481 return final;
482 }),
483 delete: Effect.fn(function* ({ output }) {
484 yield* influxdb
485 .deleteDbInstance({ identifier: output.id })
486 .pipe(
487 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
488 );
489 yield* waitForStatus(output.id, "DELETED");
490 }),
491 };
492 }),
493 );