1import * as appsync from "@distilled.cloud/aws/appsync";
2import * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import { pipe } from "effect/Function";
5import * as Schedule from "effect/Schedule";
6import { diffTags, normalizeTags } from "../../Tags.ts";
7
8/**
9 * AppSync serializes control-plane mutations per API; concurrent resolver /
10 * data-source / schema operations surface as `ConcurrentModificationException`.
11 * Ride out the contention window with a bounded retry (~40s).
12 *
13 * The helper carries an EXPLICIT return annotation so the conditional type
14 * of `Effect.retry` never leaks into declaration emit (which would widen the
15 * provider layer to `unknown` for every consumer of `AWS.providers()`).
16 */
17export const retryConcurrentModification = <A, E extends { _tag: string }, R>(
18 effect: Effect.Effect<A, E, R>,
19): Effect.Effect<A, E, R> =>
20 Effect.retry(effect, {
21 while: (error: E) => error._tag === "ConcurrentModificationException",
22 schedule: Schedule.max([
23 pipe(
24 Schedule.exponential(Duration.seconds(1), 2),
25 Schedule.modifyDelay(({ duration }) =>
26 Effect.succeed(
27 Duration.isGreaterThan(duration, Duration.seconds(8))
28 ? Duration.seconds(8)
29 : duration,
30 ),
31 ),
32 ),
33 Schedule.recurs(8),
34 ]),
35 }) as Effect.Effect<A, E, R>;
36
37/**
38 * IAM changes (fresh service roles, updated trust policies) propagate to
39 * AppSync eventually; a `createDataSource` right after `createRole` can
40 * transiently fail with a `BadRequestException` complaining that AppSync
41 * is not authorized to assume the role. Bounded retry (~20s), scoped to
42 * role-assumption messages so genuine validation errors fail fast.
43 */
44export const retryWhileRolePropagates = <
45 A,
46 E extends { _tag: string; message?: string },
47 R,
48>(
49 effect: Effect.Effect<A, E, R>,
50): Effect.Effect<A, E, R> =>
51 Effect.retry(effect, {
52 while: (error: E) =>
53 error._tag === "BadRequestException" &&
54 /assume|role|authorized/i.test(error.message ?? ""),
55 schedule: Schedule.max([
56 Schedule.fixed(Duration.seconds(2)),
57 Schedule.recurs(10),
58 ]),
59 }) as Effect.Effect<A, E, R>;
60
61/**
62 * AppSync data-source and pipeline-function names must match
63 * `[_A-Za-z][_0-9A-Za-z]*` (no dashes). Deterministic physical names are
64 * generated with dashes, so sanitize into the allowed alphabet.
65 */
66export const sanitizeAppSyncName = (name: string): string => {
67 const sanitized = name.replaceAll(/[^_0-9A-Za-z]/g, "_");
68 return /^[_A-Za-z]/.test(sanitized) ? sanitized : `_${sanitized}`;
69};
70
71/**
72 * Normalize the wire tag map (values may be `undefined`) to a plain record.
73 */
74export const tagRecord = (
75 tags: { [key: string]: string | undefined } | undefined,
76): Record<string, string> =>
77 Object.fromEntries(
78 Object.entries(tags ?? {}).filter(
79 (entry): entry is [string, string] => entry[1] !== undefined,
80 ),
81 );
82
83/**
84 * Diff observed tags against desired tags and apply only the delta via the
85 * AppSync `tagResource`/`untagResource` operations.
86 */
87export const syncAppSyncTags = Effect.fn(function* ({
88 resourceArn,
89 oldTags,
90 newTags,
91}: {
92 resourceArn: string;
93 oldTags: Record<string, string>;
94 newTags: Record<string, string>;
95}) {
96 const { removed, upsert } = diffTags(oldTags, newTags);
97 if (removed.length > 0) {
98 yield* appsync
99 .untagResource({ resourceArn, tagKeys: removed })
100 .pipe(Effect.catchTag("NotFoundException", () => Effect.void));
101 }
102 if (upsert.length > 0) {
103 yield* appsync.tagResource({
104 resourceArn,
105 tags: normalizeTags(upsert),
106 });
107 }
108});