diff --git a/packages/cfworkers/src/mod.test.ts b/packages/cfworkers/src/mod.test.ts index 0ed720606..50c63a2fa 100644 --- a/packages/cfworkers/src/mod.test.ts +++ b/packages/cfworkers/src/mod.test.ts @@ -4,6 +4,15 @@ import { WorkersKvStore, WorkersMessageQueue, } from "./mod.ts"; +import type { + MessageSendRequest, + Queue, + QueueMetrics, + QueueSendBatchOptions, + QueueSendBatchResponse, + QueueSendOptions, + QueueSendResponse, +} from "@cloudflare/workers-types"; // Mock Temporal.Duration for testing in Cloudflare Workers environment const mockDuration = (seconds: number) => ({ @@ -352,3 +361,98 @@ describe("WorkersMessageQueue", () => { expect(second.message).toEqual({ id: "second" }); }); }); + +interface MockWrappedMessage { + readonly __fedify_ordering_key__?: string; + readonly __fedify_payload__: unknown; +} + +class MockQueue implements Queue { + metrics(): Promise { + return Promise.resolve({ backlogCount: 0, backlogBytes: 0 }); + } + sentSingles: { + wrapped: MockWrappedMessage; + options?: QueueSendOptions; + }[] = []; + sentBatches: { + batch: MessageSendRequest[]; + options?: QueueSendBatchOptions; + }[] = []; + + send( + wrapped: MockWrappedMessage, + options?: QueueSendOptions, + ): Promise { + this.sentSingles.push({ wrapped, options }); + return Promise.resolve({ + metadata: { metrics: { backlogCount: 0, backlogBytes: 0 } }, + }); + } + + sendBatch( + batch: Iterable>, + options?: QueueSendBatchOptions, + ): Promise { + this.sentBatches.push({ batch: [...batch], options }); + return Promise.resolve({ + metadata: { metrics: { backlogCount: 0, backlogBytes: 0 } }, + }); + } +} + +describe("WorkersMessageQueue.enqueueMany() - wrapped message shape", () => { + it("enqueueMany() - wraps each message with body{} and contentType", async () => { + const sendingMockQueue = new MockQueue(); + const queue = new WorkersMessageQueue(sendingMockQueue); + + await queue.enqueueMany(["msg-1", "msg-2"], { orderingKey: "ferer" }); + + expect(sendingMockQueue.sentBatches).toHaveLength(1); + expect(sendingMockQueue.sentBatches[0].batch).toEqual([ + { + body: { + __fedify_ordering_key__: "ferer", + __fedify_payload__: "msg-1", + }, + contentType: "json", + }, + { + body: { + __fedify_ordering_key__: "ferer", + __fedify_payload__: "msg-2", + }, + contentType: "json", + }, + ]); + }); + + it("enqueue() and enqueueMany() - produce the same wrapped shape", async () => { + const sendingMockQueue = new MockQueue(); + const queue = new WorkersMessageQueue(sendingMockQueue); + + await queue.enqueue("msg-1", { orderingKey: "ferer" }); + await queue.enqueueMany(["msg-1"], { orderingKey: "ferer" }); + + // enqueueMany() wraps each body in a request ({ body, contentType }), so + // unwrap it before comparing against enqueue()'s bare wrapped message. + const singleWrapped = sendingMockQueue.sentSingles[0].wrapped; + const batchWrapped = sendingMockQueue.sentBatches[0].batch[0].body; + + expect(batchWrapped).toEqual(singleWrapped); + }); + + it("enqueueMany() - omits ordering key when not provided", async () => { + const sendingMockQueue = new MockQueue(); + const queue = new WorkersMessageQueue(sendingMockQueue); + + await queue.enqueueMany(["msg-1"]); + + expect(sendingMockQueue.sentBatches[0].batch[0].body).toEqual( + { + __fedify_ordering_key__: undefined, + __fedify_payload__: "msg-1", + }, + ); + }); +});