fix: deployed worker unable to invoke itself in memory queue

.changeset/spicy-seas-appear.md

@@ -0,0 +1,5 @@
+---
+"@opennextjs/cloudflare": patch
+---
+
+fix: deployed worker unable to invoke itself in memory queue

examples/e2e/app-pages-router/wrangler.json

@@ -13,5 +13,11 @@
       "binding": "NEXT_CACHE_WORKERS_KV",
       "id": "<BINDING_ID>"
     }
+  ],
+  "services": [
+    {
+      "binding": "NEXT_CACHE_REVALIDATION_WORKER",
+      "service": "app-pages-router"
+    }
   ]
 }

examples/e2e/app-router/wrangler.json

@@ -20,5 +20,11 @@
       "database_id": "NEXT_CACHE_D1",
       "database_name": "NEXT_CACHE_D1"
     }
+  ],
+  "services": [
+    {
+      "binding": "NEXT_CACHE_REVALIDATION_WORKER",
+      "service": "app-router"
+    }
   ]
 }

examples/e2e/pages-router/wrangler.json

@@ -13,5 +13,11 @@
       "binding": "NEXT_CACHE_WORKERS_KV",
       "id": "<BINDING_ID>"
     }
+  ],
+  "services": [
+    {
+      "binding": "NEXT_CACHE_REVALIDATION_WORKER",
+      "service": "pages-router"
+    }
   ]
 }

packages/cloudflare/src/api/cloudflare-context.ts

@@ -6,6 +6,7 @@ declare global {
     NEXT_CACHE_D1?: D1Database;
     NEXT_CACHE_D1_TAGS_TABLE?: string;
     NEXT_CACHE_D1_REVALIDATIONS_TABLE?: string;
+    NEXT_CACHE_REVALIDATION_WORKER?: Service;
     ASSETS?: Fetcher;
   }
 }

packages/cloudflare/src/api/memory-queue.spec.ts

@@ -1,10 +1,17 @@
 import { generateMessageGroupId } from "@opennextjs/aws/core/routing/queue.js";
 import { afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 
-import cache, { DEFAULT_REVALIDATION_TIMEOUT_MS } from "./memory-queue";
+import cache, { DEFAULT_REVALIDATION_TIMEOUT_MS } from "./memory-queue.js";
 
 vi.mock("./.next/prerender-manifest.json", () => Promise.resolve({ preview: { previewModeId: "id" } }));
 
+const mockServiceWorkerFetch = vi.fn();
+vi.mock("./cloudflare-context", () => ({
+  getCloudflareContext: () => ({
+    env: { NEXT_CACHE_REVALIDATION_WORKER: { fetch: mockServiceWorkerFetch } },
+  }),
+}));
+
 describe("MemoryQueue", () => {
   beforeAll(() => {
     vi.useFakeTimers();
@@ -21,7 +28,7 @@ describe("MemoryQueue", () => {
     });
     vi.advanceTimersByTime(DEFAULT_REVALIDATION_TIMEOUT_MS);
     await firstRequest;
-    expect(globalThis.internalFetch).toHaveBeenCalledTimes(1);
+    expect(mockServiceWorkerFetch).toHaveBeenCalledTimes(1);
 
     const secondRequest = cache.send({
       MessageBody: { host: "test.local", url: "/test" },
@@ -30,7 +37,7 @@ describe("MemoryQueue", () => {
     });
     vi.advanceTimersByTime(1);
     await secondRequest;
-    expect(globalThis.internalFetch).toHaveBeenCalledTimes(2);
+    expect(mockServiceWorkerFetch).toHaveBeenCalledTimes(2);
   });
 
   it("should process revalidations for multiple paths", async () => {
@@ -41,7 +48,7 @@ describe("MemoryQueue", () => {
     });
     vi.advanceTimersByTime(1);
     await firstRequest;
-    expect(globalThis.internalFetch).toHaveBeenCalledTimes(1);
+    expect(mockServiceWorkerFetch).toHaveBeenCalledTimes(1);
 
     const secondRequest = cache.send({
       MessageBody: { host: "test.local", url: "/test" },
@@ -50,7 +57,7 @@ describe("MemoryQueue", () => {
     });
     vi.advanceTimersByTime(1);
     await secondRequest;
-    expect(globalThis.internalFetch).toHaveBeenCalledTimes(2);
+    expect(mockServiceWorkerFetch).toHaveBeenCalledTimes(2);
   });
 
   it("should de-dupe revalidations", async () => {
@@ -68,6 +75,6 @@ describe("MemoryQueue", () => {
     ];
     vi.advanceTimersByTime(1);
     await Promise.all(requests);
-    expect(globalThis.internalFetch).toHaveBeenCalledTimes(1);
+    expect(mockServiceWorkerFetch).toHaveBeenCalledTimes(1);
   });
 });

packages/cloudflare/src/api/memory-queue.ts

@@ -1,12 +1,17 @@
 import logger from "@opennextjs/aws/logger.js";
 import type { Queue, QueueMessage } from "@opennextjs/aws/types/overrides.js";
+import { IgnorableError } from "@opennextjs/aws/utils/error.js";
+
+import { getCloudflareContext } from "./cloudflare-context";
 
 export const DEFAULT_REVALIDATION_TIMEOUT_MS = 10_000;
 
 /**
  * The Memory Queue offers basic ISR revalidation by directly requesting a revalidation of a route.
  *
  * It offers basic support for in-memory de-duping per isolate.
+ *
+ * A service binding called `NEXT_CACHE_REVALIDATION_WORKER` that points to your worker is required.
  */
 export class MemoryQueue implements Queue {
   readonly name = "memory-queue";
@@ -16,6 +21,9 @@ export class MemoryQueue implements Queue {
   constructor(private opts = { revalidationTimeoutMs: DEFAULT_REVALIDATION_TIMEOUT_MS }) {}
 
   async send({ MessageBody: { host, url }, MessageGroupId }: QueueMessage): Promise<void> {
+    const service = getCloudflareContext().env.NEXT_CACHE_REVALIDATION_WORKER;
+    if (!service) throw new IgnorableError("No service binding for cache revalidation worker");
+
     if (this.revalidatedPaths.has(MessageGroupId)) return;
 
     this.revalidatedPaths.set(
@@ -30,7 +38,7 @@ export class MemoryQueue implements Queue {
       // TODO: Drop the import - https://github.com/opennextjs/opennextjs-cloudflare/issues/361
       // @ts-ignore
       const manifest = await import("./.next/prerender-manifest.json");
-      await globalThis.internalFetch(`${protocol}://${host}${url}`, {
+      await service.fetch(`${protocol}://${host}${url}`, {
         method: "HEAD",
         headers: {
           "x-prerender-revalidate": manifest.preview.previewModeId,