1import * as databrew from "@distilled.cloud/aws/databrew";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
10import { AWSEnvironment } from "../Environment.ts";
11import type { Providers } from "../Providers.ts";
12import {
13 canonicalJson,
14 cleanMap,
15 databrewArn,
16 fetchObservedTags,
17 retryWhileConflict,
18 syncTags,
19} from "./internal.ts";
20
21/** A single transformation step in a DataBrew recipe. */
22export interface RecipeStep {
23 /** The transformation to apply. */
24 action: {
25 /**
26 * The DataBrew operation name, e.g. `UPPER_CASE`, `REMOVE_VALUES`,
27 * `RENAME`. See the DataBrew recipe-action reference for the full list.
28 */
29 operation: string;
30 /** Operation parameters, e.g. `{ sourceColumn: "name" }`. */
31 parameters?: Record<string, string>;
32 };
33 /** Conditions that must hold for the step to apply to a row. */
34 conditionExpressions?: {
35 /** The condition name, e.g. `LESS_THAN`, `IS_MISSING`. */
36 condition: string;
37 /** The value to compare against (JSON-encoded for lists). */
38 value?: string;
39 /** The column the condition applies to. */
40 targetColumn: string;
41 }[];
42}
43
44export interface RecipeProps {
45 /**
46 * Name of the recipe. If omitted, a unique name is generated. Changing
47 * the name replaces the recipe.
48 * @default a generated physical name
49 */
50 recipeName?: string;
51 /**
52 * A description of the recipe.
53 */
54 description?: string;
55 /**
56 * The ordered transformation steps applied to the data.
57 */
58 steps: RecipeStep[];
59 /**
60 * Publish a new numbered version whenever the working steps change (and
61 * once on create). Recipe jobs consume the latest *published* version by
62 * default, so set this when the recipe feeds a `DataBrew.Job`.
63 * @default false
64 */
65 publish?: boolean;
66 /**
67 * Tags to apply to the recipe. Merged with internal Alchemy tags.
68 */
69 tags?: Record<string, string>;
70}
71
72export interface Recipe extends Resource<
73 "AWS.DataBrew.Recipe",
74 RecipeProps,
75 {
76 /** Name of the recipe. */
77 recipeName: string;
78 /** ARN of the recipe. */
79 recipeArn: string;
80 /** Latest published version (e.g. `"1.0"`), or `"LATEST_WORKING"` if never published. */
81 recipeVersion: string;
82 },
83 {},
84 Providers
85> {}
86
87/**
88 * An AWS Glue DataBrew recipe — an ordered list of data-transformation steps
89 * (rename, filter, case conversion, etc.). Edits modify the `LATEST_WORKING`
90 * version; setting `publish: true` snapshots numbered versions that recipe
91 * jobs consume.
92 * ### Creating Recipes
93 * **Example:** Simple Transform Recipe
94 * ```typescript
95 * import * as AWS from "alchemy/AWS";
96 *
97 * const recipe = yield* AWS.DataBrew.Recipe("Clean", {
98 * description: "normalize customer names",
99 * steps: [
100 * {
101 * action: {
102 * operation: "UPPER_CASE",
103 * parameters: { sourceColumn: "name" },
104 * },
105 * },
106 * ],
107 * });
108 * ```
109 *
110 * **Example:** Published Recipe for Jobs
111 * ```typescript
112 * // publish: true snapshots a numbered version (1.0, 2.0, ...) whenever the
113 * // steps change — recipe jobs run the latest published version by default.
114 * const recipe = yield* AWS.DataBrew.Recipe("Clean", {
115 * publish: true,
116 * steps: [
117 * {
118 * action: {
119 * operation: "REMOVE_VALUES",
120 * parameters: { sourceColumn: "email" },
121 * },
122 * conditionExpressions: [
123 * { condition: "IS_MISSING", targetColumn: "email" },
124 * ],
125 * },
126 * ],
127 * });
128 * ```
129 *
130 * @resource
131 */
132export const Recipe = Resource<Recipe>("AWS.DataBrew.Recipe");
133
134export const buildSteps = (steps: RecipeStep[]) =>
135 steps.map((step) => ({
136 Action: {
137 Operation: step.action.operation,
138 Parameters: step.action.parameters,
139 },
140 ConditionExpressions: step.conditionExpressions?.map((expr) => ({
141 Condition: expr.condition,
142 Value: expr.value,
143 TargetColumn: expr.targetColumn,
144 })),
145 }));
146
147/** Numeric-aware max over published version strings like `"2.0"`. */
148const latestPublished = (versions: (string | undefined)[]) => {
149 const numeric = versions
150 .filter((v): v is string => v !== undefined && !v.startsWith("LATEST"))
151 .sort((a, b) => Number.parseFloat(a) - Number.parseFloat(b));
152 return numeric.at(-1);
153};
154
155export const RecipeProvider = () =>
156 Provider.effect(
157 Recipe,
158 Effect.gen(function* () {
159 const createName = Effect.fn(function* (
160 id: string,
161 props: { recipeName?: string | undefined },
162 ) {
163 return (
164 props.recipeName ??
165 (yield* createPhysicalName({ id, maxLength: 255 }))
166 );
167 });
168
169 const observeWorking = Effect.fn(function* (name: string) {
170 return yield* databrew
171 .describeRecipe({ Name: name, RecipeVersion: "LATEST_WORKING" })
172 .pipe(
173 Effect.catchTag("ResourceNotFoundException", () =>
174 Effect.succeed(undefined),
175 ),
176 );
177 });
178
179 const publishedVersions = Effect.fn(function* (name: string) {
180 const pages = yield* databrew.listRecipeVersions
181 .pages({ Name: name })
182 .pipe(Stream.runCollect);
183 return Array.from(pages)
184 .flatMap((page) => page.Recipes ?? [])
185 .map((r) => r.RecipeVersion);
186 });
187
188 return Recipe.Provider.of({
189 stables: ["recipeName", "recipeArn"],
190
191 list: () =>
192 Effect.gen(function* () {
193 const { accountId, region } = yield* AWSEnvironment.current;
194 // LATEST_WORKING enumerates every recipe (the default listing
195 // only returns recipes with a published version).
196 const workingPages = yield* databrew.listRecipes
197 .pages({ RecipeVersion: "LATEST_WORKING" })
198 .pipe(Stream.runCollect);
199 const publishedPages = yield* databrew.listRecipes
200 .pages({})
201 .pipe(Stream.runCollect);
202 const published = new Map(
203 Array.from(publishedPages)
204 .flatMap((page) => page.Recipes ?? [])
205 .map((r) => [r.Name, r.RecipeVersion]),
206 );
207 return Array.from(workingPages)
208 .flatMap((page) => page.Recipes ?? [])
209 .map((r) => ({
210 recipeName: r.Name,
211 recipeArn:
212 r.ResourceArn ??
213 databrewArn(region, accountId, "recipe", r.Name),
214 recipeVersion: published.get(r.Name) ?? "LATEST_WORKING",
215 }));
216 }),
217
218 read: Effect.fn(function* ({ id, olds, output }) {
219 const { accountId, region } = yield* AWSEnvironment.current;
220 const name =
221 output?.recipeName ?? (yield* createName(id, olds ?? {}));
222 const recipe = yield* observeWorking(name);
223 if (recipe === undefined) return undefined;
224 const arn =
225 recipe.ResourceArn ??
226 databrewArn(region, accountId, "recipe", name);
227 const versions = yield* publishedVersions(name);
228 const attrs = {
229 recipeName: name,
230 recipeArn: arn,
231 recipeVersion: latestPublished(versions) ?? "LATEST_WORKING",
232 };
233 const tags = cleanMap(recipe.Tags);
234 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
235 }),
236
237 diff: Effect.fn(function* ({ id, news, olds }) {
238 if (!isResolved(news)) return undefined;
239 const oldName = yield* createName(id, olds);
240 const newName = yield* createName(id, news);
241 if (oldName !== newName) return { action: "replace" } as const;
242 // steps/description are UpdateRecipe-able; publish is a sync step
243 }),
244
245 reconcile: Effect.fn(function* ({ id, news, output, session }) {
246 const { accountId, region } = yield* AWSEnvironment.current;
247 const name = output?.recipeName ?? (yield* createName(id, news));
248 const internalTags = yield* createInternalTags(id);
249 const desiredTags = { ...news.tags, ...internalTags };
250 const desiredSteps = buildSteps(news.steps);
251
252 // 1. OBSERVE the working copy
253 let recipe = yield* observeWorking(name);
254 let changed = false;
255
256 // 2. ENSURE
257 if (recipe === undefined) {
258 yield* databrew
259 .createRecipe({
260 Name: name,
261 Description: news.description,
262 Steps: desiredSteps,
263 Tags: desiredTags,
264 })
265 .pipe(Effect.catchTag("ConflictException", () => Effect.void));
266 changed = true;
267 recipe = yield* observeWorking(name);
268 } else if (
269 canonicalJson(recipe.Steps ?? []) !== canonicalJson(desiredSteps) ||
270 (recipe.Description ?? undefined) !== news.description
271 ) {
272 // 3. SYNC the working copy only when observed differs
273 yield* databrew.updateRecipe({
274 Name: name,
275 Description: news.description,
276 Steps: desiredSteps,
277 });
278 changed = true;
279 }
280
281 // 3b. SYNC published versions — publish when the working copy
282 // changed, or when publishing is desired but nothing is published.
283 let versions = yield* publishedVersions(name);
284 if (news.publish === true && (changed || versions.length === 0)) {
285 yield* databrew.publishRecipe({
286 Name: name,
287 Description: news.description,
288 });
289 versions = yield* publishedVersions(name);
290 }
291
292 const arn =
293 recipe?.ResourceArn ??
294 databrewArn(region, accountId, "recipe", name);
295
296 // 3c. SYNC TAGS against observed cloud tags
297 const observedTags = yield* fetchObservedTags(arn);
298 yield* syncTags(arn, observedTags, desiredTags);
299
300 yield* session.note(name);
301 return {
302 recipeName: name,
303 recipeArn: arn,
304 recipeVersion: latestPublished(versions) ?? "LATEST_WORKING",
305 };
306 }),
307
308 delete: Effect.fn(function* ({ output }) {
309 const name = output.recipeName;
310 // A recipe is deleted by deleting all of its versions: published
311 // versions first (batched), then LATEST_WORKING.
312 const versions = yield* databrew.listRecipeVersions
313 .pages({ Name: name })
314 .pipe(Stream.runCollect)
315 .pipe(
316 Effect.map((pages) =>
317 Array.from(pages)
318 .flatMap((page) => page.Recipes ?? [])
319 .map((r) => r.RecipeVersion)
320 .filter((v): v is string => v !== undefined),
321 ),
322 Effect.catchTag("ResourceNotFoundException", () =>
323 Effect.succeed([] as string[]),
324 ),
325 );
326 if (versions.length > 0) {
327 yield* retryWhileConflict(
328 databrew.batchDeleteRecipeVersion({
329 Name: name,
330 RecipeVersions: versions,
331 }),
332 ).pipe(
333 Effect.catchTag("ResourceNotFoundException", () =>
334 Effect.succeed(undefined),
335 ),
336 );
337 }
338 yield* retryWhileConflict(
339 databrew.deleteRecipeVersion({
340 Name: name,
341 RecipeVersion: "LATEST_WORKING",
342 }),
343 ).pipe(
344 Effect.catchTag("ResourceNotFoundException", () =>
345 Effect.succeed(undefined),
346 ),
347 );
348 }),
349 });
350 }),
351 );