1import * as TSQ from "@distilled.cloud/aws/timestream-query";
2import * as Effect from "effect/Effect";
3import * as Option from "effect/Option";
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 type { Providers } from "../Providers.ts";
17import { withQueryEndpoint } from "./internal.ts";
18
19export type ScheduledQueryState = TSQ.ScheduledQueryState;
20
21export interface ScheduledQueryErrorReportS3 {
22 /**
23 * Name of the S3 bucket where error reports for failed runs are written.
24 */
25 bucketName: string;
26 /**
27 * Object key prefix for error report objects.
28 */
29 objectKeyPrefix?: string;
30 /**
31 * Server-side encryption for error report objects.
32 * @default "SSE_S3"
33 */
34 encryptionOption?: TSQ.S3EncryptionOption;
35}
36
37export interface ScheduledQueryProps {
38 /**
39 * Name of the scheduled query. Must be unique within the account and
40 * region.
41 * @default ${app}-${stage}-${id}
42 */
43 name?: string;
44 /**
45 * The SQL the scheduled query runs on each invocation.
46 */
47 queryString: string;
48 /**
49 * When to run the query — a cron (`cron(0 12 * * ? *)`) or rate
50 * (`rate(1 hour)`) expression.
51 */
52 scheduleExpression: string;
53 /**
54 * ARN of the SNS topic Timestream notifies after each run.
55 */
56 notificationTopicArn: string;
57 /**
58 * ARN of the IAM role Timestream assumes to run the query, write results,
59 * publish notifications, and write error reports.
60 */
61 executionRoleArn: string;
62 /**
63 * Where results are written (a Timestream table with measure/dimension
64 * mappings). Omit for queries whose results are not materialized.
65 */
66 targetConfiguration?: TSQ.TargetConfiguration;
67 /**
68 * S3 location where error reports for failed runs are written.
69 */
70 errorReportS3: ScheduledQueryErrorReportS3;
71 /**
72 * The KMS key used to encrypt the scheduled query resource at rest. When
73 * omitted, Timestream uses an AWS-owned key.
74 */
75 kmsKeyId?: string;
76 /**
77 * Whether the schedule is active.
78 * @default "ENABLED"
79 */
80 state?: ScheduledQueryState;
81 /**
82 * Tags to associate with the scheduled query.
83 */
84 tags?: Record<string, string>;
85}
86
87export interface ScheduledQuery extends Resource<
88 "AWS.Timestream.ScheduledQuery",
89 ScheduledQueryProps,
90 {
91 /**
92 * ARN of the scheduled query.
93 */
94 scheduledQueryArn: string;
95 /**
96 * The scheduled query's physical name.
97 */
98 name: string;
99 /**
100 * Whether the schedule is currently active.
101 */
102 state: ScheduledQueryState;
103 /**
104 * Current tags reported for the scheduled query.
105 */
106 tags: Record<string, string>;
107 },
108 never,
109 Providers
110> {}
111
112/**
113 * An Amazon Timestream for LiveAnalytics scheduled query — a SQL query
114 * Timestream runs on a cron/rate schedule, materializing results into a
115 * target table and notifying an SNS topic after each run.
116 *
117 * Only the `state` (ENABLED/DISABLED) is mutable in place; changing the
118 * query, schedule, notification topic, role, target, error report location,
119 * or KMS key replaces the scheduled query. Tags sync in place.
120 *
121 * :::note
122 * Timestream for LiveAnalytics is closed to new AWS customers. Accounts that
123 * were not already onboarded receive `TimestreamNotOnboarded` (a specialized
124 * `AccessDenied`) on every operation.
125 * :::
126 * ### Creating Scheduled Queries
127 * **Example:** Hourly Rollup
128 * ```typescript
129 * import * as Timestream from "alchemy/AWS/Timestream";
130 *
131 * const rollup = yield* Timestream.ScheduledQuery("HourlyRollup", {
132 * queryString: `SELECT host, AVG(measure_value::double) AS avg_cpu
133 * FROM "metrics"."cpu"
134 * WHERE time > ago(1h) GROUP BY host`,
135 * scheduleExpression: "rate(1 hour)",
136 * notificationTopicArn: topic.topicArn,
137 * executionRoleArn: role.roleArn,
138 * errorReportS3: { bucketName: bucket.bucketName },
139 * targetConfiguration: {
140 * TimestreamConfiguration: {
141 * DatabaseName: database.databaseName,
142 * TableName: rollupTable.tableName,
143 * TimeColumn: "time",
144 * DimensionMappings: [{ Name: "host", DimensionValueType: "VARCHAR" }],
145 * MultiMeasureMappings: {
146 * TargetMultiMeasureName: "cpu_rollup",
147 * MultiMeasureAttributeMappings: [
148 * { SourceColumn: "avg_cpu", MeasureValueType: "DOUBLE" },
149 * ],
150 * },
151 * },
152 * },
153 * });
154 * ```
155 *
156 * **Example:** Pausing a Schedule
157 * ```typescript
158 * const rollup = yield* Timestream.ScheduledQuery("HourlyRollup", {
159 * // ... unchanged configuration ...
160 * state: "DISABLED",
161 * });
162 * ```
163 *
164 * @resource
165 */
166export const ScheduledQuery = Resource<ScheduledQuery>(
167 "AWS.Timestream.ScheduledQuery",
168);
169
170const createScheduledQueryName = (
171 id: string,
172 props: { name?: string | undefined },
173) =>
174 Effect.gen(function* () {
175 if (props.name) {
176 return props.name;
177 }
178 return yield* createPhysicalName({ id, maxLength: 64 });
179 });
180
181const toTagRecord = (
182 tags: Array<{ Key: string; Value: string }> | undefined,
183): Record<string, string> =>
184 Object.fromEntries((tags ?? []).map((tag) => [tag.Key, tag.Value]));
185
186const readScheduledQuery = Effect.fn(function* (scheduledQueryArn: string) {
187 const response = yield* withQueryEndpoint(
188 TSQ.describeScheduledQuery({ ScheduledQueryArn: scheduledQueryArn }),
189 ).pipe(
190 Effect.catchTag("ResourceNotFoundException", () =>
191 Effect.succeed(undefined),
192 ),
193 );
194 if (!response?.ScheduledQuery) {
195 return undefined;
196 }
197 const scheduledQuery = response.ScheduledQuery;
198 const tags = yield* withQueryEndpoint(
199 TSQ.listTagsForResource.items({ ResourceARN: scheduledQuery.Arn }).pipe(
200 EffectStream.runCollect,
201 Effect.map((chunk) => Array.from(chunk)),
202 ),
203 ).pipe(
204 Effect.catchTag("ResourceNotFoundException", () =>
205 Effect.succeed(undefined),
206 ),
207 );
208 if (!tags) {
209 return undefined;
210 }
211 return {
212 scheduledQueryArn: scheduledQuery.Arn,
213 name: scheduledQuery.Name,
214 state: scheduledQuery.State,
215 tags: toTagRecord(tags),
216 } satisfies ScheduledQuery["Attributes"];
217});
218
219const findScheduledQueryByName = Effect.fn(function* (name: string) {
220 const match = yield* withQueryEndpoint(
221 TSQ.listScheduledQueries.pages({}).pipe(
222 EffectStream.map((page) => page.ScheduledQueries ?? []),
223 EffectStream.flattenIterable,
224 EffectStream.filter((summary) => summary.Name === name),
225 EffectStream.runHead,
226 ),
227 ).pipe(Effect.map(Option.getOrUndefined));
228 if (!match) return undefined;
229 return yield* readScheduledQuery(match.Arn);
230});
231
232export const ScheduledQueryProvider = () =>
233 Provider.effect(
234 ScheduledQuery,
235 Effect.gen(function* () {
236 return {
237 stables: ["scheduledQueryArn", "name"],
238 list: () =>
239 Effect.gen(function* () {
240 const summaries = yield* withQueryEndpoint(
241 TSQ.listScheduledQueries.pages({}).pipe(EffectStream.runCollect),
242 ).pipe(
243 Effect.map((chunk) =>
244 Array.from(chunk).flatMap(
245 (page) => page.ScheduledQueries ?? [],
246 ),
247 ),
248 );
249 const hydrated = yield* Effect.forEach(
250 summaries,
251 (summary) => readScheduledQuery(summary.Arn),
252 { concurrency: 5 },
253 );
254 return hydrated.filter(
255 (attrs): attrs is ScheduledQuery["Attributes"] =>
256 attrs !== undefined,
257 );
258 }),
259 read: Effect.fn(function* ({ id, olds, output }) {
260 const state = output?.scheduledQueryArn
261 ? yield* readScheduledQuery(output.scheduledQueryArn)
262 : yield* findScheduledQueryByName(
263 yield* createScheduledQueryName(id, olds ?? {}),
264 );
265 if (!state) return undefined;
266 return (yield* hasAlchemyTags(id, state.tags as Tags))
267 ? state
268 : Unowned(state);
269 }),
270 diff: Effect.fn(function* ({ id, news, olds }) {
271 if (!isResolved(news) || olds === undefined) return;
272 const oldName = yield* createScheduledQueryName(id, olds);
273 const newName = yield* createScheduledQueryName(id, news);
274 // Only State is mutable via UpdateScheduledQuery — every other
275 // aspect requires a replacement.
276 const changed = (a: unknown, b: unknown) =>
277 JSON.stringify(a ?? null) !== JSON.stringify(b ?? null);
278 if (
279 oldName !== newName ||
280 olds.queryString !== news.queryString ||
281 olds.scheduleExpression !== news.scheduleExpression ||
282 olds.notificationTopicArn !== news.notificationTopicArn ||
283 olds.executionRoleArn !== news.executionRoleArn ||
284 olds.kmsKeyId !== news.kmsKeyId ||
285 changed(olds.targetConfiguration, news.targetConfiguration) ||
286 changed(olds.errorReportS3, news.errorReportS3)
287 ) {
288 return { action: "replace" } as const;
289 }
290 }),
291 reconcile: Effect.fn(function* ({ id, news, output, session }) {
292 if (!news) {
293 return yield* Effect.fail(
294 new Error("Timestream ScheduledQuery requires props"),
295 );
296 }
297 const name =
298 output?.name ?? (yield* createScheduledQueryName(id, news));
299 const internalTags = yield* createInternalTags(id);
300 const desiredTags = { ...internalTags, ...news.tags };
301
302 // Observe — prefer the cached ARN; fall back to a name lookup so a
303 // create whose state failed to persist is adopted, not duplicated.
304 let state = output?.scheduledQueryArn
305 ? yield* readScheduledQuery(output.scheduledQueryArn)
306 : yield* findScheduledQueryByName(name);
307
308 // Ensure — create if missing; tolerate a ConflictException race.
309 if (state === undefined) {
310 yield* withQueryEndpoint(
311 TSQ.createScheduledQuery({
312 Name: name,
313 QueryString: news.queryString,
314 ScheduleConfiguration: {
315 ScheduleExpression: news.scheduleExpression,
316 },
317 NotificationConfiguration: {
318 SnsConfiguration: { TopicArn: news.notificationTopicArn },
319 },
320 ScheduledQueryExecutionRoleArn: news.executionRoleArn,
321 TargetConfiguration: news.targetConfiguration,
322 ErrorReportConfiguration: {
323 S3Configuration: {
324 BucketName: news.errorReportS3.bucketName,
325 ObjectKeyPrefix: news.errorReportS3.objectKeyPrefix,
326 EncryptionOption: news.errorReportS3.encryptionOption,
327 },
328 },
329 KmsKeyId: news.kmsKeyId,
330 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
331 Key,
332 Value,
333 })),
334 }),
335 ).pipe(Effect.catchTag("ConflictException", () => Effect.void));
336 yield* session.note(`Creating scheduled query ${name}...`);
337 state = yield* findScheduledQueryByName(name);
338 if (state === undefined) {
339 return yield* Effect.fail(
340 new Error(`failed to read created scheduled query ${name}`),
341 );
342 }
343 }
344
345 // Sync state — the only in-place mutable aspect.
346 const desiredState = news.state ?? "ENABLED";
347 if (state.state !== desiredState) {
348 yield* withQueryEndpoint(
349 TSQ.updateScheduledQuery({
350 ScheduledQueryArn: state.scheduledQueryArn,
351 State: desiredState,
352 }),
353 );
354 yield* session.note(
355 `Updated scheduled query ${name} state to ${desiredState}`,
356 );
357 }
358
359 // Sync tags — diff against observed cloud tags.
360 const { removed, upsert } = diffTags(state.tags, desiredTags);
361 if (removed.length > 0) {
362 yield* withQueryEndpoint(
363 TSQ.untagResource({
364 ResourceARN: state.scheduledQueryArn,
365 TagKeys: removed,
366 }),
367 );
368 }
369 if (upsert.length > 0) {
370 yield* withQueryEndpoint(
371 TSQ.tagResource({
372 ResourceARN: state.scheduledQueryArn,
373 Tags: upsert.map(({ Key, Value }) => ({ Key, Value })),
374 }),
375 );
376 }
377
378 yield* session.note(state.scheduledQueryArn);
379
380 const final = yield* readScheduledQuery(state.scheduledQueryArn);
381 if (!final) {
382 return yield* Effect.fail(
383 new Error(`failed to read reconciled scheduled query ${name}`),
384 );
385 }
386 return final;
387 }),
388 delete: Effect.fn(function* ({ output }) {
389 yield* withQueryEndpoint(
390 TSQ.deleteScheduledQuery({
391 ScheduledQueryArn: output.scheduledQueryArn,
392 }),
393 ).pipe(
394 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
395 );
396 }),
397 };
398 }),
399 );