feat(cache): use vary headers to compare cached response with request headers

.gitignore

@@ -19,3 +19,4 @@ db
 .nyc*
 yarn.lock
 .tap
+file:*

lib/interceptor/cache.js

@@ -1,19 +1,21 @@
 import assert from 'node:assert'
-import { LRUCache } from 'lru-cache'
 import { DecoratorHandler, parseHeaders, parseCacheControl } from '../utils.js'
+import { DatabaseSync } from 'node:sqlite' // --experimental-sqlite
 
 class CacheHandler extends DecoratorHandler {
   #handler
   #store
   #key
+  #opts
   #value = null
 
-  constructor({ key, handler, store }) {
+  constructor({ key, handler, store, opts = [] }) {
     super(handler)
 
     this.#key = key
     this.#handler = handler
     this.#store = store
+    this.#opts = opts
   }
 
   onConnect(abort) {
@@ -64,11 +66,10 @@ class CacheHandler extends DecoratorHandler {
             (rawHeaders?.reduce((xs, x) => xs + x.length, 0) ?? 0) +
             (statusMessage?.length ?? 0) +
             64,
-          ttl: ttl * 1e3,
+          expires: Date.now() + ttl, // in ms!
         }
       }
     }
-
     return this.#handler.onHeaders(statusCode, rawHeaders, resume, statusMessage, headers)
   }
 
