1import * as TSW from "@distilled.cloud/aws/timestream-write";
2import * as Effect from "effect/Effect";
3import * as EffectStream from "effect/Stream";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import {
10 createInternalTags,
11 diffTags,
12 hasAlchemyTags,
13 type Tags,
14} from "../../Tags.ts";
15import { AWSEnvironment, type AccountID } from "../Environment.ts";
16import type { Providers } from "../Providers.ts";
17import type { RegionID } from "../Region.ts";
18import { withWriteEndpoint } from "./internal.ts";
19
20export type DatabaseArn =
21 `arn:aws:timestream:${RegionID}:${AccountID}:database/${string}`;
22
23export interface DatabaseProps {
24 /**
25 * Name of the Timestream database. Must be unique within the account and
26 * region and between 3 and 256 characters.
27 * @default ${app}-${stage}-${id}
28 */
29 databaseName?: string;
30 /**
31 * The ID (or ARN/alias) of the KMS key used to encrypt data at rest. When
32 * omitted, Timestream provisions and manages an AWS-owned KMS key.
33 */
34 kmsKeyId?: string;
35 /**
36 * Tags to associate with the database.
37 */
38 tags?: Record<string, string>;
39}
40
41export interface Database extends Resource<
42 "AWS.Timestream.Database",
43 DatabaseProps,
44 {
45 /**
46 * The database's physical name.
47 */
48 databaseName: string;
49 /**
50 * ARN of the database.
51 */
52 databaseArn: DatabaseArn;
53 /**
54 * The KMS key backing encryption at rest.
55 */
56 kmsKeyId: string | undefined;
57 /**
58 * Number of tables in the database.
59 */
60 tableCount: number | undefined;
61 /**
62 * Current tags reported for the database.
63 */
64 tags: Record<string, string>;
65 },
66 never,
67 Providers
68> {}
69
70/**
71 * An Amazon Timestream for LiveAnalytics database — the top-level container for
72 * time-series {@link Table}s.
73 *
74 * `Database` owns the database's lifecycle and its mutable configuration: the
75 * KMS key used for encryption at rest and its tags. A database name is
76 * auto-generated from the app, stage, and logical ID unless you provide one.
77 *
78 * :::note
79 * Timestream for LiveAnalytics is closed to new AWS customers. Accounts that
80 * were not already onboarded receive `TimestreamNotOnboarded` (a specialized
81 * `AccessDenied`) on every operation.
82 * :::
83 * ### Creating Databases
84 * **Example:** Basic Database
85 * ```typescript
86 * import * as Timestream from "alchemy/AWS/Timestream";
87 *
88 * const database = yield* Timestream.Database("Metrics");
89 * ```
90 *
91 * **Example:** Database with a Customer-Managed KMS Key
92 * ```typescript
93 * const database = yield* Timestream.Database("SecureMetrics", {
94 * kmsKeyId: "alias/my-timestream-key",
95 * tags: { Environment: "production" },
96 * });
97 * ```
98 *
99 * @resource
100 */
101export const Database = Resource<Database>("AWS.Timestream.Database");
102
103const createDatabaseName = (
104 id: string,
105 props: { databaseName?: string | undefined },
106) =>
107 Effect.gen(function* () {
108 if (props.databaseName) {
109 return props.databaseName;
110 }
111 return yield* createPhysicalName({ id, maxLength: 256 });
112 });
113
114const toTagRecord = (
115 tags: Array<{ Key: string; Value: string }> | undefined,
116): Record<string, string> =>
117 Object.fromEntries((tags ?? []).map((tag) => [tag.Key, tag.Value]));
118
119const readDatabase = Effect.fn(function* (databaseName: string) {
120 const response = yield* withWriteEndpoint(
121 TSW.describeDatabase({ DatabaseName: databaseName }),
122 ).pipe(
123 Effect.catchTag("ResourceNotFoundException", () =>
124 Effect.succeed(undefined),
125 ),
126 );
127 if (!response?.Database) {
128 return undefined;
129 }
130 const database = response.Database;
131 const tagsResponse = yield* withWriteEndpoint(
132 TSW.listTagsForResource({ ResourceARN: database.Arn! }),
133 ).pipe(
134 Effect.catchTag("ResourceNotFoundException", () =>
135 Effect.succeed(undefined),
136 ),
137 );
138 if (!tagsResponse) {
139 return undefined;
140 }
141 return {
142 databaseName: database.DatabaseName!,
143 databaseArn: database.Arn as DatabaseArn,
144 kmsKeyId: database.KmsKeyId,
145 tableCount: database.TableCount,
146 tags: toTagRecord(tagsResponse.Tags),
147 } satisfies Database["Attributes"];
148});
149
150export const DatabaseProvider = () =>
151 Provider.effect(
152 Database,
153 Effect.gen(function* () {
154 return {
155 stables: ["databaseName", "databaseArn"],
156 // Enumerate every Timestream database in the ambient account/region.
157 // `listDatabases` is paginated; collect every page and hydrate each
158 // into the exact `read` Attributes shape. A database that vanishes
159 // between listing and hydration is dropped (typed NotFound handled in
160 // `readDatabase`).
161 list: () =>
162 Effect.gen(function* () {
163 const names = yield* withWriteEndpoint(
164 TSW.listDatabases.pages({}).pipe(EffectStream.runCollect),
165 ).pipe(
166 Effect.map((chunk) =>
167 Array.from(chunk).flatMap((page) =>
168 (page.Databases ?? []).flatMap((db) =>
169 db.DatabaseName ? [db.DatabaseName] : [],
170 ),
171 ),
172 ),
173 );
174 const hydrated = yield* Effect.forEach(
175 names,
176 (name) => readDatabase(name),
177 { concurrency: 10 },
178 );
179 return hydrated.filter(
180 (attrs): attrs is Database["Attributes"] => attrs !== undefined,
181 );
182 }),
183 read: Effect.fn(function* ({ id, olds, output }) {
184 const databaseName =
185 output?.databaseName ?? (yield* createDatabaseName(id, olds ?? {}));
186 const state = yield* readDatabase(databaseName);
187 if (!state) return undefined;
188 return (yield* hasAlchemyTags(id, state.tags as Tags))
189 ? state
190 : Unowned(state);
191 }),
192 diff: Effect.fn(function* ({ id, news = {}, olds = {} }) {
193 if (!isResolved(news)) return;
194 const oldName = yield* createDatabaseName(id, olds);
195 const newName = yield* createDatabaseName(id, news);
196 if (oldName !== newName) {
197 return { action: "replace" } as const;
198 }
199 }),
200 reconcile: Effect.fn(function* ({ id, news = {}, output, session }) {
201 const { accountId, region } = yield* AWSEnvironment.current;
202 const databaseName =
203 output?.databaseName ?? (yield* createDatabaseName(id, news));
204 const databaseArn =
205 `arn:aws:timestream:${region}:${accountId}:database/${databaseName}` as DatabaseArn;
206 const internalTags = yield* createInternalTags(id);
207 const desiredTags = { ...internalTags, ...news.tags };
208
209 // Observe — cloud state is authoritative; output is only a name cache.
210 let state = yield* readDatabase(databaseName);
211
212 // Ensure — create if missing. Tolerate a ConflictException as a race
213 // with a peer reconciler and fall through to sync.
214 if (state === undefined) {
215 yield* withWriteEndpoint(
216 TSW.createDatabase({
217 DatabaseName: databaseName,
218 KmsKeyId: news.kmsKeyId,
219 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
220 Key,
221 Value,
222 })),
223 }),
224 ).pipe(Effect.catchTag("ConflictException", () => Effect.void));
225 yield* session.note(`Creating database ${databaseName}...`);
226 state = yield* readDatabase(databaseName);
227 if (state === undefined) {
228 return yield* Effect.fail(
229 new Error(`failed to read created database ${databaseName}`),
230 );
231 }
232 }
233
234 // Sync KMS key — only when the user pinned one and it drifted.
235 if (news.kmsKeyId !== undefined && state.kmsKeyId !== news.kmsKeyId) {
236 yield* withWriteEndpoint(
237 TSW.updateDatabase({
238 DatabaseName: databaseName,
239 KmsKeyId: news.kmsKeyId,
240 }),
241 );
242 yield* session.note(`Updated KMS key for ${databaseName}`);
243 }
244
245 // Sync tags — diff against observed cloud tags.
246 const { removed, upsert } = diffTags(state.tags, desiredTags);
247 if (removed.length > 0) {
248 yield* withWriteEndpoint(
249 TSW.untagResource({
250 ResourceARN: databaseArn,
251 TagKeys: removed,
252 }),
253 );
254 }
255 if (upsert.length > 0) {
256 yield* withWriteEndpoint(
257 TSW.tagResource({
258 ResourceARN: databaseArn,
259 Tags: upsert.map(({ Key, Value }) => ({ Key, Value })),
260 }),
261 );
262 }
263
264 yield* session.note(databaseArn);
265
266 const final = yield* readDatabase(databaseName);
267 if (!final) {
268 return yield* Effect.fail(
269 new Error(`failed to read reconciled database ${databaseName}`),
270 );
271 }
272 return final;
273 }),
274 delete: Effect.fn(function* ({ output }) {
275 yield* withWriteEndpoint(
276 TSW.deleteDatabase({ DatabaseName: output.databaseName }),
277 ).pipe(
278 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
279 );
280 }),
281 };
282 }),
283 );