initial implementation

Author: conico974

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

@@ -1,10 +1,11 @@
 import { defineCloudflareConfig } from "@opennextjs/cloudflare";
 import d1TagCache from "@opennextjs/cloudflare/d1-tag-cache";
 import kvIncrementalCache from "@opennextjs/cloudflare/kv-cache";
-import memoryQueue from "@opennextjs/cloudflare/memory-queue";
+// import memoryQueue from "@opennextjs/cloudflare/memory-queue";
+import doQueue from "@opennextjs/cloudflare/durable-queue";
 
 export default defineCloudflareConfig({
   incrementalCache: kvIncrementalCache,
   tagCache: d1TagCache,
-  queue: memoryQueue,
+  queue: doQueue,
 });

examples/e2e/app-router/wrangler.jsonc

@@ -8,6 +8,22 @@
     "directory": ".open-next/assets",
     "binding": "ASSETS"
   },
+  "durable_objects": {
+    "bindings": [
+      {
+        "name": "NEXT_CACHE_REVALIDATION_DURABLE_OBJECT",
+        "class_name": "DurableObjectQueueHandler"
+      }
+    ]
+  },
+  "migrations": [
+    {
+      "tag": "v1",
+      "new_classes": [
+        "DurableObjectQueueHandler"
+      ]
+    }
+  ],
   "kv_namespaces": [
     {
       "binding": "NEXT_CACHE_WORKERS_KV",

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

@@ -1,12 +1,15 @@
 import type { Context, RunningCodeOptions } from "node:vm";
 
+import type { DurableObjectQueueHandler } from "./durable-objects/queue";
+
 declare global {
   interface CloudflareEnv {
     NEXT_CACHE_WORKERS_KV?: KVNamespace;
     NEXT_CACHE_D1?: D1Database;
     NEXT_CACHE_D1_TAGS_TABLE?: string;
     NEXT_CACHE_D1_REVALIDATIONS_TABLE?: string;
     NEXT_CACHE_REVALIDATION_WORKER?: Service;
+    NEXT_CACHE_REVALIDATION_DURABLE_OBJECT?: DurableObjectNamespace<DurableObjectQueueHandler>;
     ASSETS?: Fetcher;
   }
 }

packages/cloudflare/src/api/durable-objects/queue.ts

@@ -0,0 +1,73 @@
+import { error } from "@opennextjs/aws/adapters/logger.js";
+import type { QueueMessage } from "@opennextjs/aws/types/overrides";
+import { IgnorableError } from "@opennextjs/aws/utils/error.js";
+import { DurableObject } from "cloudflare:workers";
+
+const MAX_REVALIDATION_BY_DURABLE_OBJECT = 5;
+const DEFAULT_REVALIDATION_TIMEOUT_MS = 10_000;
+
+interface ExtendedQueueMessage extends QueueMessage {
+  previewModeId: string;
+}
+
+export class DurableObjectQueueHandler extends DurableObject<CloudflareEnv> {
+  // Ongoing revalidations are deduped by the deduplication id
+  // Since this is running in waitUntil, we expect the durable object state to persist this during the duration of the revalidation
+  // TODO: handle incremental cache with only eventual consistency (i.e. KV or R2/D1 with the optional cache layer on top)
+  ongoingRevalidations = new Map<string, Promise<void>>();
+
+  service: NonNullable<CloudflareEnv["NEXT_CACHE_REVALIDATION_WORKER"]>;
+
+  // TODO: allow this to be configurable
+  maxRevalidations = MAX_REVALIDATION_BY_DURABLE_OBJECT;
+
+  constructor(ctx: DurableObjectState, env: CloudflareEnv) {
+    super(ctx, env);
+    const service = env.NEXT_CACHE_REVALIDATION_WORKER;
+    // If there is no service binding, we throw an error because we can't revalidate without it
+    if (!service) throw new IgnorableError("No service binding for cache revalidation worker");
+    this.service = service;
+
+  }
+  
+  async revalidate(msg: ExtendedQueueMessage) {
+    // If there is already an ongoing revalidation, we don't need to revalidate again
+    if (this.ongoingRevalidations.has(msg.MessageDeduplicationId)) return;
+
+    if(this.ongoingRevalidations.size >= MAX_REVALIDATION_BY_DURABLE_OBJECT) {
+      const ongoingRevalidations = this.ongoingRevalidations.values()
+      await this.ctx.blockConcurrencyWhile(() => Promise.race(ongoingRevalidations));
+    }
+
+    const revalidationPromise = this.executeRevalidation(msg);
+
+    // We store the promise to dedupe the revalidation
+    this.ongoingRevalidations.set(
+      msg.MessageDeduplicationId,
+      revalidationPromise
+    );
+
+    this.ctx.waitUntil(revalidationPromise);
+  }
+
+  private async executeRevalidation({MessageBody: {host, url}, MessageDeduplicationId, previewModeId}: ExtendedQueueMessage) {
+    try {
+      const protocol = host.includes("localhost") ? "http" : "https";
+
+      //TODO: handle the different types of errors that can occur during the fetch (i.e. timeout, network error, etc)
+      await this.service.fetch(`${protocol}://${host}${url}`, {
+        method: "HEAD",
+        headers: {
+          "x-prerender-revalidate": previewModeId,
+          "x-isr": "1",
+        },
+        signal: AbortSignal.timeout(DEFAULT_REVALIDATION_TIMEOUT_MS)
+      })
+    } catch (e) {
+      error(e);
+    } finally {
+      this.ongoingRevalidations.delete(MessageDeduplicationId);
+    }
+  }
+  
+}

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

@@ -0,0 +1,22 @@
+import type { Queue, QueueMessage } from "@opennextjs/aws/types/overrides";
+import { IgnorableError } from "@opennextjs/aws/utils/error.js";
+
+import { getCloudflareContext } from "./cloudflare-context";
+
+
+
+export default {
+  name: "durable-queue",
+  send: async (msg: QueueMessage) => {
+    const durableObject = getCloudflareContext().env.NEXT_CACHE_REVALIDATION_DURABLE_OBJECT;
+    if (!durableObject) throw new IgnorableError("No durable object binding for cache revalidation");
+
+    const id = durableObject.idFromName(msg.MessageGroupId);
+    const stub = durableObject.get(id);
+    const previewModeId = process.env.__NEXT_PREVIEW_MODE_ID!;
+    await stub.revalidate({
+      ...msg,
+      previewModeId,
+    });
+  }
+} satisfies Queue;

packages/cloudflare/src/cli/build/bundle-server.ts

@@ -49,14 +49,30 @@ export async function bundleServer(buildOpts: BuildOptions): Promise<void> {
   patches.copyPackageCliFiles(packageDistDir, buildOpts);
 
   const { appPath, outputDir, monorepoRoot } = buildOpts;
-  const serverFiles = path.join(
+  const baseManifestPath = path.join(
     outputDir,
     "server-functions/default",
     getPackagePath(buildOpts),
-    ".next/required-server-files.json"
+    ".next"
+  )
+  const serverFiles = path.join(
+    baseManifestPath,
+    "required-server-files.json"
   );
   const nextConfig = JSON.parse(fs.readFileSync(serverFiles, "utf-8")).config;
 
+  // TODO: This is a temporary solution to get the previewModeId from the prerender-manifest.json
+  //       We should find a better way to get this value, probably directly provided from aws
+  //       probably in an env variable exactly as for BUILD_ID
+  const prerenderManifest = path.join(
+    baseManifestPath,
+    "prerender-manifest.json"
+  );
+  const prerenderManifestContent = fs.readFileSync(prerenderManifest, "utf-8");
+  const prerenderManifestJson = JSON.parse(prerenderManifestContent);
+  const previewModeId = prerenderManifestJson.preview.previewModeId;
+
+
   console.log(`\x1b[35m⚙️ Bundling the OpenNext server...\n\x1b[0m`);
 
   await patchWebpackRuntime(buildOpts);
@@ -144,6 +160,8 @@ export async function bundleServer(buildOpts: BuildOptions): Promise<void> {
       "process.env.TURBOPACK": "false",
       // This define should be safe to use for Next 14.2+, earlier versions (13.5 and less) will cause trouble
       "process.env.__NEXT_EXPERIMENTAL_REACT": `${needsExperimentalReact(nextConfig)}`,
+      // Used for the durable object queue handler
+      "process.env.__NEXT_PREVIEW_MODE_ID": `"${previewModeId}"`,
     },
     platform: "node",
     banner: {

packages/cloudflare/src/cli/templates/worker.ts

@@ -17,6 +17,9 @@ Object.defineProperty(globalThis, Symbol.for("__cloudflare-context__"), {
   },
 });
 
+//@ts-expect-error: Will be resolved by wrangler build
+export { DurableObjectQueueHandler } from "@opennextjs/cloudflare/durable-objects/queue";
+
 // Populate process.env on the first request
 let processEnvPopulated = false;