feat: d1 adapter for the tag cache (#320)

* fix: enable using the `direct` queue for isr

* skip the ssr fetch cache test due to flakiness

* remove redundant import

* fix: enable using the `direct` queue for isr

* skip the ssr fetch cache test due to flakiness

* feat: d1 adapter for the tag cache

* re-use aws manifest output to create our manifest

* address review comments

* use results instead of mapping over

* output an sql file instead

* move file inside a cloudflare directory

* use a single insert statement

* move the d1 setup to preview and e2e so you can still use skipbuild

* use two tables for the tag cache

* json.stringify

* move where the json.stringify is being done

* insert unique tags only, and use recoverableerror

* re-use the manifest from the createcacheassets function

* re-use the useTagCache var from aws

* fix flaky test

* change type import location

* Revert "move where the json.stringify is being done"

* rename to tables

* add back comment that the revert wiped out

* rebuild lockfile

* Update packages/cloudflare/src/api/d1-tag-cache.ts

Co-authored-by: Victor Berchet <victor@suumit.com>

* Update packages/cloudflare/src/api/d1-tag-cache.ts

* Update packages/cloudflare/src/api/d1-tag-cache.ts

---------

Co-authored-by: Victor Berchet <victor@suumit.com>

Author: james-elicx

.changeset/five-balloons-walk.md

@@ -0,0 +1,5 @@
+---
+"@opennextjs/cloudflare": minor
+---
+
+feat: d1 adapter for the tag cache

examples/e2e/app-router/e2e/after.test.ts

@@ -1,8 +1,6 @@
 import { expect, test } from "@playwright/test";
 
-// Cache is currently not supported: https://github.com/opennextjs/opennextjs-cloudflare/issues/105
-// (Note: specifically this test relied on `unstable_cache`: https://github.com/opennextjs/opennextjs-cloudflare/issues/105#issuecomment-2627074820)
-test.skip("Next after", async ({ request }) => {
+test("Next after", async ({ request }) => {
   const initialSSG = await request.get("/api/after/ssg");
   expect(initialSSG.status()).toEqual(200);
   const initialSSGJson = await initialSSG.json();

examples/e2e/app-router/e2e/revalidateTag.test.ts

@@ -1,7 +1,6 @@
 import { expect, test } from "@playwright/test";
 
-// Cache (and revalidateTag) is currently not supported: https://github.com/opennextjs/opennextjs-cloudflare/issues/105
-test.skip("Revalidate tag", async ({ page, request }) => {
+test("Revalidate tag", async ({ page, request }) => {
   test.setTimeout(45000);
   // We need to hit the page twice to make sure it's properly cached
   // Turbo might cache next build result, resulting in the tag being newer than the page
@@ -69,8 +68,7 @@ test.skip("Revalidate tag", async ({ page, request }) => {
   expect(nextCacheHeaderNested).toEqual("HIT");
 });
 
-// Cache (and revalidatePath) is currently not supported: https://github.com/opennextjs/opennextjs-cloudflare/issues/105
-test.skip("Revalidate path", async ({ page, request }) => {
+test("Revalidate path", async ({ page, request }) => {
   await page.goto("/revalidate-path");
 
   let elLayout = page.getByText("RequestID:");

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

@@ -1,16 +1,16 @@
 import type { OpenNextConfig } from "@opennextjs/aws/types/open-next.js";
-import cache from "@opennextjs/cloudflare/kv-cache";
+import tagCache from "@opennextjs/cloudflare/d1-tag-cache";
+import incrementalCache 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,
+      incrementalCache: async () => incrementalCache,
+      tagCache: () => tagCache,
       queue: () => memoryQueue,
-      // Unused implementation
-      tagCache: "dummy",
     },
   },
 

examples/e2e/app-router/package.json

@@ -9,9 +9,11 @@
     "start": "next start --port 3001",
     "lint": "next lint",
     "clean": "rm -rf .turbo node_modules .next .open-next",
+    "d1:clean": "wrangler d1 execute NEXT_CACHE_D1 --command \"DROP TABLE IF EXISTS tags; DROP TABLE IF EXISTS revalidations\"",
+    "d1:setup": "wrangler d1 execute NEXT_CACHE_D1 --file .open-next/cloudflare/cache-assets-manifest.sql",
     "build:worker": "pnpm opennextjs-cloudflare",
-    "preview": "pnpm build:worker && pnpm wrangler dev",
-    "e2e": "playwright test -c e2e/playwright.config.ts"
+    "preview": "pnpm build:worker && pnpm d1:clean && pnpm d1:setup && pnpm wrangler dev",
+    "e2e": "pnpm d1:clean && pnpm d1:setup && playwright test -c e2e/playwright.config.ts"
   },
   "dependencies": {
     "@opennextjs/cloudflare": "workspace:*",

examples/e2e/app-router/wrangler.json

@@ -13,5 +13,12 @@
       "binding": "NEXT_CACHE_WORKERS_KV",
       "id": "<BINDING_ID>"
     }
+  ],
+  "d1_databases": [
+    {
+      "binding": "NEXT_CACHE_D1",
+      "database_id": "NEXT_CACHE_D1",
+      "database_name": "NEXT_CACHE_D1"
+    }
   ]
 }

packages/cloudflare/env.d.ts

@@ -7,6 +7,8 @@ declare global {
       NEXT_PRIVATE_DEBUG_CACHE?: string;
       OPEN_NEXT_ORIGIN: string;
       NODE_ENV?: string;
+      NEXT_CACHE_D1_TAGS_TABLE?: string;
+      NEXT_CACHE_D1_REVALIDATIONS_TABLE?: string;
     }
   }
 }

packages/cloudflare/package.json

@@ -73,7 +73,7 @@
   "dependencies": {
     "@ast-grep/napi": "^0.34.1",
     "@dotenvx/dotenvx": "catalog:",
-    "@opennextjs/aws": "https://pkg.pr.new/@opennextjs/aws@733",
+    "@opennextjs/aws": "https://pkg.pr.new/@opennextjs/aws@748",
     "enquirer": "^2.4.1",
     "glob": "catalog:",
     "yaml": "^2.7.0"

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

@@ -3,6 +3,9 @@ import type { Context, RunningCodeOptions } from "node:vm";
 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;
     ASSETS?: Fetcher;
   }
 }

packages/cloudflare/src/api/d1-tag-cache.ts

@@ -0,0 +1,175 @@
+import { debug, error } from "@opennextjs/aws/adapters/logger.js";
+import type { OpenNextConfig } from "@opennextjs/aws/types/open-next.js";
+import type { TagCache } from "@opennextjs/aws/types/overrides.js";
+import { RecoverableError } from "@opennextjs/aws/utils/error.js";
+
+import { getCloudflareContext } from "./cloudflare-context.js";
+
+/**
+ * An instance of the Tag Cache that uses a D1 binding (`NEXT_CACHE_D1`) as it's underlying data store.
+ *
+ * **Tag/path mappings table**
+ *
+ * Information about the relation between tags and paths is stored in a `tags` table that contains
+ * two columns; `tag`, and `path`. The table name can be configured with `NEXT_CACHE_D1_TAGS_TABLE`
+ * environment variable.
+ *
+ * This table should be populated using an SQL file that is generated during the build process.
+ *
+ * **Tag revalidations table**
+ *
+ * Revalidation times for tags are stored in a `revalidations` table that contains two columns; `tags`,
+ * and `revalidatedAt`. The table name can be configured with `NEXT_CACHE_D1_REVALIDATIONS_TABLE`
+ * environment variable.
+ */
+class D1TagCache implements TagCache {
+  public readonly name = "d1-tag-cache";
+
+  public async getByPath(rawPath: string): Promise<string[]> {
+    const { isDisabled, db, tables } = this.getConfig();
+    if (isDisabled) return [];
+
+    const path = this.getCacheKey(rawPath);
+
+    try {
+      const { success, results } = await db
+        .prepare(`SELECT tag FROM ${JSON.stringify(tables.tags)} WHERE path = ?`)
+        .bind(path)
+        .all<{ tag: string }>();
+
+      if (!success) throw new RecoverableError(`D1 select failed for ${path}`);
+
+      const tags = results?.map((item) => this.removeBuildId(item.tag));
+
+      debug("tags for path", path, tags);
+      return tags;
+    } catch (e) {
+      error("Failed to get tags by path", e);
+      return [];
+    }
+  }
+
+  public async getByTag(rawTag: string): Promise<string[]> {
+    const { isDisabled, db, tables } = this.getConfig();
+    if (isDisabled) return [];
+
+    const tag = this.getCacheKey(rawTag);
+
+    try {
+      const { success, results } = await db
+        .prepare(`SELECT path FROM ${JSON.stringify(tables.tags)} WHERE tag = ?`)
+        .bind(tag)
+        .all<{ path: string }>();
+
+      if (!success) throw new RecoverableError(`D1 select failed for ${tag}`);
+
+      const paths = results?.map((item) => this.removeBuildId(item.path));
+
+      debug("paths for tag", tag, paths);
+      return paths;
+    } catch (e) {
+      error("Failed to get by tag", e);
+      return [];
+    }
+  }
+
+  public async getLastModified(path: string, lastModified?: number): Promise<number> {
+    const { isDisabled, db, tables } = this.getConfig();
+    if (isDisabled) return lastModified ?? Date.now();
+
+    try {
+      const { success, results } = await db
+        .prepare(
+          `SELECT ${JSON.stringify(tables.revalidations)}.tag FROM ${JSON.stringify(tables.revalidations)}
+            INNER JOIN ${JSON.stringify(tables.tags)} ON ${JSON.stringify(tables.revalidations)}.tag = ${JSON.stringify(tables.tags)}.tag
+            WHERE ${JSON.stringify(tables.tags)}.path = ? AND ${JSON.stringify(tables.revalidations)}.revalidatedAt > ?;`
+        )
+        .bind(this.getCacheKey(path), lastModified ?? 0)
+        .all<{ tag: string }>();
+
+      if (!success) throw new RecoverableError(`D1 select failed for ${path} - ${lastModified ?? 0}`);
+
+      debug("revalidatedTags", results);
+      return results?.length > 0 ? -1 : (lastModified ?? Date.now());
+    } catch (e) {
+      error("Failed to get revalidated tags", e);
+      return lastModified ?? Date.now();
+    }
+  }
+
+  public async writeTags(tags: { tag: string; path: string; revalidatedAt?: number }[]): Promise<void> {
+    const { isDisabled, db, tables } = this.getConfig();
+    if (isDisabled || tags.length === 0) return;
+
+    try {
+      const uniqueTags = new Set<string>();
+      const results = await db.batch(
+        tags
+          .map(({ tag, path, revalidatedAt }) => {
+            if (revalidatedAt === 1) {
+              // new tag/path mapping from set
+              return db
+                .prepare(`INSERT INTO ${JSON.stringify(tables.tags)} (tag, path) VALUES (?, ?)`)
+                .bind(this.getCacheKey(tag), this.getCacheKey(path));
+            }
+
+            if (!uniqueTags.has(tag) && revalidatedAt !== -1) {
+              // tag was revalidated
+              uniqueTags.add(tag);
+              return db
+                .prepare(
+                  `INSERT INTO ${JSON.stringify(tables.revalidations)} (tag, revalidatedAt) VALUES (?, ?)`
+                )
+                .bind(this.getCacheKey(tag), revalidatedAt ?? Date.now());
+            }
+          })
+          .filter((stmt) => !!stmt)
+      );
+
+      const failedResults = results.filter((res) => !res.success);
+
+      if (failedResults.length > 0) {
+        throw new RecoverableError(`${failedResults.length} tags failed to write`);
+      }
+    } catch (e) {
+      error("Failed to batch write tags", e);
+    }
+  }
+
+  private getConfig() {
+    const cfEnv = getCloudflareContext().env;
+    const db = cfEnv.NEXT_CACHE_D1;
+
+    if (!db) debug("No D1 database found");
+
+    const isDisabled = !!(globalThis as unknown as { openNextConfig: OpenNextConfig }).openNextConfig
+      .dangerous?.disableTagCache;
+
+    if (!db || isDisabled) {
+      return { isDisabled: true as const };
+    }
+
+    return {
+      isDisabled: false as const,
+      db,
+      tables: {
+        tags: cfEnv.NEXT_CACHE_D1_TAGS_TABLE ?? "tags",
+        revalidations: cfEnv.NEXT_CACHE_D1_REVALIDATIONS_TABLE ?? "revalidations",
+      },
+    };
+  }
+
+  protected removeBuildId(key: string) {
+    return key.replace(`${this.getBuildId()}/`, "");
+  }
+
+  protected getCacheKey(key: string) {
+    return `${this.getBuildId()}/${key}`.replaceAll("//", "/");
+  }
+
+  protected getBuildId() {
+    return process.env.NEXT_BUILD_ID ?? "no-build-id";
+  }
+}
+
+export default new D1TagCache();

packages/cloudflare/src/cli/build/build.ts

@@ -12,6 +12,7 @@ import logger from "@opennextjs/aws/logger.js";
 
 import type { ProjectOptions } from "../project-options.js";
 import { bundleServer } from "./bundle-server.js";
+import { compileCacheAssetsManifestSqlFile } from "./open-next/compile-cache-assets-manifest.js";
 import { compileEnvFiles } from "./open-next/compile-env-files.js";
 import { copyCacheAssets } from "./open-next/copyCacheAssets.js";
 import { createServerBundle } from "./open-next/createServerBundle.js";
@@ -86,8 +87,12 @@ export async function build(projectOpts: ProjectOptions): Promise<void> {
   createStaticAssets(options);
 
   if (config.dangerous?.disableIncrementalCache !== true) {
-    createCacheAssets(options);
+    const { useTagCache, metaFiles } = createCacheAssets(options);
     copyCacheAssets(options);
+
+    if (useTagCache) {
+      compileCacheAssetsManifestSqlFile(options, metaFiles);
+    }
   }
 
   await createServerBundle(options);

packages/cloudflare/src/cli/build/open-next/compile-cache-assets-manifest.ts

@@ -0,0 +1,31 @@
+import { appendFileSync, mkdirSync, writeFileSync } from "node:fs";
+import path from "node:path";
+
+import type { BuildOptions } from "@opennextjs/aws/build/helper.js";
+import type { TagCacheMetaFile } from "@opennextjs/aws/types/cache.js";
+
+/**
+ * Generates SQL statements that can be used to initialise the cache assets manifest in an SQL data store.
+ */
+export function compileCacheAssetsManifestSqlFile(options: BuildOptions, metaFiles: TagCacheMetaFile[]) {
+  const outputPath = path.join(options.outputDir, "cloudflare/cache-assets-manifest.sql");
+
+  const tagsTable = process.env.NEXT_CACHE_D1_TAGS_TABLE || "tags";
+  const revalidationsTable = process.env.NEXT_CACHE_D1_REVALIDATIONS_TABLE || "revalidations";
+
+  mkdirSync(path.dirname(outputPath), { recursive: true });
+  writeFileSync(
+    outputPath,
+    `CREATE TABLE IF NOT EXISTS ${JSON.stringify(tagsTable)} (tag TEXT NOT NULL, path TEXT NOT NULL, UNIQUE(tag, path) ON CONFLICT REPLACE);
+     CREATE TABLE IF NOT EXISTS ${JSON.stringify(revalidationsTable)} (tag TEXT NOT NULL, revalidatedAt INTEGER NOT NULL, UNIQUE(tag) ON CONFLICT REPLACE);\n`
+  );
+
+  const values = metaFiles.map(({ tag, path }) => `(${JSON.stringify(tag.S)}, ${JSON.stringify(path.S)})`);
+
+  if (values.length) {
+    appendFileSync(
+      outputPath,
+      `INSERT INTO ${JSON.stringify(tagsTable)} (tag, path) VALUES ${values.join(", ")};`
+    );
+  }
+}

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

@@ -13,7 +13,9 @@ export function ensureCloudflareConfig(config: OpenNextConfig) {
     dftMaybeUseCache:
       config.default?.override?.incrementalCache === "dummy" ||
       typeof config.default?.override?.incrementalCache === "function",
-    dftUseDummyTagCache: config.default?.override?.tagCache === "dummy",
+    dftMaybeUseTagCache:
+      config.default?.override?.tagCache === "dummy" ||
+      typeof config.default?.override?.incrementalCache === "function",
     dftMaybeUseQueue:
       config.default?.override?.queue === "dummy" ||
       config.default?.override?.queue === "direct" ||

packages/cloudflare/tsconfig.json

@@ -15,5 +15,5 @@
     "target": "ES2022",
     "types": ["@cloudflare/workers-types", "@opennextjs/aws/types/global.d.ts"]
   },
-  "include": ["src/**/*.ts"]
+  "include": ["src/**/*.ts", "env.d.ts"]
 }