store blobs rather than texts

Author: anonrig

packages/open-next/src/adapters/composable-cache.ts

@@ -1,6 +1,5 @@
 import type { ComposableCacheEntry, ComposableCacheHandler } from "types/cache";
 import { writeTags } from "utils/cache";
-import { fromReadableStream, toReadableStream } from "utils/stream";
 import { debug } from "./logger";
 
 const pendingWritePromiseMap = new Map<string, Promise<ComposableCacheEntry>>();
@@ -50,7 +49,7 @@ export default {
 
       return {
         ...result.value,
-        value: toReadableStream(result.value.value),
+        value: result.value.value,
       };
     } catch (e) {
       debug("Cannot read composable cache entry");
@@ -59,39 +58,19 @@ export default {
   },
 
   async set(cacheKey: string, pendingEntry: Promise<ComposableCacheEntry>) {
-    const teedPromise = pendingEntry.then((entry) => {
-      // Optimization: We avoid consuming and stringifying the stream here,
-      // because it creates double copies just to be discarded when this function
-      // ends. This avoids unnecessary memory usage, and reduces GC pressure.
-      const [stream1, stream2] = entry.value.tee();
-      return [
-        { ...entry, value: stream1 },
-        { ...entry, value: stream2 },
-      ] as const;
-    });
-
-    pendingWritePromiseMap.set(
-      cacheKey,
-      teedPromise.then(([entry]) => entry),
-    );
+    // Store the entry in the pending map so concurrent get() calls can access it
+    pendingWritePromiseMap.set(cacheKey, pendingEntry);
 
-    const [, entryForStorage] = await teedPromise.finally(() => {
+    const entry = await pendingEntry.finally(() => {
       pendingWritePromiseMap.delete(cacheKey);
     });
 
-    await globalThis.incrementalCache.set(
-      cacheKey,
-      {
-        ...entryForStorage,
-        value: await fromReadableStream(entryForStorage.value),
-      },
-      "composable",
-    );
+    await globalThis.incrementalCache.set(cacheKey, entry, "composable");
 
     if (globalThis.tagCache.mode === "original") {
       const storedTags = await globalThis.tagCache.getByPath(cacheKey);
       const tagsToWrite = [];
-      for (const tag of entryForStorage.tags) {
+      for (const tag of entry.tags) {
         if (!storedTags.includes(tag)) {
           tagsToWrite.push({ tag, path: cacheKey });
         }

packages/open-next/src/types/cache.ts

@@ -1,5 +1,3 @@
-import type { ReadableStream } from "node:stream/web";
-
 interface CachedFetchValue {
   kind: "FETCH";
   data: {
@@ -143,7 +141,7 @@ export type IncrementalCacheContext =
   | SetIncrementalResponseCacheContext;
 
 export interface ComposableCacheEntry {
-  value: ReadableStream<Uint8Array>;
+  value: Blob;
   tags: string[];
   stale: number;
   timestamp: number;
@@ -152,7 +150,7 @@ export interface ComposableCacheEntry {
 }
 
 export type StoredComposableCacheEntry = Omit<ComposableCacheEntry, "value"> & {
-  value: string;
+  value: Blob;
 };
 
 export interface ComposableCacheHandler {

packages/tests-unit/tests/adapters/composable-cache.test.ts

@@ -1,8 +1,4 @@
 import ComposableCache from "@opennextjs/aws/adapters/composable-cache";
-import {
-  fromReadableStream,
-  toReadableStream,
-} from "@opennextjs/aws/utils/stream";
 import { vi } from "vitest";
 
 describe("Composable cache handler", () => {
@@ -19,7 +15,7 @@ describe("Composable cache handler", () => {
         timestamp: Date.now(),
         expire: Date.now() + 1000,
         revalidate: 3600,
-        value: "test-value",
+        value: new Blob(["test-value"]),
       },
       lastModified: Date.now(),
     }),
@@ -81,7 +77,7 @@ describe("Composable cache handler", () => {
       );
       expect(result).toBeDefined();
       expect(result?.tags).toEqual(["tag1", "tag2"]);
-      expect(result?.value).toBeInstanceOf(ReadableStream);
+      expect(result?.value).toBeInstanceOf(Blob);
     });
 
     it("should return undefined when cache entry does not exist", async () => {
@@ -132,7 +128,7 @@ describe("Composable cache handler", () => {
           type: "route",
           body: "{}",
           tags: [],
-          value: "test-value",
+          value: new Blob(["test-value"]),
         },
         lastModified: Date.now(),
       });
@@ -185,7 +181,7 @@ describe("Composable cache handler", () => {
 
     it("should return pending write promise if available", async () => {
       const pendingEntry = Promise.resolve({
-        value: toReadableStream("pending-value"),
+        value: new Blob(["pending-value"]),
         tags: ["tag1"],
         stale: 0,
         timestamp: Date.now(),
@@ -200,7 +196,7 @@ describe("Composable cache handler", () => {
       const result = await ComposableCache.get("pending-key");
 
       expect(result).toBeDefined();
-      expect(result?.value).toBeInstanceOf(ReadableStream);
+      expect(result?.value).toBeInstanceOf(Blob);
 
       // Wait for set to complete
       await setPromise;
@@ -214,8 +210,9 @@ describe("Composable cache handler", () => {
 
     it("should set cache entry and handle tags in original mode", async () => {
       tagCache.mode = "original";
+      const blob = new Blob(["test-value"]);
       const entry = {
-        value: toReadableStream("test-value"),
+        value: blob,
         tags: ["tag1", "tag2"],
         stale: 0,
         timestamp: Date.now(),
@@ -229,7 +226,7 @@ describe("Composable cache handler", () => {
         "test-key",
         expect.objectContaining({
           tags: ["tag1", "tag2"],
-          value: "test-value",
+          value: blob,
         }),
         "composable",
       );
@@ -241,7 +238,7 @@ describe("Composable cache handler", () => {
       tagCache.getByPath.mockResolvedValueOnce(["tag1"]);
 
       const entry = {
-        value: toReadableStream("test-value"),
+        value: new Blob(["test-value"]),
         tags: ["tag1", "tag2", "tag3"],
         stale: 0,
         timestamp: Date.now(),
@@ -262,7 +259,7 @@ describe("Composable cache handler", () => {
       tagCache.getByPath.mockResolvedValueOnce(["tag1", "tag2"]);
 
       const entry = {
-        value: toReadableStream("test-value"),
+        value: new Blob(["test-value"]),
         tags: ["tag1", "tag2"],
         stale: 0,
         timestamp: Date.now(),
@@ -279,7 +276,7 @@ describe("Composable cache handler", () => {
       tagCache.mode = "nextMode";
 
       const entry = {
-        value: toReadableStream("test-value"),
+        value: new Blob(["test-value"]),
         tags: ["tag1", "tag2"],
         stale: 0,
         timestamp: Date.now(),
@@ -293,9 +290,10 @@ describe("Composable cache handler", () => {
       expect(tagCache.writeTags).not.toHaveBeenCalled();
     });
 
-    it("should convert ReadableStream to string", async () => {
+    it("should store Blob directly", async () => {
+      const blob = new Blob(["test-content"]);
       const entry = {
-        value: toReadableStream("test-content"),
+        value: blob,
         tags: ["tag1"],
         stale: 0,
         timestamp: Date.now(),
@@ -308,7 +306,7 @@ describe("Composable cache handler", () => {
       expect(incrementalCache.set).toHaveBeenCalledWith(
         "test-key",
         expect.objectContaining({
-          value: "test-content",
+          value: blob,
         }),
         "composable",
       );
@@ -437,8 +435,9 @@ describe("Composable cache handler", () => {
   describe("integration tests", () => {
     it("should handle complete cache lifecycle", async () => {
       // Set a cache entry
+      const blob = new Blob(["integration-test"]);
       const entry = {
-        value: toReadableStream("integration-test"),
+        value: blob,
         tags: ["integration-tag"],
         stale: 0,
         timestamp: Date.now(),
@@ -452,7 +451,7 @@ describe("Composable cache handler", () => {
       expect(incrementalCache.set).toHaveBeenCalledWith(
         "integration-key",
         expect.objectContaining({
-          value: "integration-test",
+          value: blob,
           tags: ["integration-tag"],
         }),
         "composable",
@@ -462,7 +461,7 @@ describe("Composable cache handler", () => {
       incrementalCache.get.mockResolvedValueOnce({
         value: {
           ...entry,
-          value: "integration-test",
+          value: blob,
         },
         lastModified: Date.now(),
       });
@@ -473,14 +472,14 @@ describe("Composable cache handler", () => {
       expect(result).toBeDefined();
       expect(result?.tags).toEqual(["integration-tag"]);
 
-      // Convert the stream back to verify content
-      const content = await fromReadableStream(result!.value);
+      // Convert the blob back to verify content
+      const content = await result!.value.text();
       expect(content).toBe("integration-test");
     });
 
     it("should handle concurrent get/set operations", async () => {
       const entry1 = {
-        value: toReadableStream("concurrent-1"),
+        value: new Blob(["concurrent-1"]),
         tags: ["tag1"],
         stale: 0,
         timestamp: Date.now(),
@@ -489,7 +488,7 @@ describe("Composable cache handler", () => {
       };
 
       const entry2 = {
-        value: toReadableStream("concurrent-2"),
+        value: new Blob(["concurrent-2"]),
         tags: ["tag2"],
         stale: 0,
         timestamp: Date.now(),
@@ -513,10 +512,10 @@ describe("Composable cache handler", () => {
       expect(results[2]).toBeDefined();
       expect(results[3]).toBeDefined();
 
-      const content1 = await fromReadableStream(results[2]!.value);
+      const content1 = await results[2]!.value.text();
       expect(content1).toBe("concurrent-1");
 
-      const content2 = await fromReadableStream(results[3]!.value);
+      const content2 = await results[3]!.value.text();
       expect(content2).toBe("concurrent-2");
     });
   });