1import type * as Effect from "effect/Effect";
2import type * as Stream from "effect/Stream";
3import {
4 consumeBusEvents,
5 type EventRecord,
6 type EventRouteProps,
7} from "../EventBridge/EventSource.ts";
8
9/**
10 * The `detail` payload Amazon IVS delivers to EventBridge. Stream
11 * state-change events describe the transition (`event_name` of
12 * `Stream Start`, `Stream End`, or `Session Created` plus the
13 * `channel_name`/`stream_id`); limit-breach events carry the breached
14 * `limit_name` and `limit_value`; recording state changes carry the
15 * `recording_status` and S3 key prefix. Fields not shared by every event
16 * kind are optional (the schema grows over time).
17 */
18export interface IvsStreamEventDetail {
19 /**
20 * Stream state-change events: `Stream Start`, `Stream End`, or
21 * `Session Created`. Limit-breach events: the name of the breached
22 * limit.
23 */
24 event_name?: string;
25 /** The name of the channel the event is about. */
26 channel_name?: string;
27 /** Stream state-change events: the id of the stream session. */
28 stream_id?: string;
29 /** Limit-breach events: the name of the breached limit. */
30 limit_name?: string;
31 /** Limit-breach events: the value of the breached limit. */
32 limit_value?: number;
33 /** Recording state-change events: the recording session id. */
34 recording_session_id?: string;
35 /**
36 * Recording state-change events: `Recording Start`, `Recording End`,
37 * or `Recording Start Failure`.
38 */
39 recording_status?: string;
40 /** Additional event fields (the schema grows over time). */
41 [key: string]: unknown;
42}
43
44/** An Amazon IVS EventBridge event delivered to the handler. */
45export type IvsStreamEvent = EventRecord<IvsStreamEventDetail>;
46
47/** Which Amazon IVS notifications to subscribe to. */
48export type IvsStreamEventKind =
49 | "stream-state-change"
50 | "recording-state-change"
51 | "limit-breach";
52
53const DETAIL_TYPES: Record<IvsStreamEventKind, string> = {
54 "stream-state-change": "IVS Stream State Change",
55 "recording-state-change": "IVS Recording State Change",
56 "limit-breach": "IVS Limit Breach",
57};
58
59export interface StreamEventSourceProps extends EventRouteProps {
60 /**
61 * Logical id for the backing EventBridge rule.
62 * @default "IVSStreamEvents"
63 */
64 id?: string;
65 /**
66 * Which notifications to subscribe to: stream state changes (stream
67 * start/end, session created), recording state changes (recording
68 * start/end/failure), and/or limit breaches.
69 * @default ["stream-state-change"]
70 */
71 kinds?: readonly IvsStreamEventKind[];
72 /**
73 * Restrict to events about specific channels (matched against the
74 * event's top-level `resources`, which contains the channel's ARN).
75 */
76 channelArns?: readonly string[];
77}
78
79/**
80 * Event source connecting Amazon IVS stream notifications to the hosting
81 * compute. IVS publishes stream state changes (a broadcast starting or
82 * ending), recording state changes (an S3 recording starting, completing,
83 * or failing), and account limit breaches to the account's default
84 * EventBridge bus (source `aws.ivs`); this subscribes the host Function
85 * to those events so it can react to broadcasts going live or recordings
86 * landing in S3.
87 *
88 * IVS publishes to EventBridge automatically — no additional resource is
89 * created besides the EventBridge rule targeting the host. Provide the
90 * host-specific implementation layer (e.g. `AWS.Lambda.EventSource`) on
91 * the Function effect.
92 *
93 * ### Consuming Stream Events
94 * **Example:** React When a Broadcast Starts
95 * ```typescript
96 * import * as AWS from "alchemy/AWS";
97 *
98 * export default NotifyFunction.make(
99 * { main: import.meta.url },
100 * Effect.gen(function* () {
101 * yield* AWS.IVS.consumeStreamEvents(
102 * { kinds: ["stream-state-change"] },
103 * (events) =>
104 * Stream.runForEach(events, (event) =>
105 * event.detail.event_name === "Stream Start"
106 * ? Effect.log(`${event.detail.channel_name} went live`)
107 * : Effect.void,
108 * ),
109 * );
110 * return {};
111 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
112 * );
113 * ```
114 */
115export const consumeStreamEvents = <StreamReq = never, Req = never>(
116 props: StreamEventSourceProps,
117 process: (
118 events: Stream.Stream<IvsStreamEvent, never, StreamReq>,
119 ) => Effect.Effect<void, never, Req>,
120) =>
121 consumeBusEvents(
122 props.id ?? "IVSStreamEvents",
123 {
124 source: ["aws.ivs"],
125 "detail-type": (props.kinds ?? (["stream-state-change"] as const)).map(
126 (kind) => DETAIL_TYPES[kind],
127 ),
128 ...(props.channelArns !== undefined
129 ? { resources: [...props.channelArns] }
130 : {}),
131 },
132 { description: props.description, state: props.state },
133 process,
134 );