fix(plugins): apply subscription schema transforms once

Both ends parsed each subscription value, and the sender sent its parse
output, so the receiver parsed a schema's output as fresh input. An event
schema like z.string().transform((s) => s.length) closed the stream as
invalid. The sender now validates and sends the value as given; the
receiver's parse produces what the handler or callback sees.

packages/plugin/src/rpc.ts

@@ -109,14 +109,15 @@ export function subscribePluginRpc<InputSchema extends ZodType, EventSchema exte
   onEvent: (event: ZodOutput<EventSchema>) => void,
   options: PluginSubscribeOptions = {},
 ): PluginSubscriptionHandle {
-  const parsedInput = contract.input.parse(input);
+  // The plugin parses the input into the schema's output, so it receives the input as given.
+  contract.input.parse(input);
   let closed = false;
   function close(error?: Error) {
     if (closed) return;
     closed = true;
     options.onClosed?.(error ? { error } : {});
   }
-  const handle = subscribe(contract.name, parsedInput, {
+  const handle = subscribe(contract.name, input, {
     event(event) {
       if (closed) return;
       const parsed = contract.event.safeParse(event);

packages/server/src/server/plugins/subscription-handlers.test.ts

@@ -0,0 +1,55 @@
+import { defineSubscription } from "@getpaseo/plugin";
+import { subscribePluginRpc } from "@getpaseo/plugin/client/host";
+import type { PluginSubscriptionContext } from "@getpaseo/plugin/server";
+import type { PaseoApi } from "@getpaseo/client";
+import { expect, test } from "vitest";
+import { z } from "zod";
+import type { PluginProcessMessage } from "./plugin-process-protocol.js";
+import { PluginSubscriptionHandlers } from "./subscription-handlers.js";
+
+const lengths = defineSubscription({
+  name: "lengths",
+  input: z.object({ times: z.string().transform(Number) }),
+  event: z.string().transform((text) => text.length),
+});
+
+/** The plugin process's handlers wired to the client helper through the channel between them. */
+function connect(handler: (input: unknown, context: PluginSubscriptionContext) => unknown) {
+  const handlers = new PluginSubscriptionHandlers({
+    send: (message) => deliver(message),
+    sendAndWait: async (message) => deliver(message),
+  });
+  handlers.register(lengths, handler);
+  let observer: Parameters<Parameters<typeof subscribePluginRpc>[1]>[2] | null = null;
+  function deliver(message: PluginProcessMessage) {
+    if (message.type === "subscription.event") observer?.event(message.event);
+    if (message.type === "subscription.closed") observer?.closed();
+  }
+  return (method: string, input: unknown, next: NonNullable<typeof observer>) => {
+    observer = next;
+    handlers.start(
+      { type: "subscribe", requestId: "r", subscriptionId: "s", method, input },
+      {} as PaseoApi,
+    );
+    return Object.assign(() => handlers.release("s"), {
+      ready: Promise.resolve(),
+      release: async () => handlers.release("s"),
+    });
+  };
+}
+
+test("an event schema that transforms is applied once, on the receiving side", async () => {
+  const handled: unknown[] = [];
+  const received: number[] = [];
+  const closed: unknown[] = [];
+  const subscribe = connect(async (input, { push }) => {
+    handled.push(input);
+    await push("abc");
+  });
+  subscribePluginRpc(lengths, subscribe, { times: "2" }, (length) => received.push(length), {
+    onClosed: (result) => closed.push(result),
+  });
+  await expect.poll(() => received).toEqual([3]);
+  expect(handled).toEqual([{ times: 2 }]);
+  expect(closed).toEqual([{}]);
+});

packages/server/src/server/plugins/subscription-handlers.ts

@@ -77,12 +77,13 @@ export class PluginSubscriptionHandlers {
     });
     const push = async (event: unknown): Promise<void> => {
       if (signal.aborted) return;
-      const value: unknown = registered.contract.event.parse(event);
-      assertPluginWireValue(value, "event");
+      // The receiver parses into the schema's output; sending that output would parse it twice.
+      registered.contract.event.parse(event);
+      assertPluginWireValue(event, "event");
       // Events pushed while the handler runs follow the acknowledgement.
       await acknowledged;
       if (signal.aborted) return;
-      await this.channel.sendAndWait({ type: "subscription.event", subscriptionId, event: value });
+      await this.channel.sendAndWait({ type: "subscription.event", subscriptionId, event });
     };
     this.active.set(subscriptionId, { controller });
     let result: unknown;