Feat Automatic cache purge (#655)

* basic purge cache

* lint and fix e2e

* DO cache purge

* added unit test

* refactor and lint fix

* review fix

* review

* lint fix

* basic error handling

* changeset

* fix linting

---------

Co-authored-by: Nicolas Dorseuil <[email protected]>

Author: conico974

.changeset/deep-mails-throw.md

@@ -0,0 +1,5 @@
+---
+"@opennextjs/cloudflare": minor
+---
+
+Automatic Cache API purge

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

@@ -3,6 +3,7 @@ import r2IncrementalCache from "@opennextjs/cloudflare/overrides/incremental-cac
 import shardedTagCache from "@opennextjs/cloudflare/overrides/tag-cache/do-sharded-tag-cache";
 import doQueue from "@opennextjs/cloudflare/overrides/queue/do-queue";
 import queueCache from "@opennextjs/cloudflare/overrides/queue/queue-cache";
+import { purgeCache } from "@opennextjs/cloudflare/overrides/cache-purge/index";
 
 export default defineCloudflareConfig({
   incrementalCache: r2IncrementalCache,
@@ -17,6 +18,7 @@ export default defineCloudflareConfig({
       },
     },
   }),
+  cachePurge: purgeCache({ type: "durableObject" }),
   enableCacheInterception: true,
   queue: queueCache(doQueue),
 });

examples/e2e/app-router/wrangler.jsonc

@@ -17,13 +17,17 @@
       {
         "name": "NEXT_TAG_CACHE_DO_SHARDED",
         "class_name": "DOShardedTagCache"
+      },
+      {
+        "name": "NEXT_CACHE_DO_PURGE",
+        "class_name": "BucketCachePurge"
       }
     ]
   },
   "migrations": [
     {
       "tag": "v1",
-      "new_sqlite_classes": ["DOQueueHandler", "DOShardedTagCache"]
+      "new_sqlite_classes": ["DOQueueHandler", "DOShardedTagCache", "BucketCachePurge"]
     }
   ],
   "r2_buckets": [

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

@@ -2,6 +2,7 @@ import type { Context, RunningCodeOptions } from "node:vm";
 
 import type { GetPlatformProxyOptions } from "wrangler";
 
+import type { BucketCachePurge } from "./durable-objects/bucket-cache-purge.js";
 import type { DOQueueHandler } from "./durable-objects/queue.js";
 import type { DOShardedTagCache } from "./durable-objects/sharded-tag-cache.js";
 import type { PREFIX_ENV_NAME as KV_CACHE_PREFIX_ENV_NAME } from "./overrides/incremental-cache/kv-incremental-cache.js";
@@ -54,6 +55,16 @@ declare global {
     // Disable SQLite for the durable object queue handler
     // This can be safely used if you don't use an eventually consistent incremental cache (i.e. R2 without the regional cache for example)
     NEXT_CACHE_DO_QUEUE_DISABLE_SQLITE?: string;
+
+    // Below are the optional env variables for purging the cache
+    // Durable Object namespace to use for the durable object cache purge
+    NEXT_CACHE_DO_PURGE?: DurableObjectNamespace<BucketCachePurge>;
+    // The amount of time in seconds that the cache purge will wait before purging the cache
+    NEXT_CACHE_DO_PURGE_BUFFER_TIME_IN_SECONDS?: string;
+    // The zone ID to use for the cache purge https://developers.cloudflare.com/fundamentals/setup/find-account-and-zone-ids/
+    CACHE_PURGE_ZONE_ID?: string;
+    // The API token to use for the cache purge. It should have the `Cache Purge` permission
+    CACHE_PURGE_API_TOKEN?: string;
   }
 }
 

packages/cloudflare/src/api/config.ts

@@ -4,7 +4,12 @@ import {
   LazyLoadedOverride,
   OpenNextConfig as AwsOpenNextConfig,
 } from "@opennextjs/aws/types/open-next";
