Sqlite

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
 
-  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,13 +26,11 @@ 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)
     }
 
-    // TODO (fix): Support vary header.
     const cacheControl = parseCacheControl(headers['cache-control'])
-
     const contentLength = headers['content-length'] ? Number(headers['content-length']) : Infinity
     const maxEntrySize = this.#store.maxEntrySize ?? Infinity
 
@@ -64,11 +65,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,
         }
       }
     }
-
     return this.#handler.onHeaders(statusCode, rawHeaders, resume, statusMessage, headers)
   }
 
@@ -88,37 +88,110 @@ class CacheHandler extends DecoratorHandler {
 
   onComplete(rawTrailers) {
     if (this.#value) {
+      const reqHeaders = this.#opts
+      const resHeaders = parseHeaders(this.#value.data.rawHeaders)
+
+      // Early return if Vary = *, uncacheable.
+      if (resHeaders.vary === '*') {
+        return this.#handler.onComplete(rawTrailers)
+      }
+
       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] ?? ''])
+    .filter(([, val]) => val)
+}
+
+export class CacheStore {
+  #database
+
+  #insertquery
+  #getQuery
+  #purgeQuery
+
+  #size = 0
+  #maxSize = 128e9
+
+  constructor(location = ':memory:', opts) {
+    // TODO (fix): Validate args...
+
+    this.#maxSize = opts.maxSize ?? this.#maxSize
+    this.#database = new DatabaseSync(location)
+
+    this.#database.exec(`
+      CREATE TABLE IF NOT EXISTS cacheInterceptor(
+        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.#maybePurge()
   }
 
-  set(key, value, opts) {
-    this.cache.set(key, value, opts)
+  set(key, { data, vary, size, expires }) {
+    this.#insertquery.run(key, JSON.stringify(data), BJSON.stringify(vary), size, expires)
+
+    this.#size += size
+    this.#maybePurge()
   }
 
   get(key) {
-    return this.cache.get(key)
+    return this.#getQuery.all(key, Date.now()).map(({ data, vary, size, expires }) => ({
+      data: BJSON.parse(data),
+      vary: JSON.parse(vary),
+      size: parseInt(size), // TODO (fix): Is parseInt necessary?
+      expires: parseInt(expires), // TODO (fix): Is parseInt necessary?
+    }))
+  }
+
+  close() {
+    this.#database.close()
+  }
+
+  #maybePurge() {
+    if (this.#size == null || this.#size > this.#maxSize) {
+      this.#purgeQuery.run(Date.now())
+      this.#size = this.#database.exec('SELECT SUM(size) FROM cacheInterceptor')[0].values[0][0]
+    }
   }
 }
 
-function makeKey(opts) {
-  // NOTE: Ignores headers...
-  return `${opts.origin}:${opts.method}:${opts.path}`
+function findEntryByHeaders(entries, reqHeaders) {
+  return entries?.find(
+    (entry) => entry.vary?.every(([key, val]) => reqHeaders?.headers[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) {
@@ -149,22 +222,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 +254,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 +277,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

@@ -33,7 +33,8 @@
     "prepare": "husky install",
     "prepublishOnly": "pinst --disable",
     "postpublish": "pinst --enable",
-    "test": "tap test"
+    "test": "tap test",
+    "taprun": "tap run"
   },
   "lint-staged": {
     "*.{js,jsx,md,ts}": [