1import * as ivschat from "@distilled.cloud/aws/ivschat";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import { diffTags } from "../../Tags.ts";
5
6/**
7 * IVS Chat wire tags allow `undefined` values in the record type —
8 * flatten to a plain string record, dropping malformed entries.
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 IVS Chat resource. Best-effort — a
21 * failure (e.g. a race with deletion) reports no tags.
22 */
23export const readIvsChatTags = Effect.fn(function* (arn: string) {
24 const response = yield* ivschat
25 .listTagsForResource({ resourceArn: arn })
26 .pipe(Effect.catch(() => Effect.succeed(undefined)));
27 return toTagRecord(response?.tags);
28});
29
30/**
31 * Sync tags on an IVS Chat resource: diff the OBSERVED cloud tags
32 * against the desired set and apply only the delta.
33 */
34export const syncIvsChatTags = Effect.fn(function* (
35 arn: string,
36 desiredTags: Record<string, string>,
37) {
38 const observedTags = yield* readIvsChatTags(arn);
39 const { removed, upsert } = diffTags(observedTags, desiredTags);
40 if (upsert.length > 0) {
41 yield* ivschat
42 .tagResource({
43 resourceArn: arn,
44 tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
45 })
46 .pipe(retryWhileThrottled);
47 }
48 if (removed.length > 0) {
49 yield* ivschat
50 .untagResource({ resourceArn: arn, tagKeys: removed })
51 .pipe(retryWhileThrottled);
52 }
53});
54
55/**
56 * Explicitly-typed pipeable retry helper (inlined `Effect.retry` in a
57 * provider op widens the layer type in declaration emit). Retries
58 * `ConflictException` — e.g. mutating or deleting a logging
59 * configuration while it is mid-state-transition — on a bounded schedule.
60 */
61export const retryWhileConflict = <A, E extends { readonly _tag: string }, R>(
62 self: Effect.Effect<A, E, R>,
63): Effect.Effect<A, E, R> =>
64 Effect.retry(self, {
65 while: (e) => e._tag === "ConflictException",
66 schedule: Schedule.max([Schedule.fixed("3 seconds"), Schedule.recurs(8)]),
67 });
68
69/**
70 * Retries the `ValidationException` IVS Chat raises while a freshly-added
71 * `lambda:InvokeFunction` permission for `ivschat.amazonaws.com` is still
72 * propagating — associating a `messageReviewHandler` validates the
73 * permission, and Create/UpdateRoom can race the IAM propagation window
74 * ("Request member: uri failed to satisfy the constraints: invalid lambda
75 * permission"). Bounded (~30s).
76 */
77export const retryWhileHandlerPermissionPropagating = <A, E, R>(
78 self: Effect.Effect<A, E, R>,
79): Effect.Effect<A, E, R> =>
80 Effect.retry(self, {
81 while: (e): boolean =>
82 e instanceof ivschat.ValidationException &&
83 e.message.includes("invalid lambda permission"),
84 schedule: Schedule.max([Schedule.fixed("3 seconds"), Schedule.recurs(10)]),
85 });
86
87/**
88 * Retries `ThrottlingException` on a bounded exponential schedule. IVS
89 * APIs have very low TPS limits (e.g. ListPlaybackKeyPairs is 1 TPS), so
90 * back-to-back reconciles routinely trip 429s.
91 */
92export const retryWhileThrottled = <A, E extends { readonly _tag: string }, R>(
93 self: Effect.Effect<A, E, R>,
94): Effect.Effect<A, E, R> =>
95 Effect.retry(self, {
96 while: (e) => e._tag === "ThrottlingException",
97 schedule: Schedule.max([
98 Schedule.exponential("1 second"),
99 Schedule.recurs(6),
100 ]),
101 });