Add a plugin API for reading HTTP response bodies (#560)

packages/plugin-runtime/src/PluginInstance.ts

@@ -21,11 +21,13 @@ import type {
   GetCookieValueRequest,
   GetCookieValueResponse,
   GetHttpRequestByIdResponse,
+  GetHttpResponseBodyInfoResponse,
   GetKeyValueResponse,
   GrpcRequestAction,
   HttpAuthenticationAction,
   HttpRequest,
   HttpRequestAction,
+  HttpResponse,
   ImportResources,
   InternalEvent,
   InternalEventPayload,
@@ -37,6 +39,7 @@ import type {
   PluginContext,
   PromptFormResponse,
   PromptTextResponse,
+  ReadHttpResponseBodyChunkResponse,
   RenderGrpcRequestResponse,
   RenderHttpRequestResponse,
   SendHttpRequestResponse,
@@ -49,6 +52,22 @@ import type {
 import { applyDynamicFormInput } from "./common";
 import { EventChannel } from "./EventChannel";
 import { migrateTemplateFunctionSelectOptions } from "./migrations";
+import { createResponseBody, decodeBase64Chunk } from "./responseBody";
+
+/**
+ * A response as a plugin should see it.
+ *
+ * The host still puts `bodyPath` on the wire for its own callers, but it names
+ * a file on the host's disk — meaningless to a plugin, absent once bodies move
+ * off the filesystem, and impossible in a browser. Plugins address bodies by
+ * response id, so drop it here rather than let one grow a dependency on it.
+ */
+function forPlugin(httpResponse: HttpResponse): HttpResponse {
+  const { bodyPath: _bodyPath, ...rest } = httpResponse as HttpResponse & {
+    bodyPath?: string | null;
+  };
+  return rest;
+}
 
 export interface PluginWorkerData {
   bootRequest: BootRequest;
@@ -552,6 +571,14 @@ export class PluginInstance {
     return this.#sendPayload(context, { type: "empty_response" }, replyId);
   }
 
+  /**
+   * Send a request to the host and wait for its reply.
+   *
+   * A host that cannot answer replies with an error, which becomes a thrown
+   * error here. The alternative is handing back a reply-shaped object with
+   * none of the fields the caller destructures, and letting it fail somewhere
+   * further along with no idea why.
+   */
   #sendForReply<T extends Omit<InternalEventPayload, "type">>(
     context: PluginContext,
     payload: InternalEventPayload,
@@ -560,11 +587,16 @@ export class PluginInstance {
     const eventToSend = this.#buildEventToSend(context, payload, null);
 
     // 2. Spawn listener in background
-    const promise = new Promise<T>((resolve) => {
+    const promise = new Promise<T>((resolve, reject) => {
       const cb = (event: InternalEvent) => {
         if (event.replyId === eventToSend.id) {
           this.#appToPluginEvents.unlisten(cb); // Unlisten, now that we're done
           const { type: _, ...payload } = event.payload;
+          if (event.payload.type === "error_response") {
+            const { error } = payload as { error?: string };
+            reject(new Error(error || `Host failed to handle ${eventToSend.payload.type}`));
+            return;
+          }
           resolve(payload as T);
         }
       };
@@ -598,6 +630,33 @@ export class PluginInstance {
   }
 
   #newCtx(context: PluginContext): Context {
+    /** Read a body the host has stored, a chunk at a time, following it if it is still arriving. */
+    const storedBody = async (responseId: string) => {
+      const bodyInfo = () =>
+        this.#sendForReply<GetHttpResponseBodyInfoResponse>(context, {
+          type: "get_http_response_body_info_request",
+          responseId,
+        });
+      const info = await bodyInfo();
+
+      return createResponseBody(
+        {
+          responseId,
+          contentLength: info.contentLength,
+          contentType: info.contentType ?? null,
+          complete: info.complete,
+        },
+        async (offset, length) => {
+          const chunk = await this.#sendForReply<ReadHttpResponseBodyChunkResponse>(
+            context,
+            { type: "read_http_response_body_chunk_request", responseId, offset, length },
+          );
+          return decodeBase64Chunk(chunk.data);
+        },
+        { refresh: bodyInfo },
+      );
+    };
+
     const _windowInfo = async () => {
       if (context.label == null) {
         throw new Error("Can't get window context without an active window");
@@ -749,8 +808,9 @@ export class PluginInstance {
             context,
             payload,
           );
-          return httpResponses;
+          return httpResponses.map(forPlugin);
         },
+        body: ({ responseId }) => storedBody(responseId),
       },
       grpcRequest: {
         render: async (args) => {
@@ -782,11 +842,34 @@ export class PluginInstance {
             type: "send_http_request_request",
             ...args,
           } as const;
-          const { httpResponse } = await this.#sendForReply<SendHttpRequestResponse>(
+          const { httpResponse, body } = await this.#sendForReply<SendHttpRequestResponse>(
             context,
             payload,
           );
-          return httpResponse;
+
+          // A send with no request behind it saves nothing, so the reply
+          // carries the only copy of its body. A saved one is read back from
+          // the host like any other. Callers get the same thing either way.
+          if (body == null) {
+            return { httpResponse: forPlugin(httpResponse), body: await storedBody(httpResponse.id) };
+          }
+
+          const bytes = decodeBase64Chunk(body);
+          return {
+            httpResponse: forPlugin(httpResponse),
+            body: createResponseBody(
+              {
+                responseId: httpResponse.id,
+                contentLength: bytes.byteLength,
+                contentType:
+                  httpResponse.headers.find((h) => h.name.toLowerCase() === "content-type")
+                    ?.value ?? null,
+                // The host waited for the whole send before replying.
+                complete: true,
+              },
+              async (offset, length) => bytes.slice(offset, offset + length),
+            ),
+          };
         },
         render: async (args) => {
           const payload = {

packages/plugin-runtime/src/responseBody.ts

@@ -0,0 +1,194 @@
+import type { HttpResponseBody, ReadHttpResponseBodyOptions } from "@yaakapp/api";
+
+/** Bytes pulled from the host per round trip, when the caller doesn't say. */
+const DEFAULT_CHUNK_SIZE = 1024 * 1024;
+
+/**
+ * The most a plugin buffers by default.
+ *
+ * Reading a body used to be unbounded, so any ceiling is an improvement; this
+ * one is set well above what an API returns and well below what makes the
+ * plugin runtime fall over. `chunks()` has no ceiling, and any caller that
+ * really wants the whole thing can raise `maxBytes`.
+ */
+const DEFAULT_MAX_BYTES = 32 * 1024 * 1024;
+
+/** How long to wait, having caught up with a body still arriving, before looking again. */
+const DEFAULT_POLL_INTERVAL_MS = 100;
+
+/** Fetch one window of body bytes from the host. */
+export type ReadResponseBodyChunk = (offset: number, length: number) => Promise<Uint8Array>;
+
+export interface ResponseBodyInfo {
+  responseId: string;
+  contentLength: number;
+  contentType: string | null;
+  /** Whether the response has finished arriving, so `contentLength` is final. */
+  complete: boolean;
+}
+
+/** What can change while a body is still arriving. */
+export type ResponseBodyProgress = Pick<ResponseBodyInfo, "contentLength" | "complete">;
+
+export interface CreateResponseBodyOptions {
+  /**
+   * Ask the host where the body has got to. Needed only for a body that was
+   * not complete when opened; a reader that has caught up calls this to learn
+   * whether to wait for more or stop.
+   */
+  refresh?: () => Promise<ResponseBodyProgress>;
+  pollIntervalMs?: number;
+}
+
+export function createResponseBody(
+  info: ResponseBodyInfo,
+  readChunk: ReadResponseBodyChunk,
+  { refresh, pollIntervalMs = DEFAULT_POLL_INTERVAL_MS }: CreateResponseBodyOptions = {},
+): HttpResponseBody {
+  const { responseId, contentLength, contentType, complete } = info;
+
+  /**
+   * Yield the body from the start until it has all arrived.
+   *
+   * A complete body is read up to its known length and no further. One still
+   * arriving is followed: on catching up, ask the host whether it has finished,
+   * and if not, wait and look again. So this ends when the response does —
+   * which for a stream that never closes means it doesn't, exactly as
+   * iterating `fetch`'s body would not.
+   */
+  async function* chunks(
+    options?: Pick<ReadHttpResponseBodyOptions, "chunkSize">,
+  ): AsyncIterable<Uint8Array> {
+    const chunkSize = Math.max(1, Math.floor(options?.chunkSize ?? DEFAULT_CHUNK_SIZE));
+    let known = contentLength;
+    let done = complete;
+    let offset = 0;
+
+    while (true) {
+      if (done && offset >= known) return;
+
+      const want = done ? Math.min(chunkSize, known - offset) : chunkSize;
+      const chunk = await readChunk(offset, want);
+      if (chunk.byteLength > 0) {
+        yield chunk;
+        offset += chunk.byteLength;
+        continue;
+      }
+
+      // Caught up. A complete body that came up short simply ended sooner than
+      // the host said; one still arriving needs asking about.
+      if (done || refresh == null) return;
+      ({ contentLength: known, complete: done } = await refresh());
+      if (offset < known) continue;
+      if (done) return;
+      await new Promise((resolve) => setTimeout(resolve, pollIntervalMs));
+    }
+  }
+
+  async function readAll(accessor: string, options?: ReadHttpResponseBodyOptions) {
+    const maxBytes = options?.maxBytes ?? DEFAULT_MAX_BYTES;
+    refuseIfTooBig(accessor, contentLength, maxBytes);
+
+    const parts: Uint8Array[] = [];
+    let total = 0;
+    for await (const chunk of chunks(options)) {
+      total += chunk.byteLength;
+      // The size the host reported is a claim about a moment ago, so check the
+      // bytes actually arriving too.
+      refuseIfTooBig(accessor, total, maxBytes);
+      parts.push(chunk);
+    }
+
+    const bytes = new Uint8Array(total);
+    let offset = 0;
+    for (const part of parts) {
+      bytes.set(part, offset);
+      offset += part.byteLength;
+    }
+    return bytes;
+  }
+
+  return {
+    responseId,
+    contentLength,
+    contentType,
+    complete,
+    chunks,
+    async arrayBuffer(options) {
+      const bytes = await readAll("arrayBuffer", options);
+      return bytes.buffer as ArrayBuffer;
+    },
+    async text(options) {
+      return decodeBody(await readAll("text", options), contentType);
+    },
+    async json<T>(options?: ReadHttpResponseBodyOptions) {
+      return JSON.parse(decodeBody(await readAll("json", options), contentType)) as T;
+    },
+  };
+}
+
+function refuseIfTooBig(accessor: string, bytes: number, maxBytes: number) {
+  if (bytes <= maxBytes) return;
+  throw new Error(
+    `Response body is ${formatBytes(bytes)}, over the ${formatBytes(maxBytes)} limit for ` +
+      `${accessor}(). Read it with chunks() instead, or pass a larger maxBytes.`,
+  );
+}
+
+/**
+ * Decode using the charset the response declared.
+ *
+ * Assuming UTF-8 mangles every response that isn't, and the header is right
+ * there. An unknown label is the one case worth guessing on, since the
+ * alternative is refusing to read a body we can very likely still read.
+ */
+function decodeBody(bytes: Uint8Array, contentType: string | null): string {
+  const charset = parseCharset(contentType);
+  if (charset != null) {
+    try {
+      return new TextDecoder(charset).decode(bytes);
+    } catch {
+      // Not a label this runtime knows.
+    }
+  }
+  // TextDecoder drops a leading BOM on its own.
+  return new TextDecoder("utf-8").decode(bytes);
+}
+
+function parseCharset(contentType: string | null): string | null {
+  const match = contentType?.match(/;\s*charset\s*=\s*"?([^";]+)"?/i);
+  return match?.[1]?.trim() || null;
+}
+
+function formatBytes(bytes: number): string {
+  if (bytes === Infinity) return "unlimited";
+  if (bytes < 1024) return `${bytes} B`;
+  const units = ["KB", "MB", "GB"];
+  let value = bytes / 1024;
+  let unit = 0;
+  while (value >= 1024 && unit < units.length - 1) {
+    value /= 1024;
+    unit++;
+  }
+  return `${value.toFixed(1)} ${units[unit]}`;
+}
+
+/**
+ * Decode a chunk that arrived as base64.
+ *
+ * The desktop transport is a WebSocket carrying JSON text frames, so bytes
+ * have to be spelled out. A host that can pass an ArrayBuffer along skips this.
+ */
+export function decodeBase64Chunk(data: string): Uint8Array {
+  if (typeof Buffer !== "undefined") {
+    const buf = Buffer.from(data, "base64");
+    return new Uint8Array(buf.buffer, buf.byteOffset, buf.byteLength);
+  }
+
+  const binary = atob(data);
+  const bytes = new Uint8Array(binary.length);
+  for (let i = 0; i < binary.length; i++) {
+    bytes[i] = binary.charCodeAt(i);
+  }
+  return bytes;
+}

packages/plugin-runtime/tests/responseBody.test.ts

@@ -0,0 +1,203 @@
+import { describe, expect, test } from "vite-plus/test";
+import { createResponseBody, decodeBase64Chunk } from "../src/responseBody";
+
+/** A finished body, in a store that records every window it was asked for. */
+function fakeBody(bytes: Uint8Array, contentType: string | null) {
+  const reads: Array<[number, number]> = [];
+  const body = createResponseBody(
+    { responseId: "rs_test", contentLength: bytes.byteLength, contentType, complete: true },
+    async (offset, length) => {
+      reads.push([offset, length]);
+      return bytes.slice(offset, offset + length);
+    },
+  );
+  return { body, reads };
+}
+
+/**
+ * A body still arriving: it grows by one script step every time the reader
+ * asks the host where it has got to, and completes on the last step.
+ */
+function streamingBody(steps: string[], contentType = "text/plain") {
+  let stored = new Uint8Array();
+  let step = 0;
+  let refreshes = 0;
+  const advance = () => {
+    if (step < steps.length) {
+      const next = utf8(steps[step]!);
+      const grown = new Uint8Array(stored.byteLength + next.byteLength);
+      grown.set(stored);
+      grown.set(next, stored.byteLength);
+      stored = grown;
+      step++;
+    }
+    return { contentLength: stored.byteLength, complete: step >= steps.length };
+  };
+  const body = createResponseBody(
+    { responseId: "rs_live", contentLength: 0, contentType, complete: false },
+    async (offset, length) => stored.slice(offset, offset + length),
+    {
+      refresh: async () => {
+        refreshes++;
+        return advance();
+      },
+      pollIntervalMs: 1,
+    },
+  );
+  return { body, refreshCount: () => refreshes };
+}
+
+function utf8(text: string) {
+  return new TextEncoder().encode(text);
+}
+
+describe("response body", () => {
+  test("pulls a body in chunks and reassembles it", async () => {
+    const { body, reads } = fakeBody(utf8("abcdefghij"), "text/plain");
+
+    expect(await body.text({ chunkSize: 4 })).toEqual("abcdefghij");
+    expect(reads).toEqual([
+      [0, 4],
+      [4, 4],
+      [8, 2],
+    ]);
+  });
+
+  test("can be read more than once, unlike fetch", async () => {
+    const { body } = fakeBody(utf8('{"a":1}'), "application/json");
+
+    expect(await body.text()).toEqual('{"a":1}');
+    expect(await body.json()).toEqual({ a: 1 });
+    expect(new Uint8Array(await body.arrayBuffer())).toEqual(utf8('{"a":1}'));
+  });
+
+  test("decodes using the charset the response declared", async () => {
+    // "café naïve" as Latin-1, which is mojibake if read as UTF-8.
+    const latin1 = new Uint8Array([0x63, 0x61, 0x66, 0xe9, 0x20, 0x6e, 0x61, 0xef, 0x76, 0x65]);
+
+    const declared = fakeBody(latin1, "text/plain; charset=iso-8859-1");
+    expect(await declared.body.text()).toEqual("café naïve");
+
+    const undeclared = fakeBody(latin1, "text/plain");
+    expect(await undeclared.body.text()).not.toEqual("café naïve");
+  });
+
+  test("falls back to UTF-8 for a charset the runtime doesn't know", async () => {
+    const { body } = fakeBody(utf8("hello"), "text/plain; charset=not-a-real-charset");
+    expect(await body.text()).toEqual("hello");
+  });
+
+  test("drops a UTF-8 BOM", async () => {
+    const withBom = new Uint8Array([0xef, 0xbb, 0xbf, ...utf8('{"a":1}')]);
+    const { body } = fakeBody(withBom, "application/json");
+
+    expect(await body.text()).toEqual('{"a":1}');
+    expect(await body.json()).toEqual({ a: 1 });
+  });
+
+  test("refuses to buffer past maxBytes, and says what to do instead", async () => {
+    const { body } = fakeBody(utf8("x".repeat(100)), "text/plain");
+
+    await expect(body.text({ maxBytes: 50 })).rejects.toThrow(/chunks\(\)/);
+    await expect(body.json({ maxBytes: 50 })).rejects.toThrow(/over the/);
+    await expect(body.arrayBuffer({ maxBytes: 50 })).rejects.toThrow(/arrayBuffer\(\)/);
+
+    // The ceiling is the caller's to raise.
+    expect(await body.text({ maxBytes: 100 })).toHaveLength(100);
+  });
+
+  test("streams past maxBytes through chunks()", async () => {
+    const { body } = fakeBody(utf8("x".repeat(100)), "text/plain");
+
+    let total = 0;
+    for await (const chunk of body.chunks({ chunkSize: 10 })) {
+      total += chunk.byteLength;
+    }
+    expect(total).toEqual(100);
+  });
+
+  test("stops early when the host runs out of bytes sooner than it claimed", async () => {
+    // contentLength says 100; the store only ever hands back 10.
+    const body = createResponseBody(
+      { responseId: "rs_test", contentLength: 100, contentType: "text/plain", complete: true },
+      async (offset) => (offset === 0 ? utf8("0123456789") : new Uint8Array()),
+    );
+
+    expect(await body.text()).toEqual("0123456789");
+  });
+
+  test("a response with no body reads as empty", async () => {
+    const { body, reads } = fakeBody(new Uint8Array(), "application/json");
+
+    expect(body.contentLength).toEqual(0);
+    expect(await body.text()).toEqual("");
+    expect(reads).toEqual([]);
+  });
+
+  test("keeps binary bytes intact", async () => {
+    const png = new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0x00, 0xff]);
+    const { body } = fakeBody(png, "image/png");
+
+    expect(new Uint8Array(await body.arrayBuffer({ chunkSize: 3 }))).toEqual(png);
+  });
+});
+
+describe("a response still arriving", () => {
+  test("chunks() follows it until it finishes", async () => {
+    const { body, refreshCount } = streamingBody(["data: 1\n", "data: 2\n", "data: 3\n"]);
+    expect(body.complete).toBe(false);
+
+    const seen: string[] = [];
+    for await (const chunk of body.chunks()) {
+      seen.push(new TextDecoder().decode(chunk));
+    }
+
+    expect(seen.join("")).toEqual("data: 1\ndata: 2\ndata: 3\n");
+    // Asked once per catch-up, and stopped as soon as the host said it was done.
+    expect(refreshCount()).toEqual(3);
+  });
+
+  test("text() waits for the rest rather than returning a prefix", async () => {
+    const { body } = streamingBody(['{"token":', '"abc"}']);
+    expect(await body.json()).toEqual({ token: "abc" });
+  });
+
+  test("keeps waiting through a stretch with nothing new", async () => {
+    // Two refreshes report no growth before the body finally moves.
+    const { body } = streamingBody(["", "", "late"]);
+    expect(await body.text()).toEqual("late");
+  });
+
+  test("still refuses to buffer past maxBytes as it streams", async () => {
+    const { body } = streamingBody(["x".repeat(40), "x".repeat(40), "x".repeat(40)]);
+    await expect(body.text({ maxBytes: 100 })).rejects.toThrow(/chunks\(\)/);
+  });
+
+  test("a finished body never asks the host again", async () => {
+    let refreshes = 0;
+    const body = createResponseBody(
+      { responseId: "rs_done", contentLength: 5, contentType: null, complete: true },
+      async (offset, length) => utf8("hello").slice(offset, offset + length),
+      {
+        refresh: async () => {
+          refreshes++;
+          return { contentLength: 5, complete: true };
+        },
+      },
+    );
+    expect(await body.text()).toEqual("hello");
+    expect(refreshes).toEqual(0);
+  });
+});
+
+describe("decodeBase64Chunk", () => {
+  test("round-trips arbitrary bytes", () => {
+    const bytes = new Uint8Array([0, 1, 127, 128, 254, 255]);
+    const base64 = Buffer.from(bytes).toString("base64");
+    expect(decodeBase64Chunk(base64)).toEqual(bytes);
+  });
+
+  test("decodes an empty chunk", () => {
+    expect(decodeBase64Chunk("")).toEqual(new Uint8Array());
+  });
+});