1import * as AWS from "@/AWS";
2import {
3 IdMappingWorkflow,
4 IdNamespace,
5 MatchingWorkflow,
6 SchemaMapping,
7} from "@/AWS/EntityResolution";
8import { toTagRecord } from "@/AWS/EntityResolution/internal.ts";
9import { Database, Table } from "@/AWS/Glue";
10import { Role } from "@/AWS/IAM";
11import { Bucket } from "@/AWS/S3";
12import * as Output from "@/Output";
13import * as Test from "@/Test/Alchemy";
14import * as entityresolution from "@distilled.cloud/aws/entityresolution";
15import * as s3 from "@distilled.cloud/aws/s3";
16import { expect } from "alchemy-test";
17import * as Effect from "effect/Effect";
18import * as Schedule from "effect/Schedule";
19
20const { test } = Test.make({ providers: AWS.providers() });
21
22// Typed-error probe: proves the SDK decodes Entity Resolution errors as
23// typed tags. A missing workflow surfaces as ResourceNotFoundException.
24test.provider(
25 "getMatchingWorkflow on a nonexistent name fails with a typed error",
26 () =>
27 Effect.gen(function* () {
28 const error = yield* Effect.flip(
29 entityresolution.getMatchingWorkflow({
30 workflowName: "does-not-exist-alchemy-probe",
31 }),
32 );
33 expect(error._tag).toBe("ResourceNotFoundException");
34 }),
35 { tags: ["provider:aws", "provider:aws:entityresolution", "live"] },
36);
37
38const CUSTOMER_FIELDS: entityresolution.SchemaInputAttribute[] = [
39 { fieldName: "id", type: "UNIQUE_ID" },
40 { fieldName: "email", type: "EMAIL_ADDRESS", matchKey: "email" },
41 { fieldName: "name", type: "NAME", matchKey: "name" },
42];
43
44/**
45 * The role Entity Resolution assumes to read the Glue input table (and its
46 * underlying S3 data) and write matched output to S3.
47 */
48const workflowRole = () =>
49 Role("EntityResolutionRole", {
50 assumeRolePolicyDocument: {
51 Version: "2012-10-17",
52 Statement: [
53 {
54 Effect: "Allow",
55 Principal: { Service: "entityresolution.amazonaws.com" },
56 Action: ["sts:AssumeRole"],
57 },
58 ],
59 },
60 inlinePolicies: {
61 workflow: {
62 Version: "2012-10-17",
63 Statement: [
64 {
65 Effect: "Allow",
66 Action: [
67 "glue:GetDatabase",
68 "glue:GetTable",
69 "glue:GetPartition",
70 "glue:GetPartitions",
71 "glue:GetSchema",
72 "glue:GetSchemaVersion",
73 ],
74 Resource: ["*"],
75 },
76 {
77 Effect: "Allow",
78 Action: ["s3:GetObject", "s3:ListBucket", "s3:PutObject"],
79 Resource: ["*"],
80 },
81 ],
82 },
83 },
84 });
85
86/** The full stack under test: bucket → Glue table → schema → workflow. */
87const workflowStack = (variant: "v1" | "v2") =>
88 Effect.gen(function* () {
89 const bucket = yield* Bucket("ErBucket", { forceDestroy: true });
90 const role = yield* workflowRole();
91 const database = yield* Database("ErDb", {});
92 const table = yield* Table("Customers", {
93 databaseName: database.databaseName,
94 storageDescriptor: {
95 location: Output.interpolate`s3://${bucket.bucketName}/input/`,
96 inputFormat: "org.apache.hadoop.mapred.TextInputFormat",
97 outputFormat:
98 "org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat",
99 serdeInfo: {
100 serializationLibrary:
101 "org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe",
102 parameters: { "field.delim": ",", "skip.header.line.count": "1" },
103 },
104 columns: [
105 { name: "id", type: "string" },
106 { name: "email", type: "string" },
107 { name: "name", type: "string" },
108 ],
109 },
110 parameters: { classification: "csv" },
111 });
112 const schema = yield* SchemaMapping("CustomersSchema", {
113 description: "Customer records schema",
114 mappedInputFields: CUSTOMER_FIELDS,
115 tags: { Environment: "test" },
116 });
117 const workflow = yield* MatchingWorkflow("Dedupe", {
118 description:
119 variant === "v1"
120 ? "Match customers by email"
121 : "Match by email or name",
122 inputSourceConfig: [
123 { inputSourceARN: table.tableArn, schemaName: schema.schemaName },
124 ],
125 outputSourceConfig: [
126 {
127 outputS3Path: Output.interpolate`s3://${bucket.bucketName}/matches/`,
128 output: [{ name: "id" }, { name: "email" }, { name: "name" }],
129 },
130 ],
131 resolutionTechniques: {
132 resolutionType: "RULE_MATCHING",
133 ruleBasedProperties: {
134 rules:
135 variant === "v1"
136 ? [{ ruleName: "ByEmail", matchingKeys: ["email"] }]
137 : [
138 { ruleName: "ByEmail", matchingKeys: ["email"] },
139 { ruleName: "ByName", matchingKeys: ["name"] },
140 ],
141 attributeMatchingModel: "ONE_TO_ONE",
142 },
143 },
144 roleArn: role.roleArn,
145 tags: { Environment: "test" },
146 });
147 return { bucket, role, database, table, schema, workflow };
148 });
149
150// Schema mappings and matching workflow DEFINITIONS are cheap and instant —
151// the live lifecycle runs ungated with out-of-band verification. The
152// matching RUN (StartMatchingJob) takes many minutes and is gated below.
153test.provider(
154 "create, update, destroy a schema mapping and matching workflow",
155 (stack) =>
156 Effect.gen(function* () {
157 yield* stack.destroy();
158
159 const created = yield* stack.deploy(workflowStack("v1"));
160
161 expect(created.schema.schemaArn).toContain(":schemamapping/");
162 expect(created.workflow.workflowArn).toContain(":matchingworkflow/");
163
164 // Out-of-band verification via distilled.
165 const observedSchema = yield* entityresolution.getSchemaMapping({
166 schemaName: created.schema.schemaName,
167 });
168 expect(observedSchema.mappedInputFields).toEqual(CUSTOMER_FIELDS);
169 // Entity Resolution Get* ops do not return tags — verify via
170 // listTagsForResource.
171 const schemaTags = yield* entityresolution.listTagsForResource({
172 resourceArn: created.schema.schemaArn,
173 });
174 expect(toTagRecord(schemaTags.tags)["alchemy::id"]).toBe(
175 "CustomersSchema",
176 );
177 expect(toTagRecord(schemaTags.tags).Environment).toBe("test");
178
179 const observedWorkflow = yield* entityresolution.getMatchingWorkflow({
180 workflowName: created.workflow.workflowName,
181 });
182 expect(observedWorkflow.description).toBe("Match customers by email");
183 expect(observedWorkflow.roleArn).toBe(created.role.roleArn);
184 expect(observedWorkflow.inputSourceConfig[0]?.schemaName).toBe(
185 created.schema.schemaName,
186 );
187 expect(
188 observedWorkflow.resolutionTechniques.ruleBasedProperties?.rules.map(
189 (r) => r.ruleName,
190 ),
191 ).toEqual(["ByEmail"]);
192 const workflowTags = yield* entityresolution.listTagsForResource({
193 resourceArn: created.workflow.workflowArn,
194 });
195 expect(toTagRecord(workflowTags.tags)["alchemy::id"]).toBe("Dedupe");
196
197 // Update in place: workflow description + an extra matching rule.
198 // (The schema mapping is left untouched — it is immutable while the
199 // workflow references it.)
200 const updated = yield* stack.deploy(workflowStack("v2"));
201 expect(updated.workflow.workflowArn).toBe(created.workflow.workflowArn);
202 expect(updated.schema.schemaArn).toBe(created.schema.schemaArn);
203
204 const reobserved = yield* entityresolution.getMatchingWorkflow({
205 workflowName: created.workflow.workflowName,
206 });
207 expect(reobserved.description).toBe("Match by email or name");
208 expect(
209 reobserved.resolutionTechniques.ruleBasedProperties?.rules.map(
210 (r) => r.ruleName,
211 ),
212 ).toEqual(["ByEmail", "ByName"]);
213
214 // Destroy and verify deletion out-of-band.
215 yield* stack.destroy();
216 const workflowError = yield* Effect.flip(
217 entityresolution.getMatchingWorkflow({
218 workflowName: created.workflow.workflowName,
219 }),
220 );
221 expect(workflowError._tag).toBe("ResourceNotFoundException");
222 const schemaError = yield* Effect.flip(
223 entityresolution.getSchemaMapping({
224 schemaName: created.schema.schemaName,
225 }),
226 );
227 expect(schemaError._tag).toBe("ResourceNotFoundException");
228 }),
229 {
230 tags: [
231 "provider:aws",
232 "provider:aws:entityresolution",
233 "provider:aws:glue",
234 "provider:aws:iam",
235 "provider:aws:s3",
236 "live",
237 ],
238 timeout: 300_000,
239 },
240);
241
242// The matching RUN processes the full input through the Entity Resolution
243// service and takes several minutes even for a 4-row input — gated so an
244// explicitly-opted-in run can exercise it end-to-end.
245test.provider.skipIf(!process.env.AWS_TEST_ENTITYRESOLUTION_RUN)(
246 "run a rule-based matching job end-to-end (gated)",
247 (stack) =>
248 Effect.gen(function* () {
249 yield* stack.destroy();
250
251 const created = yield* stack.deploy(workflowStack("v1"));
252
253 // Seed input records — two obvious duplicates by email.
254 yield* s3.putObject({
255 Bucket: created.bucket.bucketName,
256 Key: "input/customers.csv",
257 Body: new TextEncoder().encode(
258 "id,email,name\n" +
259 "1,jane@example.com,Jane Doe\n" +
260 "2,jane@example.com,Jane D\n" +
261 "3,bob@example.com,Bob Roe\n" +
262 "4,alice@example.com,Alice Poe\n",
263 ),
264 ContentType: "text/csv",
265 });
266
267 const { jobId } = yield* entityresolution.startMatchingJob({
268 workflowName: created.workflow.workflowName,
269 });
270
271 const job = yield* entityresolution
272 .getMatchingJob({
273 workflowName: created.workflow.workflowName,
274 jobId,
275 })
276 .pipe(
277 Effect.repeat({
278 schedule: Schedule.spaced("15 seconds"),
279 until: (j) => j.status === "SUCCEEDED" || j.status === "FAILED",
280 times: 36,
281 }),
282 );
283 expect(job.status).toBe("SUCCEEDED");
284 expect(job.metrics?.inputRecords).toBe(4);
285
286 yield* stack.destroy();
287 }),
288 {
289 tags: [
290 "provider:aws",
291 "provider:aws:entityresolution",
292 "provider:aws:glue",
293 "provider:aws:iam",
294 "provider:aws:s3",
295 "live",
296 ],
297 timeout: 900_000,
298 },
299);