feat(wrangler): add email.sending event subscription source

.changeset/email-sending-event-subscription-source.md

@@ -0,0 +1,7 @@
+---
+"wrangler": minor
+---
+
+Add `email.sending` as an event subscription source for queues
+
+`wrangler queues subscription create` now accepts `--source email.sending` alongside two new flags, `--zone-id` and `--domain`, which identify the zone and the sending domain (zone apex or a verified subdomain) to subscribe to. Both flags are required for this source. The subscription's resource is displayed as the sending domain in `wrangler queues subscription get`.

packages/wrangler/src/__tests__/queues/queues-subscription.test.ts

@@ -61,6 +61,24 @@ describe("queues subscription", () => {
 		events: ["namespace.created"],
 	};
 
+	const mockSubscriptionEmail: EventSubscription = {
+		id: "sub-email",
+		created_at: "2024-01-01T00:00:00Z",
+		modified_at: "2024-01-01T00:00:00Z",
+		name: "Test Subscription Email",
+		enabled: true,
+		source: {
+			type: EventSourceType.EMAIL_SENDING,
+			zone_id: "zone-123",
+			domain: "example.com",
+		},
+		destination: {
+			type: "queues.queue",
+			queue_id: expectedQueueId,
+		},
+		events: ["message.delivered"],
+	};
+
 	describe("create", () => {
 		it("should show the correct help text", async ({ expect }) => {
 			await runWrangler("queues subscription create --help");
@@ -84,13 +102,15 @@ describe("queues subscription", () => {
 				  -v, --version         Show version number  [boolean]
 
 				OPTIONS
-				      --source         The event source type  [string] [required] [choices: "images", "kv", "r2", "superSlurper", "vectorize", "workersAi.model", "workersBuilds.worker", "workflows.workflow"]
+				      --source         The event source type  [string] [required] [choices: "email.sending", "images", "kv", "r2", "superSlurper", "vectorize", "workersAi.model", "workersBuilds.worker", "workflows.workflow"]
 				      --events         Comma-separated list of event types to subscribe to  [string] [required]
 				      --name           Name for the subscription (auto-generated if not provided)  [string]
 				      --enabled        Whether the subscription should be active  [boolean] [default: true]
 				      --model-name     Workers AI model name (required for workersAi.model source)  [string]
 				      --worker-name    Worker name (required for workersBuilds.worker source)  [string]
-				      --workflow-name  Workflow name (required for workflows.workflow source)  [string]"
+				      --workflow-name  Workflow name (required for workflows.workflow source)  [string]
+				      --zone-id        Zone ID (required for email.sending source)  [string]
+				      --domain         Sending domain — zone apex or verified subdomain (required for email.sending source)  [string]"
 			`);
 		});
 
@@ -187,6 +207,79 @@ describe("queues subscription", () => {
 			`);
 		});
 
+		it("should create a subscription for email.sending source", async ({
+			expect,
+		}) => {
+			const queueNameResolveRequest = mockGetQueueByNameRequest(
+				expectedQueueName,
+				{
+					queue_id: expectedQueueId,
+					queue_name: expectedQueueName,
+					created_on: "",
+					producers: [],
+					consumers: [],
+					producers_total_count: 0,
+					consumers_total_count: 0,
+					modified_on: "",
+				}
+			);
+
+			const expectedRequest: Partial<CreateEventSubscriptionRequest> = {
+				name: "testQueue email.sending",
+				enabled: true,
+				source: {
+					type: EventSourceType.EMAIL_SENDING,
+					zone_id: "zone-123",
+					domain: "example.com",
+				},
+				events: ["message.delivered"],
+			};
+
+			const createRequest = mockCreateSubscriptionRequest(
+				expectedRequest,
+				expectedQueueId
+			);
+
+			await runWrangler(
+				"queues subscription create testQueue --source email.sending --events message.delivered --zone-id zone-123 --domain example.com"
+			);
+
+			expect(queueNameResolveRequest.count).toEqual(1);
+			expect(createRequest.count).toEqual(1);
+			expect(std.err).toMatchInlineSnapshot(`""`);
+			expect(std.out).toMatchInlineSnapshot(`
+				"
+				 ⛅️ wrangler x.x.x
+				──────────────────
+				Creating event subscription for queue 'testQueue'...
+				✨ Successfully created event subscription 'testQueue email.sending' with id 'sub-123'."
+			`);
+		});
+
+		it("should show error when zone-id is missing for email.sending source", async ({
+			expect,
+		}) => {
+			await expect(
+				runWrangler(
+					"queues subscription create testQueue --source email.sending --events message.delivered --domain example.com"
+				)
+			).rejects.toThrowErrorMatchingInlineSnapshot(
+				`[Error: --zone-id is required when using source 'email.sending']`
+			);
+		});
+
+		it("should show error when domain is missing for email.sending source", async ({
+			expect,
+		}) => {
+			await expect(
+				runWrangler(
+					"queues subscription create testQueue --source email.sending --events message.delivered --zone-id zone-123"
+				)
+			).rejects.toThrowErrorMatchingInlineSnapshot(
+				`[Error: --domain is required when using source 'email.sending']`
+			);
+		});
+
 		it("should create subscription with custom name and disabled state", async ({
 			expect,
 		}) => {
@@ -254,7 +347,7 @@ describe("queues subscription", () => {
 				)
 			).rejects.toThrowErrorMatchingInlineSnapshot(`
 				[Error: Invalid values:
-				  Argument: source, Given: "invalid", Choices: "images", "kv", "r2", "superSlurper", "vectorize", "workersAi.model", "workersBuilds.worker", "workflows.workflow"]
+				  Argument: source, Given: "invalid", Choices: "email.sending", "images", "kv", "r2", "superSlurper", "vectorize", "workersAi.model", "workersBuilds.worker", "workflows.workflow"]
 			`);
 		});
 
@@ -528,6 +621,44 @@ describe("queues subscription", () => {
 			`);
 		});
 
+		it("should render the sending domain as the resource for email.sending source", async ({
+			expect,
+		}) => {
+			mockGetQueueByNameRequest("testQueue", {
+				queue_id: expectedQueueId,
+				queue_name: "testQueue",
+				created_on: "",
+				modified_on: "",
+				producers: [],
+				consumers: [],
+				producers_total_count: 0,
+				consumers_total_count: 0,
+			});
+			const getRequest = mockGetSubscriptionRequest(
+				"sub-email",
+				mockSubscriptionEmail
+			);
+
+			await runWrangler("queues subscription get testQueue --id sub-email");
+
+			expect(getRequest.count).toEqual(1);
+			expect(std.err).toMatchInlineSnapshot(`""`);
+			expect(std.out).toMatchInlineSnapshot(`
+				"
+				 ⛅️ wrangler x.x.x
+				──────────────────
+				ID:           sub-email
+				Name:         Test Subscription Email
+				Source:       email.sending
+				Resource:     example.com
+				Queue ID:     queueId
+				Events:       message.delivered
+				Enabled:      Yes
+				Created At:   1/1/2024, 12:00:00 AM
+				Modified At:  1/1/2024, 12:00:00 AM"
+			`);
+		});
+
 		it('supports valid json output with "--json" flag', async ({ expect }) => {
 			mockGetQueueByNameRequest("testQueue", {
 				queue_id: expectedQueueId,

packages/wrangler/src/queues/cli/commands/subscription/create.ts

@@ -17,9 +17,30 @@ function parseSourceArgument(
 		modelName?: string;
 		workerName?: string;
 		workflowName?: string;
+		zoneId?: string;
+		domain?: string;
 	}
 ): EventSource {
 	switch (source as EventSourceType) {
+		case EventSourceType.EMAIL_SENDING:
+			if (!args.zoneId) {
+				throw new UserError(
+					`--zone-id is required when using source '${EventSourceType.EMAIL_SENDING}'`,
+					{ telemetryMessage: "queues subscription create missing zone id" }
+				);
+			}
+			if (!args.domain) {
+				throw new UserError(
+					`--domain is required when using source '${EventSourceType.EMAIL_SENDING}'`,
+					{ telemetryMessage: "queues subscription create missing domain" }
+				);
+			}
+			return {
+				type: EventSourceType.EMAIL_SENDING,
+				zone_id: args.zoneId,
+				domain: args.domain,
+			};
+
 		case EventSourceType.IMAGES:
 			return { type: EventSourceType.IMAGES };
 
@@ -128,12 +149,23 @@ export const queuesSubscriptionCreateCommand = createCommand({
 			describe: "Workflow name (required for workflows.workflow source)",
 			type: "string",
 		},
+		"zone-id": {
+			describe: "Zone ID (required for email.sending source)",
+			type: "string",
+		},
+		domain: {
+			describe:
+				"Sending domain — zone apex or verified subdomain (required for email.sending source)",
+			type: "string",
+		},
 	},
 	async handler(args, { config }) {
 		const source = parseSourceArgument(args.source, {
 			modelName: args.modelName,
 			workerName: args.workerName,
 			workflowName: args.workflowName,
+			zoneId: args.zoneId,
+			domain: args.domain,
 		});
 
 		const events = args.events

packages/wrangler/src/queues/cli/commands/subscription/utils.ts

@@ -7,6 +7,8 @@ export function getSourceType(source: EventSource): string {
 
 export function getSourceResource(source: EventSource): string {
 	switch (source.type) {
+		case EventSourceType.EMAIL_SENDING:
+			return source.domain;
 		case EventSourceType.WORKERS_AI_MODEL:
 			return source.model_name;
 		case EventSourceType.WORKERS_BUILDS_WORKER:

packages/wrangler/src/queues/subscription-types.ts

@@ -15,6 +15,7 @@ export interface EventDestination {
 }
 
 export enum EventSourceType {
+	EMAIL_SENDING = "email.sending",
 	IMAGES = "images",
 	KV = "kv",
 	R2 = "r2",
@@ -28,6 +29,7 @@ export enum EventSourceType {
 export const EVENT_SOURCE_TYPES = Object.values(EventSourceType);
 
 export type EventSource =
+	| EmailSendingEventSource
 	| ImagesEventSource
 	| KvEventSource
 	| R2EventSource
@@ -37,6 +39,12 @@ export type EventSource =
 	| WorkersBuildsWorkerEventSource
 	| WorkflowsWorkflowEventSource;
 
+export interface EmailSendingEventSource {
+	type: EventSourceType.EMAIL_SENDING;
+	zone_id: string;
+	domain: string;
+}
+
 export interface ImagesEventSource {
 	type: EventSourceType.IMAGES;
 }