Support Queue events in tail

Queue events have an "event" section containing two fields -- "queue"
and "batchSize", which contain the queue name and number of messages in
the batch. Queue events can be distinguished from other events by the
presence of the "queue" field, so that's what I do here.

I've followed the example of RequestEvent more closely than either
ScheduledEvent or AlarmEvent here, but am happy to change up the format
if there's a reason to do so.

.changeset/warm-queens-taste.md

@@ -0,0 +1,8 @@
+---
+"wrangler": patch
+---
+
+feature: Support Queue consumer events in tail
+
+So that it's less confusing when tailing a worker that consumes events from a
+Queue.

packages/wrangler/src/__tests__/pages-deployment-tail.test.ts

@@ -14,6 +14,7 @@ import type {
 	AlarmEvent,
 	EmailEvent,
 	TailInfo,
+	QueueEvent,
 } from "../tail/createTail";
 import type { RequestInit } from "undici";
 import type WebSocket from "ws";
@@ -378,6 +379,20 @@ describe("pages deployment tail", () => {
 			expect(std.out).toMatch(deserializeToJson(serializedMessage));
 		});
 
+		it("logs queue messages in json format", async () => {
+			const api = mockTailAPIs();
+			await runWrangler(
+				"pages deployment tail mock-deployment-id --project-name mock-project --format json"
+			);
+
+			const event = generateMockQueueEvent();
+			const message = generateMockEventMessage({ event });
+			const serializedMessage = serialize(message);
+
+			api.ws.send(serializedMessage);
+			expect(std.out).toMatch(deserializeToJson(serializedMessage));
+		});
+
 		it("logs request messages in pretty format", async () => {
 			const api = mockTailAPIs();
 			await runWrangler(
@@ -486,6 +501,33 @@ describe("pages deployment tail", () => {
 			`);
 		});
 
+		it("logs queue messages in pretty format", async () => {
+			const api = mockTailAPIs();
+			await runWrangler(
+				"pages deployment tail mock-deployment-id --project-name mock-project --format pretty"
+			);
+
+			const event = generateMockQueueEvent();
+			const message = generateMockEventMessage({ event });
+			const serializedMessage = serialize(message);
+
+			api.ws.send(serializedMessage);
+			expect(
+				std.out
+					.replace(
+						new Date(mockEventTimestamp).toLocaleString(),
+						"[mock timestamp string]"
+					)
+					.replace(
+						mockTailExpiration.toLocaleString(),
+						"[mock expiration date]"
+					)
+			).toMatchInlineSnapshot(`
+					"Connected to deployment mock-deployment-id, waiting for logs...
+					Queue my-queue123 (7 messages) - Ok @ [mock timestamp string]"
+			`);
+		});
+
 		it("should not crash when the tail message has a void event", async () => {
 			const api = mockTailAPIs();
 			await runWrangler(
@@ -664,6 +706,7 @@ function isRequest(
 		| AlarmEvent
 		| EmailEvent
 		| TailInfo
+		| QueueEvent
 		| undefined
 		| null
 ): event is RequestEvent {
@@ -962,3 +1005,9 @@ function generateMockEmailEvent(opts?: Partial<EmailEvent>): EmailEvent {
 		rawSize: opts?.rawSize || mockEmailEventSize,
 	};
 }
+function generateMockQueueEvent(opts?: Partial<QueueEvent>): QueueEvent {
+	return {
+		queue: opts?.queue || "my-queue123",
+		batchSize: opts?.batchSize || 7,
+	};
+}

packages/wrangler/src/__tests__/tail.test.ts

@@ -15,6 +15,7 @@ import type {
 	AlarmEvent,
 	EmailEvent,
 	TailInfo,
+	QueueEvent,
 } from "../tail/createTail";
 import type { RequestInit } from "undici";
 import type WebSocket from "ws";
@@ -426,6 +427,18 @@ describe("tail", () => {
 			expect(std.out).toMatch(deserializeToJson(serializedMessage));
 		});
 
+		it("logs queue messages in json format", async () => {
+			const api = mockWebsocketAPIs();
+			await runWrangler("tail test-worker --format json");
+
+			const event = generateMockQueueEvent();
+			const message = generateMockEventMessage({ event });
+			const serializedMessage = serialize(message);
+
+			api.ws.send(serializedMessage);
+			expect(std.out).toMatch(deserializeToJson(serializedMessage));
+		});
+
 		it("logs request messages in pretty format", async () => {
 			const api = mockWebsocketAPIs();
 			await runWrangler("tail test-worker --format pretty");
@@ -544,6 +557,29 @@ describe("tail", () => {
 		`);
 		});
 
+		it("logs queue messages in pretty format", async () => {
+			const api = mockWebsocketAPIs();
+			await runWrangler("tail test-worker --format pretty");
+
+			const event = generateMockQueueEvent();
+			const message = generateMockEventMessage({ event });
+			const serializedMessage = serialize(message);
+
+			api.ws.send(serializedMessage);
+			expect(
+				std.out
+					.replace(
+						new Date(mockEventTimestamp).toLocaleString(),
+						"[mock timestamp string]"
+					)
+					.replace(mockTailExpiration.toISOString(), "[mock expiration date]")
+			).toMatchInlineSnapshot(`
+			        "Successfully created tail, expires at [mock expiration date]
+			        Connected to test-worker, waiting for logs...
+			        Queue my-queue123 (7 messages) - Ok @ [mock timestamp string]"
+		      `);
+		});
+
 		it("should not crash when the tail message has a void event", async () => {
 			const api = mockWebsocketAPIs();
 			await runWrangler("tail test-worker --format pretty");
@@ -711,6 +747,7 @@ function isRequest(
 		| AlarmEvent
 		| EmailEvent
 		| TailInfo
+		| QueueEvent
 		| undefined
 		| null
 ): event is RequestEvent {
@@ -1006,3 +1043,10 @@ function generateTailInfo(overload: boolean): TailInfo {
 				type: "overload-stop",
 		  };
 }
+
+function generateMockQueueEvent(opts?: Partial<QueueEvent>): QueueEvent {
+	return {
+		queue: opts?.queue || "my-queue123",
+		batchSize: opts?.batchSize || 7,
+	};
+}

packages/wrangler/src/tail/createTail.ts

@@ -242,7 +242,8 @@ export type TailEventMessage = {
 	 * The event that triggered the worker. In the case of an HTTP request,
 	 * this will be a RequestEvent. If it's a cron trigger, it'll be a
 	 * ScheduledEvent. If it's a durable object alarm, it's an AlarmEvent.
-	 * If it's a email, it'a an EmailEvent
+	 * If it's a email, it'a an EmailEvent. If it's a Queue consumer event,
+	 * it's a QueueEvent.
 	 *
 	 * Until workers-types exposes individual types for export, we'll have
 	 * to just re-define these types ourselves.
@@ -253,6 +254,7 @@ export type TailEventMessage = {
 		| AlarmEvent
 		| EmailEvent
 		| TailInfo
+		| QueueEvent
 		| undefined
 		| null;
 };
@@ -413,3 +415,18 @@ export type TailInfo = {
 	message: string;
 	type: string;
 };
+
+/*
+ * A event that was triggered by receiving a batch of messages from a Queue for consumption.
+ */
+export type QueueEvent = {
+	/**
+	 * The name of the queue that the message batch came from.
+	 */
+	queue: string;
+
+	/**
+	 * The number of messages in the batch.
+	 */
+	batchSize: number;
+};

packages/wrangler/src/tail/printing.ts

@@ -3,6 +3,7 @@ import { logger } from "../logger";
 import type {
 	AlarmEvent,
 	EmailEvent,
+	QueueEvent,
 	RequestEvent,
 	ScheduledEvent,
 	TailInfo,
@@ -56,6 +57,16 @@ export function prettyPrintLogs(data: WebSocket.RawData): void {
 		} else if (eventMessage.event.type === "overload-stop") {
 			logger.log(`${chalk.yellow.bold(eventMessage.event.message)}`);
 		}
+	} else if (isQueueEvent(eventMessage.event)) {
+		const outcome = prettifyOutcome(eventMessage.outcome);
+		const datetime = new Date(eventMessage.eventTimestamp).toLocaleString();
+		const queueName = eventMessage.event.queue;
+		const batchSize = eventMessage.event.batchSize;
+		const batchSizeMsg = `${batchSize} message${batchSize !== 1 ? "s" : ""}`;
+
+		logger.log(
+			`Queue ${queueName} (${batchSizeMsg}) - ${outcome} @ ${datetime}`
+		);
 	} else {
 		// Unknown event type
 		const outcome = prettifyOutcome(eventMessage.outcome);
@@ -97,6 +108,10 @@ function isEmailEvent(event: TailEventMessage["event"]): event is EmailEvent {
 	return Boolean(event && "mailFrom" in event);
 }
 
+function isQueueEvent(event: TailEventMessage["event"]): event is QueueEvent {
+	return Boolean(event && "queue" in event);
+}
+
 /**
  * Check to see if an event sent from a worker is an AlarmEvent.
  *