Add a Compressed Redis store

src/compressed-redis.test.ts

@@ -0,0 +1,49 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import Redis from "ioredis";
+
+import { RedisCacheStore, CompressedRedisCacheStore } from ".";
+import { testCache } from "./storetest";
+
+const { REDIS_HOST = "localhost" } = process.env;
+
+test("CompressedRedisCacheStore", async (t) => {
+  const redisClient = new Redis(REDIS_HOST);
+  t.after(() => redisClient.quit());
+
+  const cache = new CompressedRedisCacheStore(redisClient);
+
+  await testCache(t, cache);
+
+  const uncompressedCache = new RedisCacheStore(redisClient);
+  uncompressedCache.setKeyTemplate("-- test --");
+
+  await t.test(
+    "CompressedRedisCacheStore can read RedisCacheStore entries",
+    async () => {
+      const key = "object";
+      const value = { a: 1, b: "c" };
+
+      const expiresAt = Date.now() + 5000;
+      await uncompressedCache.set(key, { value, expiresAt });
+
+      assert.deepEqual(await cache.get(key), { value, expiresAt });
+
+      await cache.delete(key);
+    },
+  );
+  await t.test(
+    "RedisCacheStore treats CachedRedisCacheStore entries as misses",
+    async () => {
+      const key = "object";
+      const value = { a: 1, b: "c" };
+
+      const expiresAt = Date.now() + 5000;
+      await cache.set(key, { value, expiresAt });
+
+      assert.deepEqual(await uncompressedCache.get(key), undefined);
+
+      await cache.delete(key);
+    },
+  );
+});

src/compressed-redis.ts

@@ -0,0 +1,106 @@
+import { exponentialBuckets, Histogram, linearBuckets } from "prom-client";
+import type {
+  Redis as IORedisClient,
+  Cluster as IORedisCluster,
+} from "ioredis";
+
+import type { CacheStore, StoreEntity } from "./store";
+import { metrics } from "./metrics";
+import { brotliCompress, brotliDecompress, constants } from "node:zlib";
+import { promisify } from "node:util";
+
+const brotliCompressAsync = promisify(brotliCompress);
+const brotliDecompressAsync = promisify(brotliDecompress);
+
+const cacheValueSize = new Histogram({
+  name: "cache_compressed_redis_value_size_bytes",
+  help: "Size of bytes sent to Redis after compression",
+  labelNames: ["key"],
+  buckets: exponentialBuckets(100, 10, 5),
+  registers: [],
+});
+
+const cacheCompressionRatio = new Histogram({
+  name: "cache_compressed_ratio",
+  help: "Ratio of uncompressed:compressed data",
+  labelNames: ["key"],
+  buckets: linearBuckets(0, 0.1, 11),
+  registers: [],
+});
+
+metrics.push(cacheValueSize, cacheCompressionRatio);
+
+type Client = Pick<IORedisClient | IORedisCluster, "getBuffer" | "set" | "del">;
+
+// Brotli looks like all the interesting options are on the compressor,
+// which wouldn't change the format of the stored data. But version the magic bytes just in case.
+// Br 26-03
+const brotli2603MagicBytes = Buffer.from([0x42, 0x72, 0x26, 0x03]);
+
+export class CompressedRedisCacheStore<T> implements CacheStore<T> {
+  private keyTemplate: string | null = null;
+
+  constructor(private client: Client) {}
+
+  setKeyTemplate(keyTemplate: string) {
+    if (this.keyTemplate !== null) {
+      throw new Error("Cannot change key template");
+    }
+
+    this.keyTemplate = keyTemplate;
+  }
+
+  async get(key: string): Promise<StoreEntity<T> | undefined> {
+    const rawData = await this.client.getBuffer(key);
+
+    if (rawData === null) return undefined;
+
+    if (
+      rawData
+        .subarray(0, brotli2603MagicBytes.length)
+        .equals(brotli2603MagicBytes)
+    ) {
+      return JSON.parse(
+        (
+          await brotliDecompressAsync(
+            rawData.subarray(brotli2603MagicBytes.length),
+          )
+        ).toString("utf8"),
+      );
+    }
+
+    try {
+      return JSON.parse(rawData.toString("utf8"));
+    } catch {
+      return undefined;
+    }
+  }
+
+  async set(key: string, record: StoreEntity<T>): Promise<void> {
+    const ttl = record.expiresAt - Date.now();
+    if (ttl <= 0) return;
+
+    const jsonBuffer = Buffer.from(JSON.stringify(record), "utf8");
+    const buf = Buffer.concat([
+      brotli2603MagicBytes,
+      await brotliCompressAsync(jsonBuffer, {
+        [constants.BROTLI_PARAM_MODE]: constants.BROTLI_MODE_TEXT,
+        [constants.BROTLI_PARAM_SIZE_HINT]: jsonBuffer.length,
+      }),
+    ]);
+
+    if (this.keyTemplate !== null) {
+      cacheValueSize.observe({ key: this.keyTemplate }, buf.length);
+      cacheCompressionRatio.observe(
+        { key: this.keyTemplate },
+        buf.length / jsonBuffer.length,
+      );
+    }
+
+    await this.client.set(key, buf, "PX", ttl);
+  }
+
+  async delete(key: string): Promise<void> {
+    await this.client.del(key);
+  }
+}

src/index.ts

@@ -2,4 +2,5 @@ export * from "./store";
 export * from "./cache";
 export * from "./lru";
 export * from "./redis";
+export * from "./compressed-redis";
 export { registerMetrics } from "./metrics";