import { WsRpcGroup } from "@t3tools/contracts";import { Duration, Effect, Layer, Schedule } from "effect";import { RpcClient, RpcSerialization } from "effect/unstable/rpc";import * as Socket from "effect/unstable/socket/Socket"; import { DEFAULT_RECONNECT_BACKOFF, getReconnectDelayMs, type ReconnectBackoffConfig,} from "./reconnectBackoff"; export interface WsProtocolLifecycleHandlers { readonly onAttempt?: (socketUrl: string) => void; readonly onOpen?: () => void; readonly onError?: (message: string) => void; readonly onClose?: (details: { readonly code: number; readonly reason: string }) => void;} export interface WsRpcProtocolOptions { /** Backoff configuration for reconnect retries. */ readonly backoff?: ReconnectBackoffConfig;} export const makeWsRpcProtocolClient = RpcClient.make(WsRpcGroup);type RpcClientFactory = typeof makeWsRpcProtocolClient;export type WsRpcProtocolClient = RpcClientFactory extends Effect.Effect<infer Client, any, any> ? Client : never;export type WsRpcProtocolSocketUrlProvider = string | (() => Promise<string>); function formatSocketErrorMessage(error: unknown): string { if (error instanceof Error && error.message.trim().length > 0) { return error.message; } return String(error);} function resolveWsRpcSocketUrl(rawUrl: string): string { const resolved = new URL(rawUrl); if (resolved.protocol !== "ws:" && resolved.protocol !== "wss:") { throw new Error(`Unsupported websocket transport URL protocol: ${resolved.protocol}`); } resolved.pathname = "/ws"; return resolved.toString();} function defaultLifecycleHandlers(): Required<WsProtocolLifecycleHandlers> { return { onAttempt: () => undefined, onOpen: () => undefined, onError: () => undefined, onClose: () => undefined, };} export function createWsRpcProtocolLayer( url: WsRpcProtocolSocketUrlProvider, handlers?: WsProtocolLifecycleHandlers, options?: WsRpcProtocolOptions,) { const lifecycle = { ...defaultLifecycleHandlers(), ...handlers, }; const backoff = options?.backoff ?? DEFAULT_RECONNECT_BACKOFF; const resolvedUrl = typeof url === "function" ? Effect.promise(() => url()).pipe( Effect.map((rawUrl) => resolveWsRpcSocketUrl(rawUrl)), Effect.tapError((error) => Effect.sync(() => { lifecycle.onError(formatSocketErrorMessage(error)); }), ), Effect.orDie, ) : resolveWsRpcSocketUrl(url); const trackingWebSocketConstructorLayer = Layer.succeed( Socket.WebSocketConstructor, (socketUrl, protocols) => { lifecycle.onAttempt(socketUrl); const socket = new globalThis.WebSocket(socketUrl, protocols); socket.addEventListener( "open", () => { lifecycle.onOpen(); }, { once: true }, ); socket.addEventListener( "error", () => { lifecycle.onError("Unable to connect to the T3 server WebSocket."); }, { once: true }, ); socket.addEventListener( "close", (event) => { lifecycle.onClose({ code: event.code, reason: event.reason, }); }, { once: true }, ); return socket; }, ); const socketLayer = Socket.layerWebSocket(resolvedUrl).pipe( Layer.provide(trackingWebSocketConstructorLayer), ); const baseSchedule = backoff.maxRetries === null ? Schedule.forever : Schedule.recurs(backoff.maxRetries); const retryPolicy = Schedule.addDelay(baseSchedule, (retryCount) => Effect.succeed(Duration.millis(getReconnectDelayMs(retryCount, backoff) ?? 0)), ); const protocolLayer = Layer.effect( RpcClient.Protocol, RpcClient.makeProtocolSocket({ retryPolicy, retryTransientErrors: true, }), ); return protocolLayer.pipe(Layer.provide(Layer.mergeAll(socketLayer, RpcSerialization.layerJson)));}
import { WsRpcGroup } from "@t3tools/contracts";
import { Duration, Effect, Layer, Schedule } from "effect";
import { RpcClient, RpcSerialization } from "effect/unstable/rpc";
import * as Socket from "effect/unstable/socket/Socket";
import {
DEFAULT_RECONNECT_BACKOFF,
getReconnectDelayMs,
type ReconnectBackoffConfig,
} from "./reconnectBackoff";
export interface WsProtocolLifecycleHandlers {
readonly onAttempt?: (socketUrl: string) => void;
readonly onOpen?: () => void;
readonly onError?: (message: string) => void;
readonly onClose?: (details: { readonly code: number; readonly reason: string }) => void;
}
export interface WsRpcProtocolOptions {
/** Backoff configuration for reconnect retries. */
readonly backoff?: ReconnectBackoffConfig;
}
export const makeWsRpcProtocolClient = RpcClient.make(WsRpcGroup);
type RpcClientFactory = typeof makeWsRpcProtocolClient;
export type WsRpcProtocolClient =
RpcClientFactory extends Effect.Effect<infer Client, any, any> ? Client : never;
export type WsRpcProtocolSocketUrlProvider = string | (() => Promise<string>);
function formatSocketErrorMessage(error: unknown): string {
if (error instanceof Error && error.message.trim().length > 0) {
return error.message;
}
return String(error);
}
function resolveWsRpcSocketUrl(rawUrl: string): string {
const resolved = new URL(rawUrl);
if (resolved.protocol !== "ws:" && resolved.protocol !== "wss:") {
throw new Error(`Unsupported websocket transport URL protocol: ${resolved.protocol}`);
}
resolved.pathname = "/ws";
return resolved.toString();
}
function defaultLifecycleHandlers(): Required<WsProtocolLifecycleHandlers> {
return {
onAttempt: () => undefined,
onOpen: () => undefined,
onError: () => undefined,
onClose: () => undefined,
};
}
export function createWsRpcProtocolLayer(
url: WsRpcProtocolSocketUrlProvider,
handlers?: WsProtocolLifecycleHandlers,
options?: WsRpcProtocolOptions,
) {
const lifecycle = {
...defaultLifecycleHandlers(),
...handlers,
};
const backoff = options?.backoff ?? DEFAULT_RECONNECT_BACKOFF;
const resolvedUrl =
typeof url === "function"
? Effect.promise(() => url()).pipe(
Effect.map((rawUrl) => resolveWsRpcSocketUrl(rawUrl)),
Effect.tapError((error) =>
Effect.sync(() => {
lifecycle.onError(formatSocketErrorMessage(error));
}),
),
Effect.orDie,
)
: resolveWsRpcSocketUrl(url);
const trackingWebSocketConstructorLayer = Layer.succeed(
Socket.WebSocketConstructor,
(socketUrl, protocols) => {
lifecycle.onAttempt(socketUrl);
const socket = new globalThis.WebSocket(socketUrl, protocols);
socket.addEventListener(
"open",
() => {
lifecycle.onOpen();
},
{ once: true },
);
socket.addEventListener(
"error",
() => {
lifecycle.onError("Unable to connect to the T3 server WebSocket.");
},
{ once: true },
);
socket.addEventListener(
"close",
(event) => {
lifecycle.onClose({
code: event.code,
reason: event.reason,
});
},
{ once: true },
);
return socket;
},
);
const socketLayer = Socket.layerWebSocket(resolvedUrl).pipe(
Layer.provide(trackingWebSocketConstructorLayer),
);
const baseSchedule =
backoff.maxRetries === null ? Schedule.forever : Schedule.recurs(backoff.maxRetries);
const retryPolicy = Schedule.addDelay(baseSchedule, (retryCount) =>
Effect.succeed(Duration.millis(getReconnectDelayMs(retryCount, backoff) ?? 0)),
);
const protocolLayer = Layer.effect(
RpcClient.Protocol,
RpcClient.makeProtocolSocket({
retryPolicy,
retryTransientErrors: true,
}),
);
return protocolLayer.pipe(Layer.provide(Layer.mergeAll(socketLayer, RpcSerialization.layerJson)));
}