1import * as kv from "@distilled.cloud/aws/kinesis-video";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Stream_ from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
11import { toWireSeconds } from "../../Util/Duration.ts";
12import type { Providers } from "../Providers.ts";
13import {
14 compactTags,
15 retryWhileResourceInUse,
16 retryWhileSettling,
17 waitForChannelActive,
18 waitForChannelGone,
19} from "./internal.ts";
20
21export interface SignalingChannelProps {
22 /**
23 * Name of the signaling channel. Changing the name replaces the channel.
24 * Must be unique per account and region.
25 * @default a generated physical name
26 */
27 channelName?: string;
28 /**
29 * The type of the channel. `SINGLE_MASTER` is the only fully supported
30 * type. Changing the type replaces the channel.
31 * @default "SINGLE_MASTER"
32 */
33 type?: "SINGLE_MASTER" | "FULL_MESH";
34 /**
35 * How long an undelivered signaling message is retained (5–120 seconds).
36 * Accepts any `Duration.Input` (e.g. `"30 seconds"`, `Duration.seconds(30)`;
37 * a bare number is milliseconds); the wire unit is whole seconds. Updated
38 * in place.
39 * @default 60 seconds
40 */
41 messageTtl?: Duration.Input;
42 /**
43 * Tags to apply to the channel. Merged with internal Alchemy tags.
44 */
45 tags?: Record<string, string>;
46}
47
48export interface SignalingChannel extends Resource<
49 "AWS.KinesisVideo.SignalingChannel",
50 SignalingChannelProps,
51 {
52 channelName: string;
53 channelArn: string;
54 },
55 {},
56 Providers
57> {}
58
59/**
60 * An Amazon Kinesis Video Streams WebRTC signaling channel — the rendezvous
61 * point that WebRTC master and viewer peers use to exchange SDP offers,
62 * answers, and ICE candidates.
63 *
64 * `messageTtl` is updated in place; changing `channelName` or `type`
65 * replaces the channel.
66 * ### Creating Channels
67 * **Example:** Basic Signaling Channel
68 * ```typescript
69 * import * as AWS from "alchemy/AWS";
70 *
71 * const channel = yield* AWS.KinesisVideo.SignalingChannel("Doorbell");
72 * ```
73 *
74 * **Example:** Channel with Message TTL
75 * ```typescript
76 * const channel = yield* AWS.KinesisVideo.SignalingChannel("Doorbell", {
77 * messageTtl: "30 seconds",
78 * tags: { Environment: "production" },
79 * });
80 * ```
81 *
82 * ### WebRTC Connectivity
83 * **Example:** ICE Server Configuration
84 * ```typescript
85 * // init
86 * const getIceServers = yield* AWS.KinesisVideo.GetIceServerConfig(channel);
87 *
88 * // runtime — TURN URIs + short-lived credentials for a WebRTC peer
89 * const { IceServerList } = yield* getIceServers({ ClientId: "viewer-1" });
90 * ```
91 *
92 * @resource
93 */
94export const SignalingChannel = Resource<SignalingChannel>(
95 "AWS.KinesisVideo.SignalingChannel",
96);
97
98export const SignalingChannelProvider = () =>
99 Provider.effect(
100 SignalingChannel,
101 Effect.gen(function* () {
102 const createName = Effect.fn(function* (
103 id: string,
104 props: { channelName?: string | undefined },
105 ) {
106 return (
107 props.channelName ??
108 (yield* createPhysicalName({ id, maxLength: 256 }))
109 );
110 });
111
112 const observeChannel = Effect.fn(function* (channelName: string) {
113 return yield* kv
114 .describeSignalingChannel({ ChannelName: channelName })
115 .pipe(
116 Effect.map((r) => r.ChannelInfo),
117 Effect.catchTag("ResourceNotFoundException", () =>
118 Effect.succeed(undefined),
119 ),
120 );
121 });
122
123 const fetchObservedTags = Effect.fn(function* (channelArn: string) {
124 return yield* kv.listTagsForResource({ ResourceARN: channelArn }).pipe(
125 Effect.map((r) => compactTags(r.Tags)),
126 Effect.catchTag("ResourceNotFoundException", () =>
127 Effect.succeed({} as Record<string, string>),
128 ),
129 );
130 });
131
132 return SignalingChannel.Provider.of({
133 stables: ["channelName", "channelArn"],
134
135 list: () =>
136 Effect.gen(function* () {
137 const pages = yield* kv.listSignalingChannels
138 .pages({})
139 .pipe(Stream_.runCollect);
140 return Array.from(pages)
141 .flatMap((page) => page.ChannelInfoList ?? [])
142 .filter(
143 (info) =>
144 info.ChannelName !== undefined &&
145 info.ChannelARN !== undefined,
146 )
147 .map((info) => ({
148 channelName: info.ChannelName!,
149 channelArn: info.ChannelARN!,
150 }));
151 }),
152
153 read: Effect.fn(function* ({ id, olds, output }) {
154 const channelName =
155 output?.channelName ?? (yield* createName(id, olds ?? {}));
156 const info = yield* observeChannel(channelName);
157 if (
158 info?.ChannelARN === undefined ||
159 info.ChannelStatus === "DELETING"
160 ) {
161 return undefined;
162 }
163 const attrs = {
164 channelName,
165 channelArn: info.ChannelARN,
166 };
167 const tags = yield* fetchObservedTags(info.ChannelARN);
168 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
169 }),
170
171 diff: Effect.fn(function* ({ id, news, olds }) {
172 if (!isResolved(news)) return undefined;
173 const oldName = yield* createName(id, olds ?? {});
174 const newName = yield* createName(id, news ?? {});
175 if (oldName !== newName) {
176 return { action: "replace" } as const;
177 }
178 // The channel type is bound at creation and cannot change in place.
179 const oldType = olds?.type ?? "SINGLE_MASTER";
180 const newType = news?.type ?? "SINGLE_MASTER";
181 if (oldType !== newType) {
182 return { action: "replace" } as const;
183 }
184 // messageTtl/tags fall through to the default update path.
185 }),
186
187 reconcile: Effect.fn(function* ({ id, news, output, session }) {
188 // `news` is undefined when the resource is declared without props
189 // (`SignalingChannel("Id")`) — normalize WITHOUT a destructuring
190 // default (defaults widen the inferred Props type).
191 const props = news ?? {};
192 const channelName =
193 output?.channelName ?? (yield* createName(id, props));
194 const internalTags = yield* createInternalTags(id);
195 const desiredTags = { ...props.tags, ...internalTags };
196 // Wire unit is whole seconds (MessageTtlSeconds).
197 const desiredTtl = toWireSeconds(props.messageTtl);
198
199 // 1. OBSERVE
200 let info = yield* observeChannel(channelName);
201
202 // A previous incarnation still in DELETING blocks re-creation of
203 // the same name — wait for the purge to finish, then recreate.
204 if (info?.ChannelStatus === "DELETING") {
205 yield* session.note(
206 `waiting for previous channel ${channelName} to finish deleting...`,
207 );
208 yield* waitForChannelGone(channelName);
209 info = undefined;
210 }
211
212 // 2. ENSURE — create if missing; tolerate the concurrent-create
213 // race and wait for ACTIVE (creation is async but fast).
214 if (info === undefined) {
215 yield* retryWhileResourceInUse(
216 kv.createSignalingChannel({
217 ChannelName: channelName,
218 ChannelType: props.type ?? "SINGLE_MASTER",
219 SingleMasterConfiguration:
220 desiredTtl !== undefined
221 ? { MessageTtlSeconds: desiredTtl }
222 : undefined,
223 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
224 Key,
225 Value,
226 })),
227 }),
228 ).pipe(
229 Effect.catchTag("ResourceInUseException", () =>
230 Effect.succeed({ ChannelARN: undefined }),
231 ),
232 );
233 yield* session.note(`creating signaling channel ${channelName}...`);
234 info = yield* waitForChannelActive(channelName);
235 } else if (info.ChannelStatus !== "ACTIVE") {
236 info = yield* waitForChannelActive(channelName);
237 }
238 const channelArn = info.ChannelARN!;
239
240 // 3. SYNC — message TTL (versioned update; only on observed drift)
241 const observedTtl = info.SingleMasterConfiguration?.MessageTtlSeconds;
242 if (desiredTtl !== undefined && desiredTtl !== observedTtl) {
243 yield* retryWhileSettling(
244 kv.updateSignalingChannel({
245 ChannelARN: channelArn,
246 CurrentVersion: info.Version!,
247 SingleMasterConfiguration: { MessageTtlSeconds: desiredTtl },
248 }),
249 );
250 // UpdateSignalingChannel is async: wait for the version bump AND
251 // ACTIVE before any further versioned mutation.
252 info = yield* waitForChannelActive(channelName, info.Version);
253 }
254
255 // 3b. SYNC TAGS — diff against OBSERVED cloud tags
256 const observedTags = yield* fetchObservedTags(channelArn);
257 const { upsert, removed } = diffTags(observedTags, desiredTags);
258 if (upsert.length > 0) {
259 yield* retryWhileSettling(
260 kv.tagResource({
261 ResourceARN: channelArn,
262 Tags: upsert.map((t) => ({ Key: t.Key, Value: t.Value })),
263 }),
264 );
265 }
266 if (removed.length > 0) {
267 yield* retryWhileSettling(
268 kv.untagResource({
269 ResourceARN: channelArn,
270 TagKeyList: removed,
271 }),
272 );
273 }
274
275 yield* session.note(channelArn);
276 return { channelName, channelArn };
277 }),
278
279 delete: Effect.fn(function* ({ output }) {
280 // A CREATING/UPDATING channel rejects deletion with
281 // ResourceInUseException — retry through the transition (bounded).
282 yield* retryWhileResourceInUse(
283 kv.deleteSignalingChannel({ ChannelARN: output.channelArn }),
284 ).pipe(
285 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
286 );
287 yield* waitForChannelGone(output.channelName);
288 }),
289 });
290 }),
291 );