diff --git a/packages/plugin/src/rpc.ts b/packages/plugin/src/rpc.ts index ab08e446bd..856b11de61 100644 --- a/packages/plugin/src/rpc.ts +++ b/packages/plugin/src/rpc.ts @@ -109,14 +109,15 @@ export function subscribePluginRpc) => 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); diff --git a/packages/server/src/server/plugins/subscription-handlers.test.ts b/packages/server/src/server/plugins/subscription-handlers.test.ts new file mode 100644 index 0000000000..ed8b881a8f --- /dev/null +++ b/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[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) => { + 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([{}]); +}); diff --git a/packages/server/src/server/plugins/subscription-handlers.ts b/packages/server/src/server/plugins/subscription-handlers.ts index ec6e7baaec..1d2c4fefdb 100644 --- a/packages/server/src/server/plugins/subscription-handlers.ts +++ b/packages/server/src/server/plugins/subscription-handlers.ts @@ -77,12 +77,13 @@ export class PluginSubscriptionHandlers { }); const push = async (event: unknown): Promise => { 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;