1import * as entityresolution from "@distilled.cloud/aws/entityresolution";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import { diffTags } from "../../Tags.ts";
5
6/**
7 * Coerce an Entity Resolution wire tag map (values decode as
8 * `string | undefined`) into a plain `Record<string, string>`.
9 */
10export const toTagRecord = (
11 tags: { [key: string]: string | undefined } | undefined,
12): Record<string, string> =>
13 Object.fromEntries(
14 Object.entries(tags ?? {}).filter(
15 (entry): entry is [string, string] => typeof entry[1] === "string",
16 ),
17 );
18
19/**
20 * Read the observed tags of an Entity Resolution resource by ARN. The
21 * `Get*` operations do NOT return tags even when they exist, so ownership
22 * checks must go through `listTagsForResource`. Tag reads are best-effort —
23 * a failure (e.g. a race with deletion) reports no tags.
24 */
25export const readEntityResolutionTags = Effect.fn(function* (arn: string) {
26 const response = yield* entityresolution
27 .listTagsForResource({ resourceArn: arn })
28 .pipe(Effect.catch(() => Effect.succeed(undefined)));
29 return toTagRecord(response?.tags);
30});
31
32/**
33 * Sync tags on an Entity Resolution resource: diff the OBSERVED cloud tags
34 * against the desired set and apply only the delta.
35 */
36export const syncEntityResolutionTags = Effect.fn(function* (
37 arn: string,
38 desiredTags: Record<string, string>,
39) {
40 const observedTags = yield* readEntityResolutionTags(arn);
41 const { removed, upsert } = diffTags(observedTags, desiredTags);
42 if (upsert.length > 0) {
43 yield* entityresolution.tagResource({
44 resourceArn: arn,
45 tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
46 });
47 }
48 if (removed.length > 0) {
49 yield* entityresolution.untagResource({
50 resourceArn: arn,
51 tagKeys: removed,
52 });
53 }
54});
55
56/**
57 * Retry the IAM-propagation race on workflow create/update: a freshly created
58 * role is transiently rejected until IAM propagates. The race surfaces as
59 * `AccessDeniedException` with either `Exception in assuming the passed
60 * role ...` (the role itself hasn't propagated) or `The service does not
61 * have access to read your data in Glue/S3 ...` (the role resolved but its
62 * just-attached policy hasn't). Bounded; explicitly typed as a pipeable
63 * helper so declaration emit stays clean (inlining `Effect.retry` erases
64 * `E`/`R` to `unknown` for every consumer).
65 */
66export const retryRolePropagation = <A, E, R>(
67 self: Effect.Effect<A, E, R>,
68): Effect.Effect<A, E, R> =>
69 Effect.retry(self, {
70 while: (e) =>
71 e instanceof entityresolution.AccessDeniedException &&
72 typeof e.message === "string" &&
73 (e.message.includes("assuming the passed role") ||
74 e.message.includes("does not have access")),
75 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(20)]),
76 });