-import type { IncrementalCache, Queue, TagCache } from "@opennextjs/aws/types/overrides";
+import type {
+  CDNInvalidationHandler,
+  IncrementalCache,
+  Queue,
+  TagCache,
+} from "@opennextjs/aws/types/overrides";
 
 export type Override<T extends BaseOverride> = "dummy" | T | LazyLoadedOverride<T>;
 
@@ -29,6 +34,11 @@ export type CloudflareOverrides = {
    */
   queue?: "direct" | Override<Queue>;
 
+  /**
+   * Sets the automatic cache purge implementation
+   */
+  cachePurge?: Override<CDNInvalidationHandler>;
+
   /**
    * Enable cache interception
    * Should be `false` when PPR is used
@@ -44,7 +54,7 @@ export type CloudflareOverrides = {
  * @returns the OpenNext configuration object
  */
 export function defineCloudflareConfig(config: CloudflareOverrides = {}): OpenNextConfig {
-  const { incrementalCache, tagCache, queue, enableCacheInterception = false } = config;
+  const { incrementalCache, tagCache, queue, cachePurge, enableCacheInterception = false } = config;
 
   return {
     default: {
@@ -55,6 +65,7 @@ export function defineCloudflareConfig(config: CloudflareOverrides = {}): OpenNe
         incrementalCache: resolveIncrementalCache(incrementalCache),
         tagCache: resolveTagCache(tagCache),
         queue: resolveQueue(queue),
+        cdnInvalidation: resolveCdnInvalidation(cachePurge),
       },
       routePreloadingBehavior: "withWaitUntil",
     },
@@ -104,6 +115,14 @@ function resolveQueue(value: CloudflareOverrides["queue"] = "dummy") {
   return typeof value === "function" ? value : () => value;
 }
 
+function resolveCdnInvalidation(value: CloudflareOverrides["cachePurge"] = "dummy") {
+  if (typeof value === "string") {
+    return value;
+  }
+
+  return typeof value === "function" ? value : () => value;
+}
+
 interface OpenNextConfig extends AwsOpenNextConfig {
   cloudflare?: {
     /**

packages/cloudflare/src/api/durable-objects/bucket-cache-purge.spec.ts

@@ -0,0 +1,149 @@
+import { describe, expect, it, vi } from "vitest";
+
+import * as internal from "../overrides/internal";
+import { BucketCachePurge } from "./bucket-cache-purge";
+
+vi.mock("cloudflare:workers", () => ({
+  DurableObject: class {
+    constructor(
+      public ctx: DurableObjectState,
+      public env: CloudflareEnv
+    ) {}
+  },
+}));
+
+const createBucketCachePurge = () => {
+  const mockState = {
+    waitUntil: vi.fn(),
+    blockConcurrencyWhile: vi.fn().mockImplementation(async (fn) => fn()),
+    storage: {
+      setAlarm: vi.fn(),
+      getAlarm: vi.fn(),
+      sql: {
+        exec: vi.fn().mockImplementation(() => ({
+          one: vi.fn(),
+          toArray: vi.fn().mockReturnValue([]),
+        })),
+      },
+    },
+  };
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  return new BucketCachePurge(mockState as any, {});
+};
+
+describe("BucketCachePurge", () => {
+  it("should block concurrency while creating the table", async () => {
+    const cache = createBucketCachePurge();
+    // @ts-expect-error - testing private method
+    expect(cache.ctx.blockConcurrencyWhile).toHaveBeenCalled();
+    // @ts-expect-error - testing private method
+    expect(cache.ctx.storage.sql.exec).toHaveBeenCalledWith(
+      expect.stringContaining("CREATE TABLE IF NOT EXISTS cache_purge")
+    );
+  });
+
+  describe("purgeCacheByTags", () => {
+    it("should insert tags into the sql table", async () => {
+      const cache = createBucketCachePurge();
+      const tags = ["tag1", "tag2"];
+      await cache.purgeCacheByTags(tags);
+      // @ts-expect-error - testing private method
+      expect(cache.ctx.storage.sql.exec).toHaveBeenCalledWith(
+        expect.stringContaining("INSERT OR REPLACE INTO cache_purge"),
+        [tags[0]]
+      );
+      // @ts-expect-error - testing private method
+      expect(cache.ctx.storage.sql.exec).toHaveBeenCalledWith(
+        expect.stringContaining("INSERT OR REPLACE INTO cache_purge"),
+        [tags[1]]
+      );
+    });
+
+    it("should set an alarm if no alarm is set", async () => {
+      const cache = createBucketCachePurge();
+      // @ts-expect-error - testing private method
+      cache.ctx.storage.getAlarm.mockResolvedValueOnce(null);
+      await cache.purgeCacheByTags(["tag"]);
+      // @ts-expect-error - testing private method
+      expect(cache.ctx.storage.setAlarm).toHaveBeenCalled();
+    });
+
+    it("should not set an alarm if one is already set", async () => {
+      const cache = createBucketCachePurge();
+      // @ts-expect-error - testing private method
+      cache.ctx.storage.getAlarm.mockResolvedValueOnce(true);
+      await cache.purgeCacheByTags(["tag"]);
+      // @ts-expect-error - testing private method
+      expect(cache.ctx.storage.setAlarm).not.toHaveBeenCalled();
+    });
+  });
+
+  describe("alarm", () => {
+    it("should purge cache by tags and delete them from the sql table", async () => {
+      const cache = createBucketCachePurge();
+      // @ts-expect-error - testing private method
+      cache.ctx.storage.sql.exec.mockReturnValueOnce({
+        toArray: () => [{ tag: "tag1" }, { tag: "tag2" }],
+      });
+      await cache.alarm();
+      // @ts-expect-error - testing private method
+      expect(cache.ctx.storage.sql.exec).toHaveBeenCalledWith(
+        expect.stringContaining("DELETE FROM cache_purge"),
+        ["tag1", "tag2"]
+      );
+    });
+    it("should not purge cache if no tags are found", async () => {
+      const cache = createBucketCachePurge();
+      // @ts-expect-error - testing private method
+      cache.ctx.storage.sql.exec.mockReturnValueOnce({
+        toArray: () => [],
+      });
+      await cache.alarm();
+      // @ts-expect-error - testing private method
+      expect(cache.ctx.storage.sql.exec).not.toHaveBeenCalledWith(
+        expect.stringContaining("DELETE FROM cache_purge"),
+        []
+      );
+    });
+
+    it("should call internalPurgeCacheByTags with the correct tags", async () => {
+      const cache = createBucketCachePurge();
+      const tags = ["tag1", "tag2"];
+      // @ts-expect-error - testing private method
+      cache.ctx.storage.sql.exec.mockReturnValueOnce({
+        toArray: () => tags.map((tag) => ({ tag })),
+      });
+      const internalPurgeCacheByTagsSpy = vi.spyOn(internal, "internalPurgeCacheByTags");
+      await cache.alarm();
+      expect(internalPurgeCacheByTagsSpy).toHaveBeenCalledWith(
+        // @ts-expect-error - testing private method
+        cache.env,
+        tags
+      );
+      // @ts-expect-error - testing private method 1st is constructor, 2nd is to get the tags and 3rd is to delete them
+      expect(cache.ctx.storage.sql.exec).toHaveBeenCalledTimes(3);
+    });
+
+    it("should continue until all tags are purged", async () => {
+      const cache = createBucketCachePurge();
+      const tags = Array.from({ length: 100 }, (_, i) => `tag${i}`);
+      // @ts-expect-error - testing private method
+      cache.ctx.storage.sql.exec.mockReturnValueOnce({
+        toArray: () => tags.map((tag) => ({ tag })),
+      });
+      const internalPurgeCacheByTagsSpy = vi.spyOn(internal, "internalPurgeCacheByTags");
+      await cache.alarm();
+      expect(internalPurgeCacheByTagsSpy).toHaveBeenCalledWith(
+        // @ts-expect-error - testing private method
+        cache.env,
+        tags
+      );
+      // @ts-expect-error - testing private method 1st is constructor, 2nd is to get the tags and 3rd is to delete them, 4th is to get the next 100 tags
+      expect(cache.ctx.storage.sql.exec).toHaveBeenCalledTimes(4);
+      // @ts-expect-error - testing private method
+      expect(cache.ctx.storage.sql.exec).toHaveBeenLastCalledWith(
+        expect.stringContaining("SELECT * FROM cache_purge LIMIT 100")
+      );
+    });
+  });
+});

packages/cloudflare/src/api/durable-objects/bucket-cache-purge.ts

@@ -0,0 +1,95 @@
+import { DurableObject } from "cloudflare:workers";
+
+import { internalPurgeCacheByTags } from "../overrides/internal";
+
+const DEFAULT_BUFFER_TIME_IN_SECONDS = 5;
+// https://developers.cloudflare.com/cache/how-to/purge-cache/#hostname-tag-prefix-url-and-purge-everything-limits
+const MAX_NUMBER_OF_TAGS_PER_PURGE = 100;
+
+export class BucketCachePurge extends DurableObject<CloudflareEnv> {
+  bufferTimeInSeconds: number;
+
+  constructor(state: DurableObjectState, env: CloudflareEnv) {
+    super(state, env);
+    this.bufferTimeInSeconds = env.NEXT_CACHE_DO_PURGE_BUFFER_TIME_IN_SECONDS
+      ? parseInt(env.NEXT_CACHE_DO_PURGE_BUFFER_TIME_IN_SECONDS)
+      : DEFAULT_BUFFER_TIME_IN_SECONDS; // Default buffer time
+
+    // Initialize the sql table if it doesn't exist
+    state.blockConcurrencyWhile(async () => {
+      state.storage.sql.exec(`
+      CREATE TABLE IF NOT EXISTS cache_purge (
+        tag TEXT NOT NULL
+      );
+      CREATE UNIQUE INDEX IF NOT EXISTS tag_index ON cache_purge (tag);
+      `);
+    });
+  }
+
+  async purgeCacheByTags(tags: string[]) {
+    for (const tag of tags) {
+      // Insert the tag into the sql table
+      this.ctx.storage.sql.exec(
+        `
+        INSERT OR REPLACE INTO cache_purge (tag)
+        VALUES (?)`,
+        [tag]
+      );
+    }
+    const nextAlarm = await this.ctx.storage.getAlarm();
+    if (!nextAlarm) {
+      // Set an alarm to trigger the cache purge
+      this.ctx.storage.setAlarm(Date.now() + this.bufferTimeInSeconds * 1000);
+    }
+  }
+
+  override async alarm() {
+    let tags = this.ctx.storage.sql
+      .exec<{ tag: string }>(
+        `
+      SELECT * FROM cache_purge LIMIT ${MAX_NUMBER_OF_TAGS_PER_PURGE}
+    `
+      )
+      .toArray();
+    do {
+      if (tags.length === 0) {
+        // No tags to purge, we can stop
+        return;
+      }
+      const result = await internalPurgeCacheByTags(
+        this.env,
+        tags.map((row) => row.tag)
+      );
+      // For every other error, we just remove the tags from the sql table
+      // and continue
+      if (result === "rate-limit-exceeded") {
+        // Rate limit exceeded, we need to wait for the next alarm
+        // and try again
+        // We throw here to take advantage of the built-in retry
+        throw new Error("Rate limit exceeded");
+      }
+
+      // Delete the tags from the sql table
+      this.ctx.storage.sql.exec(
+        `
+        DELETE FROM cache_purge
+        WHERE tag IN (${tags.map(() => "?").join(",")})
+      `,
+        tags.map((row) => row.tag)
+      );
+      if (tags.length < MAX_NUMBER_OF_TAGS_PER_PURGE) {
+        // If we have less than MAX_NUMBER_OF_TAGS_PER_PURGE tags, we can stop
+        tags = [];
+      } else {
+        // Otherwise, we need to get the next 100 tags
+        tags = this.ctx.storage.sql
+          .exec<{ tag: string }>(
+            `
+          SELECT * FROM cache_purge LIMIT ${MAX_NUMBER_OF_TAGS_PER_PURGE}
+        `
+          )
+          .toArray();
+      }
+    } while (tags.length >= 0);
+  }
+}