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 readEntityResolutionTags,
14 retryRolePropagation,
15 syncEntityResolutionTags,
16} from "./internal.ts";
17
18export interface MatchingWorkflowProps {
19 /**
20 * Name of the matching 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 match: each entry names a Glue table
31 * (`inputSourceARN`) and the schema mapping (`schemaName`) that describes
32 * its columns.
33 */
34 inputSourceConfig: entityresolution.InputSource[];
35 /**
36 * Where matched output records are written: an S3 path (`outputS3Path`),
37 * the columns to include (`output`), and an optional KMS key (`KMSArn`).
38 */
39 outputSourceConfig: entityresolution.OutputSource[];
40 /**
41 * How records are matched: `RULE_MATCHING` with `ruleBasedProperties`,
42 * `ML_MATCHING`, or a `PROVIDER` service.
43 */
44 resolutionTechniques: entityresolution.ResolutionTechniques;
45 /**
46 * Incremental-run configuration. Only supported for `RULE_MATCHING`
47 * workflows.
48 */
49 incrementalRunConfig?: entityresolution.IncrementalRunConfig;
50 /**
51 * The ARN of the IAM role Entity Resolution assumes to read the input Glue
52 * tables and write the S3 output on your behalf.
53 */
54 roleArn: string;
55 /**
56 * Tags to apply to the workflow. Merged with internal Alchemy tags.
57 */
58 tags?: Record<string, string>;
59}
60
61export interface MatchingWorkflow extends Resource<
62 "AWS.EntityResolution.MatchingWorkflow",
63 MatchingWorkflowProps,
64 {
65 /**
66 * The name of the matching workflow.
67 */
68 workflowName: string;
69 /**
70 * The ARN of the matching workflow.
71 */
72 workflowArn: string;
73 },
74 never,
75 Providers
76> {}
77
78/**
79 * An AWS Entity Resolution matching workflow — a data-processing job
80 * definition that matches and deduplicates records across Glue table input
81 * sources using rule-based or ML-powered matching, writing matched record
82 * groups to S3.
83 *
84 * The workflow definition itself is cheap and instant; a matching RUN
85 * (`StartMatchingJob`) processes the full input and takes many minutes.
86 *
87 * ### Creating Workflows
88 * **Example:** Rule-based matching over a Glue table
89 * ```typescript
90 * import * as AWS from "alchemy/AWS";
91 *
92 * const workflow = yield* AWS.EntityResolution.MatchingWorkflow("Dedupe", {
93 * inputSourceConfig: [
94 * { inputSourceARN: table.tableArn, schemaName: schema.schemaName },
95 * ],
96 * outputSourceConfig: [
97 * {
98 * outputS3Path: `s3://${bucket.bucketName}/matches/`,
99 * output: [{ name: "id" }, { name: "email" }],
100 * },
101 * ],
102 * resolutionTechniques: {
103 * resolutionType: "RULE_MATCHING",
104 * ruleBasedProperties: {
105 * rules: [{ ruleName: "ByEmail", matchingKeys: ["email"] }],
106 * attributeMatchingModel: "ONE_TO_ONE",
107 * },
108 * },
109 * roleArn: role.roleArn,
110 * });
111 * ```
112 *
113 * **Example:** ML-powered matching
114 * ```typescript
115 * const workflow = yield* AWS.EntityResolution.MatchingWorkflow("MlDedupe", {
116 * inputSourceConfig: [
117 * { inputSourceARN: table.tableArn, schemaName: schema.schemaName },
118 * ],
119 * outputSourceConfig: [
120 * {
121 * outputS3Path: `s3://${bucket.bucketName}/ml-matches/`,
122 * output: [{ name: "id" }],
123 * },
124 * ],
125 * resolutionTechniques: { resolutionType: "ML_MATCHING" },
126 * roleArn: role.roleArn,
127 * });
128 * ```
129 *
130 * @resource
131 */
132export const MatchingWorkflow = Resource<MatchingWorkflow>(
133 "AWS.EntityResolution.MatchingWorkflow",
134);
135
136export const MatchingWorkflowProvider = () =>
137 Provider.effect(
138 MatchingWorkflow,
139 Effect.gen(function* () {
140 const createName = Effect.fn(function* (
141 id: string,
142 props: { workflowName?: string | undefined },
143 ) {
144 return (
145 props.workflowName ??
146 (yield* createPhysicalName({ id, maxLength: 255 }))
147 );
148 });
149
150 /** Get a workflow by name; typed not-found → undefined. */
151 const getByName = Effect.fn(function* (workflowName: string) {
152 return yield* entityresolution
153 .getMatchingWorkflow({ workflowName })
154 .pipe(
155 Effect.catchTag("ResourceNotFoundException", () =>
156 Effect.succeed(undefined),
157 ),
158 );
159 });
160
161 return {
162 stables: ["workflowName", "workflowArn"],
163
164 read: Effect.fn(function* ({ id, olds, output }) {
165 const name =
166 output?.workflowName ?? (yield* createName(id, olds ?? {}));
167 const workflow = yield* getByName(name);
168 if (workflow === undefined) return undefined;
169 const attrs = {
170 workflowName: workflow.workflowName,
171 workflowArn: workflow.workflowArn,
172 };
173 // getMatchingWorkflow does not return tags — read them via
174 // listTagsForResource for the ownership check.
175 const tags = yield* readEntityResolutionTags(workflow.workflowArn);
176 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
177 }),
178
179 diff: Effect.fn(function* ({ id, news, olds }) {
180 if (!isResolved(news)) return undefined;
181 const oldName = yield* createName(id, olds);
182 const newName = yield* createName(id, news);
183 if (oldName !== newName) {
184 return { action: "replace" } as const;
185 }
186 }),
187
188 reconcile: Effect.fn(function* ({ id, news, output, session }) {
189 const name = output?.workflowName ?? (yield* createName(id, news));
190 const internalTags = yield* createInternalTags(id);
191 const desiredTags = { ...news.tags, ...internalTags };
192
193 // 1. Observe — names are unique, so the name IS the identity.
194 let workflow = yield* getByName(name);
195
196 // 2. Ensure — create when missing. A freshly-created IAM role is
197 // transiently rejected while IAM propagates, so retry that
198 // typed AccessDeniedException on a bounded schedule; tolerate
199 // the concurrent-create race, then re-observe: the create
200 // response is a structural subset of the Get shape (it omits
201 // createdAt/updatedAt).
202 if (workflow === undefined) {
203 yield* entityresolution
204 .createMatchingWorkflow({
205 workflowName: name,
206 description: news.description,
207 inputSourceConfig: news.inputSourceConfig,
208 outputSourceConfig: news.outputSourceConfig,
209 resolutionTechniques: news.resolutionTechniques,
210 incrementalRunConfig: news.incrementalRunConfig,
211 roleArn: news.roleArn,
212 tags: desiredTags,
213 })
214 .pipe(
215 retryRolePropagation,
216 Effect.catchTag("ConflictException", () => Effect.void),
217 );
218 workflow = yield* entityresolution.getMatchingWorkflow({
219 workflowName: name,
220 });
221 }
222
223 // 3. Sync — everything except the name updates in place via a
224 // full-PUT update. Only call the API on an actual delta
225 // between OBSERVED and desired configuration.
226 const desired = {
227 description: news.description ?? undefined,
228 inputSourceConfig: news.inputSourceConfig,
229 outputSourceConfig: news.outputSourceConfig,
230 resolutionTechniques: news.resolutionTechniques,
231 incrementalRunConfig: news.incrementalRunConfig ?? undefined,
232 roleArn: news.roleArn,
233 };
234 const observed = {
235 description: workflow.description ?? undefined,
236 inputSourceConfig: workflow.inputSourceConfig,
237 outputSourceConfig: workflow.outputSourceConfig,
238 resolutionTechniques: workflow.resolutionTechniques,
239 incrementalRunConfig: workflow.incrementalRunConfig ?? undefined,
240 roleArn: workflow.roleArn,
241 };
242 if (!deepEqual(observed, desired)) {
243 // The update response omits workflowArn/createdAt/updatedAt;
244 // the ARN is stable, so keep the observed Get shape.
245 yield* entityresolution
246 .updateMatchingWorkflow({
247 workflowName: name,
248 description: news.description,
249 inputSourceConfig: news.inputSourceConfig,
250 outputSourceConfig: news.outputSourceConfig,
251 resolutionTechniques: news.resolutionTechniques,
252 incrementalRunConfig: news.incrementalRunConfig,
253 roleArn: news.roleArn,
254 })
255 .pipe(retryRolePropagation);
256 }
257
258 const workflowArn = workflow.workflowArn;
259
260 // 3b. Sync tags against OBSERVED cloud tags (adoption-safe).
261 yield* syncEntityResolutionTags(workflowArn, desiredTags);
262
263 yield* session.note(workflowArn);
264 return { workflowName: name, workflowArn };
265 }),
266
267 delete: Effect.fn(function* ({ output }) {
268 // deleteMatchingWorkflow succeeds even when the workflow is
269 // already gone; a running matching job conflicts — retry briefly.
270 yield* entityresolution
271 .deleteMatchingWorkflow({ workflowName: output.workflowName })
272 .pipe(
273 Effect.retry({
274 while: (e) => e._tag === "ConflictException",
275 schedule: Schedule.max([
276 Schedule.fixed("2 seconds"),
277 Schedule.recurs(10),
278 ]),
279 }),
280 );
281 }),
282
283 list: () =>
284 entityresolution.listMatchingWorkflows.items({}).pipe(
285 Stream.map((summary) => ({
286 workflowName: summary.workflowName,
287 workflowArn: summary.workflowArn,
288 })),
289 Stream.runCollect,
290 Effect.map((chunk) => Array.from(chunk)),
291 ),
292 };
293 }),
294 );