import { Effect, Exit, PubSub, Scope, Stream } from "effect";import { WS_METHODS, WsRpcGroup } from "@t3tools/contracts";import { RpcSerialization, RpcServer } from "effect/unstable/rpc"; type RpcServerInstance = RpcServer.RpcServer<any>; type BrowserWsClient = { send: (data: string) => void;}; export type NormalizedWsRpcRequestBody = { _tag: string; [key: string]: unknown;}; type UnaryResolverResult = unknown | Promise<unknown>; interface BrowserWsRpcHarnessOptions { readonly resolveUnary?: (request: NormalizedWsRpcRequestBody) => UnaryResolverResult; readonly getInitialStreamValues?: ( request: NormalizedWsRpcRequestBody, ) => ReadonlyArray<unknown> | undefined;} const STREAM_METHODS = new Set<string>([ WS_METHODS.gitRunStackedAction, WS_METHODS.subscribeOrchestrationDomainEvents, WS_METHODS.subscribeTerminalEvents, WS_METHODS.subscribeServerConfig, WS_METHODS.subscribeServerLifecycle,]); const ALL_RPC_METHODS = Array.from(WsRpcGroup.requests.keys()); function normalizeRequest(tag: string, payload: unknown): NormalizedWsRpcRequestBody { if (payload && typeof payload === "object" && !Array.isArray(payload)) { return { _tag: tag, ...(payload as Record<string, unknown>), }; } return { _tag: tag, payload };} function asEffect(result: UnaryResolverResult): Effect.Effect<unknown> { if (result instanceof Promise) { return Effect.promise(() => result); } return Effect.succeed(result);} export class BrowserWsRpcHarness { readonly requests: Array<NormalizedWsRpcRequestBody> = []; private readonly parser = RpcSerialization.json.makeUnsafe(); private client: BrowserWsClient | null = null; private scope: Scope.Closeable | null = null; private serverReady: Promise<RpcServerInstance> | null = null; private resolveUnary: NonNullable<BrowserWsRpcHarnessOptions["resolveUnary"]> = () => ({}); private getInitialStreamValues: NonNullable< BrowserWsRpcHarnessOptions["getInitialStreamValues"] > = () => []; private streamPubSubs = new Map<string, PubSub.PubSub<unknown>>(); async reset(options?: BrowserWsRpcHarnessOptions): Promise<void> { await this.disconnect(); this.requests.length = 0; this.resolveUnary = options?.resolveUnary ?? (() => ({})); this.getInitialStreamValues = options?.getInitialStreamValues ?? (() => []); this.initializeStreamPubSubs(); } connect(client: BrowserWsClient): void { if (this.scope) { void Effect.runPromise(Scope.close(this.scope, Exit.void)).catch(() => undefined); } if (this.streamPubSubs.size === 0) { this.initializeStreamPubSubs(); } this.client = client; this.scope = Effect.runSync(Scope.make()); this.serverReady = Effect.runPromise( Scope.provide(this.scope)( RpcServer.makeNoSerialization(WsRpcGroup, this.makeServerOptions()), ).pipe(Effect.provide(this.makeLayer())), ) as Promise<RpcServerInstance>; } async disconnect(): Promise<void> { if (this.scope) { await Effect.runPromise(Scope.close(this.scope, Exit.void)).catch(() => undefined); this.scope = null; } for (const pubsub of this.streamPubSubs.values()) { Effect.runSync(PubSub.shutdown(pubsub)); } this.streamPubSubs.clear(); this.serverReady = null; this.client = null; } private initializeStreamPubSubs(): void { this.streamPubSubs = new Map( Array.from(STREAM_METHODS, (method) => [method, Effect.runSync(PubSub.unbounded<unknown>())]), ); } async onMessage(rawData: string): Promise<void> { const server = await this.serverReady; if (!server) { throw new Error("RPC test server is not connected"); } const messages = this.parser.decode(rawData); for (const message of messages) { await Effect.runPromise(server.write(0, message as never)); } } emitStreamValue(method: string, value: unknown): void { const pubsub = this.streamPubSubs.get(method); if (!pubsub) { throw new Error(`No stream registered for ${method}`); } Effect.runSync(PubSub.publish(pubsub, value)); } private makeLayer() { const handlers: Record<string, (payload: unknown) => unknown> = {}; for (const method of ALL_RPC_METHODS) { handlers[method] = STREAM_METHODS.has(method) ? (payload) => this.handleStream(method, payload) : (payload) => this.handleUnary(method, payload); } return WsRpcGroup.toLayer(handlers as never); } private makeServerOptions() { return { onFromServer: (response: unknown) => Effect.sync(() => { if (!this.client) { return; } const encoded = this.parser.encode(response); if (typeof encoded === "string") { this.client.send(encoded); } }), }; } private handleUnary(method: string, payload: unknown) { const request = normalizeRequest(method, payload); this.requests.push(request); return asEffect(this.resolveUnary(request)); } private handleStream(method: string, payload: unknown) { const request = normalizeRequest(method, payload); this.requests.push(request); const pubsub = this.streamPubSubs.get(method); if (!pubsub) { throw new Error(`No stream registered for ${method}`); } return Stream.fromIterable(this.getInitialStreamValues(request) ?? []).pipe( Stream.concat(Stream.fromPubSub(pubsub)), ); }}
import { Effect, Exit, PubSub, Scope, Stream } from "effect";
import { WS_METHODS, WsRpcGroup } from "@t3tools/contracts";
import { RpcSerialization, RpcServer } from "effect/unstable/rpc";
type RpcServerInstance = RpcServer.RpcServer<any>;
type BrowserWsClient = {
send: (data: string) => void;
};
export type NormalizedWsRpcRequestBody = {
_tag: string;
[key: string]: unknown;
};
type UnaryResolverResult = unknown | Promise<unknown>;
interface BrowserWsRpcHarnessOptions {
readonly resolveUnary?: (request: NormalizedWsRpcRequestBody) => UnaryResolverResult;
readonly getInitialStreamValues?: (
request: NormalizedWsRpcRequestBody,
) => ReadonlyArray<unknown> | undefined;
}
const STREAM_METHODS = new Set<string>([
WS_METHODS.gitRunStackedAction,
WS_METHODS.subscribeOrchestrationDomainEvents,
WS_METHODS.subscribeTerminalEvents,
WS_METHODS.subscribeServerConfig,
WS_METHODS.subscribeServerLifecycle,
]);
const ALL_RPC_METHODS = Array.from(WsRpcGroup.requests.keys());
function normalizeRequest(tag: string, payload: unknown): NormalizedWsRpcRequestBody {
if (payload && typeof payload === "object" && !Array.isArray(payload)) {
return {
_tag: tag,
...(payload as Record<string, unknown>),
};
}
return { _tag: tag, payload };
}
function asEffect(result: UnaryResolverResult): Effect.Effect<unknown> {
if (result instanceof Promise) {
return Effect.promise(() => result);
}
return Effect.succeed(result);
}
export class BrowserWsRpcHarness {
readonly requests: Array<NormalizedWsRpcRequestBody> = [];
private readonly parser = RpcSerialization.json.makeUnsafe();
private client: BrowserWsClient | null = null;
private scope: Scope.Closeable | null = null;
private serverReady: Promise<RpcServerInstance> | null = null;
private resolveUnary: NonNullable<BrowserWsRpcHarnessOptions["resolveUnary"]> = () => ({});
private getInitialStreamValues: NonNullable<
BrowserWsRpcHarnessOptions["getInitialStreamValues"]
> = () => [];
private streamPubSubs = new Map<string, PubSub.PubSub<unknown>>();
async reset(options?: BrowserWsRpcHarnessOptions): Promise<void> {
await this.disconnect();
this.requests.length = 0;
this.resolveUnary = options?.resolveUnary ?? (() => ({}));
this.getInitialStreamValues = options?.getInitialStreamValues ?? (() => []);
this.initializeStreamPubSubs();
}
connect(client: BrowserWsClient): void {
if (this.scope) {
void Effect.runPromise(Scope.close(this.scope, Exit.void)).catch(() => undefined);
}
if (this.streamPubSubs.size === 0) {
this.initializeStreamPubSubs();
}
this.client = client;
this.scope = Effect.runSync(Scope.make());
this.serverReady = Effect.runPromise(
Scope.provide(this.scope)(
RpcServer.makeNoSerialization(WsRpcGroup, this.makeServerOptions()),
).pipe(Effect.provide(this.makeLayer())),
) as Promise<RpcServerInstance>;
}
async disconnect(): Promise<void> {
if (this.scope) {
await Effect.runPromise(Scope.close(this.scope, Exit.void)).catch(() => undefined);
this.scope = null;
}
for (const pubsub of this.streamPubSubs.values()) {
Effect.runSync(PubSub.shutdown(pubsub));
}
this.streamPubSubs.clear();
this.serverReady = null;
this.client = null;
}
private initializeStreamPubSubs(): void {
this.streamPubSubs = new Map(
Array.from(STREAM_METHODS, (method) => [method, Effect.runSync(PubSub.unbounded<unknown>())]),
);
}
async onMessage(rawData: string): Promise<void> {
const server = await this.serverReady;
if (!server) {
throw new Error("RPC test server is not connected");
}
const messages = this.parser.decode(rawData);
for (const message of messages) {
await Effect.runPromise(server.write(0, message as never));
}
}
emitStreamValue(method: string, value: unknown): void {
const pubsub = this.streamPubSubs.get(method);
if (!pubsub) {
throw new Error(`No stream registered for ${method}`);
}
Effect.runSync(PubSub.publish(pubsub, value));
}
private makeLayer() {
const handlers: Record<string, (payload: unknown) => unknown> = {};
for (const method of ALL_RPC_METHODS) {
handlers[method] = STREAM_METHODS.has(method)
? (payload) => this.handleStream(method, payload)
: (payload) => this.handleUnary(method, payload);
}
return WsRpcGroup.toLayer(handlers as never);
}
private makeServerOptions() {
return {
onFromServer: (response: unknown) =>
Effect.sync(() => {
if (!this.client) {
return;
}
const encoded = this.parser.encode(response);
if (typeof encoded === "string") {
this.client.send(encoded);
}
}),
};
}
private handleUnary(method: string, payload: unknown) {
const request = normalizeRequest(method, payload);
this.requests.push(request);
return asEffect(this.resolveUnary(request));
}
private handleStream(method: string, payload: unknown) {
const request = normalizeRequest(method, payload);
this.requests.push(request);
const pubsub = this.streamPubSubs.get(method);
if (!pubsub) {
throw new Error(`No stream registered for ${method}`);
}
return Stream.fromIterable(this.getInitialStreamValues(request) ?? []).pipe(
Stream.concat(Stream.fromPubSub(pubsub)),
);
}
}