Use ReadableStreamLinks for normal data transport, too (#424)

* stream deferred useSuspenseQuery

* use readableStream everywhere

* add missing transform call

* change defer delay to the same 1000 everywhere

* remove some obsolete code

* freeze transported data immediately when queuing injection

* `QueryEvent`: reuse `ReadableStreamLinkEvent`type & shape

* update resolution

* only start streaming on link hit

* clean up console.log

* fixup

* fix SSR test

* address slight asynchronity in new solution

* restart query using a link query

* more progress

* move links into base class

* no-cache

* update AC version to 3.13.0-rc.0

* remove unneccessary property

* remove exports keyword, slight type tweaks

integration-test/nextjs/src/app/cc/dynamic/dynamic.test.ts

@@ -5,6 +5,7 @@ const regex_connection_closed_early =
   /streaming connection closed before server query could be fully transported, rerunning/i;
 const regex_query_error_restart =
   /query failed on server, rerunning in browser/i;
+const reactErr419 = /(Minified React error #419|Switched to client rendering)/;
 
 test.describe("CC dynamic", () => {
   test.describe("useSuspenseQuery", () => {
@@ -40,7 +41,7 @@ test.describe("CC dynamic", () => {
         return regex_query_error_restart.test(message.text());
       });
       await page.waitForEvent("pageerror", (error) => {
-        return error.message.includes("Minified React error #419");
+        return reactErr419.test(error.message);
       });
 
       await hydrationFinished;

integration-test/nextjs/src/app/rsc/dynamic/PreloadQuery/PreloadQuery.test.ts

@@ -1,6 +1,8 @@
 import { expect } from "@playwright/test";
 import { test } from "../../../../../fixture";
 
+const reactErr419 = /(Minified React error #419|Switched to client rendering)/;
+
 test.describe("PreloadQuery", () => {
   for (const [description, path] of [
     ["with useSuspenseQuery", "useSuspenseQuery"],
@@ -37,10 +39,7 @@ test.describe("PreloadQuery", () => {
         await expect(page).toBeInitiallyLoading(true);
 
         await page.waitForEvent("pageerror", (error) => {
-          return (
-            /* prod */ error.message.includes("Minified React error #419") ||
-            /* dev */ error.message.includes("Query failed upstream.")
-          );
+          return reactErr419.test(error.message);
         });
 
         await expect(page.getByText("loading")).not.toBeVisible();
@@ -94,10 +93,7 @@ test.describe("PreloadQuery", () => {
           await expect(page).toBeInitiallyLoading(true);
 
           await page.waitForEvent("pageerror", (error) => {
-            return (
-              /* prod */ error.message.includes("Minified React error #419") ||
-              /* dev */ error.message.includes("Query failed upstream.")
-            );
+            return reactErr419.test(error.message);
           });
 
           await expect(page.getByText("loading")).not.toBeVisible();

integration-test/react-router/app/routes/home.tsx

@@ -11,7 +11,7 @@ import { useTransition } from "react";
 
 export const loader = apolloLoader<Route.LoaderArgs>()(({ preloadQuery }) => {
   const queryRef = preloadQuery(DEFERRED_QUERY, {
-    variables: { delayDeferred: 500 },
+    variables: { delayDeferred: 1000 },
   });
   return {
     queryRef,

integration-test/tanstack-start/app/router.tsx

@@ -27,11 +27,12 @@ export const delayLink = new ApolloLink((operation, forward) => {
   });
 });
 
-const link = delayLink.concat(
+const link = ApolloLink.from([
+  delayLink,
   typeof window === "undefined"
     ? new IncrementalSchemaLink({ schema })
-    : new HttpLink({ uri: "/api/graphql" })
-);
+    : new HttpLink({ uri: "/api/graphql" }),
+]);
 
 export function createRouter() {
   const apolloClient = new ApolloClient({

integration-test/tanstack-start/app/routes/loader-defer.tsx

@@ -11,7 +11,7 @@ export const Route = createFileRoute("/loader-defer")({
   component: LoaderDeferPage,
   loader: async ({ context: { preloadQuery } }) => {
     const queryRef = preloadQuery(DEFERRED_QUERY, {
-      variables: { delayDeferred: 500 },
+      variables: { delayDeferred: 1000 },
     });
     return {
       queryRef,

integration-test/tanstack-start/app/routes/useSuspenseQuery.tsx

@@ -1,6 +1,6 @@
 import { createFileRoute } from "@tanstack/react-router";
-import { QUERY } from "@integration-test/shared/queries";
-import { useSuspenseQuery } from "@apollo/client/index.js";
+import { DEFERRED_QUERY } from "@integration-test/shared/queries";
+import { useApolloClient, useSuspenseQuery } from "@apollo/client/index.js";
 import { useTransition } from "react";
 
 export const Route = createFileRoute("/useSuspenseQuery")({
@@ -9,19 +9,43 @@ export const Route = createFileRoute("/useSuspenseQuery")({
 
 function RouteComponent() {
   const [refetching, startTransition] = useTransition();
-  const { data, refetch } = useSuspenseQuery(QUERY);
+  const client = useApolloClient();
+  const { data, refetch } = useSuspenseQuery(DEFERRED_QUERY, {
+    variables: { delayDeferred: 1000 },
+  });
 
   return (
     <>
       <ul>
-        {data.products.map(({ id, title }) => (
-          <li key={id}>{title}</li>
+        {data.products.map(({ id, title, rating }) => (
+          <li key={id}>
+            {title}
+            <br />
+            Rating:{" "}
+            <div style={{ display: "inline-block", verticalAlign: "text-top" }}>
+              {rating?.value || ""}
+              <br />
+              {rating ? `Queried in ${rating.env} environment` : "loading..."}
+            </div>
+          </li>
         ))}
       </ul>
       <p>Queried in {data.env} environment</p>
       <button
         disabled={refetching}
         onClick={() => {
+          client.cache.batch({
+            update(cache) {
+              for (const product of data.products) {
+                cache.modify({
+                  id: cache.identify(product),
+                  fields: {
+                    rating: () => null,
+                  },
+                });
+              }
+            },
+          });
           startTransition(() => {
             refetch();
           });

package.json

@@ -28,7 +28,7 @@
   },
   "resolutions": {
     "@microsoft/api-documenter": "7.24.1",
-    "@apollo/client": "3.12.6",
+    "@apollo/client": "3.13.0-rc.0",
     "graphql": "17.0.0-alpha.2",
     "react-router": "patch:react-router@npm%3A7.0.2#~/.yarn/patches/react-router-npm-7.0.2-b96f2bd13c.patch",
     "@tanstack/start": "1.99.6",

packages/client-react-streaming/package.json

@@ -160,7 +160,7 @@
     "zen-observable-ts": "1.1.0"
   },
   "peerDependencies": {
-    "@apollo/client": "^3.10.4",
+    "@apollo/client": ">=3.13.0-rc.0",
     "graphql": "^16 || >=17.0.0-alpha.2",
     "react": "^18 || >=19.0.0-rc",
     "react-dom": "^18 || >=19.0.0-rc"

packages/client-react-streaming/src/DataTransportAbstraction/DataTransportAbstraction.ts

@@ -1,7 +1,8 @@
 import type React from "react";
-import type { FetchResult, Observable } from "@apollo/client/index.js";
+import type { Observable } from "@apollo/client/index.js";
 import { createContext } from "react";
 import type { TransportedOptions } from "./transportedOptions.js";
+import type { ReadableStreamLinkEvent } from "../ReadableStreamLink.ts";
 
 interface DataTransportAbstraction {
   /**
@@ -74,20 +75,8 @@ export type QueryEvent =
       options: TransportedOptions;
       id: TransportIdentifier;
     }
-  | {
-      type: "data";
-      id: TransportIdentifier;
-      result: FetchResult;
-    }
-  | {
-      type: "error";
-      id: TransportIdentifier;
-      // for now we don't transport the error itself, as it might leak some sensitive information
-      // this is similar to how React handles errors during SSR
-    }
-  | {
-      type: "complete";
+  | (ReadableStreamLinkEvent & {
       id: TransportIdentifier;
-    };
+    });
 
 export type ProgressEvent = Exclude<QueryEvent, { type: "started" }>;

packages/client-react-streaming/src/DataTransportAbstraction/WrappedApolloClient.test.tsx

@@ -55,12 +55,12 @@ describe(
     };
     const FIRST_RESULT = { me: "User" };
     const EVENT_DATA: QueryEvent = {
-      type: "data",
+      type: "next",
       id: "1" as any,
-      result: { data: FIRST_RESULT },
+      value: { data: FIRST_RESULT },
     };
     const EVENT_COMPLETE: QueryEvent = {
-      type: "complete",
+      type: "completed",
       id: "1" as any,
     };
     const FIRST_HOOK_RESULT = {
@@ -127,7 +127,9 @@ describe(
           )
         );
 
-        assert.deepStrictEqual(events, [EVENT_STARTED]);
+        // these are an uuid, not just an incremental number, so we need to fix our EVENT_* constants with the real random uuid
+        const id = events[0].id;
+        assert.deepStrictEqual(events, [{ ...EVENT_STARTED, id }]);
         assert.deepStrictEqual(staticData, []);
 
         await act(async () =>
@@ -137,9 +139,9 @@ describe(
         await findByText("User");
 
         assert.deepStrictEqual(events, [
-          EVENT_STARTED,
-          EVENT_DATA,
-          EVENT_COMPLETE,
+          { ...EVENT_STARTED, id },
+          { ...EVENT_DATA, id },
+          { ...EVENT_COMPLETE, id },
         ]);
         assert.deepStrictEqual(
           staticData,
@@ -313,6 +315,7 @@ describe(
         appendToBody`<div hidden id="S:1"><div id="user">User</div></div>`;
         $RS("S:1", "P:1");
 
+        await new Promise((resolve) => setTimeout(resolve, 10));
         // meanwhile, in the browser, the cache is modified
         client.cache.writeQuery({
           query: QUERY_ME,