feat(yt): integrate AdaptiveCache into redis

lib/adaptive-cache.test.ts

@@ -0,0 +1,166 @@
+import { describe, expect, mock, test } from "bun:test";
+import { AdaptiveCache } from "./adaptive-cache";
+
+class FakeRedis {
+  readonly strings = new Map<string, { value: string; expires: number }>();
+  readonly sortedSets = new Map<string, Map<string, number>>();
+  readonly commands: { command: string; args: string[] }[] = [];
+
+  async get(key: string) {
+    const entry = this.strings.get(key);
+    if (!entry || entry.expires <= Date.now()) {
+      this.strings.delete(key);
+      return null;
+    }
+    return entry.value;
+  }
+
+  async send(command: string, args: string[]): Promise<unknown> {
+    this.commands.push({ command, args });
+
+    if (command === "DEL") return this.strings.delete(args[0]) ? 1 : 0;
+    if (command !== "EVAL") throw new Error(`Unsupported command: ${command}`);
+
+    if (args[0].includes("return {")) {
+      const requests = this.sortedSet(args[2]);
+      const entries = this.sortedSet(args[3]);
+      const cutoff = Number(args[4]);
+      const now = Number(args[5]);
+
+      for (const [member, score] of requests)
+        if (score < cutoff) requests.delete(member);
+      for (const [member, score] of entries)
+        if (score <= now) entries.delete(member);
+
+      return [requests.size, entries.size];
+    }
+
+    const responseKey = args[2];
+    const requests = this.sortedSet(args[3]);
+    const entries = this.sortedSet(args[4]);
+    const startedAt = Number(args[5]);
+    const requestMember = args[6];
+    const cached = args[8];
+    const remainingTtl = Number(args[9]);
+    const expiresAt = Number(args[10]);
+    const hash = args[11];
+
+    requests.set(requestMember, startedAt);
+    if (remainingTtl > 0) {
+      this.strings.set(responseKey, {
+        value: cached,
+        expires: Date.now() + remainingTtl,
+      });
+      entries.set(hash, expiresAt);
+    }
+    return null;
+  }
+
+  private sortedSet(key: string) {
+    let set = this.sortedSets.get(key);
+    if (!set) {
+      set = new Map();
+      this.sortedSets.set(key, set);
+    }
+    return set;
+  }
+}
+
+const options = {
+  name: "test",
+  maxRequests: 4,
+  period: 1_000,
+  lookback: 500,
+  ttl: 1_000,
+};
+
+describe("AdaptiveCache", () => {
+  test("shares quota and cached responses between instances", async () => {
+    const redis = new FakeRedis();
+    const firstReplica = new AdaptiveCache(options, redis);
+    const restartedReplica = new AdaptiveCache(options, redis);
+    const upstream = mock(
+      async () =>
+        new Response("first", {
+          status: 201,
+          statusText: "Created",
+          headers: { "x-cache-test": "yes" },
+        }),
+    );
+    const url = "https://example.com/data?secret=do-not-leak";
+
+    await firstReplica.fetch(url, undefined, upstream);
+    const response = await restartedReplica.fetch(url, undefined, upstream);
+
+    expect(upstream).toHaveBeenCalledTimes(1);
+    expect(response.status).toBe(201);
+    expect(response.statusText).toBe("Created");
+    expect(response.headers.get("x-cache-test")).toBe("yes");
+    expect(await response.text()).toBe("first");
+    expect(redis.commands.flatMap(({ args }) => args).join(" ")).not.toContain(
+      "do-not-leak",
+    );
+    expect(await restartedReplica.debugStats()).toContain(
+      "HOT | recent=1 | projected=2.0/4",
+    );
+  });
+
+  test("does not count or cache unsuccessful responses", async () => {
+    const redis = new FakeRedis();
+    const cache = new AdaptiveCache(options, redis);
+
+    await cache.fetch(
+      "https://example.com/failure",
+      undefined,
+      async () => new Response("failure", { status: 500 }),
+    );
+
+    expect(await cache.debugStats()).toContain("recent=0");
+    expect(redis.strings.size).toBe(0);
+  });
+
+  test("fetches upstream without writing when Redis is unavailable", async () => {
+    const unavailable = {
+      get: async () => {
+        throw new Error("offline");
+      },
+      send: mock(async () => {
+        throw new Error("offline");
+      }),
+    };
+    const cache = new AdaptiveCache(options, unavailable);
+    const error = console.error;
+    console.error = mock(() => {});
+
+    try {
+      const response = await cache.fetch(
+        "https://example.com/data",
+        undefined,
+        async () => new Response("upstream"),
+      );
+      expect(await response.text()).toBe("upstream");
+      expect(unavailable.send).toHaveBeenCalledTimes(1);
+    } finally {
+      console.error = error;
+    }
+  });
+
+  test("expires response entries without losing request history", async () => {
+    const redis = new FakeRedis();
+    const cache = new AdaptiveCache({ ...options, ttl: 5 }, redis);
+    let calls = 0;
+    const upstream = async () => new Response(`${++calls}`);
+
+    await cache.fetch("https://example.com/data", undefined, upstream);
+    await Bun.sleep(10);
+    const response = await cache.fetch(
+      "https://example.com/data",
+      undefined,
+      upstream,
+    );
+
+    expect(await response.text()).toBe("2");
+    expect(calls).toBe(2);
+    expect(await cache.debugStats()).toContain("recent=2");
+  });
+});

