1import * as kv from "@distilled.cloud/aws/kinesis-video";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5
6/**
7 * Raised when a Kinesis Video stream or signaling channel fails to reach
8 * `ACTIVE` within the bounded polling budget after a create/update.
9 */
10export class KinesisVideoNotConverged extends Data.TaggedError(
11 "KinesisVideoNotConverged",
12)<{
13 readonly resource: string;
14 readonly status: string | undefined;
15}> {}
16
17/**
18 * Raised when `GetSignalingChannelEndpoint` returns no endpoint for the
19 * requested protocol.
20 */
21export class SignalingEndpointUnavailable extends Data.TaggedError(
22 "SignalingEndpointUnavailable",
23)<{
24 readonly channelArn: string;
25 readonly protocol: string;
26}> {}
27
28/**
29 * Bounded retry through transient `ResourceInUseException` states — e.g.
30 * deleting a stream or channel that is still `CREATING`/`UPDATING`, or
31 * re-creating one whose previous incarnation is still `DELETING`.
32 *
33 * Expressed as an explicitly-typed module-scope helper: inlining
34 * `Effect.retry` in lifecycle code leaves its conditional return type
35 * unresolved in the provider's declaration emit, which widens the
36 * `AWS.providers()` layer type for every downstream consumer.
37 */
38export const retryWhileResourceInUse = <A, E extends { _tag: string }, R>(
39 self: Effect.Effect<A, E, R>,
40): Effect.Effect<A, E, R> =>
41 Effect.retry(self, {
42 while: (e) => e._tag === "ResourceInUseException",
43 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(30)]),
44 });
45
46/**
47 * Bounded retry through the transient states a mutation can hit while a
48 * previous async transition settles: `ResourceInUseException` and the
49 * synthetic `StreamNotActive` (Kinesis Video overloads
50 * `ResourceNotFoundException` with "not found or not active" while a
51 * stream/channel is CREATING/UPDATING — patched into a typed tag in
52 * distilled). Explicitly typed for the declaration-emit reason above.
53 */
54export const retryWhileSettling = <A, E extends { _tag: string }, R>(
55 self: Effect.Effect<A, E, R>,
56): Effect.Effect<A, E, R> =>
57 Effect.retry(self, {
58 while: (e) =>
59 e._tag === "ResourceInUseException" || e._tag === "StreamNotActive",
60 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(15)]),
61 });
62
63/**
64 * Bounded retry through `ResourceNotFoundException` — a freshly-created
65 * stream/channel can be invisible to `Describe*` for a few seconds
66 * (eventual consistency). Explicitly typed for the same declaration-emit
67 * reason as above.
68 */
69export const retryWhileNotFound = <A, E extends { _tag: string }, R>(
70 self: Effect.Effect<A, E, R>,
71): Effect.Effect<A, E, R> =>
72 Effect.retry(self, {
73 while: (e) => e._tag === "ResourceNotFoundException",
74 schedule: Schedule.max([Schedule.fixed("1 second"), Schedule.recurs(10)]),
75 });
76
77/**
78 * Repeat a describe poll until `done` holds (bounded — streams and channels
79 * typically become active within seconds). Explicitly typed for the
80 * declaration-emit reason above.
81 */
82const untilConverged = <A, E, R>(
83 self: Effect.Effect<A, E, R>,
84 done: (a: A) => boolean,
85 times = 30,
86): Effect.Effect<A, E, R> =>
87 Effect.repeat(self, {
88 schedule: Schedule.spaced("2 seconds"),
89 until: done,
90 times,
91 });
92
93/**
94 * Poll `DescribeStream` until the stream is `ACTIVE`. Tolerates the brief
95 * post-create window where the stream is not yet describable.
96 *
97 * Mutations (`UpdateStream`/`UpdateDataRetention`) are asynchronous AND bump
98 * the stream version — a describe issued immediately after can still show
99 * the pre-mutation `ACTIVE` state with the stale version. Pass
100 * `previousVersion` after a mutation to also wait for the version bump, so
101 * the returned `Version` is safe to use in the next versioned call.
102 */
103export const waitForStreamActive = Effect.fn(
104 "AWS.KinesisVideo.waitForStreamActive",
105)(function* (streamName: string, previousVersion?: string) {
106 const done = (info: kv.StreamInfo | undefined): boolean =>
107 info?.Status === "ACTIVE" &&
108 (previousVersion === undefined || info.Version !== previousVersion);
109 const info = yield* untilConverged(
110 retryWhileNotFound(kv.describeStream({ StreamName: streamName })).pipe(
111 Effect.map((r) => r.StreamInfo),
112 ),
113 done,
114 );
115 if (!done(info)) {
116 return yield* Effect.fail(
117 new KinesisVideoNotConverged({
118 resource: streamName,
119 status: info?.Status,
120 }),
121 );
122 }
123 return info!;
124});
125
126/**
127 * Poll `DescribeSignalingChannel` until the channel is `ACTIVE`. Tolerates
128 * the brief post-create window where the channel is not yet describable.
129 * Pass `previousVersion` after `UpdateSignalingChannel` to also wait for the
130 * version bump (see {@link waitForStreamActive}).
131 */
132export const waitForChannelActive = Effect.fn(
133 "AWS.KinesisVideo.waitForChannelActive",
134)(function* (channelName: string, previousVersion?: string) {
135 const done = (info: kv.ChannelInfo | undefined): boolean =>
136 info?.ChannelStatus === "ACTIVE" &&
137 (previousVersion === undefined || info.Version !== previousVersion);
138 const info = yield* untilConverged(
139 retryWhileNotFound(
140 kv.describeSignalingChannel({ ChannelName: channelName }),
141 ).pipe(Effect.map((r) => r.ChannelInfo)),
142 done,
143 );
144 if (!done(info)) {
145 return yield* Effect.fail(
146 new KinesisVideoNotConverged({
147 resource: channelName,
148 status: info?.ChannelStatus,
149 }),
150 );
151 }
152 return info!;
153});
154
155/**
156 * Poll `DescribeStream` until the stream is fully purged (NotFound).
157 * A stream in `DELETING` blocks re-creation of the same name with
158 * `ResourceInUseException`, so reconcilers wait it out before recreating.
159 */
160export const waitForStreamGone = Effect.fn(
161 "AWS.KinesisVideo.waitForStreamGone",
162)(function* (streamName: string) {
163 const info = yield* untilConverged(
164 kv.describeStream({ StreamName: streamName }).pipe(
165 Effect.map((r) => r.StreamInfo),
166 Effect.catchTag("ResourceNotFoundException", () =>
167 Effect.succeed(undefined),
168 ),
169 ),
170 (i) => i === undefined,
171 // full purge can take noticeably longer than activation (~90s bound)
172 45,
173 );
174 if (info !== undefined) {
175 return yield* Effect.fail(
176 new KinesisVideoNotConverged({
177 resource: streamName,
178 status: info.Status,
179 }),
180 );
181 }
182});
183
184/**
185 * Poll `DescribeSignalingChannel` until the channel is fully purged
186 * (NotFound) — see {@link waitForStreamGone}.
187 */
188export const waitForChannelGone = Effect.fn(
189 "AWS.KinesisVideo.waitForChannelGone",
190)(function* (channelName: string) {
191 const info = yield* untilConverged(
192 kv.describeSignalingChannel({ ChannelName: channelName }).pipe(
193 Effect.map((r) => r.ChannelInfo),
194 Effect.catchTag("ResourceNotFoundException", () =>
195 Effect.succeed(undefined),
196 ),
197 ),
198 (i) => i === undefined,
199 45,
200 );
201 if (info !== undefined) {
202 return yield* Effect.fail(
203 new KinesisVideoNotConverged({
204 resource: channelName,
205 status: info.ChannelStatus,
206 }),
207 );
208 }
209});
210
211/**
212 * Drop `undefined` values from a distilled tag map (`{ [key]: string |
213 * undefined }`) so it can be diffed as a plain `Record<string, string>`.
214 */
215export const compactTags = (
216 tags: { [key: string]: string | undefined } | undefined,
217): Record<string, string> => {
218 const out: Record<string, string> = {};
219 for (const [key, value] of Object.entries(tags ?? {})) {
220 if (value !== undefined) {
221 out[key] = value;
222 }
223 }
224 return out;
225};
226
227// Per-stream data endpoints are stable for the life of the stream — cache
228// them per (arn, API) so repeated data-plane calls skip the extra
229// GetDataEndpoint round-trip.
230const dataEndpointCache = new Map<string, string>();
231
232/**
233 * Discover (and cache) the per-stream data endpoint for `apiName`.
234 * Parameterized over the `GetDataEndpoint` operation so binding layers can
235 * pass an operation captured via yield-first (`yield* op`) whose calls are
236 * requirement-free.
237 */
238export const discoverDataEndpoint = <E, R>(
239 streamArn: string,
240 apiName: kv.APIName,
241 getDataEndpoint: (
242 input: kv.GetDataEndpointInput,
243 ) => Effect.Effect<kv.GetDataEndpointOutput, E, R>,
244): Effect.Effect<string, E, R> =>
245 Effect.gen(function* () {
246 const key = `${streamArn}#${apiName}`;
247 const cached = dataEndpointCache.get(key);
248 if (cached !== undefined) {
249 return cached;
250 }
251 const response = yield* getDataEndpoint({
252 StreamARN: streamArn,
253 APIName: apiName,
254 });
255 const url = response.DataEndpoint!;
256 dataEndpointCache.set(key, url);
257 return url;
258 });
259
260// Signaling endpoints are likewise stable per (channel, protocol, role).
261const signalingEndpointCache = new Map<string, string>();
262
263/**
264 * Discover (and cache) the per-channel signaling endpoint for `protocol` +
265 * `role` via `GetSignalingChannelEndpoint`. Parameterized over the operation
266 * for the same yield-first reason as {@link discoverDataEndpoint}.
267 */
268export const discoverSignalingEndpoint = <E, R>(
269 channelArn: string,
270 protocol: kv.ChannelProtocol,
271 role: kv.ChannelRole,
272 getSignalingChannelEndpoint: (
273 input: kv.GetSignalingChannelEndpointInput,
274 ) => Effect.Effect<kv.GetSignalingChannelEndpointOutput, E, R>,
275): Effect.Effect<string, E | SignalingEndpointUnavailable, R> =>
276 Effect.gen(function* () {
277 const key = `${channelArn}#${protocol}#${role}`;
278 const cached = signalingEndpointCache.get(key);
279 if (cached !== undefined) {
280 return cached;
281 }
282 const response = yield* getSignalingChannelEndpoint({
283 ChannelARN: channelArn,
284 SingleMasterChannelEndpointConfiguration: {
285 Protocols: [protocol],
286 Role: role,
287 },
288 });
289 const endpoint = (response.ResourceEndpointList ?? []).find(
290 (item) => item.Protocol === protocol,
291 )?.ResourceEndpoint;
292 if (endpoint === undefined) {
293 return yield* Effect.fail(
294 new SignalingEndpointUnavailable({ channelArn, protocol }),
295 );
296 }
297 signalingEndpointCache.set(key, endpoint);
298 return endpoint;
299 });