1import type * as cf from "@cloudflare/workers-types";
2import { Webhooks } from "@distilled.cloud/stripe";
3import * as Context from "effect/Context";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Redacted from "effect/Redacted";
7import { Worker, isWorkerEvent } from "../Cloudflare/Workers/Worker.ts";
8import * as Namespace from "../Namespace.ts";
9import * as Output from "../Output.ts";
10import {
11 isRedactedMarker,
12 sanitizeKey,
13 unpackEnvValue,
14 type RuntimeContext,
15} from "../RuntimeContext.ts";
16import type { StripeEventClass, StripeEventInstance } from "./Events.ts";
17import { WebhookEndpoint } from "./WebhookEndpoint.ts";
18
19export interface ConsumeEventsProps<
20 E extends readonly StripeEventClass[] = readonly StripeEventClass[],
21> {
22 /**
23 * Event classes to subscribe to. The handler `event` parameter is the
24 * union of these classes.
25 */
26 events: E;
27 /**
28 * Path on the host Worker. Defaults to `/webhooks/stripe`.
29 */
30 path?: string;
31}
32
33export type SelectedStripeEvent<E extends readonly StripeEventClass[]> =
34 InstanceType<E[number]>;
35
36export const webhookPath = (path?: string): string =>
37 path ?? "/webhooks/stripe";
38
39export const webhookSecretEnvName = (path?: string): string =>
40 `STRIPE_WEBHOOK_SECRET_${webhookPath(path).replaceAll(/[^a-zA-Z0-9]/g, "_")}`;
41
42/**
43 * Deterministic logical id for the {@link WebhookEndpoint} created by
44 * {@link consumeEvents} when the caller does not pass an explicit id.
45 * Derived from the delivery path so two subscriptions on different paths
46 * don't collide.
47 */
48export const webhookEndpointLogicalId = (path?: string): string =>
49 `WebhookEndpoint${sanitizeKey(webhookPath(path))}`;
50
51/**
52 * Attach a webhook signing secret to a Worker as `secret_text`.
53 *
54 * {@link consumeEvents} calls this automatically for the endpoint it
55 * provisions. Declare a {@link WebhookEndpoint} manually only when you need
56 * a URL you already own; then bind the minted secret yourself.
57 */
58export const bindWebhookSecret = (
59 host: Worker,
60 secret: WebhookEndpoint["secret"],
61 path?: string,
62): Effect.Effect<void> =>
63 host.bind(`stripe-webhook:${webhookPath(path)}`, {
64 env: {
65 [webhookSecretEnvName(path)]: secret,
66 },
67 });
68
69/**
70 * Subscribe to Stripe webhook events on the host Worker.
71 *
72 * `consumeEvents` provisions the {@link WebhookEndpoint} itself: it derives
73 * the delivery URL from the Worker's `url`, enables the selected event
74 * types, and binds the endpoint's minted signing secret onto the Worker so
75 * deliveries are verified. No separate endpoint declaration is needed.
76 *
77 * Provide {@link ConsumeEventsLive} on the Worker Effect.
78 *
79 * ### Handling events
80 * **Example:** Customer created and invoice paid
81 * ```typescript
82 * yield* Stripe.consumeEvents("Events", {
83 * events: [Stripe.CustomerCreated, Stripe.InvoicePaid],
84 * }, Effect.fn(function* (event) {
85 * // event: CustomerCreated | InvoicePaid
86 * yield* Effect.log(event.type);
87 * }));
88 * ```
89 *
90 * @binding
91 * @product Webhook
92 */
93export function consumeEvents<
94 const E extends readonly StripeEventClass[],
95 Req = never,
96>(
97 props: ConsumeEventsProps<E>,
98 process: (
99 event: SelectedStripeEvent<E>,
100 ) => Effect.Effect<void, never, Req | RuntimeContext>,
101): Effect.Effect<void, never, EventSource>;
102export function consumeEvents<
103 const E extends readonly StripeEventClass[],
104 Req = never,
105>(
106 id: string,
107 props: ConsumeEventsProps<E>,
108 process: (
109 event: SelectedStripeEvent<E>,
110 ) => Effect.Effect<void, never, Req | RuntimeContext>,
111): Effect.Effect<void, never, EventSource>;
112export function consumeEvents(
113 idOrProps: string | ConsumeEventsProps,
114 propsOrProcess:
115 | ConsumeEventsProps
116 | ((event: StripeEventInstance) => Effect.Effect<void, never, any>),
117 maybeProcess?: (
118 event: StripeEventInstance,
119 ) => Effect.Effect<void, never, any>,
120): Effect.Effect<void, never, EventSource> {
121 const [id, props, process] =
122 typeof idOrProps === "string"
123 ? [
124 idOrProps,
125 propsOrProcess as ConsumeEventsProps,
126 maybeProcess as (
127 event: StripeEventInstance,
128 ) => Effect.Effect<void, never, any>,
129 ]
130 : [
131 undefined,
132 idOrProps as ConsumeEventsProps,
133 propsOrProcess as (
134 event: StripeEventInstance,
135 ) => Effect.Effect<void, never, any>,
136 ];
137 return EventSource.use((source) =>
138 source(id ?? webhookEndpointLogicalId(props.path), props, process),
139 );
140}
141
142export type EventSourceService = <
143 E extends readonly StripeEventClass[],
144 Req = never,
145>(
146 id: string,
147 props: ConsumeEventsProps<E>,
148 process: (event: SelectedStripeEvent<E>) => Effect.Effect<void, never, Req>,
149) => Effect.Effect<void, never, never>;
150
151export class EventSource extends Context.Service<
152 EventSource,
153 EventSourceService
154>()("Stripe.EventSource") {}
155
156/**
157 * Cloudflare Worker implementation of {@link consumeEvents}.
158 *
159 * Deploy-time: provisions a {@link WebhookEndpoint} pointing at this Worker
160 * (at the subscribed path) and binds its minted signing secret onto the
161 * Worker so deliveries can be verified. Runtime: registers a `fetch`
162 * listener that claims requests on that path, verifies `Stripe-Signature`,
163 * and runs the handler once per event.
164 *
165 * @layer
166 * @product Webhook
167 * @provides Stripe.EventSource
168 */
169export const ConsumeEventsLive = Layer.effect(
170 EventSource,
171 Effect.gen(function* () {
172 const host = yield* Worker;
173
174 return Effect.fn(function* (
175 id: string,
176 props: ConsumeEventsProps,
177 process: (
178 event: StripeEventInstance,
179 ) => Effect.Effect<void, never, never>,
180 ) {
181 const path = webhookPath(props.path);
182 const secretKey = webhookSecretEnvName(path);
183 const byType = new Map(
184 props.events.map((event) => [event.type, event] as const),
185 );
186
187 if (!globalThis.__ALCHEMY_RUNTIME__) {
188 yield* Namespace.push(
189 host.LogicalId,
190 Effect.gen(function* () {
191 const endpoint = yield* WebhookEndpoint(id, {
192 url: Output.interpolate`${host.url}${path}`,
193 enabledEvents: [...props.events],
194 });
195 yield* bindWebhookSecret(host, endpoint.secret, props.path);
196 }),
197 );
198 }
199
200 yield* host.listen((event) => {
201 if (!isWorkerEvent(event) || event.type !== "fetch") return;
202 const request = event.input as cf.Request;
203 let pathname: string;
204 try {
205 pathname = new URL(request.url).pathname;
206 } catch {
207 return;
208 }
209 if (pathname !== path) return;
210 const env = (event.env ?? {}) as Record<string, unknown>;
211 return handleDelivery(request, env, secretKey, byType, process);
212 });
213 }) as EventSourceService;
214 }),
215);
216
217const asWebhookSecret = (
218 raw: unknown,
219): Redacted.Redacted<string> | undefined => {
220 if (raw === undefined || raw === null || raw === "") return undefined;
221 if (Redacted.isRedacted(raw)) {
222 const value = Redacted.value(raw);
223 return typeof value === "string" && value.length > 0
224 ? (raw as Redacted.Redacted<string>)
225 : undefined;
226 }
227 if (isRedactedMarker(raw) && typeof raw.value === "string") {
228 return raw.value.length > 0 ? Redacted.make(raw.value) : undefined;
229 }
230 if (typeof raw === "string") return Redacted.make(raw);
231 return undefined;
232};
233
234const handleDelivery = <Req>(
235 request: cf.Request,
236 env: Record<string, any>,
237 secretKey: string,
238 byType: Map<string, StripeEventClass>,
239 process: (event: StripeEventInstance) => Effect.Effect<void, never, Req>,
240): Effect.Effect<Response, never, Req> =>
241 Effect.gen(function* () {
242 if (request.method !== "POST") {
243 return new Response("method not allowed", { status: 405 });
244 }
245 const payload = yield* Effect.promise(() =>
246 (request as unknown as Request).text(),
247 );
248 const signature = request.headers.get("stripe-signature") ?? "";
249 const resolved = asWebhookSecret(
250 unpackEnvValue(env[secretKey] as string | undefined) ?? env[secretKey],
251 );
252 if (resolved === undefined) {
253 return new Response("webhook secret missing", { status: 500 });
254 }
255 const parsed = yield* Webhooks.constructEvent({
256 payload,
257 signature,
258 secret: Redacted.value(resolved),
259 }).pipe(
260 Effect.catchTag(
261 ["StripeWebhookSignatureError", "StripeWebhookPayloadParseError"],
262 () => Effect.succeed(undefined),
263 ),
264 );
265 if (parsed === undefined) {
266 return new Response("invalid signature", { status: 401 });
267 }
268 const Ctor = byType.get(parsed.type ?? "");
269 if (Ctor === undefined) {
270 return new Response(null, { status: 200 });
271 }
272 const data =
273 typeof parsed.data === "object" &&
274 parsed.data !== null &&
275 "object" in parsed.data
276 ? (parsed.data as { object: unknown }).object
277 : parsed.data;
278 yield* process(new Ctor(data as never)).pipe(Effect.orDie);
279 return new Response(null, { status: 200 });
280 });