lib/adaptive-cache.ts

@@ -1,6 +1,9 @@
-// gpt'd at 3am, used as-is by another module
+import { redis } from "@/lib/db/redis";
 
 interface AdaptiveCacheOptions {
+  /** Stable namespace shared by every replica. */
+  name: string;
+
   /** Maximum requests allowed per period. */
   maxRequests: number;
 
@@ -20,88 +23,199 @@ interface AdaptiveCacheOptions {
   key?(req: Request): string;
 }
 
+type RedisClient = {
+  get(key: string): Promise<string | null>;
+  send(command: string, args: string[]): Promise<unknown>;
+};
+
+type Fetcher = (
+  input: RequestInfo | URL,
+  init?: RequestInit,
+) => Promise<Response>;
+
+type CachedResponse = {
+  version: 1;
+  status: number;
+  statusText: string;
+  headers: [string, string][];
+  body: string;
+};
+
+const STATS_SCRIPT = `
+redis.call("ZREMRANGEBYSCORE", KEYS[1], "-inf", "(" .. ARGV[1])
+redis.call("ZREMRANGEBYSCORE", KEYS[2], "-inf", ARGV[2])
+return { redis.call("ZCARD", KEYS[1]), redis.call("ZCARD", KEYS[2]) }
+`;
+
+const STORE_SCRIPT = `
+redis.call("ZADD", KEYS[2], ARGV[1], ARGV[2])
+redis.call("PEXPIRE", KEYS[2], ARGV[3])
+
+if tonumber(ARGV[5]) > 0 then
+  redis.call("SET", KEYS[1], ARGV[4], "PX", ARGV[5])
+  redis.call("ZADD", KEYS[3], ARGV[6], ARGV[7])
+  redis.call("PEXPIRE", KEYS[3], ARGV[8])
+end
+`;
+
 export class AdaptiveCache {
-  private readonly cache = new Map<
-    string,
-    {
-      expires: number;
-      response: Response;
-    }
-  >();
+  private readonly lookback: number;
+  private readonly threshold: number;
+  private readonly key: (req: Request) => string;
+  private readonly prefix: string;
 
-  private readonly requests: number[] = [];
-
-  constructor(private readonly options: AdaptiveCacheOptions) {
-    options.lookback ??= options.period / 2;
-    options.threshold ??= 0.5;
-    options.key ??= (req) => req.url;
+  constructor(
+    private readonly options: AdaptiveCacheOptions,
+    private readonly client: RedisClient | null = redis,
+  ) {
+    this.lookback = options.lookback ?? options.period / 2;
+    this.threshold = options.threshold ?? 0.5;
+    this.key = options.key ?? ((req) => req.url);
+    this.prefix = `adaptive-cache:v1:${options.name}`;
   }
 
   async fetch(
     input: RequestInfo | URL,
     init?: RequestInit,
-    fetcher: typeof fetch = fetch,
+    fetcher: Fetcher = fetch,
   ): Promise<Response> {
     const req = input instanceof Request ? input : new Request(input, init);
+    const startedAt = Date.now();
 
-    const now = Date.now();
+    if (!this.client) return fetcher(req);
 
-    this.cleanup(now);
+    const hash = await this.hash(this.key(req));
+    const responseKey = `${this.prefix}:response:${hash}`;
 
-    const key = this.options.key!(req);
+    try {
+      const { requests } = await this.stats(startedAt);
 
-    if (this.shouldCache()) {
-      const cached = this.cache.get(key);
-      if (cached && cached.expires > now) return cached.response.clone();
+      if (this.shouldCache(requests)) {
+        const cached = await this.client.get(responseKey);
+        if (cached) {
+          try {
+            return this.deserialize(cached);
+          } catch (error) {
+            console.error("Invalid AdaptiveCache response in Redis", error);
+            await this.client.send("DEL", [responseKey]);
+          }
+        }
+      }
+    } catch (error) {
+      console.error("AdaptiveCache Redis read failed", error);
+      return fetcher(req);
     }
 
     const res = await fetcher(req);
 
     if (res.ok) {
-      this.requests.push(now);
+      const expiresAt = startedAt + this.options.ttl;
+      const remainingTtl = Math.max(0, expiresAt - Date.now());
+      let cached = "";
 
-      this.cache.set(key, {
-        expires: now + this.options.ttl,
-        response: res.clone(),
-      });
+      try {
+        cached = await this.serialize(res.clone());
+      } catch (error) {
+        console.error("AdaptiveCache response serialization failed", error);
+      }
+
+      try {
+        await this.client.send("EVAL", [
+          STORE_SCRIPT,
+          "3",
+          responseKey,
+          `${this.prefix}:requests`,
+          `${this.prefix}:entries`,
+          `${startedAt}`,
+          `${startedAt}:${crypto.randomUUID()}`,
+          `${this.lookback}`,
+          cached,
+          `${cached ? remainingTtl : 0}`,
+          `${expiresAt}`,
+          hash,
+          `${this.options.ttl}`,
+        ]);
+      } catch (error) {
+        console.error("AdaptiveCache Redis write failed", error);
+      }
     }
 
     return res;
   }
 
-  private shouldCache() {
-    const projected =
-      this.requests.length * (this.options.period / this.options.lookback!);
+  async debugStats() {
+    if (!this.client)
+      throw new Error("Redis is not available in this environment.");
 
-    return projected >= this.options.maxRequests * this.options.threshold!;
-  }
-
-  private cleanup(now: number) {
-    const cutoff = now - this.options.lookback!;
-
-    while (this.requests.length && this.requests[0] < cutoff)
-      this.requests.shift();
-
-    for (const [key, value] of this.cache)
-      if (value.expires <= now) this.cache.delete(key);
-  }
-
-  debugStats() {
-    const projected =
-      this.requests.length * (this.options.period / this.options.lookback!);
+    const { requests, entries } = await this.stats(Date.now());
+    const projected = requests * (this.options.period / this.lookback);
 
     return [
       `AdaptiveCache`,
-      this.shouldCache() ? "HOT" : "COLD",
-      `recent=${this.requests.length}`,
+      this.shouldCache(requests) ? "HOT" : "COLD",
+      `recent=${requests}`,
       `projected=${projected.toFixed(1)}/${this.options.maxRequests}`,
       `usage=${((projected / this.options.maxRequests) * 100).toFixed(0)}%`,
-      `entries=${this.cache.size}`,
+      `entries=${entries}`,
     ].join(" | ");
   }
+
+  private shouldCache(requests: number) {
+    const projected = requests * (this.options.period / this.lookback);
+    return projected >= this.options.maxRequests * this.threshold;
+  }
+
+  private async stats(now: number) {
+    const result = await this.client!.send("EVAL", [
+      STATS_SCRIPT,
+      "2",
+      `${this.prefix}:requests`,
+      `${this.prefix}:entries`,
+      `${now - this.lookback}`,
+      `${now}`,
+    ]);
+
+    if (!Array.isArray(result) || result.length !== 2)
+      throw new Error("Unexpected AdaptiveCache stats response from Redis.");
+
+    return { requests: Number(result[0]), entries: Number(result[1]) };
+  }
+
+  private async hash(value: string) {
+    const digest = await crypto.subtle.digest(
+      "SHA-256",
+      new TextEncoder().encode(value),
+    );
+    return Array.from(new Uint8Array(digest), (byte) =>
+      byte.toString(16).padStart(2, "0"),
+    ).join("");
+  }
+
+  private async serialize(response: Response) {
+    const cached: CachedResponse = {
+      version: 1,
+      status: response.status,
+      statusText: response.statusText,
+      headers: Array.from(response.headers.entries()),
+      body: Buffer.from(await response.arrayBuffer()).toString("base64"),
+    };
+    return JSON.stringify(cached);
+  }
+
+  private deserialize(value: string) {
+    const cached = JSON.parse(value) as CachedResponse;
+    if (cached.version !== 1) throw new Error("Unsupported cache version.");
+
+    return new Response(Buffer.from(cached.body, "base64"), {
+      status: cached.status,
+      statusText: cached.statusText,
+      headers: cached.headers,
+    });
+  }
 }
 
 export const youtubeCache = new AdaptiveCache({
+  name: "youtube",
   maxRequests: 100, // YouTube search.list requests/day
   period: 24 * 60 * 60_000,
   ttl: 15 * 60_000,