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 { toWireHours } from "../../Util/Duration.ts";
12import type { Providers } from "../Providers.ts";
13import {
14 compactTags,
15 retryWhileResourceInUse,
16 retryWhileSettling,
17 waitForStreamActive,
18 waitForStreamGone,
19} from "./internal.ts";
20
21export interface StreamProps {
22 /**
23 * Name of the video stream. Changing the name replaces the stream.
24 * Must be unique per account and region.
25 * @default a generated physical name
26 */
27 streamName?: string;
28 /**
29 * Name of the device that writes to the stream. Kinesis Video Streams
30 * does not currently use this name, but it is stored with the stream
31 * metadata and can be updated in place.
32 */
33 deviceName?: string;
34 /**
35 * Media type of the stream as a MIME type (e.g. `video/h264`). Consumers
36 * can use it to determine how to process the stream. Updated in place.
37 */
38 mediaType?: string;
39 /**
40 * The ID (or alias/ARN) of the KMS key used to encrypt stream data.
41 * Changing the key replaces the stream.
42 * @default the AWS-managed `aws/kinesisvideo` key
43 */
44 kmsKeyId?: string;
45 /**
46 * How long stream data is retained. Zero duration means no retention —
47 * data is only available live. Accepts any `Duration.Input` (e.g.
48 * `"24 hours"`, `Duration.hours(24)`; a bare number is milliseconds); the
49 * wire unit is whole hours. Adjusted in place via `UpdateDataRetention`.
50 * @default 0
51 */
52 dataRetention?: Duration.Input;
53 /**
54 * Tags to apply to the stream. Merged with internal Alchemy tags.
55 */
56 tags?: Record<string, string>;
57}
58
59export interface Stream extends Resource<
60 "AWS.KinesisVideo.Stream",
61 StreamProps,
62 {
63 streamName: string;
64 streamArn: string;
65 },
66 {},
67 Providers
68> {}
69
70/**
71 * An Amazon Kinesis Video Stream for ingesting, storing, and consuming
72 * live video.
73 *
74 * Stream creation is asynchronous — the provider waits (bounded) for the
75 * stream to become `ACTIVE` before returning. `deviceName` and `mediaType`
76 * are updated in place; `dataRetention` converges via `UpdateDataRetention`;
77 * changing `streamName` or `kmsKeyId` replaces the stream.
78 * ### Creating Streams
79 * **Example:** Basic Video Stream
80 * ```typescript
81 * import * as AWS from "alchemy/AWS";
82 *
83 * const stream = yield* AWS.KinesisVideo.Stream("Camera");
84 * ```
85 *
86 * **Example:** Stream with Retention and Media Type
87 * ```typescript
88 * const stream = yield* AWS.KinesisVideo.Stream("Camera", {
89 * mediaType: "video/h264",
90 * dataRetention: "24 hours",
91 * tags: { Environment: "production" },
92 * });
93 * ```
94 *
95 * ### Reading Media
96 * Bind data-plane read operations in the init phase and use them in
97 * runtime handlers. The bindings resolve the per-stream data endpoint
98 * (`GetDataEndpoint`) automatically.
99 *
100 * **Example:** HLS Playback URL
101 * ```typescript
102 * // init
103 * const getHls = yield* AWS.KinesisVideo.GetHLSStreamingSessionURL(stream);
104 *
105 * // runtime
106 * const { HLSStreamingSessionURL } = yield* getHls({ PlaybackMode: "LIVE" });
107 * ```
108 *
109 * **Example:** Raw Media
110 * ```typescript
111 * // init
112 * const getMedia = yield* AWS.KinesisVideo.GetMedia(stream);
113 *
114 * // runtime
115 * const media = yield* getMedia({
116 * StartSelector: { StartSelectorType: "EARLIEST" },
117 * });
118 * ```
119 *
120 * @resource
121 */
122export const Stream = Resource<Stream>("AWS.KinesisVideo.Stream");
123
124export const StreamProvider = () =>
125 Provider.effect(
126 Stream,
127 Effect.gen(function* () {
128 const createName = Effect.fn(function* (
129 id: string,
130 props: { streamName?: string | undefined },
131 ) {
132 return (
133 props.streamName ??
134 (yield* createPhysicalName({ id, maxLength: 256 }))
135 );
136 });
137
138 const observeStream = Effect.fn(function* (streamName: string) {
139 return yield* kv.describeStream({ StreamName: streamName }).pipe(
140 Effect.map((r) => r.StreamInfo),
141 Effect.catchTag("ResourceNotFoundException", () =>
142 Effect.succeed(undefined),
143 ),
144 );
145 });
146
147 const fetchObservedTags = Effect.fn(function* (streamArn: string) {
148 return yield* kv.listTagsForStream({ StreamARN: streamArn }).pipe(
149 Effect.map((r) => compactTags(r.Tags)),
150 Effect.catchTag("ResourceNotFoundException", () =>
151 Effect.succeed({} as Record<string, string>),
152 ),
153 );
154 });
155
156 return Stream.Provider.of({
157 stables: ["streamName", "streamArn"],
158
159 list: () =>
160 Effect.gen(function* () {
161 const pages = yield* kv.listStreams
162 .pages({})
163 .pipe(Stream_.runCollect);
164 return Array.from(pages)
165 .flatMap((page) => page.StreamInfoList ?? [])
166 .filter(
167 (info) =>
168 info.StreamName !== undefined && info.StreamARN !== undefined,
169 )
170 .map((info) => ({
171 streamName: info.StreamName!,
172 streamArn: info.StreamARN!,
173 }));
174 }),
175
176 read: Effect.fn(function* ({ id, olds, output }) {
177 const streamName =
178 output?.streamName ?? (yield* createName(id, olds ?? {}));
179 const info = yield* observeStream(streamName);
180 if (info?.StreamARN === undefined || info.Status === "DELETING") {
181 return undefined;
182 }
183 const attrs = {
184 streamName,
185 streamArn: info.StreamARN,
186 };
187 const tags = yield* fetchObservedTags(info.StreamARN);
188 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
189 }),
190
191 diff: Effect.fn(function* ({ id, news, olds }) {
192 if (!isResolved(news)) return undefined;
193 const oldName = yield* createName(id, olds ?? {});
194 const newName = yield* createName(id, news ?? {});
195 if (oldName !== newName) {
196 return { action: "replace" } as const;
197 }
198 // The KMS key is bound at creation and cannot be changed in place.
199 if (olds?.kmsKeyId !== news?.kmsKeyId) {
200 return { action: "replace" } as const;
201 }
202 // deviceName/mediaType/dataRetention/tags fall through to
203 // the default update path (reconcile applies the deltas).
204 }),
205
206 reconcile: Effect.fn(function* ({ id, news, output, session }) {
207 // `news` is undefined when the resource is declared without props
208 // (`Stream("Id")`) — normalize WITHOUT a destructuring default
209 // (defaults widen the inferred Props type).
210 const props = news ?? {};
211 const streamName =
212 output?.streamName ?? (yield* createName(id, props));
213 const internalTags = yield* createInternalTags(id);
214 const desiredTags = { ...props.tags, ...internalTags };
215 // Wire unit is whole hours (DataRetentionInHours).
216 const desiredRetention = toWireHours(props.dataRetention);
217
218 // 1. OBSERVE — cloud state is authoritative
219 let info = yield* observeStream(streamName);
220
221 // A previous incarnation still in DELETING blocks re-creation of
222 // the same name — wait for the purge to finish, then recreate.
223 if (info?.Status === "DELETING") {
224 yield* session.note(
225 `waiting for previous stream ${streamName} to finish deleting...`,
226 );
227 yield* waitForStreamGone(streamName);
228 info = undefined;
229 }
230
231 // 2. ENSURE — create if missing; a `ResourceInUseException` means a
232 // peer created it concurrently (or a prior incarnation is still
233 // DELETING — retried inside the helper), so fall through to
234 // observation. Creation is async; wait for ACTIVE (bounded).
235 if (info === undefined) {
236 yield* retryWhileResourceInUse(
237 kv.createStream({
238 StreamName: streamName,
239 DeviceName: props.deviceName,
240 MediaType: props.mediaType,
241 KmsKeyId: props.kmsKeyId,
242 DataRetentionInHours: desiredRetention ?? 0,
243 Tags: desiredTags,
244 }),
245 ).pipe(
246 Effect.catchTag("ResourceInUseException", () =>
247 Effect.succeed({ StreamARN: undefined }),
248 ),
249 );
250 yield* session.note(`creating stream ${streamName}...`);
251 info = yield* waitForStreamActive(streamName);
252 } else if (info.Status !== "ACTIVE") {
253 // A CREATING/UPDATING stream rejects mutations — wait it out.
254 info = yield* waitForStreamActive(streamName);
255 }
256 const streamArn = info.StreamARN!;
257
258 // 3. SYNC — deviceName/mediaType (versioned update; only when the
259 // observed metadata differs from the desired state)
260 if (
261 (props.deviceName !== undefined &&
262 props.deviceName !== info.DeviceName) ||
263 (props.mediaType !== undefined &&
264 props.mediaType !== info.MediaType)
265 ) {
266 yield* retryWhileSettling(
267 kv.updateStream({
268 StreamARN: streamArn,
269 CurrentVersion: info.Version!,
270 DeviceName: props.deviceName,
271 MediaType: props.mediaType,
272 }),
273 );
274 // UpdateStream is async: it bumps the version and transitions
275 // through UPDATING. Wait for BOTH the version bump and ACTIVE so
276 // the next versioned mutation uses a fresh version.
277 info = yield* waitForStreamActive(streamName, info.Version);
278 }
279
280 // 3b. SYNC — data retention (delta-based API: INCREASE/DECREASE by
281 // the difference between observed and desired)
282 const observedRetention = info.DataRetentionInHours ?? 0;
283 if (
284 desiredRetention !== undefined &&
285 desiredRetention !== observedRetention
286 ) {
287 yield* retryWhileSettling(
288 kv.updateDataRetention({
289 StreamARN: streamArn,
290 CurrentVersion: info.Version!,
291 Operation:
292 desiredRetention > observedRetention
293 ? "INCREASE_DATA_RETENTION"
294 : "DECREASE_DATA_RETENTION",
295 DataRetentionChangeInHours: Math.abs(
296 desiredRetention - observedRetention,
297 ),
298 }),
299 );
300 info = yield* waitForStreamActive(streamName, info.Version);
301 }
302
303 // 3c. SYNC TAGS — diff against OBSERVED cloud tags so adoption
304 // converges
305 const observedTags = yield* fetchObservedTags(streamArn);
306 const { upsert, removed } = diffTags(observedTags, desiredTags);
307 if (upsert.length > 0) {
308 yield* retryWhileSettling(
309 kv.tagStream({
310 StreamARN: streamArn,
311 Tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
312 }),
313 );
314 }
315 if (removed.length > 0) {
316 yield* retryWhileSettling(
317 kv.untagStream({
318 StreamARN: streamArn,
319 TagKeyList: removed,
320 }),
321 );
322 }
323
324 yield* session.note(streamArn);
325 return { streamName, streamArn };
326 }),
327
328 delete: Effect.fn(function* ({ output }) {
329 // A CREATING/UPDATING stream rejects deletion with
330 // ResourceInUseException — retry through the transition (bounded).
331 yield* retryWhileResourceInUse(
332 kv.deleteStream({ StreamARN: output.streamArn }),
333 ).pipe(
334 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
335 );
336 // DeleteStream is accepted immediately; the name stays reserved
337 // until the stream is fully purged. Wait it out so destroy (and
338 // the suite's out-of-band list) don't race a still-ACTIVE row.
339 yield* waitForStreamGone(output.streamName);
340 }),
341 });
342 }),
343 );