| 1 | import * as datasync from "@distilled.cloud/aws/datasync"; |
| 2 | import * as Effect from "effect/Effect"; |
| 3 | import * as Schedule from "effect/Schedule"; |
| 4 | import * as Stream from "effect/Stream"; |
| 5 | import { diffTags } from "../../Tags.ts"; |
| 6 | |
| 7 | /** |
| 8 | * A freshly-created IAM role (or a just-attached inline policy) takes a |
| 9 | * while to propagate to DataSync: `CreateLocation*` transiently rejects it |
| 10 | * as `LocationRoleNotAssumable` (patched from `InvalidRequestException` + |
| 11 | * "Invalid IAM role") or fails its location access test as |
| 12 | * `LocationAccessTestFailed` (patched from `InvalidRequestException` + |
| 13 | * "location access test failed"). Bounded retry (~60s), explicitly typed so |
| 14 | * declaration emit never widens the provider layer (see PATTERNS §7). |
| 15 | */ |
| 16 | export const retryWhileRoleNotAssumable = <A, E extends { _tag: string }, R>( |
| 17 | self: Effect.Effect<A, E, R>, |
| 18 | ): Effect.Effect<A, E, R> => |
| 19 | Effect.retry(self, { |
| 20 | while: (e) => |
| 21 | e._tag === "LocationRoleNotAssumable" || |
| 22 | e._tag === "LocationAccessTestFailed", |
| 23 | schedule: Schedule.max([Schedule.fixed("3 seconds"), Schedule.recurs(20)]), |
| 24 | }); |
| 25 | |
| 26 | /** Convert a DataSync tag list to a plain record. */ |
| 27 | export const dsTagsToRecord = ( |
| 28 | tags: readonly datasync.TagListEntry[] | undefined, |
| 29 | ): Record<string, string> => |
| 30 | Object.fromEntries((tags ?? []).map((t) => [t.Key, t.Value ?? ""])); |
| 31 | |
| 32 | /** Read the observed tags currently attached to a DataSync resource. */ |
| 33 | export const readObservedTags = Effect.fn(function* (resourceArn: string) { |
| 34 | const res = yield* datasync.listTagsForResource({ ResourceArn: resourceArn }); |
| 35 | return dsTagsToRecord(res.Tags); |
| 36 | }); |
| 37 | |
| 38 | /** |
| 39 | * Diff observed cloud tags against the desired set and apply the delta. |
| 40 | * DataSync `tagResource` upserts; `untagResource` removes by key. |
| 41 | */ |
| 42 | export const syncTags = Effect.fn(function* ( |
| 43 | resourceArn: string, |
| 44 | observed: Record<string, string>, |
| 45 | desired: Record<string, string>, |
| 46 | ) { |
| 47 | const { upsert, removed } = diffTags(observed, desired); |
| 48 | if (upsert.length > 0) { |
| 49 | yield* datasync.tagResource({ |
| 50 | ResourceArn: resourceArn, |
| 51 | Tags: upsert.map((t) => ({ Key: t.Key, Value: t.Value })), |
| 52 | }); |
| 53 | } |
| 54 | if (removed.length > 0) { |
| 55 | yield* datasync.untagResource({ ResourceArn: resourceArn, Keys: removed }); |
| 56 | } |
| 57 | }); |
| 58 | |
| 59 | /** |
| 60 | * Scan every DataSync location and return the ARN of the one whose |
| 61 | * `LocationUri` matches (ignoring a trailing slash). DataSync has no |
| 62 | * create-idempotency token, so this makes reconcile idempotent across state |
| 63 | * loss. |
| 64 | */ |
| 65 | export const findLocationArnByUri = Effect.fn(function* (expectedUri: string) { |
| 66 | const strip = (u: string) => u.replace(/\/+$/, ""); |
| 67 | const target = strip(expectedUri); |
| 68 | const locations = yield* datasync.listLocations.pages({}).pipe( |
| 69 | Stream.runCollect, |
| 70 | Effect.map((chunk) => Array.from(chunk).flatMap((p) => p.Locations ?? [])), |
| 71 | ); |
| 72 | return locations.find( |
| 73 | (l) => l.LocationUri !== undefined && strip(l.LocationUri) === target, |
| 74 | )?.LocationArn; |
| 75 | }); |