add local in memory cache

Author: conico974

packages/cloudflare/src/api/queue-cache.spec.ts

@@ -0,0 +1,112 @@
+import type { Queue } from "@opennextjs/aws/types/overrides";
+import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
+
+import queueCache from "./queue-cache";
+
+const mockedQueue = {
+  name: "mocked-queue",
+  send: vi.fn(),
+} satisfies Queue;
+
+const generateMessage = () => ({
+  MessageGroupId: "test",
+  MessageBody: {
+    eTag: "test",
+    url: "test",
+    host: "test",
+    lastModified: Date.now(),
+  },
+  MessageDeduplicationId: "test",
+});
+
+const mockedPut = vi.fn();
+const mockedMatch = vi.fn().mockReturnValue(null);
+
+describe("queue-cache", () => {
+  beforeEach(() => {
+    // @ts-ignore
+    globalThis.caches = {
+      open: vi.fn().mockReturnValue({
+        put: mockedPut,
+        match: mockedMatch,
+      }),
+    };
+  });
+
+  afterEach(() => {
+    vi.resetAllMocks();
+  });
+  test("should send the message to the original queue", async () => {
+    const msg = generateMessage();
+    const queue = queueCache(mockedQueue, {});
+    expect(queue.name).toBe("cached-mocked-queue");
+    await queue.send(msg);
+    expect(mockedQueue.send).toHaveBeenCalledWith(msg);
+  });
+
+  test("should use the local cache", async () => {
+    const msg = generateMessage();
+    const queue = queueCache(mockedQueue, {});
+    await queue.send(msg);
+
+    expect(queue.localCache.size).toBe(1);
+    expect(queue.localCache.has(`queue/test/test`)).toBe(true);
+    expect(mockedPut).toHaveBeenCalled();
+
+    const spiedHas = vi.spyOn(queue.localCache, "has");
+    await queue.send(msg);
+    expect(spiedHas).toHaveBeenCalled();
+
+    expect(mockedQueue.send).toHaveBeenCalledTimes(1);
+
+    expect(mockedMatch).toHaveBeenCalledTimes(1);
+  });
+
+  test("should clear the local cache after 5s", async () => {
+    vi.useFakeTimers();
+    const msg = generateMessage();
+    const queue = queueCache(mockedQueue, {});
+    await queue.send(msg);
+    expect(queue.localCache.size).toBe(1);
+    expect(queue.localCache.has(`queue/test/test`)).toBe(true);
+
+    vi.advanceTimersByTime(5001);
+    const alteredMsg = generateMessage();
+    alteredMsg.MessageGroupId = "test2";
+    await queue.send(alteredMsg);
+    expect(queue.localCache.size).toBe(1);
+    console.log(queue.localCache);
+    expect(queue.localCache.has(`queue/test2/test`)).toBe(true);
+    expect(queue.localCache.has(`queue/test/test`)).toBe(false);
+    vi.useRealTimers();
+  });
+
+  test("should use the regional cache if not in local cache", async () => {
+    const msg = generateMessage();
+    const queue = queueCache(mockedQueue, {});
+    await queue.send(msg);
+
+    expect(mockedMatch).toHaveBeenCalledTimes(1);
+    expect(mockedPut).toHaveBeenCalledTimes(1);
+    expect(queue.localCache.size).toBe(1);
+    expect(queue.localCache.has(`queue/test/test`)).toBe(true);
+    // We need to delete the local cache to test the regional cache
+    queue.localCache.delete(`queue/test/test`);
+
+    const spiedHas = vi.spyOn(queue.localCache, "has");
+    await queue.send(msg);
+    expect(spiedHas).toHaveBeenCalled();
+    expect(mockedMatch).toHaveBeenCalledTimes(2);
+  });
+
+  test("should return early if the message is in the regional cache", async () => {
+    const msg = generateMessage();
+    const queue = queueCache(mockedQueue, {});
+
+    mockedMatch.mockReturnValueOnce(new Response(null, { status: 200 }));
+
+    const spiedSend = mockedQueue.send;
+    await queue.send(msg);
+    expect(spiedSend).not.toHaveBeenCalled();
+  });
+});

packages/cloudflare/src/api/queue-cache.ts

@@ -51,6 +51,8 @@ class QueueCache implements Queue {
       }
     } catch (e) {
       error("Error sending message to queue", e);
+    } finally {
+      this.clearLocalCache();
     }
   }
 
@@ -81,7 +83,12 @@ class QueueCache implements Queue {
 
   private async isInCache(msg: QueueMessage) {
     if (this.localCache.has(this.getCacheUrlString(msg))) {
-      return true;
+      const insertedAt = this.localCache.get(this.getCacheUrlString(msg))!;
+      if (Date.now() - insertedAt < this.regionalCacheTtlSec * 1000) {
+        return true;
+      }
+      this.localCache.delete(this.getCacheUrlString(msg));
+      return false;
     }
     const cacheKey = this.getCacheKey(msg);
     const cache = await this.getCache();