1import * as TSW from "@distilled.cloud/aws/timestream-write";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as EffectStream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import {
11 createInternalTags,
12 diffTags,
13 hasAlchemyTags,
14 type Tags,
15} from "../../Tags.ts";
16import { toWireDays, toWireHours } from "../../Util/Duration.ts";
17import { AWSEnvironment, type AccountID } from "../Environment.ts";
18import type { Providers } from "../Providers.ts";
19import type { RegionID } from "../Region.ts";
20import { withWriteEndpoint } from "./internal.ts";
21
22export type TableArn =
23 `arn:aws:timestream:${RegionID}:${AccountID}:database/${string}/table/${string}`;
24
25export type TableStatus = "ACTIVE" | "DELETING" | "RESTORING";
26
27export interface TableRetentionProperties {
28 /**
29 * How long data stays in the memory store before moving to the magnetic
30 * store. Rounded to whole hours on the wire
31 * (`MemoryStoreRetentionPeriodInHours`).
32 */
33 memoryStoreRetention: Duration.Input;
34 /**
35 * How long data stays in the magnetic store before deletion. Rounded to
36 * whole days on the wire (`MagneticStoreRetentionPeriodInDays`).
37 */
38 magneticStoreRetention: Duration.Input;
39}
40
41export interface TableProps {
42 /**
43 * Name of the database that owns this table. Pass the `databaseName` of a
44 * {@link Database} resource.
45 */
46 databaseName: string;
47 /**
48 * Name of the table. Must be unique within the database and between 3 and
49 * 256 characters.
50 * @default ${app}-${stage}-${id}
51 */
52 tableName?: string;
53 /**
54 * Retention configuration for the table's memory and magnetic stores.
55 * @default { memoryStoreRetention: "6 hours", magneticStoreRetention: "73000 days" }
56 */
57 retentionProperties?: TableRetentionProperties;
58 /**
59 * Magnetic store write configuration, including whether late-arriving data
60 * is written to the magnetic store and where rejected records are logged.
61 */
62 magneticStoreWriteProperties?: TSW.MagneticStoreWriteProperties;
63 /**
64 * Partitioning schema for the table.
65 */
66 schema?: TSW.Schema;
67 /**
68 * Tags to associate with the table.
69 */
70 tags?: Record<string, string>;
71}
72
73export interface Table extends Resource<
74 "AWS.Timestream.Table",
75 TableProps,
76 {
77 /**
78 * The table's physical name.
79 */
80 tableName: string;
81 /**
82 * Name of the database that owns the table.
83 */
84 databaseName: string;
85 /**
86 * ARN of the table.
87 */
88 tableArn: TableArn;
89 /**
90 * Current lifecycle status of the table.
91 */
92 tableStatus: TableStatus;
93 /**
94 * Effective retention configuration for the table.
95 */
96 retentionProperties: TSW.RetentionProperties | undefined;
97 /**
98 * Effective magnetic store write configuration.
99 */
100 magneticStoreWriteProperties: TSW.MagneticStoreWriteProperties | undefined;
101 /**
102 * Current tags reported for the table.
103 */
104 tags: Record<string, string>;
105 },
106 never,
107 Providers
108> {}
109
110/**
111 * An Amazon Timestream for LiveAnalytics table — a time-series store inside a
112 * {@link Database}.
113 *
114 * `Table` owns the table's lifecycle and its mutable configuration: memory and
115 * magnetic store retention, magnetic store write behavior, and tags. A table
116 * name is auto-generated from the app, stage, and logical ID unless you provide
117 * one.
118 *
119 * :::note
120 * Timestream for LiveAnalytics is closed to new AWS customers. Accounts that
121 * were not already onboarded receive `TimestreamNotOnboarded` on every
122 * operation.
123 * :::
124 * ### Creating Tables
125 * **Example:** Basic Table
126 * ```typescript
127 * import * as Timestream from "alchemy/AWS/Timestream";
128 *
129 * const database = yield* Timestream.Database("Metrics");
130 * const table = yield* Timestream.Table("Cpu", {
131 * databaseName: database.databaseName,
132 * });
133 * ```
134 *
135 * **Example:** Table with Retention Tuning
136 * ```typescript
137 * const table = yield* Timestream.Table("Cpu", {
138 * databaseName: database.databaseName,
139 * retentionProperties: {
140 * memoryStoreRetention: "24 hours",
141 * magneticStoreRetention: "365 days",
142 * },
143 * });
144 * ```
145 *
146 * ### Writing Points
147 * **Example:** Write records from a handler
148 * ```typescript
149 * // init
150 * const writeRecords = yield* Timestream.WriteRecords(table);
151 *
152 * return {
153 * fetch: Effect.gen(function* () {
154 * // runtime
155 * yield* writeRecords({
156 * Records: [
157 * {
158 * Dimensions: [{ Name: "host", Value: "web-1" }],
159 * MeasureName: "cpu",
160 * MeasureValue: "42.0",
161 * MeasureValueType: "DOUBLE",
162 * Time: `${Date.now()}`,
163 * TimeUnit: "MILLISECONDS",
164 * },
165 * ],
166 * });
167 * return HttpServerResponse.text("ok");
168 * }),
169 * };
170 * ```
171 *
172 * @resource
173 */
174export const Table = Resource<Table>("AWS.Timestream.Table");
175
176const createTableName = (
177 id: string,
178 props: { tableName?: string | undefined },
179) =>
180 Effect.gen(function* () {
181 if (props.tableName) {
182 return props.tableName;
183 }
184 return yield* createPhysicalName({ id, maxLength: 256 });
185 });
186
187const toTagRecord = (
188 tags: Array<{ Key: string; Value: string }> | undefined,
189): Record<string, string> =>
190 Object.fromEntries((tags ?? []).map((tag) => [tag.Key, tag.Value]));
191
192/**
193 * Convert the alchemy-shaped {@link TableRetentionProperties} (Duration
194 * inputs) to Timestream's wire shape (whole hours / whole days).
195 */
196const toWireRetention = (
197 retention: TableRetentionProperties | undefined,
198): TSW.RetentionProperties | undefined =>
199 retention === undefined
200 ? undefined
201 : {
202 MemoryStoreRetentionPeriodInHours: toWireHours(
203 retention.memoryStoreRetention,
204 )!,
205 MagneticStoreRetentionPeriodInDays: toWireDays(
206 retention.magneticStoreRetention,
207 )!,
208 };
209
210const readTable = Effect.fn(function* (
211 databaseName: string,
212 tableName: string,
213) {
214 const response = yield* withWriteEndpoint(
215 TSW.describeTable({ DatabaseName: databaseName, TableName: tableName }),
216 ).pipe(
217 Effect.catchTag("ResourceNotFoundException", () =>
218 Effect.succeed(undefined),
219 ),
220 );
221 if (!response?.Table) {
222 return undefined;
223 }
224 const table = response.Table;
225 const tagsResponse = yield* withWriteEndpoint(
226 TSW.listTagsForResource({ ResourceARN: table.Arn! }),
227 ).pipe(
228 Effect.catchTag("ResourceNotFoundException", () =>
229 Effect.succeed(undefined),
230 ),
231 );
232 if (!tagsResponse) {
233 return undefined;
234 }
235 return {
236 tableName: table.TableName!,
237 databaseName: table.DatabaseName!,
238 tableArn: table.Arn as TableArn,
239 tableStatus: (table.TableStatus ?? "ACTIVE") as TableStatus,
240 retentionProperties: table.RetentionProperties,
241 magneticStoreWriteProperties: table.MagneticStoreWriteProperties,
242 tags: toTagRecord(tagsResponse.Tags),
243 } satisfies Table["Attributes"];
244});
245
246export const TableProvider = () =>
247 Provider.effect(
248 Table,
249 Effect.gen(function* () {
250 return {
251 stables: ["tableName", "databaseName", "tableArn"],
252 // Enumerate every Timestream table in the ambient account/region.
253 // `listTables` with no `DatabaseName` lists across all databases; it is
254 // paginated, so collect every page and hydrate each into the `read`
255 // Attributes shape.
256 list: () =>
257 Effect.gen(function* () {
258 const refs = yield* withWriteEndpoint(
259 TSW.listTables.pages({}).pipe(EffectStream.runCollect),
260 ).pipe(
261 Effect.map((chunk) =>
262 Array.from(chunk).flatMap((page) =>
263 (page.Tables ?? []).flatMap((table) =>
264 table.DatabaseName && table.TableName
265 ? [
266 {
267 databaseName: table.DatabaseName,
268 tableName: table.TableName,
269 },
270 ]
271 : [],
272 ),
273 ),
274 ),
275 );
276 const hydrated = yield* Effect.forEach(
277 refs,
278 (ref) => readTable(ref.databaseName, ref.tableName),
279 { concurrency: 10 },
280 );
281 return hydrated.filter(
282 (attrs): attrs is Table["Attributes"] => attrs !== undefined,
283 );
284 }),
285 read: Effect.fn(function* ({ id, olds, output }) {
286 const databaseName = output?.databaseName ?? olds?.databaseName;
287 if (!databaseName) return undefined;
288 const tableName =
289 output?.tableName ?? (yield* createTableName(id, olds ?? {}));
290 const state = yield* readTable(databaseName, tableName);
291 if (!state) return undefined;
292 return (yield* hasAlchemyTags(id, state.tags as Tags))
293 ? state
294 : Unowned(state);
295 }),
296 diff: Effect.fn(function* ({ id, news, olds = {} }) {
297 if (!isResolved(news)) return;
298 const oldName = yield* createTableName(id, olds);
299 const newName = yield* createTableName(id, news);
300 // Table name or its owning database is immutable — either requires a
301 // replacement.
302 if (oldName !== newName || olds.databaseName !== news.databaseName) {
303 return { action: "replace" } as const;
304 }
305 }),
306 reconcile: Effect.fn(function* ({ id, news, output, session }) {
307 if (!news?.databaseName) {
308 return yield* Effect.fail(
309 new Error("Timestream Table requires a databaseName"),
310 );
311 }
312 const { accountId, region } = yield* AWSEnvironment.current;
313 const databaseName = news.databaseName;
314 const tableName =
315 output?.tableName ?? (yield* createTableName(id, news));
316 const tableArn =
317 `arn:aws:timestream:${region}:${accountId}:database/${databaseName}/table/${tableName}` as TableArn;
318 const internalTags = yield* createInternalTags(id);
319 const desiredTags = { ...internalTags, ...news.tags };
320 const desiredRetention = toWireRetention(news.retentionProperties);
321
322 // Observe.
323 let state = yield* readTable(databaseName, tableName);
324
325 // Ensure — create if missing; tolerate a ConflictException race.
326 if (state === undefined) {
327 yield* withWriteEndpoint(
328 TSW.createTable({
329 DatabaseName: databaseName,
330 TableName: tableName,
331 RetentionProperties: desiredRetention,
332 MagneticStoreWriteProperties: news.magneticStoreWriteProperties,
333 Schema: news.schema,
334 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
335 Key,
336 Value,
337 })),
338 }),
339 ).pipe(Effect.catchTag("ConflictException", () => Effect.void));
340 yield* session.note(`Creating table ${tableName}...`);
341 state = yield* readTable(databaseName, tableName);
342 if (state === undefined) {
343 return yield* Effect.fail(
344 new Error(`failed to read created table ${tableName}`),
345 );
346 }
347 }
348
349 // Sync retention / magnetic-store-write config — diff the desired
350 // wire values against observed cloud state and skip the API call
351 // entirely on a no-op.
352 const retentionDrifted =
353 desiredRetention !== undefined &&
354 (state.retentionProperties?.MemoryStoreRetentionPeriodInHours !==
355 desiredRetention.MemoryStoreRetentionPeriodInHours ||
356 state.retentionProperties?.MagneticStoreRetentionPeriodInDays !==
357 desiredRetention.MagneticStoreRetentionPeriodInDays);
358 if (
359 retentionDrifted ||
360 news.magneticStoreWriteProperties !== undefined
361 ) {
362 yield* withWriteEndpoint(
363 TSW.updateTable({
364 DatabaseName: databaseName,
365 TableName: tableName,
366 RetentionProperties: desiredRetention,
367 MagneticStoreWriteProperties: news.magneticStoreWriteProperties,
368 Schema: news.schema,
369 }),
370 );
371 yield* session.note(`Updated table ${tableName} configuration`);
372 }
373
374 // Sync tags — diff against observed cloud tags.
375 const { removed, upsert } = diffTags(state.tags, desiredTags);
376 if (removed.length > 0) {
377 yield* withWriteEndpoint(
378 TSW.untagResource({ ResourceARN: tableArn, TagKeys: removed }),
379 );
380 }
381 if (upsert.length > 0) {
382 yield* withWriteEndpoint(
383 TSW.tagResource({
384 ResourceARN: tableArn,
385 Tags: upsert.map(({ Key, Value }) => ({ Key, Value })),
386 }),
387 );
388 }
389
390 yield* session.note(tableArn);
391
392 const final = yield* readTable(databaseName, tableName);
393 if (!final) {
394 return yield* Effect.fail(
395 new Error(`failed to read reconciled table ${tableName}`),
396 );
397 }
398 return final;
399 }),
400 delete: Effect.fn(function* ({ output }) {
401 yield* withWriteEndpoint(
402 TSW.deleteTable({
403 DatabaseName: output.databaseName,
404 TableName: output.tableName,
405 }),
406 ).pipe(
407 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
408 );
409 }),
410 };
411 }),
412 );