feat: basic in-memory de-duping revalidation queue

.changeset/witty-baboons-smile.md

@@ -0,0 +1,5 @@
+---
+"@opennextjs/cloudflare": minor
+---
+
+feat: basic in-memory de-duping revalidation queue

examples/e2e/app-router/open-next.config.ts

@@ -1,13 +1,14 @@
 import type { OpenNextConfig } from "@opennextjs/aws/types/open-next.js";
 import cache from "@opennextjs/cloudflare/kv-cache";
+import memoryQueue from "@opennextjs/cloudflare/memory-queue";
 
 const config: OpenNextConfig = {
   default: {
     override: {
       wrapper: "cloudflare-node",
       converter: "edge",
       incrementalCache: async () => cache,
-      queue: "direct",
+      queue: () => memoryQueue,
       // Unused implementation
       tagCache: "dummy",
     },

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

@@ -0,0 +1,36 @@
+import { generateMessageGroupId } from "@opennextjs/aws/core/routing/queue.js";
+import { beforeAll, describe, expect, it, vi } from "vitest";
+
+import cache from "./memory-queue";
+
+vi.mock("./.next/prerender-manifest.json", () => Promise.resolve({ preview: { previewModeId: "id" } }));
+
+const defaultOpts = {
+  MessageBody: { host: "test.local", url: "/test" },
+  MessageGroupId: generateMessageGroupId("/test"),
+  MessageDeduplicationId: "",
+};
+
+describe("MemoryQueue", () => {
+  beforeAll(() => {
+    vi.useFakeTimers();
+    globalThis.internalFetch = vi.fn().mockReturnValue(new Promise((res) => setTimeout(() => res(true), 1)));
+  });
+
+  it("should de-dupe revalidations", async () => {
+    const firstBatch = [cache.send(defaultOpts), cache.send(defaultOpts)];
+    vi.advanceTimersByTime(1);
+    await Promise.all(firstBatch);
+    expect(globalThis.internalFetch).toHaveBeenCalledTimes(1);
+
+    const secondBatch = [cache.send(defaultOpts)];
+    vi.advanceTimersByTime(10_000);
+    await Promise.all(secondBatch);
+    expect(globalThis.internalFetch).toHaveBeenCalledTimes(2);
+
+    const thirdBatch = [cache.send({ ...defaultOpts, MessageGroupId: generateMessageGroupId("/other") })];
+    vi.advanceTimersByTime(1);
+    await Promise.all(thirdBatch);
+    expect(globalThis.internalFetch).toHaveBeenCalledTimes(3);
+  });
+});

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

@@ -0,0 +1,45 @@
+import logger from "@opennextjs/aws/logger.js";
+import type { Queue, QueueMessage } from "@opennextjs/aws/types/overrides.js";
+
+/**
+ * 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.
+ */
+class MemoryQueue implements Queue {
+  readonly name = "memory-queue";
+
+  public revalidatedPaths = new Map<string, ReturnType<typeof setTimeout>>();
+
+  public async send({ MessageBody: { host, url }, MessageGroupId }: QueueMessage): Promise<void> {
+    if (this.revalidatedPaths.has(MessageGroupId)) return;
+
+    this.revalidatedPaths.set(
+      MessageGroupId,
+      // force remove to allow new revalidations incase something went wrong
+      setTimeout(() => this.revalidatedPaths.delete(MessageGroupId), 10_000)
+    );
+
+    try {
+      const protocol = host.includes("localhost") ? "http" : "https";
+
+      // 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}`, {
+        method: "HEAD",
+        headers: {
+          "x-prerender-revalidate": manifest.preview.previewModeId,
+          "x-isr": "1",
+        },
+      });
+    } catch (e) {
+      logger.error(e);
+    } finally {
+      clearTimeout(this.revalidatedPaths.get(MessageGroupId));
+      this.revalidatedPaths.delete(MessageGroupId);
+    }
+  }
+}
+
+export default new MemoryQueue();

packages/cloudflare/src/cli/build/utils/ensure-cf-config.ts

@@ -15,7 +15,9 @@ export function ensureCloudflareConfig(config: OpenNextConfig) {
       typeof config.default?.override?.incrementalCache === "function",
     dftUseDummyTagCache: config.default?.override?.tagCache === "dummy",
     dftMaybeUseQueue:
-      config.default?.override?.queue === "dummy" || config.default?.override?.queue === "direct",
+      config.default?.override?.queue === "dummy" ||
+      config.default?.override?.queue === "direct" ||
+      typeof config.default?.override?.incrementalCache === "function",
     disableCacheInterception: config.dangerous?.enableCacheInterception !== true,
     mwIsMiddlewareExternal: config.middleware?.external == true,
     mwUseCloudflareWrapper: config.middleware?.override?.wrapper === "cloudflare-edge",
@@ -37,7 +39,7 @@ export function ensureCloudflareConfig(config: OpenNextConfig) {
               converter: "edge",
               incrementalCache: "dummy" | function,
               tagCache: "dummy",
-              queue: "dummy" | "direct",
+              queue: "dummy" | "direct" | function,
             },
           },
 

packages/cloudflare/tsconfig.json

@@ -13,7 +13,7 @@
     "noPropertyAccessFromIndexSignature": false,
     "outDir": "./dist",
     "target": "ES2022",
-    "types": ["@cloudflare/workers-types"]
+    "types": ["@cloudflare/workers-types", "@opennextjs/aws/types/global.d.ts"]
   },
   "include": ["src/**/*.ts"]
 }