1import * as queues from "@distilled.cloud/cloudflare/queues";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Effectable from "effect/Effectable";
5import * as Option from "effect/Option";
6import * as Predicate from "effect/Predicate";
7import * as Stream from "effect/Stream";
8import { Unowned } from "../../AdoptPolicy.ts";
9import { isResolved } from "../../Diff.ts";
10import type { PropsInput } from "../../Input.ts";
11import { createPhysicalName } from "../../PhysicalName.ts";
12import * as Provider from "../../Provider.ts";
13import {
14 isResourceOfType,
15 Resource,
16 type ResourceClass,
17} from "../../Resource.ts";
18import { CloudflareEnvironment } from "../CloudflareEnvironment.ts";
19import type { Providers } from "../Providers.ts";
20import type { Model } from "../AI/Model.ts";
21import type { Variant } from "../Images/Variant.ts";
22import type { Namespace } from "../KV/Namespace.ts";
23import type { Bucket } from "../R2/Bucket.ts";
24import type { SuperSlurperJob } from "../R2/SuperSlurperJob.ts";
25import type { Index } from "../Vectorize/VectorizeIndex.ts";
26import type { Worker } from "../Workers/Worker.ts";
27import type {
28 WorkflowBinding,
29 WorkflowResource,
30} from "../Workflows/Workflow.ts";
31
32const TypeId = "Cloudflare.Queues.Subscription" as const;
33type TypeId = typeof TypeId;
34
35/**
36 * The Cloudflare product whose events the subscription delivers into the
37 * Queue. Cloudflare allows a single subscription per source per account,
38 * and the source is fixed at creation — changing it triggers a
39 * replacement.
40 */
41export type SubscriptionSource =
42 | {
43 /** Cloudflare Images events. */
44 type: "images";
45 }
46 | {
47 /** Workers KV namespace events. */
48 type: "kv";
49 }
50 | {
51 /** R2 bucket events. */
52 type: "r2";
53 }
54 | {
55 /** Super Slurper migration events. */
56 type: "superSlurper";
57 }
58 | {
59 /** Vectorize index events. */
60 type: "vectorize";
61 }
62 | {
63 /** Workers AI model events for a specific model. */
64 type: "workersAi.model";
65 /** Name of the Workers AI model to subscribe to. */
66 modelName: string;
67 }
68 | {
69 /** Workers Builds events for a specific Worker. */
70 type: "workersBuilds.worker";
71 /** Name of the Worker whose build events to subscribe to. */
72 workerName: string;
73 }
74 | {
75 /** Workflows events for a specific workflow. */
76 type: "workflows.workflow";
77 /** Name of the workflow to subscribe to. */
78 workflowName: string;
79 };
80
81export type SubscriptionProps = {
82 /**
83 * Human readable name of the subscription. If omitted, a unique name is
84 * generated from the app, stage, and logical ID.
85 * @default ${app}-${stage}-${id}
86 */
87 name?: string;
88 /**
89 * The event source to subscribe to (e.g. `{ type: "r2" }` for R2 bucket
90 * events). The constructor also accepts Workflow bindings and supported
91 * Cloudflare resources or yielded `.ref(...)` references. Images variants,
92 * KV namespaces, R2 buckets, Super Slurper jobs, and Vectorize indexes select
93 * all events of that product in the account, not just the supplied resource.
94 * Models, Workers, and Workflows select their deferred physical names.
95 * References read persisted state without owning their sources.
96 * Fixed at creation — changing it triggers a replacement. Cloudflare
97 * allows at most one subscription per source per account.
98 */
99 source: SubscriptionSource;
100 /**
101 * Account of a resource source, captured by the constructor. Resource sources
102 * must belong to the subscription's ambient Cloudflare account.
103 */
104 sourceAccountId?: string;
105 /**
106 * Event types to deliver, scoped to the source (e.g. `bucket.created`
107 * and `bucket.deleted` for the `r2` source, `namespace.created` for
108 * `kv`). Must contain at least one event type valid for the source.
109 */
110 events: string[];
111 /**
112 * The ID of the Queue that receives the events
113 * (the destination, e.g. `queue.queueId`).
114 */
115 queueId: string;
116 /**
117 * Whether the subscription is active.
118 * @default true
119 */
120 enabled?: boolean;
121};
122
123export type SubscriptionAttributes = {
124 /**
125 * Unique identifier for the subscription.
126 */
127 subscriptionId: string;
128 /**
129 * The Cloudflare account the subscription belongs to.
130 */
131 accountId: string;
132 /**
133 * Human readable name of the subscription.
134 */
135 name: string;
136 /**
137 * The event source the subscription listens to.
138 */
139 source: SubscriptionSource;
140 /**
141 * Event types delivered by this subscription.
142 */
143 events: string[];
144 /**
145 * The ID of the destination Queue.
146 */
147 queueId: string;
148 /**
149 * Whether the subscription is active.
150 */
151 enabled: boolean;
152 /**
153 * When the subscription was created.
154 */
155 createdAt: string;
156 /**
157 * When the subscription was last modified.
158 */
159 modifiedAt: string;
160};
161
162export type Subscription = Resource<
163 TypeId,
164 SubscriptionProps,
165 SubscriptionAttributes,
166 never,
167 Providers
168>;
169
170/** Resources whose product events can be delivered to a Queue. */
171export type SubscriptionResourceSource =
172 | WorkflowResource
173 | Variant
174 | Namespace
175 | Bucket
176 | SuperSlurperJob
177 | Index
178 | Model
179 | Worker;
180
181/** Constructor inputs; resource sources are normalized before registration. */
182export type SubscriptionInput = Omit<
183 PropsInput<SubscriptionProps>,
184 "source" | "sourceAccountId"
185> & {
186 /** An explicit source, Workflow binding, resource, or yielded resource reference. */
187 source:
188 | PropsInput<SubscriptionProps>["source"]
189 | WorkflowBinding
190 | SubscriptionResourceSource;
191};
192
193type SubscriptionConstructor<Req = never> = {
194 Type: TypeId;
195 Props: SubscriptionProps;
196 <const Methods extends Record<string, any>>(
197 methods: Methods,
198 ): SubscriptionClass & Methods;
199 (
200 id: string,
201 props: SubscriptionInput,
202 ): Effect.Effect<Subscription, never, Req>;
203 <PropsReq = never>(
204 id: string,
205 props: Effect.Effect<SubscriptionInput, never, PropsReq>,
206 ): Effect.Effect<Subscription, never, PropsReq | Req>;
207};
208
209type SubscriptionClass = SubscriptionConstructor<Providers> &
210 Effect.Effect<SubscriptionConstructor> &
211 Pick<ResourceClass<Subscription>, "Self" | "Provider" | "Aliases" | "ref">;
212
213const SubscriptionResource = Resource<Subscription>(TypeId, {
214 aliases: ["Cloudflare.Queue.Subscription"],
215});
216
217const isSourceResource = <Type extends SubscriptionResourceSource["Type"]>(
218 source: SubscriptionInput["source"],
219 type: Type,
220): source is Extract<SubscriptionResourceSource, { Type: Type }> =>
221 isResourceOfType(source, type);
222
223const isWorkflowBinding = (
224 source: SubscriptionInput["source"],
225): source is WorkflowBinding =>
226 typeof source === "object" &&
227 source !== null &&
228 (source as WorkflowBinding).kind === "Cloudflare.Workflow";
229
230const normalizeSubscriptionProps = (
231 props: SubscriptionInput,
232): PropsInput<SubscriptionProps> => {
233 const source = props.source;
234 // Resource references are Output proxies; inspect their type before binding fields.
235 if (isSourceResource(source, "Cloudflare.Workflow")) {
236 return {
237 ...props,
238 sourceAccountId: source.accountId,
239 source: { type: "workflows.workflow", workflowName: source.workflowName },
240 };
241 }
242 if (isSourceResource(source, "Cloudflare.Worker")) {
243 return {
244 ...props,
245 sourceAccountId: source.accountId,
246 source: { type: "workersBuilds.worker", workerName: source.workerName },
247 };
248 }
249 if (isSourceResource(source, "Cloudflare.AI.Model")) {
250 return {
251 ...props,
252 sourceAccountId: source.accountId,
253 source: { type: "workersAi.model", modelName: source.modelName },
254 };
255 }
256 if (isSourceResource(source, "Cloudflare.Images.Variant")) {
257 return {
258 ...props,
259 sourceAccountId: source.accountId,
260 source: { type: "images" },
261 };
262 }
263 if (isSourceResource(source, "Cloudflare.KV.Namespace")) {
264 return {
265 ...props,
266 sourceAccountId: source.accountId,
267 source: { type: "kv" },
268 };
269 }
270 if (isSourceResource(source, "Cloudflare.R2.Bucket")) {
271 return {
272 ...props,
273 sourceAccountId: source.accountId,
274 source: { type: "r2" },
275 };
276 }
277 if (isSourceResource(source, "Cloudflare.R2.SuperSlurperJob")) {
278 return {
279 ...props,
280 sourceAccountId: source.accountId,
281 source: { type: "superSlurper" },
282 };
283 }
284 if (isSourceResource(source, "Cloudflare.VectorizeIndex")) {
285 return {
286 ...props,
287 sourceAccountId: source.accountId,
288 source: { type: "vectorize" },
289 };
290 }
291 if (isWorkflowBinding(source)) {
292 return {
293 ...props,
294 source: { type: "workflows.workflow", workflowName: source.workflowName },
295 };
296 }
297 return { ...props, source };
298};
299
300/** A resource source cannot select events in a different Cloudflare account. */
301export class SubscriptionSourceAccountMismatch extends Data.TaggedError(
302 "SubscriptionSourceAccountMismatch",
303)<{ readonly accountId: string; readonly sourceAccountId: string }> {}
304
305const validateSourceAccount = (accountId: string, sourceAccountId?: string) =>
306 sourceAccountId !== undefined && sourceAccountId !== accountId
307 ? Effect.fail(
308 new SubscriptionSourceAccountMismatch({ accountId, sourceAccountId }),
309 )
310 : Effect.void;
311
312/**
313 * A Cloudflare Queues event subscription — delivers platform events
314 * (R2 bucket events, KV namespace events, Workers Builds, Workflows,
315 * etc.) into a Queue as messages.
316 *
317 * The `source` selects which product emits the events and is fixed at
318 * creation (changing it replaces the subscription). `name`, `events`,
319 * `enabled`, and the destination `queueId` are all mutable in place.
320 * Cloudflare allows at most one subscription per source per account.
321 * ### Creating a Subscription
322 * **Example:** R2 bucket events into a Queue
323 * ```typescript
324 * const queue = yield* Cloudflare.Queues.Queue("EventsQueue");
325 *
326 * const subscription = yield* Cloudflare.Queues.Subscription("R2Events", {
327 * source: { type: "r2" },
328 * events: ["bucket.created", "bucket.deleted"],
329 * queueId: queue.queueId,
330 * });
331 * ```
332 *
333 * **Example:** KV namespace events with an explicit name
334 * ```typescript
335 * const subscription = yield* Cloudflare.Queues.Subscription("KvEvents", {
336 * name: "kv-events",
337 * source: { type: "kv" },
338 * events: ["namespace.created"],
339 * queueId: queue.queueId,
340 * });
341 * ```
342 *
343 * **Example:** Workers Builds events for one Worker
344 * ```typescript
345 * const subscription = yield* Cloudflare.Queues.Subscription("BuildEvents", {
346 * source: { type: "workersBuilds.worker", workerName: "my-worker" },
347 * events: ["build.started", "build.succeeded"],
348 * queueId: queue.queueId,
349 * });
350 * ```
351 *
352 * **Example:** Workflow lifecycle events, from a Workflow bound in this stack
353 * Pass the Workflow binding from the host Worker's `env` directly. Its
354 * physical name remains an `Output`, so the subscription works on the
355 * Workflow's first deployment and follows later renames. Only the source
356 * type and physical workflow name are persisted, not the binding metadata.
357 * ```typescript
358 * const worker = yield* Cloudflare.Worker("Worker", {
359 * main: "./src/worker.ts",
360 * env: { INGESTION: Cloudflare.Workflow("Ingestion", { className: "IngestionWorkflow" }) },
361 * });
362 *
363 * const subscription = yield* Cloudflare.Queues.Subscription("WorkflowEvents", {
364 * source: worker.env.INGESTION,
365 * events: ["instance.completed", "instance.errored"],
366 * queueId: queue.queueId,
367 * });
368 * ```
369 *
370 * **Example:** Workflow lifecycle events from a persisted resource reference
371 * Use the logical resource ID, including any namespace. References read
372 * persisted state, so deploy the host first. Omitting the options uses the
373 * current stack and stage; the reference does not take ownership of the host.
374 * ```typescript
375 * const subscription = yield* Cloudflare.Queues.Subscription("WorkflowEvents", {
376 * source: yield* Cloudflare.Workflow.ref("Ingestion", {
377 * stack: "workflow-host",
378 * stage: "production",
379 * }),
380 * events: ["instance.completed", "instance.errored"],
381 * queueId: queue.queueId,
382 * });
383 * ```
384 *
385 * **Example:** Workflow lifecycle events by an existing physical name
386 * ```typescript
387 * const subscription = yield* Cloudflare.Queues.Subscription("WorkflowEvents", {
388 * source: { type: "workflows.workflow", workflowName: "existing-ingestion" },
389 * events: ["instance.completed", "instance.errored"],
390 * queueId: queue.queueId,
391 * });
392 * ```
393 *
394 * ### Resource and Reference Sources
395 * **Example:** Account-wide KV events from a namespace
396 * ```typescript
397 * const namespace = yield* Cloudflare.KV.Namespace("Cache");
398 * yield* Cloudflare.Queues.Subscription("NamespaceEvents", {
399 * source: namespace,
400 * events: ["namespace.created", "namespace.deleted"],
401 * queueId: queue.queueId,
402 * });
403 * ```
404 * KV, R2, Images, Vectorize, and Super Slurper sources are account-wide.
405 * Passing a resource retains its account and deployment dependency, but does
406 * not filter events to that resource. The subscription can miss the source's
407 * initial creation or final deletion because it depends on that source.
408 * Use an explicit product descriptor when the subscription must exist first.
409 *
410 * **Example:** R2 bucket reference from another stack
411 * ```typescript
412 * yield* Cloudflare.Queues.Subscription("BucketEvents", {
413 * source: yield* Cloudflare.R2.Bucket.ref("Uploads", {
414 * stack: "storage",
415 * stage: "production",
416 * }),
417 * events: ["bucket.created", "bucket.deleted"],
418 * queueId: queue.queueId,
419 * });
420 * ```
421 * All resource forms accept yielded `.ref` references with optional stack
422 * and stage selectors. The referenced resource must already be deployed in
423 * the same Cloudflare account. Removing a subscription does not remove its
424 * referenced resource.
425 *
426 * **Example:** Images upload events from a variant reference
427 * ```typescript
428 * yield* Cloudflare.Queues.Subscription("ImageEvents", {
429 * source: yield* Cloudflare.Images.Variant.ref("Thumbnail"),
430 * events: ["image.uploaded"],
431 * queueId: queue.queueId,
432 * });
433 * ```
434 * The variant selects its Images account; uploads are not filtered by variant.
435 *
436 * **Example:** Vectorize index events
437 * ```typescript
438 * yield* Cloudflare.Queues.Subscription("IndexEvents", {
439 * source: yield* Cloudflare.Vectorize.Index.ref("Search"),
440 * events: ["index.created", "index.deleted"],
441 * queueId: queue.queueId,
442 * });
443 * ```
444 *
445 * **Example:** Super Slurper migration events
446 * ```typescript
447 * yield* Cloudflare.Queues.Subscription("MigrationEvents", {
448 * source: yield* Cloudflare.R2.SuperSlurperJob.ref("Migration"),
449 * events: ["job.completed", "job.aborted"],
450 * queueId: queue.queueId,
451 * });
452 * ```
453 * This selects all migration jobs in the account, not one job's objects.
454 *
455 * **Example:** Workers AI batch events
456 * ```typescript
457 * const model = yield* Cloudflare.AI.Model("Embeddings", {
458 * modelName: "@cf/baai/bge-m3",
459 * });
460 * yield* Cloudflare.Queues.Subscription("BatchEvents", {
461 * source: model,
462 * events: ["batch.queued", "batch.succeeded", "batch.failed"],
463 * queueId: queue.queueId,
464 * });
465 * ```
466 * The model is a non-owning catalog handle. These events require asynchronous
467 * batch inference; ordinary synchronous inference does not emit them.
468 *
469 * **Example:** Workers Builds events from a Worker reference
470 * ```typescript
471 * yield* Cloudflare.Queues.Subscription("BuildEvents", {
472 * source: yield* Cloudflare.Worker.ref("Website"),
473 * events: ["build.started", "build.succeeded", "build.failed"],
474 * queueId: queue.queueId,
475 * });
476 * ```
477 * The Worker must have a Workers Builds integration to emit build events.
478 * An ordinary Alchemy Worker upload is not a Workers Builds run.
479 *
480 * Event delivery can lag subscription creation or replacement even after the
481 * destination Queue accepts messages. Deployment confirms configuration, not
482 * delivery readiness; verify delivery before emitting events that must be observed.
483 * During Vectorize subscription replacement or a destination Queue update,
484 * Cloudflare can still route new events to the previous Queue. Replacement events
485 * can carry the deleted subscription's ID; updates retain the same subscription ID.
486 * A single early event does not prove that routing has fully propagated. Keep the
487 * previous destination available during the transition and verify the receiving
488 * Queue, `metadata.eventSubscriptionId`, and the event's resource identity.
489 *
490 * ### Pausing delivery
491 * **Example:** Disable a subscription without deleting it
492 * ```typescript
493 * const subscription = yield* Cloudflare.Queues.Subscription("R2Events", {
494 * source: { type: "r2" },
495 * events: ["bucket.created"],
496 * queueId: queue.queueId,
497 * enabled: false,
498 * });
499 * ```
500 *
501 * @see https://developers.cloudflare.com/queues/event-subscriptions/
502 *
503 * @resource
504 * @product Queues
505 * @category Storage & Databases
506 */
507export const Subscription: SubscriptionClass = Object.assign(
508 (
509 ...args:
510 | [
511 id: string,
512 props:
513 | SubscriptionInput
514 | Effect.Effect<SubscriptionInput, never, any>,
515 ]
516 | [methods: Record<string, any>]
517 ) => {
518 if (typeof args[0] === "object") {
519 return Object.assign(Subscription, args[0]);
520 }
521 const [id, props] = args as [
522 string,
523 SubscriptionInput | Effect.Effect<SubscriptionInput, never, any>,
524 ];
525 // Resource supplies Self while evaluating Effect-valued props.
526 return Effect.isEffect(props)
527 ? SubscriptionResource(id, Effect.map(props, normalizeSubscriptionProps))
528 : SubscriptionResource(id, normalizeSubscriptionProps(props));
529 },
530 SubscriptionResource,
531 Effectable.Prototype({
532 label: `Resource<${TypeId}>`,
533 evaluate: (): Effect.Effect<SubscriptionConstructor> =>
534 Effect.succeed(Subscription),
535 }),
536) as SubscriptionClass;
537
538/**
539 * Returns true if the given value is a Subscription resource.
540 */
541export const isSubscription = (value: unknown): value is Subscription =>
542 Predicate.hasProperty(value, "Type") && value.Type === TypeId;
543
544export const SubscriptionProvider = () =>
545 Provider.succeed(SubscriptionResource, {
546 stables: ["subscriptionId", "accountId", "source", "createdAt"],
547 diff: Effect.fn(function* ({ olds, news, output }) {
548 const { accountId } = yield* yield* CloudflareEnvironment;
549 if (isResolved(news)) {
550 yield* validateSourceAccount(accountId, news.sourceAccountId);
551 }
552 if ((output?.accountId ?? accountId) !== accountId) {
553 return { action: "replace" } as const;
554 }
555 // An unresolved immutable source may change with its upstream resource.
556 // Delete first in case it resolves to the same account-unique source.
557 if (!("source" in news) || !isResolved(news.source)) {
558 return { action: "replace", deleteFirst: true } as const;
559 }
560 // The source is fixed at creation.
561 const oldSource = output?.source ?? olds?.source;
562 if (oldSource && !sameSource(oldSource, news.source)) {
563 return { action: "replace" } as const;
564 }
565 return undefined;
566 }),
567 read: Effect.fn(function* ({ id, output, olds }) {
568 const { accountId } = yield* yield* CloudflareEnvironment;
569 if (!output) {
570 yield* validateSourceAccount(accountId, olds?.sourceAccountId);
571 }
572 const acct = output?.accountId ?? accountId;
573
574 if (output?.subscriptionId) {
575 const observed = yield* getSubscriptionOrUndefined(
576 acct,
577 output.subscriptionId,
578 );
579 return observed ? toAttributes(observed, acct) : undefined;
580 }
581 // Cold read — recover from lost state by matching the deterministic
582 // physical name. Names are not enforced unique server-side; an exact
583 // match on our generated/explicit name is the best identity we have.
584 const name = yield* createSubscriptionName(id, olds?.name);
585 const match = yield* findByName(acct, name);
586 if (!match) return undefined;
587 const attributes = toAttributes(match, acct);
588 return olds?.name !== undefined ? Unowned(attributes) : attributes;
589 }),
590 reconcile: Effect.fn(function* ({ id, news, output }) {
591 const { accountId } = yield* yield* CloudflareEnvironment;
592 yield* validateSourceAccount(accountId, news.sourceAccountId);
593 const acct = output?.accountId ?? accountId;
594 const name = yield* createSubscriptionName(id, news.name);
595
596 // Observe — the subscriptionId cached on `output` is a hint, not a
597 // guarantee: a missing subscription falls through to create.
598 let observed = output?.subscriptionId
599 ? yield* getSubscriptionOrUndefined(acct, output.subscriptionId)
600 : undefined;
601
602 // A source collision does not establish ownership of the existing subscription.
603 if (!observed) {
604 observed = yield* queues.createSubscription({
605 accountId: acct,
606 name,
607 enabled: news.enabled ?? true,
608 events: news.events,
609 source: news.source,
610 destination: { type: "queues.queue", queueId: news.queueId },
611 });
612 }
613
614 // Sync — diff observed cloud state against desired and patch only
615 // the delta; skip the API call entirely on a no-op.
616 const desired = {
617 name,
618 enabled: news.enabled ?? true,
619 events: news.events,
620 queueId: news.queueId,
621 };
622 const dirty =
623 observed.name !== desired.name ||
624 observed.enabled !== desired.enabled ||
625 observed.destination.queueId !== desired.queueId ||
626 !sameEvents(observed.events, desired.events);
627
628 const final = dirty
629 ? yield* queues.patchSubscription({
630 accountId: acct,
631 subscriptionId: observed.id,
632 name: desired.name,
633 enabled: desired.enabled,
634 events: desired.events,
635 destination: { type: "queues.queue", queueId: desired.queueId },
636 })
637 : observed;
638
639 return toAttributes(final, acct);
640 }),
641 delete: Effect.fn(function* ({ output }) {
642 yield* queues
643 .deleteSubscription({
644 accountId: output.accountId,
645 subscriptionId: output.subscriptionId,
646 })
647 .pipe(Effect.catchTag("SubscriptionNotFound", () => Effect.void));
648 }),
649 // Account collection — subscriptions are account-scoped. Exhaustively
650 // paginate `listSubscriptions` (response array is `result`) and hydrate
651 // each row into the same Attributes shape `read` returns.
652 list: Effect.fn(function* () {
653 const { accountId } = yield* yield* CloudflareEnvironment;
654 return yield* queues.listSubscriptions.pages({ accountId }).pipe(
655 Stream.runCollect,
656 Effect.map((chunk) =>
657 Array.from(chunk).flatMap((page) =>
658 (page.result ?? []).map((sub) => toAttributes(sub, accountId)),
659 ),
660 ),
661 );
662 }),
663 });
664
665type ObservedSubscription =
666 | queues.GetSubscriptionResponse
667 | queues.CreateSubscriptionResponse
668 | queues.PatchSubscriptionResponse;
669
670/**
671 * Read a subscription by ID, mapping "gone" (`SubscriptionNotFound`,
672 * HTTP 404 "No subscription with this ID") to `undefined`.
673 */
674const getSubscriptionOrUndefined = (
675 accountId: string,
676 subscriptionId: string,
677) =>
678 queues
679 .getSubscription({ accountId, subscriptionId })
680 .pipe(
681 Effect.catchTag("SubscriptionNotFound", () => Effect.succeed(undefined)),
682 );
683
684/**
685 * Find a subscription by exact name. Cloudflare's list endpoint has no
686 * name filter, so scan the paginated stream.
687 */
688const findByName = (accountId: string, name: string) =>
689 queues.listSubscriptions.items({ accountId }).pipe(
690 Stream.filter((s) => s.name === name),
691 Stream.runHead,
692 Effect.map(Option.getOrUndefined),
693 );
694
695const createSubscriptionName = (id: string, name: string | undefined) =>
696 Effect.gen(function* () {
697 return name ?? (yield* createPhysicalName({ id, lowercase: true }));
698 });
699
700/**
701 * The wire shape of a subscription source — a flattened view of the
702 * response union (only the member matching `type` carries its name field).
703 */
704type WireSource = {
705 type?: string | null;
706 modelName?: string | null;
707 workerName?: string | null;
708 workflowName?: string | null;
709};
710
711const toSource = (wire: unknown): SubscriptionSource => {
712 const source = wire as WireSource;
713 switch (source.type) {
714 case "workersAi.model":
715 return { type: "workersAi.model", modelName: source.modelName ?? "" };
716 case "workersBuilds.worker":
717 return {
718 type: "workersBuilds.worker",
719 workerName: source.workerName ?? "",
720 };
721 case "workflows.workflow":
722 return {
723 type: "workflows.workflow",
724 workflowName: source.workflowName ?? "",
725 };
726 case "images":
727 case "kv":
728 case "r2":
729 case "superSlurper":
730 case "vectorize":
731 return { type: source.type };
732 default:
733 // Unknown/new source types added server-side: surface the raw type
734 // so diff treats it as a foreign source rather than crashing.
735 return { type: source.type as never };
736 }
737};
738
739const sameSource = (a: SubscriptionSource, b: SubscriptionSource): boolean => {
740 if (a.type !== b.type) return false;
741 switch (a.type) {
742 case "workersAi.model":
743 return a.modelName === (b as { modelName?: string }).modelName;
744 case "workersBuilds.worker":
745 return a.workerName === (b as { workerName?: string }).workerName;
746 case "workflows.workflow":
747 return a.workflowName === (b as { workflowName?: string }).workflowName;
748 default:
749 return true;
750 }
751};
752
753const sameEvents = (observed: readonly string[], desired: readonly string[]) =>
754 observed.length === desired.length &&
755 [...observed].sort().join(",") === [...desired].sort().join(",");
756
757const toAttributes = (
758 subscription: ObservedSubscription,
759 accountId: string,
760): SubscriptionAttributes => ({
761 subscriptionId: subscription.id,
762 accountId,
763 name: subscription.name,
764 source: toSource(subscription.source),
765 events: [...subscription.events],
766 queueId: subscription.destination.queueId,
767 enabled: subscription.enabled,
768 createdAt: subscription.createdAt,
769 modifiedAt: subscription.modifiedAt,
770});