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,22 @@
 import assert from 'node:assert'
-import { LRUCache } from 'lru-cache'
 import { DecoratorHandler, parseHeaders, parseCacheControl } from '../utils.js'
+import { DatabaseSync } from 'node:sqlite' // --experimental-sqlite
+import * as BJSON from 'buffer-json'
 
 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) {
@@ -23,7 +26,7 @@ class CacheHandler extends DecoratorHandler {
   }
 
   onHeaders(statusCode, rawHeaders, resume, statusMessage, headers = parseHeaders(rawHeaders)) {
-    if (statusCode !== 307) {
+    if (statusCode !== 307 && statusCode !== 200) {
       return this.#handler.onHeaders(statusCode, rawHeaders, resume, statusMessage, headers)
     }
 
@@ -64,18 +67,16 @@ 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)
   }
 
   onData(chunk) {
     if (this.#value) {
       this.#value.size += chunk.bodyLength
-
       const maxEntrySize = this.#store.maxEntrySize ?? Infinity
       if (this.#value.size > maxEntrySize) {
         this.#value = null
@@ -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
+
       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 {
+  #insertquery
+  #getQuery
+  #purgeQuery
+  #database
+  #sizeQuery
+  #size = 0
+  #maxSize = 128e9
+  #maxTTL = Infinity
+
+  constructor(location = ':memory:', opts = {}) {
+    this.#maxSize = opts.maxSize ?? this.#maxSize
+    this.#maxTTL = opts.maxTTL ?? this.#maxTTL
+
+    this.#database = new DatabaseSync(location)
+
+    this.#database.exec(`
+      CREATE TABLE IF NOT EXISTS cacheInterceptor(
+        id INTEGER PRIMARY KEY,
+        key TEXT,
+        data TEXT,
+        vary TEXT,
+        size INTEGER,
+        expires INTEGER
+      ) STRICT
+    `)
+
+    this.#insertquery = this.#database.prepare(
+      'INSERT INTO cacheInterceptor (key, data, vary, size, expires) VALUES (?, ?, ?, ?, ?)',
+    )
+
+    this.#getQuery = this.#database.prepare(
+      'SELECT * FROM cacheInterceptor WHERE key = ? AND expires > ? ',
+    )
+
+    this.#purgeQuery = this.#database.prepare('DELETE FROM cacheInterceptor WHERE expires < ?')
+
+    this.#sizeQuery = this.#database.prepare('SELECT SUM(size) FROM cacheInterceptor')
   }
 
-  set(key, value, opts) {
-    this.cache.set(key, value, opts)
+  set(key, { data, vary, size, expires }) {
+    if (expires < Date.now()) {
+      return
+    }
+
+    const maxExpires = Date.now() + this.#maxTTL
+
+    expires = expires > maxExpires ? maxExpires : expires
+
+    this.#insertquery.run(key, BJSON.stringify(data), BJSON.stringify(vary), size, expires)
+
+    this.#size += size
+
+    this.#maybePurge()
   }
 
   get(key) {
-    return this.cache.get(key)
+    const rows = this.#getQuery.all(key, Date.now()).map(({ data, vary, size, expires }) => {
+      return {
+        data: BJSON.parse(data),
+        vary: JSON.parse(vary),
+        size: parseInt(size),
+        expires: parseInt(expires),
+      }
+    })
+
+    return rows
+  }
+
+  deleteAll() {
+    this.#database.exec('DELETE FROM cacheInterceptor')
+  }
+
+  close() {
+    this.#database.close()
+  }
+
+  #maybePurge() {
+    if (this.#size == null || this.#size > this.#maxSize) {
+      this.#purgeQuery.run(Date.now())
+
+      this.#size = Object.values(this.#sizeQuery.get())[0]
+
+      // In the case where the cache is full but has no expired entries yet, delete 10% of the cache, ordered by
+      // the oldest entries according to the 'expires' column.
+      if (this.#size > this.#maxSize) {
+        this.#database.exec(`
+          DELETE FROM cacheInterceptor
+          WHERE id IN (
+              SELECT id
+              FROM cacheInterceptor
+              ORDER BY expires ASC
+              LIMIT (SELECT ROUND(COUNT(*) * 0.10 + 1) FROM cacheInterceptor)
+          );
+        `)
+
+        this.#size = Object.values(this.#sizeQuery.get())[0]
+      }
+    }
   }
 }
 
-function makeKey(opts) {
-  // NOTE: Ignores headers...
-  return `${opts.origin}:${opts.method}:${opts.path}`
+function findEntryByHeaders(entries, request) {
+  return entries?.find(
+    (entry) =>
+      entry.vary?.every(([key, val]) => {
+        return request?.headers[key] === val || request[key] === val
+      }) ?? true,
+  )
 }
 
-const DEFAULT_CACHE_STORE = new CacheStore({ maxSize: 128 * 1024, maxEntrySize: 1024 })
+const DEFAULT_CACHE_STORE = new CacheStore()
 
 export default (opts) => (dispatch) => (opts, handler) => {
   if (!opts.cache || opts.upgrade) {
@@ -148,23 +260,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 +292,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 +315,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

@@ -9,10 +9,11 @@
     "lib/*"
   ],
   "dependencies": {
+    "buffer-json": "^2.0.0",
     "cache-control-parser": "^2.0.6",
     "cacheable-lookup": "^7.0.0",
     "http-errors": "^2.0.0",
-    "lru-cache": "^11.0.0",
+    "sqlite3": "^5.1.7",
     "undici": "^6.19.5"
   },
   "devDependencies": {