1import * as stream from "@distilled.cloud/cloudflare/stream";
2import * as Effect from "effect/Effect";
3import * as Predicate from "effect/Predicate";
4import * as Redacted from "effect/Redacted";
5
6import { Unowned } from "../../AdoptPolicy.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { CloudflareEnvironment } from "../CloudflareEnvironment.ts";
10import type { Providers } from "../Providers.ts";
11
12const TypeId = "Cloudflare.Stream.Webhook" as const;
13type TypeId = typeof TypeId;
14
15export type WebhookProps = {
16 /**
17 * The URL where Stream webhook notifications (e.g. video ready,
18 * live input connected/disconnected) are sent. Mutable — updated in
19 * place via Cloudflare's PUT upsert.
20 */
21 notificationUrl: string;
22};
23
24export type WebhookAttributes = {
25 /**
26 * The Cloudflare account the webhook belongs to.
27 */
28 accountId: string;
29 /**
30 * The URL where webhook notifications are sent.
31 */
32 notificationUrl: string;
33 /**
34 * The date and time the webhook was last modified.
35 */
36 modified: string | undefined;
37 /**
38 * The HMAC secret used to verify webhook request signatures
39 * (`Webhook-Signature` header).
40 */
41 secret: Redacted.Redacted<string>;
42};
43
44export type Webhook = Resource<
45 TypeId,
46 WebhookProps,
47 WebhookAttributes,
48 never,
49 Providers
50>;
51
52/**
53 * The Cloudflare Stream webhook — an **account-level singleton** that
54 * receives notifications when videos finish processing or live inputs
55 * connect/disconnect.
56 *
57 * Each account has at most one Stream webhook, so creating this
58 * resource takes over the account's webhook slot; an existing webhook
59 * configured outside Alchemy is only adopted when `--adopt` is set.
60 * Destroying the resource deletes the webhook configuration.
61 *
62 * Requires the Stream subscription to be enabled on the account.
63 * ### Configuring the webhook
64 * **Example:** Receive Stream notifications
65 * ```typescript
66 * const webhook = yield* Cloudflare.Stream.Webhook("Notifications", {
67 * notificationUrl: "https://example.com/hooks/stream",
68 * });
69 *
70 * // Verify the Webhook-Signature header with the HMAC secret:
71 * const secret = webhook.secret; // Redacted<string>
72 * ```
73 *
74 * @see https://developers.cloudflare.com/stream/manage-video-library/using-webhooks/
75 *
76 * @resource
77 * @product Stream
78 * @category Media
79 */
80export const Webhook = Resource<Webhook>(TypeId);
81
82/**
83 * Returns true if the given value is a Webhook resource.
84 */
85export const isWebhook = (value: unknown): value is Webhook =>
86 Predicate.hasProperty(value, "Type") && value.Type === TypeId;
87
88export const WebhookProvider = () =>
89 Provider.succeed(Webhook, {
90 stables: ["accountId"],
91
92 diff: Effect.fn(function* ({ output }) {
93 const { accountId } = yield* yield* CloudflareEnvironment;
94 // The webhook is a singleton per account — moving accounts is a
95 // replacement.
96 if (output !== undefined && output.accountId !== accountId) {
97 return { action: "replace" } as const;
98 }
99 return undefined;
100 }),
101
102 read: Effect.fn(function* ({ output }) {
103 const { accountId } = yield* yield* CloudflareEnvironment;
104 const acct = output?.accountId ?? accountId;
105 const observed = yield* getWebhook(acct);
106 if (observed === undefined) return undefined;
107 const attrs = toAttributes(observed, acct);
108 // The webhook is an account singleton with no ownership markers.
109 // On a cold read (no prior output) an existing webhook was
110 // configured outside Alchemy — brand it `Unowned` so the engine
111 // refuses to take over unless `--adopt` is set.
112 if (output === undefined) return Unowned(attrs);
113 return attrs;
114 }),
115
116 list: Effect.fn(function* () {
117 const { accountId } = yield* yield* CloudflareEnvironment;
118 // Account-level singleton: the account has at most one Stream
119 // webhook, so enumerate by reading that single slot — return a
120 // one-element array when configured, [] when unset. Exactly
121 // mirrors `read`.
122 const observed = yield* getWebhook(accountId);
123 if (observed === undefined) return [];
124 return [toAttributes(observed, accountId)];
125 }),
126
127 reconcile: Effect.fn(function* ({ news, output }) {
128 const { accountId } = yield* yield* CloudflareEnvironment;
129 const acct = output?.accountId ?? accountId;
130
131 // Observe — read the live webhook config (`WebhookNotFound` means
132 // the account has no webhook yet).
133 const observed = yield* getWebhook(acct);
134
135 // Sync — PUT is a true upsert, so create and update are the same
136 // call. Skip the API entirely when the observed URL already
137 // matches the desired one.
138 if (
139 observed !== undefined &&
140 observed.notificationUrl === news.notificationUrl
141 ) {
142 return toAttributes(observed, acct);
143 }
144 const updated = yield* stream.putWebhook({
145 accountId: acct,
146 notificationUrl: news.notificationUrl,
147 });
148 return toAttributes(updated, acct);
149 }),
150
151 delete: Effect.fn(function* ({ output }) {
152 // DELETE on an already-absent webhook returns success, so the
153 // operation is naturally idempotent; tolerate `WebhookNotFound`
154 // anyway.
155 yield* stream
156 .deleteWebhook({ accountId: output.accountId })
157 .pipe(Effect.catchTag("WebhookNotFound", () => Effect.void));
158 }),
159 });
160
161/**
162 * Read the account's webhook, mapping "not configured"
163 * (`WebhookNotFound`, Cloudflare error code 10003) to `undefined`.
164 */
165const getWebhook = (accountId: string) =>
166 stream
167 .getWebhook({ accountId })
168 .pipe(Effect.catchTag("WebhookNotFound", () => Effect.succeed(undefined)));
169
170const toAttributes = (
171 webhook: stream.GetWebhookResponse | stream.PutWebhookResponse,
172 accountId: string,
173): WebhookAttributes => ({
174 accountId,
175 notificationUrl: webhook.notificationUrl ?? "",
176 modified: webhook.modified ?? undefined,
177 secret: Redacted.make(webhook.secret ?? ""),
178});