@@ -88,37 +89,148 @@ class CacheHandler extends DecoratorHandler {
 
   onComplete(rawTrailers) {
     if (this.#value) {
+      const resHeaders = parseHeaders(this.#value.data.rawHeaders)
+
+      // Early return if Vary = *, uncacheable.
+      if (resHeaders.vary === '*') {
+        return this.#handler.onComplete(rawTrailers)
+      }
+
+      const reqHeaders = this.#opts
+
+      // If Range header present, assume that the response varies based on Range.
+      if (reqHeaders.headers?.range) {
+        resHeaders.vary += ', Range'
+      }
+
       this.#value.data.rawTrailers = rawTrailers
-      this.#value.size += rawTrailers?.reduce((xs, x) => xs + x.length, 0) ?? 0
-      this.#store.set(this.#key, this.#value.data, { ttl: this.#value.ttl, size: this.#value.size })
+      this.#value.size = this.#value.size
+        ? this.#value.size + rawTrailers?.reduce((xs, x) => xs + x.length, 0)
+        : 0
+
+      this.#value.vary = formatVaryData(resHeaders, reqHeaders)
+
+      this.#store.set(this.#key, this.#value)
     }
     return this.#handler.onComplete(rawTrailers)
   }
 }
 
-// TODO (fix): Async filesystem cache.
-class CacheStore {
-  constructor({ maxSize = 1024 * 1024, maxEntrySize = 128 * 1024 }) {
-    this.maxSize = maxSize
-    this.maxEntrySize = maxEntrySize
-    this.cache = new LRUCache({ maxSize })
+function formatVaryData(resHeaders, reqHeaders) {
+  return resHeaders.vary
+    ?.split(',')
+    .map((key) => key.trim().toLowerCase())
+    .map((key) => [key, reqHeaders[key] ?? reqHeaders.headers[key]])
+    .filter(([_key, val]) => val)
+}
+
+export class CacheStore {
+  constructor() {
+    this.database = null
+    this.init()
+  }
+
+  init() {
+    this.database = new DatabaseSync('file:memdb1?mode=memory&cache=shared')
+
+    this.database.exec(`
+      CREATE TABLE IF NOT EXISTS cacheInterceptor(
+        key TEXT,
+        data TEXT,
+        vary TEXT,
+        size INTEGER,
+        expires INTEGER
+      ) STRICT
+    `)
   }
 
-  set(key, value, opts) {
-    this.cache.set(key, value, opts)
+  set(key, entry) {
+    if (!this.database) {
+      throw new Error('Database not initialized')
+    }
+
+    entry.data = JSON.stringify(entry.data)
+    entry.vary = JSON.stringify(entry.vary)
+
+    const insert = this.database.prepare(
+      'INSERT INTO cacheInterceptor (key, data, vary, size, expires) VALUES (?, ?, ?, ?, ?)',
+    )
+
+    insert.run(key, entry.data, entry.vary, entry.size, entry.expires)
+
+    this.purge()
   }
 
   get(key) {
-    return this.cache.get(key)
+    if (!this.database) {
+      throw new Error('Database not initialized')
+    }
+    this.purge()
+    const query = this.database.prepare(
+      'SELECT * FROM cacheInterceptor WHERE key = ? AND expires > ? ',
+    )
+    const rows = query.all(key, Date.now())
+    rows.map((i) => {
+      i.data = JSON.parse(i.data)
+      i.vary = JSON.parse(i.vary)
+      i.data = {
+        ...i.data,
+        // JSON.parse doesn't convert a Buffer object back to a Buffer object once it has been stringified.
+        body: this.#convertToBuffer(i.data.body),
+        rawHeaders: this.#convertToBuffer(i.data.rawHeaders),
+        rawTrailers: this.#convertToBuffer(i.data.rawTrailers),
+      }
+      return i
+    })
+
+    return rows
+  }
+
+  purge() {
+    const query = this.database.prepare('DELETE FROM cacheInterceptor WHERE expires < ?')
+    query.run(Date.now())
+  }
+
+  deleteAll() {
+    const query = this.database.prepare('DELETE FROM cacheInterceptor')
+    query.run()
+  }
+
+  #convertToBuffer(bufferArray) {
+    if (Array.isArray(bufferArray) && bufferArray.length > 0) {
+      return bufferArray.map((ba) => {
+        return typeof ba === 'object' ? Buffer.from(ba.data) : ba
+      })
+    }
+    return []
   }
 }
 
-function makeKey(opts) {
-  // NOTE: Ignores headers...
-  return `${opts.origin}:${opts.method}:${opts.path}`
+/*
+  Sort entries by number of vary headers in descending order, because
+  we need to compare the most complex response to the request first.
+  A cached response with an empty ´vary´ field will otherwise win every time.
+*/
+function sortEntriesByVary(entries) {
+  entries.sort((a, b) => {
+    const lengthA = a.vary ? a.vary.length : 0
+    const lengthB = b.vary ? b.vary.length : 0
+    return lengthB - lengthA
+  })
 }
 
-const DEFAULT_CACHE_STORE = new CacheStore({ maxSize: 128 * 1024, maxEntrySize: 1024 })
+function findEntryByHeaders(entries, reqHeaders) {
+  sortEntriesByVary(entries)
+
+  return entries?.find(
+    (entry) =>
+      entry.vary?.every(([key, val]) => {
+        return reqHeaders?.headers[key] === val
+      }) ?? true,
+  )
+}
+
+const DEFAULT_CACHE_STORE = new CacheStore()
 
 export default (opts) => (dispatch) => (opts, handler) => {
   if (!opts.cache || opts.upgrade) {
@@ -149,22 +261,27 @@ export default (opts) => (dispatch) => (opts, handler) => {
   // Dump body...
   opts.body?.on('error', () => {}).resume()
 
+  opts.host = opts.host ?? new URL(opts.origin).host
+
+  if (!opts.headers) {
+    opts.headers = {}
+  }
+
+  // idea: use DEFAULT_CACHE_STORE by default if 'cache' not specified, since the cache interceptor was already specified to be used.
   const store = opts.cache === true ? DEFAULT_CACHE_STORE : opts.cache
 
   if (!store) {
     throw new Error(`Cache store not provided.`)
   }
 
-  let key = makeKey(opts)
-  let value = store.get(key)
+  const key = `${opts.method}:${opts.path}`
 
-  if (value == null && opts.method === 'HEAD') {
-    key = makeKey({ ...opts, method: 'GET' })
-    value = store.get(key)
-  }
+  const entries = (store.get(key) ?? opts.method === 'HEAD') ? store.get(`GET:${opts.path}`) : null
 
-  if (value) {
-    const { statusCode, statusMessage, rawHeaders, rawTrailers, body } = value
+  const entry = findEntryByHeaders(entries, opts)
+
+  if (entry) {
+    const { statusCode, statusMessage, rawHeaders, rawTrailers, body } = entry.data
     const ac = new AbortController()
     const signal = ac.signal
 
@@ -176,11 +293,14 @@ export default (opts) => (dispatch) => (opts, handler) => {
     try {
       handler.onConnect(abort)
       signal.throwIfAborted()
+
       handler.onHeaders(statusCode, rawHeaders, resume, statusMessage)
       signal.throwIfAborted()
+
       if (opts.method !== 'HEAD') {
         for (const chunk of body) {
           const ret = handler.onData(chunk)
+
           signal.throwIfAborted()
           if (ret === false) {
             // TODO (fix): back pressure...
@@ -196,6 +316,6 @@ export default (opts) => (dispatch) => (opts, handler) => {
 
     return true
   } else {
-    return dispatch(opts, new CacheHandler({ handler, store, key: makeKey(opts) }))
+    return dispatch(opts, new CacheHandler({ handler, store, key, opts }))
   }
 }

lib/interceptor/proxy.js

@@ -14,6 +14,7 @@ class Handler extends DecoratorHandler {
   }
 
   onUpgrade(statusCode, rawHeaders, socket) {
+    console.log('Proxy onUpgrade')
     return this.#handler.onUpgrade(
       statusCode,
       reduceHeaders(
@@ -34,6 +35,7 @@ class Handler extends DecoratorHandler {
   }
 
   onHeaders(statusCode, rawHeaders, resume, statusMessage) {
+    console.log('Proxy onHeaders')
     return this.#handler.onHeaders(
       statusCode,
       reduceHeaders(
@@ -164,6 +166,7 @@ function printIp(address, port) {
 }
 
 export default (opts) => (dispatch) => (opts, handler) => {
+  console.log('Proxy default dispatch')
   if (!opts.proxy) {
     return dispatch(opts, handler)
   }

lib/interceptor/redirect.js

@@ -36,6 +36,7 @@ class Handler extends DecoratorHandler {
   }
 
   onConnect(abort) {
+    console.log('Redirect onConnect')
     if (this.#aborted) {
       abort(this.#reason)
     } else {
@@ -48,6 +49,7 @@ class Handler extends DecoratorHandler {
   }
 
   onHeaders(statusCode, rawHeaders, resume, statusText, headers = parseHeaders(rawHeaders)) {
+    console.log('Redirect onHeaders')
     if (redirectableStatusCodes.indexOf(statusCode) === -1) {
       assert(!this.#headersSent)
       this.#headersSent = true
@@ -109,6 +111,7 @@ class Handler extends DecoratorHandler {
   }
 
   onData(chunk) {
+    console.log('Redirect onData')
     if (this.#location) {
       /*
         https://tools.ietf.org/html/rfc7231#section-6.4

package.json

@@ -12,7 +12,6 @@
     "cache-control-parser": "^2.0.6",
     "cacheable-lookup": "^7.0.0",
     "http-errors": "^2.0.0",
-    "lru-cache": "^11.0.0",
     "undici": "^6.19.5"
   },
   "devDependencies": {