1import * as personalize from "@distilled.cloud/aws/personalize";
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 { readPersonalizeTags, syncPersonalizeTags } from "./internal.ts";
13
14export interface DatasetProps {
15 /**
16 * Name of the dataset. If omitted, a unique name is generated from the app,
17 * stage, and logical ID. Changing the name replaces the dataset.
18 */
19 name?: string;
20 /**
21 * ARN of the schema that describes this dataset's fields. Immutable —
22 * changing it replaces the dataset.
23 */
24 schemaArn: string;
25 /**
26 * ARN of the dataset group this dataset belongs to. Immutable — changing it
27 * replaces the dataset.
28 */
29 datasetGroupArn: string;
30 /**
31 * The type of dataset: `Interactions`, `Items`, `Users`, `Actions`, or
32 * `Action_Interactions`. Immutable — changing it replaces the dataset.
33 */
34 datasetType: string;
35 /**
36 * User-defined tags for the dataset.
37 */
38 tags?: Record<string, string>;
39}
40
41export interface Dataset extends Resource<
42 "AWS.Personalize.Dataset",
43 DatasetProps,
44 {
45 /**
46 * ARN of the dataset.
47 */
48 datasetArn: string;
49 /**
50 * Name of the dataset.
51 */
52 name: string;
53 /**
54 * Dataset type (`Interactions`, `Items`, `Users`, or `Actions`).
55 */
56 datasetType: string;
57 /**
58 * Dataset status (e.g. `ACTIVE`, `CREATE PENDING`).
59 */
60 status: string;
61 /**
62 * ARN of the schema the dataset conforms to.
63 */
64 schemaArn: string;
65 /**
66 * ARN of the dataset group the dataset belongs to.
67 */
68 datasetGroupArn: string;
69 },
70 never,
71 Providers
72> {}
73
74/**
75 * An Amazon Personalize dataset — a typed collection (Interactions, Items,
76 * Users, …) inside a dataset group, backed by a schema. Creating the dataset
77 * is a cheap metadata operation; bulk imports and training happen through
78 * separate import jobs and solutions.
79 *
80 * ### Creating a Dataset
81 * **Example:** Interactions Dataset
82 * ```typescript
83 * const dataset = yield* Personalize.Dataset("Interactions", {
84 * schemaArn: schema.schemaArn,
85 * datasetGroupArn: group.datasetGroupArn,
86 * datasetType: "Interactions",
87 * });
88 * ```
89 *
90 * @resource
91 */
92export const Dataset = Resource<Dataset>("AWS.Personalize.Dataset");
93
94export const DatasetProvider = () =>
95 Provider.effect(
96 Dataset,
97 Effect.gen(function* () {
98 const createName = Effect.fn(function* (id: string, props: DatasetProps) {
99 return props.name ?? (yield* createPhysicalName({ id, maxLength: 63 }));
100 });
101
102 const describe = Effect.fn(function* (datasetArn: string) {
103 const response = yield* personalize
104 .describeDataset({ datasetArn })
105 .pipe(
106 Effect.catchTag("ResourceNotFoundException", () =>
107 Effect.succeed(undefined),
108 ),
109 );
110 return response?.dataset;
111 });
112
113 const waitActive = Effect.fn(function* (datasetArn: string) {
114 const dataset = yield* describe(datasetArn).pipe(
115 Effect.repeat({
116 schedule: Schedule.max([
117 Schedule.fixed("2 seconds"),
118 Schedule.recurs(40),
119 ]),
120 until: (d) =>
121 d?.status === "ACTIVE" || (d?.status ?? "").includes("FAILED"),
122 }),
123 );
124 if (dataset?.status !== "ACTIVE") {
125 return yield* Effect.fail(
126 new Error(
127 `Personalize dataset ${datasetArn} did not become ACTIVE (status: ${dataset?.status})`,
128 ),
129 );
130 }
131 return dataset;
132 });
133
134 /** Find an existing dataset's ARN by name within its dataset group. */
135 const findArnByName = Effect.fn(function* (
136 name: string,
137 datasetGroupArn: string,
138 ) {
139 const pages = yield* personalize.listDatasets
140 .pages({ datasetGroupArn })
141 .pipe(Stream.runCollect);
142 return Array.from(pages)
143 .flatMap((page) => page.datasets ?? [])
144 .find((summary) => summary.name === name)?.datasetArn;
145 });
146
147 const toAttrs = (dataset: personalize.Dataset) => ({
148 datasetArn: dataset.datasetArn!,
149 name: dataset.name!,
150 datasetType: dataset.datasetType!,
151 status: dataset.status!,
152 schemaArn: dataset.schemaArn!,
153 datasetGroupArn: dataset.datasetGroupArn!,
154 });
155
156 return {
157 stables: ["datasetArn", "name"],
158
159 diff: Effect.fn(function* ({ id, olds, news }) {
160 if (!isResolved(news)) return undefined;
161 const oldName = yield* createName(id, olds);
162 const newName = yield* createName(id, news);
163 if (
164 oldName !== newName ||
165 (olds.schemaArn ?? undefined) !== (news.schemaArn ?? undefined) ||
166 (olds.datasetGroupArn ?? undefined) !==
167 (news.datasetGroupArn ?? undefined) ||
168 (olds.datasetType ?? undefined) !== (news.datasetType ?? undefined)
169 ) {
170 return { action: "replace" } as const;
171 }
172 }),
173
174 read: Effect.fn(function* ({ id, output }) {
175 if (!output?.datasetArn) return undefined;
176 const dataset = yield* describe(output.datasetArn);
177 if (dataset === undefined) return undefined;
178 const attrs = toAttrs(dataset);
179 const tags = yield* readPersonalizeTags(dataset.datasetArn!);
180 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
181 }),
182
183 reconcile: Effect.fn(function* ({ id, news, output, session }) {
184 const name = yield* createName(id, news);
185 const internalTags = yield* createInternalTags(id);
186 const desiredTags = { ...internalTags, ...news.tags };
187
188 let dataset =
189 output?.datasetArn !== undefined
190 ? yield* describe(output.datasetArn)
191 : undefined;
192
193 // Ensure — create if missing. A crashed prior run may have left a
194 // same-named dataset behind with no persisted state — adopt it by
195 // name and converge.
196 if (dataset === undefined) {
197 const arn = yield* personalize
198 .createDataset({
199 name,
200 schemaArn: news.schemaArn,
201 datasetGroupArn: news.datasetGroupArn,
202 datasetType: news.datasetType,
203 tags: Object.entries(desiredTags).map(([tagKey, tagValue]) => ({
204 tagKey,
205 tagValue,
206 })),
207 })
208 .pipe(
209 Effect.map((created) => created.datasetArn!),
210 Effect.catchTag("ResourceAlreadyExistsException", (error) =>
211 findArnByName(name, news.datasetGroupArn).pipe(
212 Effect.flatMap((existing) =>
213 existing === undefined
214 ? Effect.fail(error)
215 : Effect.succeed(existing),
216 ),
217 ),
218 ),
219 );
220 dataset = yield* waitActive(arn);
221 }
222
223 // Sync tags — diff against OBSERVED cloud tags (no-op after a fresh
224 // create; converges an adopted leftover).
225 yield* syncPersonalizeTags(dataset.datasetArn!, desiredTags);
226
227 yield* session.note(dataset.datasetArn!);
228 return toAttrs(dataset);
229 }),
230
231 delete: Effect.fn(function* ({ output }) {
232 yield* personalize
233 .deleteDataset({ datasetArn: output.datasetArn })
234 .pipe(
235 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
236 Effect.retry({
237 while: (e) => e._tag === "ResourceInUseException",
238 schedule: Schedule.max([
239 Schedule.fixed("3 seconds"),
240 Schedule.recurs(20),
241 ]),
242 }),
243 );
244 // Deletion is asynchronous (DELETE PENDING) — wait until the dataset
245 // is actually gone so the schema and dataset group it references can
246 // delete without exhausting their ResourceInUse retries.
247 const remaining = yield* describe(output.datasetArn).pipe(
248 Effect.repeat({
249 schedule: Schedule.max([
250 Schedule.fixed("3 seconds"),
251 Schedule.recurs(40),
252 ]),
253 until: (dataset): boolean => dataset === undefined,
254 }),
255 );
256 if (remaining !== undefined) {
257 return yield* Effect.fail(
258 new Error(
259 `Personalize dataset ${output.datasetArn} was not deleted (status: ${remaining.status})`,
260 ),
261 );
262 }
263 }),
264
265 // Datasets are keyed by their parent dataset group (ListDatasets
266 // requires a datasetGroupArn), so account-wide enumeration walks the
267 // dataset groups first and lists each group's datasets. Without this,
268 // orphaned datasets are invisible to nuke and permanently block both
269 // their schema and their dataset group from deleting.
270 list: () =>
271 personalize.listDatasetGroups.pages({}).pipe(
272 Stream.runCollect,
273 Effect.map((chunk) =>
274 Array.from(chunk).flatMap((page) => page.datasetGroups ?? []),
275 ),
276 Effect.flatMap(
277 Effect.forEach(
278 (group) =>
279 personalize.listDatasets
280 .pages({ datasetGroupArn: group.datasetGroupArn })
281 .pipe(
282 Stream.runCollect,
283 Effect.map((chunk) =>
284 Array.from(chunk).flatMap(
285 (page) => page.datasets ?? [],
286 ),
287 ),
288 ),
289 { concurrency: 4 },
290 ),
291 ),
292 Effect.map((groups) => groups.flat()),
293 Effect.flatMap(
294 Effect.forEach(
295 (summary) =>
296 describe(summary.datasetArn!).pipe(
297 Effect.map((d) => (d ? toAttrs(d) : undefined)),
298 ),
299 { concurrency: 4 },
300 ),
301 ),
302 Effect.map((items) => items.filter((item) => item !== undefined)),
303 ),
304 };
305 }),
306 );