feat: edge gateway extend cdn resolution with r2

packages/edge-gateway/ava.config.js

@@ -6,4 +6,5 @@ export default {
   timeout: '5m',
   concurrency: 1,
   nodeArguments: ['--experimental-vm-modules'],
+  require: ['./test/_setup-browser-env.js'],
 }

packages/edge-gateway/package.json

@@ -34,6 +34,7 @@
     "@sentry/cli": "^1.71.0",
     "@web-std/fetch": "^4.0.0",
     "ava": "^3.15.0",
+    "browser-env": "^3.3.0",
     "delay": "^5.0.0",
     "esbuild": "^0.14.2",
     "execa": "^6.0.0",

packages/edge-gateway/src/env.js

@@ -21,6 +21,7 @@ import { Logging } from './logs.js'
  * @property {Object} CIDSTRACKER
  * @property {Object} GATEWAYREDIRECTCOUNTER
  * @property {KVNamespace} DENYLIST
+ * @property {R2Bucket} SUPERHOT
  *
  * @typedef {Object} EnvTransformed
  * @property {string} IPFS_GATEWAY_HOSTNAME
@@ -118,4 +119,19 @@ function getSentry(request, env, ctx) {
  * }} DurableObjectStub
  *
  * @typedef {{ get: (key: string) => Promise<string | null> }} KVNamespace
+ *
+ * @typedef {Object} R2PutOptions
+ * @property {Headers} [httpMetadata]
+ * @property {Record<string, string>} [customMetadata]
+ *
+ * @typedef {Object} R2Object
+ * @property {Date} uploaded
+ * @property {Headers} [httpMetadata]
+ * @property {Record<string, string>} [customMetadata]
+ *
+ * @typedef {Object} R2Bucket
+ * @property {(key: string) => Promise<R2Object | null>} head
+ * @property {(key: string) => Promise<Response & R2Object | null>} get
+ * @property {(key: string, value: ReadableStream | ArrayBuffer | ArrayBufferView | string | null, options?: R2PutOptions) => Promise<R2Object>} put
+ * @property {(key: string) => Promise<void>} delete
  */

packages/edge-gateway/src/gateway.js

