1import * as forecast from "@distilled.cloud/aws/forecast";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import * as Stream 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 { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
11import type { Providers } from "../Providers.ts";
12import {
13 readForecastTags,
14 syncForecastTags,
15 toForecastName,
16} from "./internal.ts";
17
18export interface SchemaAttribute {
19 /** Name of the field in the dataset (e.g. `item_id`, `timestamp`). */
20 attributeName: string;
21 /** Type of the field: `string`, `integer`, `float`, `timestamp`, or `geolocation`. */
22 attributeType: string;
23}
24
25export interface DatasetSchema {
26 /** The ordered list of fields that make up the dataset. */
27 attributes: SchemaAttribute[];
28}
29
30export interface DatasetEncryptionConfig {
31 /** ARN of the IAM role Forecast assumes to access the KMS key. */
32 roleArn: string;
33 /** ARN of the customer-managed KMS key. */
34 kmsKeyArn: string;
35}
36
37export interface DatasetProps {
38 /**
39 * Name of the dataset. If omitted, a unique name is generated from the app,
40 * stage, and logical ID. Changing the name replaces the dataset.
41 */
42 datasetName?: string;
43 /**
44 * The forecasting domain: `RETAIL`, `CUSTOM`, `INVENTORY_PLANNING`,
45 * `EC2_CAPACITY`, `WORK_FORCE`, `WEB_TRAFFIC`, or `METRICS`. Immutable —
46 * changing it replaces the dataset.
47 */
48 domain: string;
49 /**
50 * The dataset type: `TARGET_TIME_SERIES`, `RELATED_TIME_SERIES`, or
51 * `ITEM_METADATA`. Immutable — changing it replaces the dataset.
52 */
53 datasetType: string;
54 /**
55 * The frequency of data collection (e.g. `D` for daily, `H` for hourly).
56 * Required for time-series datasets. Immutable — changing it replaces the
57 * dataset.
58 */
59 dataFrequency?: string;
60 /**
61 * The schema describing the dataset's fields. Immutable — changing it
62 * replaces the dataset.
63 */
64 schema: DatasetSchema;
65 /**
66 * Customer-managed KMS encryption configuration. Immutable — changing it
67 * replaces the dataset.
68 */
69 encryptionConfig?: DatasetEncryptionConfig;
70 /**
71 * User-defined tags for the dataset.
72 */
73 tags?: Record<string, string>;
74}
75
76export interface Dataset extends Resource<
77 "AWS.Forecast.Dataset",
78 DatasetProps,
79 {
80 /** The ARN of the dataset. */
81 datasetArn: string;
82 /** The name of the dataset. */
83 datasetName: string;
84 /** The forecasting domain of the dataset, e.g. `RETAIL` or `CUSTOM`. */
85 domain: string;
86 /** The dataset type, e.g. `TARGET_TIME_SERIES`. */
87 datasetType: string;
88 /** The dataset status, e.g. `ACTIVE`. */
89 status: string;
90 },
91 never,
92 Providers
93> {}
94
95/**
96 * An Amazon Forecast dataset — a typed, domain-scoped collection of
97 * time-series (or metadata) records described by a schema. Creating the
98 * dataset is a cheap metadata operation; bulk imports and training happen
99 * through separate import jobs and predictors.
100 *
101 * ### Creating a Dataset
102 * **Example:** Target Time-Series Dataset
103 * ```typescript
104 * const dataset = yield* Forecast.Dataset("Demand", {
105 * domain: "CUSTOM",
106 * datasetType: "TARGET_TIME_SERIES",
107 * dataFrequency: "D",
108 * schema: {
109 * attributes: [
110 * { attributeName: "item_id", attributeType: "string" },
111 * { attributeName: "timestamp", attributeType: "timestamp" },
112 * { attributeName: "target_value", attributeType: "float" },
113 * ],
114 * },
115 * });
116 * ```
117 *
118 * @resource
119 */
120export const Dataset = Resource<Dataset>("AWS.Forecast.Dataset");
121
122export const DatasetProvider = () =>
123 Provider.effect(
124 Dataset,
125 Effect.gen(function* () {
126 const createName = Effect.fn(function* (id: string, props: DatasetProps) {
127 return (
128 props.datasetName ??
129 toForecastName(yield* createPhysicalName({ id, maxLength: 63 }))
130 );
131 });
132
133 const describe = Effect.fn(function* (datasetArn: string) {
134 return yield* forecast
135 .describeDataset({ DatasetArn: datasetArn })
136 .pipe(
137 Effect.catchTag("ResourceNotFoundException", () =>
138 Effect.succeed(undefined),
139 ),
140 );
141 });
142
143 const toAttrs = (dataset: forecast.DescribeDatasetResponse) => ({
144 datasetArn: dataset.DatasetArn!,
145 datasetName: dataset.DatasetName!,
146 domain: dataset.Domain!,
147 datasetType: dataset.DatasetType!,
148 status: dataset.Status!,
149 });
150
151 return {
152 stables: ["datasetArn", "datasetName"],
153
154 diff: Effect.fn(function* ({ id, olds, news }) {
155 if (!isResolved(news)) return undefined;
156 const oldName = yield* createName(id, olds);
157 const newName = yield* createName(id, news);
158 if (
159 oldName !== newName ||
160 (olds.domain ?? undefined) !== (news.domain ?? undefined) ||
161 (olds.datasetType ?? undefined) !==
162 (news.datasetType ?? undefined) ||
163 (olds.dataFrequency ?? undefined) !==
164 (news.dataFrequency ?? undefined) ||
165 JSON.stringify(olds.schema ?? null) !==
166 JSON.stringify(news.schema ?? null)
167 ) {
168 return { action: "replace" } as const;
169 }
170 }),
171
172 read: Effect.fn(function* ({ id, output }) {
173 if (!output?.datasetArn) return undefined;
174 const dataset = yield* describe(output.datasetArn);
175 if (dataset === undefined) return undefined;
176 const attrs = toAttrs(dataset);
177 const tags = yield* readForecastTags(dataset.DatasetArn!);
178 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
179 }),
180
181 reconcile: Effect.fn(function* ({ id, news, output, session }) {
182 const name = yield* createName(id, news);
183 const internalTags = yield* createInternalTags(id);
184 const desiredTags = { ...internalTags, ...news.tags };
185
186 let dataset =
187 output?.datasetArn !== undefined
188 ? yield* describe(output.datasetArn)
189 : undefined;
190
191 if (dataset === undefined) {
192 const created = yield* forecast.createDataset({
193 DatasetName: name,
194 Domain: news.domain,
195 DatasetType: news.datasetType,
196 DataFrequency: news.dataFrequency,
197 Schema: {
198 Attributes: news.schema.attributes.map((a) => ({
199 AttributeName: a.attributeName,
200 AttributeType: a.attributeType,
201 })),
202 },
203 EncryptionConfig: news.encryptionConfig
204 ? {
205 RoleArn: news.encryptionConfig.roleArn,
206 KMSKeyArn: news.encryptionConfig.kmsKeyArn,
207 }
208 : undefined,
209 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
210 Key,
211 Value,
212 })),
213 });
214 dataset = yield* describe(created.DatasetArn!);
215 } else {
216 yield* syncForecastTags(dataset.DatasetArn!, desiredTags);
217 }
218
219 yield* session.note(dataset!.DatasetArn!);
220 return toAttrs(dataset!);
221 }),
222
223 delete: Effect.fn(function* ({ output }) {
224 yield* forecast.deleteDataset({ DatasetArn: output.datasetArn }).pipe(
225 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
226 Effect.retry({
227 while: (e) => e._tag === "ResourceInUseException",
228 schedule: Schedule.max([
229 Schedule.fixed("3 seconds"),
230 Schedule.recurs(20),
231 ]),
232 }),
233 );
234 }),
235
236 list: () =>
237 forecast.listDatasets.pages({}).pipe(
238 Stream.runCollect,
239 Effect.map((chunk) =>
240 Array.from(chunk).flatMap((page) => page.Datasets ?? []),
241 ),
242 Effect.flatMap(
243 Effect.forEach(
244 (summary) =>
245 describe(summary.DatasetArn!).pipe(
246 Effect.map((d) => (d ? toAttrs(d) : undefined)),
247 ),
248 { concurrency: 4 },
249 ),
250 ),
251 Effect.map((items) => items.filter((item) => item !== undefined)),
252 ),
253 };
254 }),
255 );