feat: basic in-memory de-duping revalidation queue (#360)

* feat: basic in-memory de-duping revalidation queue

* change to pr for package

* add tests and fix mistake

* remove dependency on memory queue from kv cache

* Revert "remove dependency on memory queue from kv cache"

This reverts commit 42739ae.

* remove dependency on memory queue from kv cache again

* move manifest retrievel to own util for mocking

* get test to be reliable

* configurable revalidation timeout

* split up tests

* review comments

Author: james-elicx

.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,73 @@
+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";
+
+vi.mock("./.next/prerender-manifest.json", () => Promise.resolve({ preview: { previewModeId: "id" } }));
+
+describe("MemoryQueue", () => {
+  beforeAll(() => {
+    vi.useFakeTimers();
+    globalThis.internalFetch = vi.fn().mockReturnValue(new Promise((res) => setTimeout(() => res(true), 1)));
+  });
+
+  afterEach(() => vi.clearAllMocks());
+
+  it("should process revalidations for a path", async () => {
+    const firstRequest = cache.send({
+      MessageBody: { host: "test.local", url: "/test" },
+      MessageGroupId: generateMessageGroupId("/test"),
+      MessageDeduplicationId: "",
+    });
+    vi.advanceTimersByTime(DEFAULT_REVALIDATION_TIMEOUT_MS);
+    await firstRequest;
+    expect(globalThis.internalFetch).toHaveBeenCalledTimes(1);
+
+    const secondRequest = cache.send({
+      MessageBody: { host: "test.local", url: "/test" },
+      MessageGroupId: generateMessageGroupId("/test"),
+      MessageDeduplicationId: "",
+    });
+    vi.advanceTimersByTime(1);
+    await secondRequest;
+    expect(globalThis.internalFetch).toHaveBeenCalledTimes(2);
+  });
+
+  it("should process revalidations for multiple paths", async () => {
+    const firstRequest = cache.send({
+      MessageBody: { host: "test.local", url: "/test" },
+      MessageGroupId: generateMessageGroupId("/test"),
+      MessageDeduplicationId: "",
+    });
+    vi.advanceTimersByTime(1);
+    await firstRequest;
+    expect(globalThis.internalFetch).toHaveBeenCalledTimes(1);
+
+    const secondRequest = cache.send({
+      MessageBody: { host: "test.local", url: "/test" },
+      MessageGroupId: generateMessageGroupId("/other"),
+      MessageDeduplicationId: "",
+    });
+    vi.advanceTimersByTime(1);
+    await secondRequest;
+    expect(globalThis.internalFetch).toHaveBeenCalledTimes(2);
+  });
+
+  it("should de-dupe revalidations", async () => {
+    const requests = [
+      cache.send({
+        MessageBody: { host: "test.local", url: "/test" },
+        MessageGroupId: generateMessageGroupId("/test"),
+        MessageDeduplicationId: "",
+      }),
+      cache.send({
+        MessageBody: { host: "test.local", url: "/test" },
+        MessageGroupId: generateMessageGroupId("/test"),
+        MessageDeduplicationId: "",
+      }),
+    ];
+    vi.advanceTimersByTime(1);
+    await Promise.all(requests);
+    expect(globalThis.internalFetch).toHaveBeenCalledTimes(1);
+  });
+});

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

@@ -0,0 +1,49 @@
+import logger from "@opennextjs/aws/logger.js";
+import type { Queue, QueueMessage } from "@opennextjs/aws/types/overrides.js";
+
+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.
+ */
+export class MemoryQueue implements Queue {
+  readonly name = "memory-queue";
+
+  revalidatedPaths = new Map<string, ReturnType<typeof setTimeout>>();
+
+  constructor(private opts = { revalidationTimeoutMs: DEFAULT_REVALIDATION_TIMEOUT_MS }) {}
+
+  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), this.opts.revalidationTimeoutMs)
+    );
+
+    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?.queue === "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"]
 }