@@ -12,7 +12,6 @@ import { toDenyListAnchor } from './utils/deny-list.js'
 import {
   CIDS_TRACKER_ID,
   SUMMARY_METRICS_ID,
-  GATEWAY_RATE_LIMIT_ID,
   REDIRECT_COUNTER_METRICS_ID,
   CF_CACHE_MAX_OBJECT_SIZE,
   HTTP_STATUS_RATE_LIMITED,
@@ -52,6 +51,7 @@ export async function gatewayGet(request, env, ctx) {
   const cid = getCidFromSubdomainUrl(reqUrl)
   const pathname = reqUrl.pathname
 
+  // Validation layer
   if (env.DENYLIST) {
     const anchor = await toDenyListAnchor(cid)
     // TODO: Remove once https://github.com/nftstorage/nftstorage.link/issues/51 is fixed
@@ -69,9 +69,9 @@ export async function gatewayGet(request, env, ctx) {
     }
   }
 
+  // 1st layer resolution - CDN
   const cache = caches.default
-  const res = await cache.match(request.url)
-
+  const res = await cdnResolution(request, env, cache)
   if (res) {
     // Update cache metrics in background
     const responseTime = Date.now() - startTs
@@ -80,7 +80,7 @@ export async function gatewayGet(request, env, ctx) {
     return res
   }
 
-  // Prepare IPFS gateway requests
+  // 2nd layer resolution - Public Gateways race
   const gatewayReqs = env.ipfsGateways.map((gwUrl) =>
     gatewayFetch(gwUrl, cid, request, {
       pathname,
@@ -176,6 +176,33 @@ export async function gatewayGet(request, env, ctx) {
   }
 }
 
+/**
+ * CDN url resolution.
+ *
+ * @param {Request} request
+ * @param {Env} env
+ * @param {Cache} cache
+ */
+async function cdnResolution(request, env, cache) {
+  let res
+
+  try {
+    res = await pAny(
+      [cache.match(request.url), env.SUPERHOT.get(request.url)],
+      {
+        filter: (res) => !!res,
+      }
+    )
+  } catch (err) {
+    if (err instanceof FilterError || err instanceof AggregateError) {
+      return undefined
+    }
+    throw err
+  }
+
+  return res
+}
+
 /**
  * Store metrics for winner gateway response
  *

packages/edge-gateway/test/_setup-browser-env.js

@@ -0,0 +1,10 @@
+// Per https://github.com/avajs/ava/blob/main/docs/recipes/browser-testing.md
+
+import browserEnv from 'browser-env'
+browserEnv()
+
+// Add web response for R2 mocking
+import { Response } from '@web-std/fetch'
+
+// @ts-ignore response does not have all the types
+global.Response = Response

packages/edge-gateway/test/cache.spec.js → packages/edge-gateway/test/cdn.spec.js

@@ -1,5 +1,6 @@
 import test from 'ava'
 import { getMiniflare } from './utils.js'
+import { fromString } from 'uint8arrays'
 
 test.beforeEach((t) => {
   // Create a new Miniflare environment for each test
@@ -8,19 +9,33 @@ test.beforeEach((t) => {
   }
 })
 
-// Miniflare cache sometimes is not yet setup...
-test.skip('Caches content', async (t) => {
+test('Caches content on resolve', async (t) => {
   const url =
     'https://bafkreidyeivj7adnnac6ljvzj2e3rd5xdw3revw4da7mx2ckrstapoupoq.ipfs.localhost:8787/'
   const content = 'Hello nft.storage! 😎'
   const { mf } = t.context
 
-  const caches = await mf.getCaches()
-
   const response = await mf.dispatchFetch(url)
   await response.waitUntil()
   t.is(await response.text(), content)
 
+  const caches = await mf.getCaches()
   const cachedRes = await caches.default.match(url)
   t.is(await cachedRes.text(), content)
 })
+
+test('Get content from Perma cache if existing', async (t) => {
+  const { mf } = t.context
+  const url =
+    'https://bafybeic2hr75ukgwhnasdl3sucxyfedfyp3dijq3oakzx6o23urcs4eige.ipfs.localhost:8787/'
+
+  const bindings = await mf.getBindings()
+  const r2Bucket = bindings.SUPERHOT
+
+  const content = 'Hello perma cache!'
+  await r2Bucket.put(url, new Response(fromString(content)).body)
+
+  const response = await mf.dispatchFetch(url)
+  await response.waitUntil()
+  t.is(await response.text(), content)
+})

packages/edge-gateway/test/utils.js

@@ -1,3 +1,4 @@
+import { concat } from 'uint8arrays/concat'
 import { Miniflare } from 'miniflare'
 
 export function getMiniflare() {
@@ -12,6 +13,9 @@ export function getMiniflare() {
     buildCommand: undefined,
     wranglerConfigEnv: 'test',
     modules: true,
+    bindings: {
+      SUPERHOT: createR2Bucket(),
+    },
   })
 }
 
@@ -26,3 +30,62 @@ export function getGatewayRateLimitsName() {
 export function getSummaryMetricsName() {
   return 'SUMMARYMETRICS'
 }
+
+function createR2Bucket() {
+  const bucket = new Map()
+
+  return {
+    put: async (key, value, putOpts = {}) => {
+      let data = new Uint8Array([])
+      const reader = value.getReader()
+      try {
+        while (true) {
+          const { done, value } = await reader.read()
+          if (done) {
+            break
+          }
+          data = concat([data, value])
+        }
+      } finally {
+        reader.releaseLock()
+      }
+
+      bucket.set(key, {
+        body: data,
+        httpMetadata: putOpts.httpMetadata || {},
+        customMetadata: putOpts.customMetadata || {},
+      })
+
+      return Promise.resolve({
+        httpMetadata: putOpts.httpMetadata,
+        customMetadata: putOpts.customMetadata,
+      })
+    },
+    get: async (key) => {
+      const value = bucket.get(key)
+      if (!value) {
+        return undefined
+      }
+
+      const response = new Response(value.body, { status: 200 })
+
+      return Promise.resolve(
+        Object.assign(response, {
+          httpMetadata: value.httpMetadata || {},
+          customMetadata: value.customMetadata || {},
+        })
+      )
+    },
+    head: async (key) => {
+      const value = bucket.get(key)
+      if (!value) {
+        return undefined
+      }
+
+      return Promise.resolve({
+        httpMetadata: value.httpMetadata || {},
+        customMetadata: value.customMetadata || {},
+      })
+    },
+  }
+}

packages/edge-gateway/wrangler.toml

@@ -7,7 +7,7 @@ type = "javascript"
 
 account_id = ""
 watch_dir = "src"
-compatibility_date = "2021-12-03"
+compatibility_date = "2022-04-26"
 
 [build]
 command = "npm run build"
@@ -46,6 +46,10 @@ bindings = [
   {name = "GATEWAYREDIRECTCOUNTER", class_name = "GatewayRedirectCounter0"}
 ]
 
+[[env.production.r2_buckets]]
+bucket_name = "super-hot"
+binding = "SUPERHOT"
+
 # Staging!
 [env.staging]
 # name = "gateway-nft-storage-staging"
@@ -68,6 +72,10 @@ bindings = [
   {name = "GATEWAYREDIRECTCOUNTER", class_name = "GatewayRedirectCounter0"}
 ]
 
+[[env.staging.r2_buckets]]
+bucket_name = "super-hot-staging"
+binding = "SUPERHOT"
+
 # Test!
 [env.test]
 workers_dev = true