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 DatasetGroupProps {
15 /**
16 * Name of the dataset group. If omitted, a unique name is generated from the
17 * app, stage, and logical ID. Changing the name replaces the group.
18 */
19 name?: string;
20 /**
21 * ARN of an IAM role Personalize assumes to access a customer-managed KMS
22 * key. Immutable — changing it replaces the group.
23 */
24 roleArn?: string;
25 /**
26 * ARN of a customer-managed KMS key used to encrypt the datasets. Immutable
27 * — changing it replaces the group.
28 */
29 kmsKeyArn?: string;
30 /**
31 * The domain of a domain dataset group (`ECOMMERCE` or `VIDEO_ON_DEMAND`).
32 * Omit for a custom dataset group. Immutable — changing it replaces the
33 * group.
34 */
35 domain?: string;
36 /**
37 * User-defined tags for the dataset group.
38 */
39 tags?: Record<string, string>;
40}
41
42export interface DatasetGroup extends Resource<
43 "AWS.Personalize.DatasetGroup",
44 DatasetGroupProps,
45 {
46 /**
47 * ARN of the dataset group.
48 */
49 datasetGroupArn: string;
50 /**
51 * Name of the dataset group.
52 */
53 name: string;
54 /**
55 * Dataset group status (e.g. `ACTIVE`, `CREATE PENDING`).
56 */
57 status: string;
58 /**
59 * Domain of the dataset group (`ECOMMERCE` or `VIDEO_ON_DEMAND`) when
60 * it is a domain dataset group.
61 */
62 domain: string | undefined;
63 },
64 never,
65 Providers
66> {}
67
68/**
69 * An Amazon Personalize dataset group — the top-level container that holds the
70 * datasets, solutions, and campaigns for a single use case. Creating a dataset
71 * group is cheap and fast; the expensive training work lives in solutions and
72 * campaigns provisioned separately.
73 *
74 * ### Creating a Dataset Group
75 * **Example:** Custom Dataset Group
76 * ```typescript
77 * const group = yield* Personalize.DatasetGroup("Recommendations", {});
78 * ```
79 *
80 * **Example:** Domain Dataset Group with Encryption
81 * ```typescript
82 * const group = yield* Personalize.DatasetGroup("Storefront", {
83 * domain: "ECOMMERCE",
84 * roleArn: role.roleArn,
85 * kmsKeyArn: key.keyArn,
86 * tags: { team: "growth" },
87 * });
88 * ```
89 *
90 * @resource
91 */
92export const DatasetGroup = Resource<DatasetGroup>(
93 "AWS.Personalize.DatasetGroup",
94);
95
96export const DatasetGroupProvider = () =>
97 Provider.effect(
98 DatasetGroup,
99 Effect.gen(function* () {
100 const createName = Effect.fn(function* (
101 id: string,
102 props: DatasetGroupProps,
103 ) {
104 return props.name ?? (yield* createPhysicalName({ id, maxLength: 63 }));
105 });
106
107 const describe = Effect.fn(function* (datasetGroupArn: string) {
108 const response = yield* personalize
109 .describeDatasetGroup({ datasetGroupArn })
110 .pipe(
111 Effect.catchTag("ResourceNotFoundException", () =>
112 Effect.succeed(undefined),
113 ),
114 );
115 return response?.datasetGroup;
116 });
117
118 /** Poll until the group reaches ACTIVE; fail fast on CREATE FAILED. */
119 const waitActive = Effect.fn(function* (datasetGroupArn: string) {
120 const group = yield* describe(datasetGroupArn).pipe(
121 Effect.repeat({
122 schedule: Schedule.max([
123 Schedule.fixed("2 seconds"),
124 Schedule.recurs(40),
125 ]),
126 until: (g) =>
127 g?.status === "ACTIVE" || (g?.status ?? "").includes("FAILED"),
128 }),
129 );
130 if (group?.status !== "ACTIVE") {
131 return yield* Effect.fail(
132 new Error(
133 `Personalize dataset group ${datasetGroupArn} did not become ACTIVE (status: ${group?.status}, reason: ${group?.failureReason})`,
134 ),
135 );
136 }
137 return group;
138 });
139
140 /** Find an existing dataset group's ARN by its (deterministic) name. */
141 const findArnByName = Effect.fn(function* (name: string) {
142 const pages = yield* personalize.listDatasetGroups
143 .pages({})
144 .pipe(Stream.runCollect);
145 return Array.from(pages)
146 .flatMap((page) => page.datasetGroups ?? [])
147 .find((summary) => summary.name === name)?.datasetGroupArn;
148 });
149
150 const toAttrs = (group: personalize.DatasetGroup) => ({
151 datasetGroupArn: group.datasetGroupArn!,
152 name: group.name!,
153 status: group.status!,
154 domain: group.domain,
155 });
156
157 /** Bounded retry schedule for ResourceInUse while async deletes drain. */
158 const inUseRetry = {
159 while: (e: { _tag: string }) => e._tag === "ResourceInUseException",
160 schedule: Schedule.max([
161 Schedule.fixed("3 seconds"),
162 Schedule.recurs(20),
163 ]),
164 };
165
166 const listChildren = Effect.fn(function* (datasetGroupArn: string) {
167 const [trackers, filters, solutions, datasets] = yield* Effect.all(
168 [
169 personalize.listEventTrackers.pages({ datasetGroupArn }).pipe(
170 Stream.runCollect,
171 Effect.map((chunk) =>
172 Array.from(chunk).flatMap((page) => page.eventTrackers ?? []),
173 ),
174 ),
175 personalize.listFilters.pages({ datasetGroupArn }).pipe(
176 Stream.runCollect,
177 Effect.map((chunk) =>
178 Array.from(chunk).flatMap((page) => page.Filters ?? []),
179 ),
180 ),
181 personalize.listSolutions.pages({ datasetGroupArn }).pipe(
182 Stream.runCollect,
183 Effect.map((chunk) =>
184 Array.from(chunk).flatMap((page) => page.solutions ?? []),
185 ),
186 ),
187 personalize.listDatasets.pages({ datasetGroupArn }).pipe(
188 Stream.runCollect,
189 Effect.map((chunk) =>
190 Array.from(chunk).flatMap((page) => page.datasets ?? []),
191 ),
192 ),
193 ],
194 { concurrency: 4 },
195 );
196 return { trackers, filters, solutions, datasets };
197 });
198
199 /**
200 * DeleteDatasetGroup requires all children (event trackers, filters,
201 * campaigns, solutions, datasets) to be deleted first. Children created
202 * out-of-band (e.g. via CreateSolution/CreateCampaign runtime bindings)
203 * are not tracked in state, so the group drains any that remain. Each
204 * child delete is idempotent (NotFound tolerated) and asynchronous, so
205 * after issuing the deletes we poll (bounded) until the group is empty.
206 */
207 const drainChildren = Effect.fn(function* (datasetGroupArn: string) {
208 const { trackers, filters, solutions, datasets } =
209 yield* listChildren(datasetGroupArn);
210
211 yield* Effect.forEach(
212 trackers,
213 (t) =>
214 personalize
215 .deleteEventTracker({ eventTrackerArn: t.eventTrackerArn! })
216 .pipe(
217 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
218 Effect.retry(inUseRetry),
219 ),
220 { concurrency: 4 },
221 );
222
223 yield* Effect.forEach(
224 filters,
225 (f) =>
226 personalize.deleteFilter({ filterArn: f.filterArn! }).pipe(
227 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
228 Effect.retry(inUseRetry),
229 ),
230 { concurrency: 4 },
231 );
232
233 // Campaigns block their solution's deletion, so drain them per
234 // solution before deleting the solution.
235 yield* Effect.forEach(
236 solutions,
237 Effect.fn(function* (s) {
238 const campaigns = yield* personalize.listCampaigns
239 .pages({ solutionArn: s.solutionArn })
240 .pipe(
241 Stream.runCollect,
242 Effect.map((chunk) =>
243 Array.from(chunk).flatMap((page) => page.campaigns ?? []),
244 ),
245 );
246 yield* Effect.forEach(
247 campaigns,
248 (c) =>
249 personalize
250 .deleteCampaign({ campaignArn: c.campaignArn! })
251 .pipe(
252 Effect.catchTag(
253 "ResourceNotFoundException",
254 () => Effect.void,
255 ),
256 Effect.retry(inUseRetry),
257 ),
258 { concurrency: 4 },
259 );
260 yield* personalize
261 .deleteSolution({ solutionArn: s.solutionArn! })
262 .pipe(
263 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
264 // Campaign deletion is async — retry while it drains.
265 Effect.retry(inUseRetry),
266 );
267 }),
268 { concurrency: 2 },
269 );
270
271 yield* Effect.forEach(
272 datasets,
273 (d) =>
274 personalize.deleteDataset({ datasetArn: d.datasetArn! }).pipe(
275 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
276 Effect.retry(inUseRetry),
277 ),
278 { concurrency: 4 },
279 );
280
281 // All child deletes are asynchronous — wait (bounded) until the group
282 // reports no remaining children so DeleteDatasetGroup succeeds.
283 const remaining = yield* listChildren(datasetGroupArn).pipe(
284 Effect.repeat({
285 schedule: Schedule.max([
286 Schedule.fixed("3 seconds"),
287 Schedule.recurs(40),
288 ]),
289 until: (children): boolean =>
290 children.trackers.length === 0 &&
291 children.filters.length === 0 &&
292 children.solutions.length === 0 &&
293 children.datasets.length === 0,
294 }),
295 );
296 if (
297 remaining.trackers.length +
298 remaining.filters.length +
299 remaining.solutions.length +
300 remaining.datasets.length >
301 0
302 ) {
303 return yield* Effect.fail(
304 new Error(
305 `Personalize dataset group ${datasetGroupArn} still has children after drain (trackers: ${remaining.trackers.length}, filters: ${remaining.filters.length}, solutions: ${remaining.solutions.length}, datasets: ${remaining.datasets.length})`,
306 ),
307 );
308 }
309 });
310
311 return {
312 stables: ["datasetGroupArn", "name"],
313
314 diff: Effect.fn(function* ({ id, olds = {}, news }) {
315 if (!isResolved(news)) return undefined;
316 const oldName = yield* createName(id, olds);
317 const newName = yield* createName(id, news);
318 if (
319 oldName !== newName ||
320 (olds.roleArn ?? undefined) !== (news.roleArn ?? undefined) ||
321 (olds.kmsKeyArn ?? undefined) !== (news.kmsKeyArn ?? undefined) ||
322 (olds.domain ?? undefined) !== (news.domain ?? undefined)
323 ) {
324 return { action: "replace" } as const;
325 }
326 }),
327
328 read: Effect.fn(function* ({ id, output }) {
329 if (!output?.datasetGroupArn) return undefined;
330 const group = yield* describe(output.datasetGroupArn);
331 if (group === undefined) return undefined;
332 const attrs = toAttrs(group);
333 const tags = yield* readPersonalizeTags(group.datasetGroupArn!);
334 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
335 }),
336
337 reconcile: Effect.fn(function* ({ id, news = {}, output, session }) {
338 const name = yield* createName(id, news);
339 const internalTags = yield* createInternalTags(id);
340 const desiredTags = { ...internalTags, ...news.tags };
341
342 // 1. Observe — cloud state is authoritative; output is an ARN cache.
343 let group =
344 output?.datasetGroupArn !== undefined
345 ? yield* describe(output.datasetGroupArn)
346 : undefined;
347
348 // 2. Ensure — create if missing, then wait for ACTIVE. A crashed
349 // prior run may have left a same-named group behind with no
350 // persisted state — adopt it by name and converge.
351 if (group === undefined) {
352 const arn = yield* personalize
353 .createDatasetGroup({
354 name,
355 roleArn: news.roleArn,
356 kmsKeyArn: news.kmsKeyArn,
357 domain: news.domain,
358 tags: Object.entries(desiredTags).map(([tagKey, tagValue]) => ({
359 tagKey,
360 tagValue,
361 })),
362 })
363 .pipe(
364 Effect.map((created) => created.datasetGroupArn!),
365 Effect.catchTag("ResourceAlreadyExistsException", (error) =>
366 findArnByName(name).pipe(
367 Effect.flatMap((existing) =>
368 existing === undefined
369 ? Effect.fail(error)
370 : Effect.succeed(existing),
371 ),
372 ),
373 ),
374 );
375 group = yield* waitActive(arn);
376 }
377
378 // 3. Sync tags — diff against OBSERVED cloud tags (no-op after a
379 // fresh create; converges an adopted leftover).
380 yield* syncPersonalizeTags(group.datasetGroupArn!, desiredTags);
381
382 yield* session.note(group.datasetGroupArn!);
383 return toAttrs(group);
384 }),
385
386 delete: Effect.fn(function* ({ output }) {
387 // DeleteDatasetGroup requires every child — event trackers, filters,
388 // campaigns, solutions, and datasets — to be deleted first, and each
389 // child delete is itself asynchronous. Deletion ordering can't
390 // guarantee children-before-parent (bindings/tests create children
391 // out-of-band that are not tracked in state), so drain the group's
392 // remaining children before deleting the group itself. Every step is
393 // idempotent and NotFound-tolerant.
394 const group = yield* describe(output.datasetGroupArn);
395 if (group !== undefined) {
396 yield* drainChildren(output.datasetGroupArn);
397 }
398 yield* personalize
399 .deleteDatasetGroup({ datasetGroupArn: output.datasetGroupArn })
400 .pipe(
401 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
402 // Tolerate the eventual-consistency window while the drained
403 // children finish their own asynchronous deletions.
404 Effect.retry({
405 while: (e) => e._tag === "ResourceInUseException",
406 schedule: Schedule.max([
407 Schedule.fixed("3 seconds"),
408 Schedule.recurs(20),
409 ]),
410 }),
411 );
412 // Deletion is asynchronous (DELETE IN_PROGRESS) — wait until the
413 // group is actually gone so the run leaves no lingering resources
414 // (the group's auto-created event schema is removed with it).
415 const remaining = yield* describe(output.datasetGroupArn).pipe(
416 Effect.repeat({
417 schedule: Schedule.max([
418 Schedule.fixed("3 seconds"),
419 Schedule.recurs(40),
420 ]),
421 until: (group): boolean => group === undefined,
422 }),
423 );
424 if (remaining !== undefined) {
425 return yield* Effect.fail(
426 new Error(
427 `Personalize dataset group ${output.datasetGroupArn} was not deleted (status: ${remaining.status})`,
428 ),
429 );
430 }
431 }),
432
433 list: () =>
434 personalize.listDatasetGroups.pages({}).pipe(
435 Stream.runCollect,
436 Effect.map((chunk) =>
437 Array.from(chunk).flatMap((page) => page.datasetGroups ?? []),
438 ),
439 Effect.flatMap(
440 Effect.forEach(
441 (summary) =>
442 describe(summary.datasetGroupArn!).pipe(
443 Effect.map((g) => (g ? toAttrs(g) : undefined)),
444 ),
445 { concurrency: 4 },
446 ),
447 ),
448 Effect.map((items) => items.filter((item) => item !== undefined)),
449 ),
450 };
451 }),
452 );