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 syncEntityResolutionTags,
15} from "./internal.ts";
16
17export interface SchemaMappingProps {
18 /**
19 * Name of the schema mapping. Must be unique per account/region and match
20 * `[a-zA-Z_0-9-]*`. If omitted, a unique name is generated from the app,
21 * stage, and logical ID. Changing the name replaces the schema mapping.
22 */
23 schemaName?: string;
24 /**
25 * A description of the schema mapping.
26 */
27 description?: string;
28 /**
29 * The columns of the input records table and how Entity Resolution should
30 * interpret them: the attribute `type` (e.g. `UNIQUE_ID`, `EMAIL_ADDRESS`,
31 * `NAME`, `PHONE_NUMBER`), an optional `matchKey` used by rule-based
32 * matching, and optional `groupName`/`subType`/`hashed` metadata.
33 *
34 * Mutable in place — but a schema mapping becomes immutable while it is
35 * referenced by a matching workflow.
36 */
37 mappedInputFields: entityresolution.SchemaInputAttribute[];
38 /**
39 * Tags to apply to the schema mapping. Merged with internal Alchemy tags.
40 */
41 tags?: Record<string, string>;
42}
43
44export interface SchemaMapping extends Resource<
45 "AWS.EntityResolution.SchemaMapping",
46 SchemaMappingProps,
47 {
48 /**
49 * The name of the schema mapping.
50 */
51 schemaName: string;
52 /**
53 * The ARN of the schema mapping.
54 */
55 schemaArn: string;
56 },
57 never,
58 Providers
59> {}
60
61/**
62 * An AWS Entity Resolution schema mapping — the schema of an input customer
63 * records table. It tells Entity Resolution the attribute type of each column
64 * (name, email, phone, unique id, …) and which columns rule-based matching
65 * compares via `matchKey`.
66 *
67 * ### Creating Schema Mappings
68 * **Example:** Customer records schema
69 * ```typescript
70 * import * as AWS from "alchemy/AWS";
71 *
72 * const schema = yield* AWS.EntityResolution.SchemaMapping("Customers", {
73 * mappedInputFields: [
74 * { fieldName: "id", type: "UNIQUE_ID" },
75 * { fieldName: "email", type: "EMAIL_ADDRESS", matchKey: "email" },
76 * { fieldName: "name", type: "NAME", matchKey: "name" },
77 * ],
78 * });
79 * ```
80 *
81 * ### Matching Workflows
82 * **Example:** Use the schema in a matching workflow input source
83 * ```typescript
84 * const workflow = yield* AWS.EntityResolution.MatchingWorkflow("Dedupe", {
85 * inputSourceConfig: [
86 * { inputSourceARN: table.tableArn, schemaName: schema.schemaName },
87 * ],
88 * // ...
89 * });
90 * ```
91 *
92 * @resource
93 */
94export const SchemaMapping = Resource<SchemaMapping>(
95 "AWS.EntityResolution.SchemaMapping",
96);
97
98export const SchemaMappingProvider = () =>
99 Provider.effect(
100 SchemaMapping,
101 Effect.gen(function* () {
102 const createName = Effect.fn(function* (
103 id: string,
104 props: { schemaName?: string | undefined },
105 ) {
106 return (
107 props.schemaName ??
108 (yield* createPhysicalName({ id, maxLength: 255 }))
109 );
110 });
111
112 /** Get a schema mapping by name; typed not-found → undefined. */
113 const getByName = Effect.fn(function* (schemaName: string) {
114 return yield* entityresolution
115 .getSchemaMapping({ schemaName })
116 .pipe(
117 Effect.catchTag("ResourceNotFoundException", () =>
118 Effect.succeed(undefined),
119 ),
120 );
121 });
122
123 return {
124 stables: ["schemaName", "schemaArn"],
125
126 read: Effect.fn(function* ({ id, olds, output }) {
127 const name =
128 output?.schemaName ?? (yield* createName(id, olds ?? {}));
129 const mapping = yield* getByName(name);
130 if (mapping === undefined) return undefined;
131 const attrs = {
132 schemaName: mapping.schemaName,
133 schemaArn: mapping.schemaArn,
134 };
135 // getSchemaMapping does not return tags — read them via
136 // listTagsForResource for the ownership check.
137 const tags = yield* readEntityResolutionTags(mapping.schemaArn);
138 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
139 }),
140
141 diff: Effect.fn(function* ({ id, news, olds }) {
142 if (!isResolved(news)) return undefined;
143 const oldName = yield* createName(id, olds);
144 const newName = yield* createName(id, news);
145 if (oldName !== newName) {
146 return { action: "replace" } as const;
147 }
148 }),
149
150 reconcile: Effect.fn(function* ({ id, news, output, session }) {
151 const name = output?.schemaName ?? (yield* createName(id, news));
152 const internalTags = yield* createInternalTags(id);
153 const desiredTags = { ...news.tags, ...internalTags };
154
155 // 1. Observe — names are unique, so the name IS the identity.
156 let mapping = yield* getByName(name);
157
158 // 2. Ensure — create when missing; tolerate the concurrent-create
159 // race, then re-observe: the create response is a structural
160 // subset of the Get shape (it omits createdAt/updatedAt/
161 // hasWorkflows).
162 if (mapping === undefined) {
163 yield* entityresolution
164 .createSchemaMapping({
165 schemaName: name,
166 description: news.description,
167 mappedInputFields: news.mappedInputFields,
168 tags: desiredTags,
169 })
170 .pipe(Effect.catchTag("ConflictException", () => Effect.void));
171 mapping = yield* entityresolution.getSchemaMapping({
172 schemaName: name,
173 });
174 }
175
176 // 3. Sync — description and mapped fields update in place (fails
177 // with a typed ConflictException while a workflow references
178 // the schema). Only call the API on an actual delta.
179 if (
180 !deepEqual(mapping.mappedInputFields, news.mappedInputFields) ||
181 (mapping.description || undefined) !==
182 (news.description ?? undefined)
183 ) {
184 // The update response omits createdAt/updatedAt/hasWorkflows;
185 // name and ARN are stable, so keep the observed Get shape.
186 yield* entityresolution.updateSchemaMapping({
187 schemaName: name,
188 description: news.description,
189 mappedInputFields: news.mappedInputFields,
190 });
191 }
192
193 // 3b. Sync tags against OBSERVED cloud tags (adoption-safe).
194 yield* syncEntityResolutionTags(mapping.schemaArn, desiredTags);
195
196 yield* session.note(mapping.schemaArn);
197 return {
198 schemaName: mapping.schemaName,
199 schemaArn: mapping.schemaArn,
200 };
201 }),
202
203 delete: Effect.fn(function* ({ output }) {
204 // deleteSchemaMapping succeeds even when the mapping is already
205 // gone; it conflicts while a matching workflow still references
206 // the schema (the engine deletes dependents first — retry the
207 // eventual-consistency window).
208 yield* entityresolution
209 .deleteSchemaMapping({ schemaName: output.schemaName })
210 .pipe(
211 Effect.retry({
212 while: (e) => e._tag === "ConflictException",
213 schedule: Schedule.max([
214 Schedule.fixed("2 seconds"),
215 Schedule.recurs(10),
216 ]),
217 }),
218 );
219 }),
220
221 list: () =>
222 entityresolution.listSchemaMappings.items({}).pipe(
223 Stream.map((summary) => ({
224 schemaName: summary.schemaName,
225 schemaArn: summary.schemaArn,
226 })),
227 Stream.runCollect,
228 Effect.map((chunk) => Array.from(chunk)),
229 ),
230 };
231 }),
232 );