1import * as entityresolution from "@distilled.cloud/aws/entityresolution";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { deepEqual, 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 {
13 retryRolePropagation,
14 syncEntityResolutionTags,
15 toTagRecord,
16} from "./internal.ts";
17
18export interface IdMappingWorkflowProps {
19 /**
20 * Name of the ID mapping workflow. Must be unique per account/region and
21 * match `[a-zA-Z_0-9-]*`. If omitted, a unique name is generated from the
22 * app, stage, and logical ID. Changing the name replaces the workflow.
23 */
24 workflowName?: string;
25 /**
26 * A description of the workflow.
27 */
28 description?: string;
29 /**
30 * The input records to map: for rule-based mapping, entries reference ID
31 * namespace ARNs (`inputSourceARN`) with their role (`type: "SOURCE"` or
32 * `"TARGET"`); for provider-based mapping, entries may reference Glue
33 * tables with a `schemaName`.
34 */
35 inputSourceConfig: entityresolution.IdMappingWorkflowInputSource[];
36 /**
37 * Where mapped output records are written: an S3 path (`outputS3Path`) and
38 * an optional KMS key (`KMSArn`). Optional — rule-based workflows can also
39 * receive per-job output configuration on `StartIdMappingJob`.
40 */
41 outputSourceConfig?: entityresolution.IdMappingWorkflowOutputSource[];
42 /**
43 * How records are mapped: `RULE_BASED` with `ruleBasedProperties`
44 * (definition type, attribute/record matching models) or a `PROVIDER`
45 * service (e.g. LiveRamp).
46 */
47 idMappingTechniques: entityresolution.IdMappingTechniques;
48 /**
49 * Incremental-run configuration. Note the service currently rejects
50 * incremental processing for ID mapping workflows.
51 */
52 incrementalRunConfig?: entityresolution.IdMappingIncrementalRunConfig;
53 /**
54 * The ARN of the IAM role Entity Resolution assumes to read Glue-table
55 * input sources and write the S3 output on your behalf. Optional when all
56 * input sources are ID namespaces that carry their own roles.
57 */
58 roleArn?: string;
59 /**
60 * Tags to apply to the workflow. Merged with internal Alchemy tags.
61 */
62 tags?: Record<string, string>;
63}
64
65export interface IdMappingWorkflow extends Resource<
66 "AWS.EntityResolution.IdMappingWorkflow",
67 IdMappingWorkflowProps,
68 {
69 /**
70 * The name of the ID mapping workflow.
71 */
72 workflowName: string;
73 /**
74 * The ARN of the ID mapping workflow.
75 */
76 workflowArn: string;
77 },
78 never,
79 Providers
80> {}
81
82/**
83 * An AWS Entity Resolution ID mapping workflow — a data-processing job
84 * definition that translates record identifiers between a `SOURCE` and a
85 * `TARGET` ID namespace, using rule-based matching or a provider service.
86 *
87 * The workflow definition itself is cheap and instant; a mapping RUN
88 * (`StartIdMappingJob`) processes the full input and takes many minutes.
89 *
90 * ### Creating ID Mapping Workflows
91 * **Example:** Rule-based ID mapping between two namespaces
92 * ```typescript
93 * import * as AWS from "alchemy/AWS";
94 *
95 * const workflow = yield* AWS.EntityResolution.IdMappingWorkflow("Map", {
96 * inputSourceConfig: [
97 * { inputSourceARN: source.idNamespaceArn, type: "SOURCE" },
98 * { inputSourceARN: target.idNamespaceArn, type: "TARGET" },
99 * ],
100 * idMappingTechniques: {
101 * idMappingType: "RULE_BASED",
102 * ruleBasedProperties: {
103 * ruleDefinitionType: "TARGET",
104 * attributeMatchingModel: "ONE_TO_ONE",
105 * recordMatchingModel: "ONE_SOURCE_TO_ONE_TARGET",
106 * },
107 * },
108 * outputSourceConfig: [
109 * { outputS3Path: `s3://${bucket.bucketName}/idmapping/` },
110 * ],
111 * roleArn: role.roleArn,
112 * });
113 * ```
114 *
115 * @resource
116 */
117export const IdMappingWorkflow = Resource<IdMappingWorkflow>(
118 "AWS.EntityResolution.IdMappingWorkflow",
119);
120
121export const IdMappingWorkflowProvider = () =>
122 Provider.effect(
123 IdMappingWorkflow,
124 Effect.gen(function* () {
125 const createName = Effect.fn(function* (
126 id: string,
127 props: { workflowName?: string | undefined },
128 ) {
129 return (
130 props.workflowName ??
131 (yield* createPhysicalName({ id, maxLength: 255 }))
132 );
133 });
134
135 /** Get an ID mapping workflow by name; typed not-found → undefined. */
136 const getByName = Effect.fn(function* (workflowName: string) {
137 return yield* entityresolution
138 .getIdMappingWorkflow({ workflowName })
139 .pipe(
140 Effect.catchTag("ResourceNotFoundException", () =>
141 Effect.succeed(undefined),
142 ),
143 );
144 });
145
146 return {
147 stables: ["workflowName", "workflowArn"],
148
149 read: Effect.fn(function* ({ id, olds, output }) {
150 const name =
151 output?.workflowName ?? (yield* createName(id, olds ?? {}));
152 const workflow = yield* getByName(name);
153 if (workflow === undefined) return undefined;
154 const attrs = {
155 workflowName: workflow.workflowName,
156 workflowArn: workflow.workflowArn,
157 };
158 // getIdMappingWorkflow returns tags — use the observed tags for
159 // the ownership check.
160 const tags = toTagRecord(workflow.tags);
161 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
162 }),
163
164 diff: Effect.fn(function* ({ id, news, olds }) {
165 if (!isResolved(news)) return undefined;
166 const oldName = yield* createName(id, olds);
167 const newName = yield* createName(id, news);
168 if (oldName !== newName) {
169 return { action: "replace" } as const;
170 }
171 }),
172
173 reconcile: Effect.fn(function* ({ id, news, output, session }) {
174 const name = output?.workflowName ?? (yield* createName(id, news));
175 const internalTags = yield* createInternalTags(id);
176 const desiredTags = { ...news.tags, ...internalTags };
177
178 // 1. Observe — names are unique, so the name IS the identity.
179 let workflow = yield* getByName(name);
180
181 // 2. Ensure — create when missing. A freshly-created IAM role is
182 // transiently rejected while IAM propagates; tolerate the
183 // concurrent-create race, then re-observe: the create response
184 // is a structural subset of the Get shape (it omits
185 // createdAt/updatedAt/tags).
186 if (workflow === undefined) {
187 yield* entityresolution
188 .createIdMappingWorkflow({
189 workflowName: name,
190 description: news.description,
191 inputSourceConfig: news.inputSourceConfig,
192 outputSourceConfig: news.outputSourceConfig,
193 idMappingTechniques: news.idMappingTechniques,
194 incrementalRunConfig: news.incrementalRunConfig,
195 roleArn: news.roleArn,
196 tags: desiredTags,
197 })
198 .pipe(
199 retryRolePropagation,
200 Effect.catchTag("ConflictException", () => Effect.void),
201 );
202 workflow = yield* entityresolution.getIdMappingWorkflow({
203 workflowName: name,
204 });
205 }
206
207 // 3. Sync — everything except the name updates in place via a
208 // full-PUT update. Only call the API on an actual delta between
209 // OBSERVED and desired configuration.
210 const desired = {
211 description: news.description ?? undefined,
212 inputSourceConfig: news.inputSourceConfig,
213 outputSourceConfig: news.outputSourceConfig ?? undefined,
214 idMappingTechniques: news.idMappingTechniques,
215 incrementalRunConfig: news.incrementalRunConfig ?? undefined,
216 roleArn: news.roleArn ?? undefined,
217 };
218 const observed = {
219 description: workflow.description ?? undefined,
220 inputSourceConfig: workflow.inputSourceConfig,
221 outputSourceConfig: workflow.outputSourceConfig ?? undefined,
222 idMappingTechniques: workflow.idMappingTechniques,
223 incrementalRunConfig: workflow.incrementalRunConfig ?? undefined,
224 roleArn: workflow.roleArn ?? undefined,
225 };
226 if (!deepEqual(observed, desired)) {
227 // The update response omits createdAt/updatedAt/tags; the ARN is
228 // stable, so keep the observed Get shape.
229 yield* entityresolution
230 .updateIdMappingWorkflow({
231 workflowName: name,
232 description: news.description,
233 inputSourceConfig: news.inputSourceConfig,
234 outputSourceConfig: news.outputSourceConfig,
235 idMappingTechniques: news.idMappingTechniques,
236 incrementalRunConfig: news.incrementalRunConfig,
237 roleArn: news.roleArn,
238 })
239 .pipe(retryRolePropagation);
240 }
241
242 const workflowArn = workflow.workflowArn;
243
244 // 3b. Sync tags against OBSERVED cloud tags (adoption-safe).
245 yield* syncEntityResolutionTags(workflowArn, desiredTags);
246
247 yield* session.note(workflowArn);
248 return { workflowName: name, workflowArn };
249 }),
250
251 delete: Effect.fn(function* ({ output }) {
252 // deleteIdMappingWorkflow succeeds even when the workflow is
253 // already gone; a running mapping job conflicts — retry briefly.
254 yield* entityresolution
255 .deleteIdMappingWorkflow({ workflowName: output.workflowName })
256 .pipe(
257 Effect.retry({
258 while: (e) => e._tag === "ConflictException",
259 schedule: Schedule.max([
260 Schedule.fixed("2 seconds"),
261 Schedule.recurs(10),
262 ]),
263 }),
264 );
265 }),
266
267 list: () =>
268 entityresolution.listIdMappingWorkflows.items({}).pipe(
269 Stream.map((summary) => ({
270 workflowName: summary.workflowName,
271 workflowArn: summary.workflowArn,
272 })),
273 Stream.runCollect,
274 Effect.map((chunk) => Array.from(chunk)),
275 ),
276 };
277 }),
278 );