1import * as mediapackagev2 from "@distilled.cloud/aws/mediapackagev2";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import * as Stream from "effect/Stream";
5import { diffTags } from "../../Tags.ts";
6
7/**
8 * Coerce a MediaPackage wire tag map (values are `string | undefined`) into a
9 * plain `Record<string, string>`, dropping any undefined values.
10 */
11export const toMpTagRecord = (
12 tags: { [key: string]: string | undefined } | undefined,
13): Record<string, string> =>
14 Object.fromEntries(
15 Object.entries(tags ?? {}).filter(
16 (entry): entry is [string, string] => typeof entry[1] === "string",
17 ),
18 );
19
20/**
21 * Sync tags on a MediaPackage resource: diff the OBSERVED cloud tags against
22 * the desired set and apply only the delta.
23 */
24export const syncMpTags = Effect.fn(function* (
25 arn: string,
26 observedTags: Record<string, string>,
27 desiredTags: Record<string, string>,
28) {
29 const { removed, upsert } = diffTags(observedTags, desiredTags);
30 if (upsert.length > 0) {
31 yield* mediapackagev2.tagResource({
32 ResourceArn: arn,
33 Tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
34 });
35 }
36 if (removed.length > 0) {
37 yield* mediapackagev2.untagResource({
38 ResourceArn: arn,
39 TagKeys: removed,
40 });
41 }
42});
43
44/**
45 * Explicitly-typed pipeable retry helper. Inlining `Effect.retry` in a
46 * provider lifecycle op leaks `Retry.Return`'s conditional into declaration
47 * emit and widens the provider layer to `unknown` R for every consumer of
48 * `AWS.providers()`.
49 *
50 * MediaPackage rejects deleting a parent whose children are still being
51 * cleaned up (and concurrent mutations of the same resource) with
52 * `ConflictException`; both settle within seconds.
53 */
54export const retryWhileMpConflict = <A, E extends { readonly _tag: string }, R>(
55 self: Effect.Effect<A, E, R>,
56): Effect.Effect<A, E, R> =>
57 Effect.retry(self, {
58 while: (e) => e._tag === "ConflictException",
59 schedule: Schedule.max([Schedule.fixed("3 seconds"), Schedule.recurs(15)]),
60 });
61
62/**
63 * Enumerate every channel group in the account/region.
64 */
65export const listAllChannelGroups = Effect.fn(function* () {
66 return yield* mediapackagev2.listChannelGroups.pages({}).pipe(
67 Stream.runCollect,
68 Effect.map((chunk) =>
69 Array.from(chunk).flatMap((page) => page.Items ?? []),
70 ),
71 );
72});
73
74/**
75 * Enumerate every channel in a channel group; a missing group yields `[]`
76 * (the typed not-found), so callers can race against a concurrent delete.
77 */
78export const listGroupChannels = Effect.fn(function* (
79 channelGroupName: string,
80) {
81 return yield* mediapackagev2.listChannels
82 .pages({ ChannelGroupName: channelGroupName })
83 .pipe(
84 Stream.runCollect,
85 Effect.map((chunk) =>
86 Array.from(chunk).flatMap((page) => page.Items ?? []),
87 ),
88 Effect.catchTag("ResourceNotFoundException", () =>
89 Effect.succeed([] as mediapackagev2.ChannelListConfiguration[]),
90 ),
91 );
92});
93
94/**
95 * Enumerate every origin endpoint in a channel; a missing channel/group
96 * yields `[]` (the typed not-found).
97 */
98export const listChannelEndpoints = Effect.fn(function* (
99 channelGroupName: string,
100 channelName: string,
101) {
102 return yield* mediapackagev2.listOriginEndpoints
103 .pages({ ChannelGroupName: channelGroupName, ChannelName: channelName })
104 .pipe(
105 Stream.runCollect,
106 Effect.map((chunk) =>
107 Array.from(chunk).flatMap((page) => page.Items ?? []),
108 ),
109 Effect.catchTag("ResourceNotFoundException", () =>
110 Effect.succeed([] as mediapackagev2.OriginEndpointListConfiguration[]),
111 ),
112 );
113});
114
115/**
116 * Delete a channel after reaping its origin endpoints. MediaPackage refuses
117 * to delete a channel that still has endpoints, so an orphan sweep that
118 * targets a channel without enumerating its children would Conflict until
119 * the retry budget ran out and leak the channel. A normal stack destroy
120 * deletes the endpoints first, so the reap observes nothing and is free.
121 * Every step is idempotent (deleting a missing endpoint/channel succeeds).
122 */
123export const deleteChannelWithEndpoints = Effect.fn(function* (
124 channelGroupName: string,
125 channelName: string,
126) {
127 const endpoints = yield* listChannelEndpoints(channelGroupName, channelName);
128 yield* Effect.forEach(
129 endpoints,
130 (endpoint) =>
131 mediapackagev2.deleteOriginEndpoint({
132 ChannelGroupName: channelGroupName,
133 ChannelName: channelName,
134 OriginEndpointName: endpoint.OriginEndpointName,
135 }),
136 { concurrency: 5, discard: true },
137 );
138 // The channel transiently rejects deletion with a Conflict while its
139 // just-deleted endpoints are cleaned up.
140 yield* mediapackagev2
141 .deleteChannel({
142 ChannelGroupName: channelGroupName,
143 ChannelName: channelName,
144 })
145 .pipe(retryWhileMpConflict);
146});
147
148/**
149 * Compare two IAM policy documents for semantic equality: parse both as JSON
150 * and compare the normalized serialization, so whitespace/key-order changes
151 * introduced by AWS never register as drift. Falls back to strict string
152 * equality when either side is not valid JSON.
153 */
154export const policiesEqual = (
155 a: string | undefined,
156 b: string | undefined,
157): boolean => {
158 if (a === undefined || b === undefined) return a === b;
159 try {
160 return JSON.stringify(JSON.parse(a)) === JSON.stringify(JSON.parse(b));
161 } catch {
162 return a === b;
163 }
164};
165
166/**
167 * Structural "desired is a subset of observed" comparison used for drift
168 * detection. Only keys present (and defined) in `desired` are compared, so
169 * server-side defaults on the observed state never register as drift. Arrays
170 * must match pairwise and in length so removed/added items are detected.
171 */
172export const matchesDesired = (
173 desired: unknown,
174 observed: unknown,
175): boolean => {
176 if (desired === undefined) return true;
177 if (desired === null) return observed === null;
178 if (desired instanceof Date) {
179 return observed instanceof Date && desired.getTime() === observed.getTime();
180 }
181 if (Array.isArray(desired)) {
182 if (!Array.isArray(observed) || observed.length !== desired.length) {
183 return false;
184 }
185 return desired.every((item, i) => matchesDesired(item, observed[i]));
186 }
187 if (typeof desired === "object") {
188 if (
189 typeof observed !== "object" ||
190 observed === null ||
191 Array.isArray(observed)
192 ) {
193 return false;
194 }
195 return Object.entries(desired as Record<string, unknown>).every(
196 ([key, value]) =>
197 value === undefined
198 ? true
199 : matchesDesired(value, (observed as Record<string, unknown>)[key]),
200 );
201 }
202 return desired === observed;
203};