fix(msteams): stream graph success responses

extensions/msteams/src/graph.test.ts

@@ -92,6 +92,33 @@ function mockTextFetchResponse(body: string, init?: ResponseInit) {
   mockFetch(async () => textResponse(body, init));
 }
 
+function graphStreamResponse(body: unknown): {
+  response: Response;
+  arrayBuffer: ReturnType<typeof vi.fn>;
+} {
+  const encoded = new TextEncoder().encode(JSON.stringify(body));
+  const stream = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(encoded);
+      controller.close();
+    },
+  });
+  const arrayBuffer = vi.fn(async () => {
+    throw new Error("Graph response must stay streaming");
+  });
+  return {
+    response: {
+      ok: true,
+      status: 200,
+      statusText: "OK",
+      headers: new Headers({ "content-type": "application/json" }),
+      body: stream,
+      arrayBuffer,
+    } as unknown as Response,
+    arrayBuffer,
+  };
+}
+
 function graphCollection<T>(...items: T[]) {
   return { value: items };
 }
@@ -229,6 +256,20 @@ describe("msteams graph helpers", () => {
     );
   });
 
+  it("keeps successful Graph responses streaming for bounded JSON parsing", async () => {
+    const { response, arrayBuffer } = graphStreamResponse(graphCollection(groupOne));
+    mockFetch(async () => response);
+
+    await expect(
+      fetchGraphJson<{ value: Array<{ id: string }> }>({
+        token: graphToken,
+        path: "/groups?$select=id",
+      }),
+    ).resolves.toEqual(graphCollection(groupOne));
+
+    expect(arrayBuffer).not.toHaveBeenCalled();
+  });
+
   it("posts Graph JSON to v1 and beta roots and treats empty mutation responses as undefined", async () => {
     mockFetch(async (input) => {
       if (requestUrl(input).startsWith("https://graph.microsoft.com/beta")) {

extensions/msteams/src/graph.ts

@@ -31,6 +31,50 @@ type GraphChannel = {
 
 export type GraphResponse<T> = { value?: T[] };
 
+function responseWithRelease(response: Response, release: () => Promise<void>): Response {
+  let released = false;
+  const releaseOnce = async () => {
+    if (released) {
+      return;
+    }
+    released = true;
+    await release();
+  };
+
+  if (!response.body || NULL_BODY_STATUSES.has(response.status)) {
+    void releaseOnce();
+    return response;
+  }
+
+  const reader = response.body.getReader();
+  const body = new ReadableStream<Uint8Array>({
+    async pull(controller) {
+      try {
+        const next = await reader.read();
+        if (next.done) {
+          controller.close();
+          await releaseOnce();
+          return;
+        }
+        controller.enqueue(next.value);
+      } catch (error) {
+        await releaseOnce();
+        throw error;
+      }
+    },
+    async cancel(reason) {
+      void reader.cancel(reason).catch(() => undefined);
+      await releaseOnce();
+    },
+  });
+
+  return new Response(body, {
+    status: response.status,
+    statusText: response.statusText,
+    headers: response.headers,
+  });
+}
+
 export function normalizeQuery(value?: string | null): string {
   return value?.trim() ?? "";
 }
@@ -66,6 +110,7 @@ async function requestGraph(params: {
     },
     auditContext: "msteams.graph",
   });
+  let releaseInFinally = true;
   try {
     if (!response.ok) {
       throw await createMSTeamsHttpError(
@@ -73,14 +118,12 @@ async function requestGraph(params: {
         `${params.errorPrefix ?? "Graph"} ${params.path} failed`,
       );
     }
-    const body = NULL_BODY_STATUSES.has(response.status) ? null : await response.arrayBuffer();
-    return new Response(body, {
-      status: response.status,
-      statusText: response.statusText,
-      headers: new Headers(response.headers),
-    });
+    releaseInFinally = false;
+    return responseWithRelease(response, release);
   } finally {
-    await release();
+    if (releaseInFinally) {
+      await release();
+    }
   }
 }