1import * as databrew from "@distilled.cloud/aws/databrew";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import { diffTags } from "../../Tags.ts";
5
6/** ARN partition — the testing account is on the standard AWS partition. */
7const PARTITION = "aws";
8
9/** `arn:aws:databrew:{region}:{account}:{type}/{name}` */
10export const databrewArn = (
11 region: string,
12 accountId: string,
13 type: "dataset" | "recipe" | "project" | "job" | "ruleset" | "schedule",
14 name: string,
15) => `arn:${PARTITION}:databrew:${region}:${accountId}:${type}/${name}`;
16
17/**
18 * Fetch the observed DataBrew tags for a resource ARN as a plain record.
19 * Tolerate a missing/untaggable resource as `{}`.
20 */
21export const fetchObservedTags = Effect.fn("AWS.DataBrew.fetchObservedTags")(
22 function* (resourceArn: string) {
23 const response = yield* databrew
24 .listTagsForResource({ ResourceArn: resourceArn })
25 .pipe(Effect.catch(() => Effect.succeed({ Tags: undefined })));
26 return cleanMap(response.Tags);
27 },
28);
29
30/**
31 * Sync a DataBrew resource's tags: diff OBSERVED cloud tags against desired
32 * and apply the delta via `TagResource` (map upsert) / `UntagResource`
33 * (removed keys).
34 */
35export const syncTags = Effect.fn("AWS.DataBrew.syncTags")(function* (
36 resourceArn: string,
37 observed: Record<string, string>,
38 desired: Record<string, string>,
39) {
40 const { upsert, removed } = diffTags(observed, desired);
41 if (upsert.length > 0) {
42 yield* databrew.tagResource({
43 ResourceArn: resourceArn,
44 Tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
45 });
46 }
47 if (removed.length > 0) {
48 yield* databrew.untagResource({
49 ResourceArn: resourceArn,
50 TagKeys: removed,
51 });
52 }
53});
54
55/**
56 * Bounded retry through `ConflictException` — DataBrew rejects mutations
57 * while a conflicting operation is in flight (e.g. deleting a job whose run
58 * is still starting, or racing reconciles on the same name). The budget is
59 * ~2 minutes because deletes run in parallel: a dataset/recipe delete can
60 * legitimately conflict ("is used in job …") for the whole window in which
61 * the associated job is still waiting for its last run to wind down before
62 * deleting itself. Explicitly typed so the conditional `Retry.Return` type
63 * does not leak into the provider's declaration emit and widen
64 * `AWS.providers()` for downstream consumers.
65 */
66export const retryWhileConflict = <A, E extends { _tag: string }, R>(
67 self: Effect.Effect<A, E, R>,
68): Effect.Effect<A, E, R> =>
69 Effect.retry(self, {
70 while: (e) => e._tag === "ConflictException",
71 schedule: Schedule.max([Schedule.fixed("5 seconds"), Schedule.recurs(24)]),
72 });
73
74/**
75 * Bounded retry through `DataBrewRoleNotAssumable` — a freshly-created IAM
76 * role is not yet visible to DataBrew (IAM propagation), which surfaces as a
77 * message-discriminated `ValidationException` ("DataBrew is not a trusted
78 * entity for the data access role ...", patched into a typed tag).
79 * Explicitly typed for the same declaration-emit reason as above.
80 */
81export const retryWhileRoleNotAssumable = <A, E extends { _tag: string }, R>(
82 self: Effect.Effect<A, E, R>,
83): Effect.Effect<A, E, R> =>
84 Effect.retry(self, {
85 // A freshly-attached S3 policy on the role also propagates lazily and
86 // surfaces as AccessDeniedException ("Access denied to s3:GetObject for
87 // arn:...:role/... Error: Forbidden") while DataBrew validates the
88 // dataset source — retry both bounded.
89 while: (e) =>
90 e._tag === "DataBrewRoleNotAssumable" ||
91 e._tag === "AccessDeniedException",
92 schedule: Schedule.max([Schedule.fixed("5 seconds"), Schedule.recurs(12)]),
93 });
94
95/** DataBrew free-form maps arrive with `undefined` values erased. */
96export const cleanMap = (
97 map: Record<string, string | undefined> | undefined,
98): Record<string, string> =>
99 Object.fromEntries(
100 Object.entries(map ?? {}).filter(
101 (entry): entry is [string, string] => entry[1] !== undefined,
102 ),
103 );
104
105/**
106 * Canonical (key-sorted) JSON — used to compare an observed wire-shaped
107 * structure against the desired one so no-op updates can skip API calls
108 * and recipe publishing only happens when the working copy actually changed.
109 */
110export const canonicalJson = (value: unknown): string =>
111 JSON.stringify(sortKeys(value));
112
113const sortKeys = (value: unknown): unknown => {
114 if (Array.isArray(value)) return value.map(sortKeys);
115 if (value !== null && typeof value === "object") {
116 return Object.fromEntries(
117 Object.entries(value as Record<string, unknown>)
118 .filter(([, v]) => v !== undefined)
119 .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0))
120 .map(([k, v]) => [k, sortKeys(v)]),
121 );
122 }
123 return value;
124};