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 { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import type { Providers } from "../Providers.ts";
10
11export interface SchemaProps {
12 /**
13 * Name of the schema. If omitted, a unique name is generated from the app,
14 * stage, and logical ID. Changing the name replaces the schema.
15 */
16 name?: string;
17 /**
18 * The Avro schema definition, as a JSON string, describing the fields of the
19 * dataset this schema applies to. Immutable — changing it replaces the
20 * schema.
21 *
22 * @example
23 * ```json
24 * {
25 * "type": "record",
26 * "name": "Interactions",
27 * "namespace": "com.amazonaws.personalize.schema",
28 * "fields": [
29 * { "name": "USER_ID", "type": "string" },
30 * { "name": "ITEM_ID", "type": "string" },
31 * { "name": "TIMESTAMP", "type": "long" }
32 * ],
33 * "version": "1.0"
34 * }
35 * ```
36 */
37 schema: string;
38 /**
39 * The domain of a domain-specific dataset group this schema belongs to
40 * (`ECOMMERCE` or `VIDEO_ON_DEMAND`). Omit for a custom (non-domain) schema.
41 * Immutable — changing it replaces the schema.
42 */
43 domain?: string;
44}
45
46export interface Schema extends Resource<
47 "AWS.Personalize.Schema",
48 SchemaProps,
49 {
50 /**
51 * ARN of the schema.
52 */
53 schemaArn: string;
54 /**
55 * Name of the schema.
56 */
57 name: string;
58 /**
59 * Avro schema definition as a JSON string.
60 */
61 schema: string;
62 /**
63 * Domain of the schema (`ECOMMERCE` or `VIDEO_ON_DEMAND`) when it is a
64 * domain schema.
65 */
66 domain: string | undefined;
67 },
68 never,
69 Providers
70> {}
71
72/**
73 * An Amazon Personalize schema — an Avro definition that describes the fields
74 * of a dataset (Interactions, Items, Users, …). Schemas are immutable once
75 * created; changing any property replaces the schema.
76 *
77 * ### Creating a Schema
78 * **Example:** Interactions Schema
79 * ```typescript
80 * const schema = yield* Personalize.Schema("Interactions", {
81 * schema: JSON.stringify({
82 * type: "record",
83 * name: "Interactions",
84 * namespace: "com.amazonaws.personalize.schema",
85 * fields: [
86 * { name: "USER_ID", type: "string" },
87 * { name: "ITEM_ID", type: "string" },
88 * { name: "TIMESTAMP", type: "long" },
89 * ],
90 * version: "1.0",
91 * }),
92 * });
93 * ```
94 *
95 * @resource
96 */
97export const Schema = Resource<Schema>("AWS.Personalize.Schema");
98
99export const SchemaProvider = () =>
100 Provider.effect(
101 Schema,
102 Effect.gen(function* () {
103 const createName = Effect.fn(function* (id: string, props: SchemaProps) {
104 return props.name ?? (yield* createPhysicalName({ id, maxLength: 63 }));
105 });
106
107 /** Describe a schema by ARN; typed not-found → undefined. */
108 const describe = Effect.fn(function* (schemaArn: string) {
109 const response = yield* personalize
110 .describeSchema({ schemaArn })
111 .pipe(
112 Effect.catchTag("ResourceNotFoundException", () =>
113 Effect.succeed(undefined),
114 ),
115 );
116 return response?.schema;
117 });
118
119 /** Find an existing schema's ARN by its (deterministic) name. */
120 const findArnByName = Effect.fn(function* (name: string) {
121 const pages = yield* personalize.listSchemas
122 .pages({})
123 .pipe(Stream.runCollect);
124 return Array.from(pages)
125 .flatMap((page) => page.schemas ?? [])
126 .find((summary) => summary.name === name)?.schemaArn;
127 });
128
129 return {
130 stables: ["schemaArn", "name"],
131
132 diff: Effect.fn(function* ({ id, olds, news }) {
133 if (!isResolved(news)) return undefined;
134 const oldName = yield* createName(id, olds);
135 const newName = yield* createName(id, news);
136 // Schemas are entirely immutable — any change replaces.
137 if (
138 oldName !== newName ||
139 (olds.schema ?? undefined) !== (news.schema ?? undefined) ||
140 (olds.domain ?? undefined) !== (news.domain ?? undefined)
141 ) {
142 return { action: "replace" } as const;
143 }
144 }),
145
146 read: Effect.fn(function* ({ output }) {
147 if (!output?.schemaArn) return undefined;
148 const schema = yield* describe(output.schemaArn);
149 if (schema === undefined) return undefined;
150 return {
151 schemaArn: schema.schemaArn!,
152 name: schema.name!,
153 schema: schema.schema!,
154 domain: schema.domain,
155 };
156 }),
157
158 reconcile: Effect.fn(function* ({ id, news, output, session }) {
159 const name = yield* createName(id, news);
160
161 // 1. Observe — cloud state is authoritative; output is an ARN cache.
162 let schema =
163 output?.schemaArn !== undefined
164 ? yield* describe(output.schemaArn)
165 : undefined;
166
167 // 2. Ensure — schemas are immutable, so only create when missing.
168 // A crashed prior run may have left a same-named schema behind
169 // with no persisted state — adopt it by name.
170 if (schema === undefined) {
171 const arn = yield* personalize
172 .createSchema({
173 name,
174 schema: news.schema,
175 domain: news.domain,
176 })
177 .pipe(
178 Effect.map((created) => created.schemaArn!),
179 Effect.catchTag("ResourceAlreadyExistsException", (error) =>
180 findArnByName(name).pipe(
181 Effect.flatMap((existing) =>
182 existing === undefined
183 ? Effect.fail(error)
184 : Effect.succeed(existing),
185 ),
186 ),
187 ),
188 );
189 schema = yield* describe(arn);
190 }
191
192 yield* session.note(schema!.schemaArn!);
193 return {
194 schemaArn: schema!.schemaArn!,
195 name: schema!.name!,
196 schema: schema!.schema!,
197 domain: schema!.domain,
198 };
199 }),
200
201 delete: Effect.fn(function* ({ output }) {
202 yield* personalize.deleteSchema({ schemaArn: output.schemaArn }).pipe(
203 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
204 // A schema still referenced by a dataset rejects deletion. The
205 // engine deletes dependent datasets first, but dataset deletion is
206 // asynchronous (DELETE PENDING) and may be draining concurrently
207 // (e.g. via a dataset group's child cascade) — retry bounded while
208 // the referencing datasets finish deleting.
209 Effect.retry({
210 while: (e) => e._tag === "ResourceInUseException",
211 schedule: Schedule.max([
212 Schedule.fixed("3 seconds"),
213 Schedule.recurs(40),
214 ]),
215 }),
216 );
217 }),
218
219 list: () =>
220 personalize.listSchemas.pages({}).pipe(
221 Stream.runCollect,
222 Effect.map((chunk) =>
223 Array.from(chunk).flatMap((page) => page.schemas ?? []),
224 ),
225 Effect.flatMap(
226 Effect.forEach(
227 (summary) =>
228 describe(summary.schemaArn!).pipe(
229 Effect.map((s) =>
230 s
231 ? {
232 schemaArn: s.schemaArn!,
233 name: s.name!,
234 schema: s.schema!,
235 domain: s.domain,
236 }
237 : undefined,
238 ),
239 ),
240 { concurrency: 4 },
241 ),
242 ),
243 Effect.map((items) => items.filter((item) => item !== undefined)),
244 ),
245 };
246 }),
247 );