import { type WsDecodeDiagnostic, type WsPush, type WsPushChannel, type WsPushMessage, WebSocketResponse, type WsResponse as WsResponseMessage, WsResponse as WsResponseSchema,} from "@t3tools/contracts";import { Schema, SchemaIssue } from "effect"; type PushListener<C extends WsPushChannel> = (message: WsPushMessage<C>) => void; interface PendingRequest { resolve: (result: unknown) => void; reject: (error: Error) => void; timeout: ReturnType<typeof setTimeout>;} interface SubscribeOptions { readonly replayLatest?: boolean;} type TransportState = "connecting" | "open" | "reconnecting" | "closed" | "disposed"; const REQUEST_TIMEOUT_MS = 60_000;const RECONNECT_DELAYS_MS = [500, 1_000, 2_000, 4_000, 8_000];const decodeWsResponse = Schema.decodeUnknownSync(WsResponseSchema);const isWebSocketResponseEnvelope = Schema.is(WebSocketResponse); const isWsPushMessage = (value: WsResponseMessage): value is WsPush => "type" in value && value.type === "push"; interface WsRequestEnvelope { id: string; body: { _tag: string; [key: string]: unknown; };} function describeValue(value: unknown): string { if (value === null) return "null"; if (Array.isArray(value)) return "array"; return typeof value;} function parseJsonOffset(error: unknown): number | undefined { if (!(error instanceof Error)) { return undefined; } const match = /position\s+(\d+)/i.exec(error.message); if (!match) { return undefined; } const parsed = Number(match[1]); return Number.isFinite(parsed) && parsed >= 0 ? parsed : undefined;} function makeEnvelopeDiagnostic(raw: unknown, reason: string): WsDecodeDiagnostic { return { code: "invalid-envelope", reason, rawKind: describeValue(raw), expected: "WsResponse", actual: describeValue(raw), };} function decodeInboundMessage( raw: unknown,): { readonly ok: true; readonly message: WsResponseMessage } | { readonly ok: false; readonly diagnostic: WsDecodeDiagnostic } { if (typeof raw !== "string") { return { ok: false, diagnostic: { code: "invalid-envelope", reason: "Expected a text WebSocket frame.", rawKind: describeValue(raw), expected: "string", actual: describeValue(raw), }, }; } let parsed: unknown; try { parsed = JSON.parse(raw); } catch (error) { return { ok: false, diagnostic: { code: "invalid-json", reason: error instanceof Error ? error.message : "Failed to parse JSON.", rawKind: "string", expected: "valid JSON string", actual: raw, ...(parseJsonOffset(error) !== undefined ? { jsonOffset: parseJsonOffset(error) } : {}), }, }; } try { return { ok: true, message: decodeWsResponse(parsed) }; } catch (error) { return { ok: false, diagnostic: makeEnvelopeDiagnostic( parsed, typeof error === "object" && error !== null && "issue" in error ? SchemaIssue.makeFormatterDefault()((error as Schema.SchemaError).issue) : error instanceof Error ? error.message : "Failed to decode WebSocket envelope.", ), }; }} function asError(value: unknown, fallback: string): Error { if (value instanceof Error) { return value; } return new Error(fallback);} export class WsTransport { private ws: WebSocket | null = null; private nextId = 1; private readonly pending = new Map<string, PendingRequest>(); private readonly listeners = new Map<string, Set<(message: WsPush) => void>>(); private readonly latestPushByChannel = new Map<string, WsPush>(); private readonly outboundQueue: string[] = []; private reconnectAttempt = 0; private reconnectTimer: ReturnType<typeof setTimeout> | null = null; private disposed = false; private state: TransportState = "connecting"; private readonly url: string; constructor(url?: string) { const bridgeUrl = window.desktopBridge?.getWsUrl(); const envUrl = import.meta.env.VITE_WS_URL as string | undefined; this.url = url ?? (bridgeUrl && bridgeUrl.length > 0 ? bridgeUrl : envUrl && envUrl.length > 0 ? envUrl : `${window.location.protocol === "https:" ? "wss" : "ws"}://${window.location.hostname}:${window.location.port}`); this.connect(); } async request<T = unknown>(method: string, params?: unknown): Promise<T> { if (typeof method !== "string" || method.length === 0) { throw new Error("Request method is required"); } const id = String(this.nextId++); const body = params != null ? { ...params, _tag: method } : { _tag: method }; const message: WsRequestEnvelope = { id, body }; const encoded = JSON.stringify(message); return new Promise<T>((resolve, reject) => { const timeout = setTimeout(() => { this.pending.delete(id); reject(new Error(`Request timed out: ${method}`)); }, REQUEST_TIMEOUT_MS); this.pending.set(id, { resolve: resolve as (result: unknown) => void, reject, timeout, }); this.send(encoded); }); } subscribe<C extends WsPushChannel>( channel: C, listener: PushListener<C>, options?: SubscribeOptions, ): () => void { let channelListeners = this.listeners.get(channel); if (!channelListeners) { channelListeners = new Set<(message: WsPush) => void>(); this.listeners.set(channel, channelListeners); } const wrappedListener = (message: WsPush) => { listener(message as WsPushMessage<C>); }; channelListeners.add(wrappedListener); if (options?.replayLatest) { const latest = this.latestPushByChannel.get(channel); if (latest) { wrappedListener(latest); } } return () => { channelListeners?.delete(wrappedListener); if (channelListeners?.size === 0) { this.listeners.delete(channel); } }; } getLatestPush<C extends WsPushChannel>(channel: C): WsPushMessage<C> | null { const latest = this.latestPushByChannel.get(channel); return latest ? (latest as WsPushMessage<C>) : null; } getState(): TransportState { return this.state; } dispose() { this.disposed = true; this.state = "disposed"; if (this.reconnectTimer !== null) { clearTimeout(this.reconnectTimer); this.reconnectTimer = null; } for (const pending of this.pending.values()) { clearTimeout(pending.timeout); pending.reject(new Error("Transport disposed")); } this.pending.clear(); this.outboundQueue.length = 0; this.ws?.close(); this.ws = null; } private connect() { if (this.disposed) { return; } this.state = this.reconnectAttempt > 0 ? "reconnecting" : "connecting"; const ws = new WebSocket(this.url); ws.addEventListener("open", () => { this.ws = ws; this.state = "open"; this.reconnectAttempt = 0; this.flushQueue(); }); ws.addEventListener("message", (event) => { this.handleMessage(event.data); }); ws.addEventListener("close", () => { if (this.ws === ws) { this.ws = null; } if (this.disposed) { this.state = "disposed"; return; } this.state = "closed"; this.scheduleReconnect(); }); ws.addEventListener("error", () => { // close will follow }); } private handleMessage(raw: unknown) { const decoded = decodeInboundMessage(raw); if (!decoded.ok) { console.warn("Dropped inbound WebSocket envelope", decoded.diagnostic); return; } const message = decoded.message; if (isWsPushMessage(message)) { this.latestPushByChannel.set(message.channel, message); const channelListeners = this.listeners.get(message.channel); if (channelListeners) { for (const listener of channelListeners) { try { listener(message); } catch { // Swallow listener errors } } } return; } if (!isWebSocketResponseEnvelope(message)) { return; } const pending = this.pending.get(message.id); if (!pending) { return; } clearTimeout(pending.timeout); this.pending.delete(message.id); if (message.error) { pending.reject(new Error(message.error.message)); return; } pending.resolve(message.result); } private send(encodedMessage: string) { if (this.disposed) { return; } this.outboundQueue.push(encodedMessage); try { this.flushQueue(); } catch { // Swallow: flushQueue has queued the message for retry on reconnect } } private flushQueue() { if (this.ws?.readyState !== WebSocket.OPEN) { return; } while (this.outboundQueue.length > 0) { const message = this.outboundQueue.shift(); if (!message) { continue; } try { this.ws.send(message); } catch (error) { this.outboundQueue.unshift(message); throw asError(error, "Failed to send WebSocket request."); } } } private scheduleReconnect() { if (this.disposed || this.reconnectTimer !== null) { return; } const delay = RECONNECT_DELAYS_MS[Math.min(this.reconnectAttempt, RECONNECT_DELAYS_MS.length - 1)] ?? RECONNECT_DELAYS_MS[0]!; this.reconnectAttempt += 1; this.reconnectTimer = setTimeout(() => { this.reconnectTimer = null; this.connect(); }, delay); }}
import {
type WsDecodeDiagnostic,
type WsPush,
type WsPushChannel,
type WsPushMessage,
WebSocketResponse,
type WsResponse as WsResponseMessage,
WsResponse as WsResponseSchema,
} from "@t3tools/contracts";
import { Schema, SchemaIssue } from "effect";
type PushListener<C extends WsPushChannel> = (message: WsPushMessage<C>) => void;
interface PendingRequest {
resolve: (result: unknown) => void;
reject: (error: Error) => void;
timeout: ReturnType<typeof setTimeout>;
}
interface SubscribeOptions {
readonly replayLatest?: boolean;
}
type TransportState = "connecting" | "open" | "reconnecting" | "closed" | "disposed";
const REQUEST_TIMEOUT_MS = 60_000;
const RECONNECT_DELAYS_MS = [500, 1_000, 2_000, 4_000, 8_000];
const decodeWsResponse = Schema.decodeUnknownSync(WsResponseSchema);
const isWebSocketResponseEnvelope = Schema.is(WebSocketResponse);
const isWsPushMessage = (value: WsResponseMessage): value is WsPush =>
"type" in value && value.type === "push";
interface WsRequestEnvelope {
id: string;
body: {
_tag: string;
[key: string]: unknown;
};
}
function describeValue(value: unknown): string {
if (value === null) return "null";
if (Array.isArray(value)) return "array";
return typeof value;
}
function parseJsonOffset(error: unknown): number | undefined {
if (!(error instanceof Error)) {
return undefined;
}
const match = /position\s+(\d+)/i.exec(error.message);
if (!match) {
return undefined;
}
const parsed = Number(match[1]);
return Number.isFinite(parsed) && parsed >= 0 ? parsed : undefined;
}
function makeEnvelopeDiagnostic(raw: unknown, reason: string): WsDecodeDiagnostic {
return {
code: "invalid-envelope",
reason,
rawKind: describeValue(raw),
expected: "WsResponse",
actual: describeValue(raw),
};
}
function decodeInboundMessage(
raw: unknown,
): { readonly ok: true; readonly message: WsResponseMessage } | { readonly ok: false; readonly diagnostic: WsDecodeDiagnostic } {
if (typeof raw !== "string") {
return {
ok: false,
diagnostic: {
code: "invalid-envelope",
reason: "Expected a text WebSocket frame.",
rawKind: describeValue(raw),
expected: "string",
actual: describeValue(raw),
},
};
}
let parsed: unknown;
try {
parsed = JSON.parse(raw);
} catch (error) {
return {
ok: false,
diagnostic: {
code: "invalid-json",
reason: error instanceof Error ? error.message : "Failed to parse JSON.",
rawKind: "string",
expected: "valid JSON string",
actual: raw,
...(parseJsonOffset(error) !== undefined ? { jsonOffset: parseJsonOffset(error) } : {}),
},
};
}
try {
return { ok: true, message: decodeWsResponse(parsed) };
} catch (error) {
return {
ok: false,
diagnostic: makeEnvelopeDiagnostic(
parsed,
typeof error === "object" && error !== null && "issue" in error
? SchemaIssue.makeFormatterDefault()((error as Schema.SchemaError).issue)
: error instanceof Error
? error.message
: "Failed to decode WebSocket envelope.",
),
};
}
}
function asError(value: unknown, fallback: string): Error {
if (value instanceof Error) {
return value;
}
return new Error(fallback);
}
export class WsTransport {
private ws: WebSocket | null = null;
private nextId = 1;
private readonly pending = new Map<string, PendingRequest>();
private readonly listeners = new Map<string, Set<(message: WsPush) => void>>();
private readonly latestPushByChannel = new Map<string, WsPush>();
private readonly outboundQueue: string[] = [];
private reconnectAttempt = 0;
private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
private disposed = false;
private state: TransportState = "connecting";
private readonly url: string;
constructor(url?: string) {
const bridgeUrl = window.desktopBridge?.getWsUrl();
const envUrl = import.meta.env.VITE_WS_URL as string | undefined;
this.url =
url ??
(bridgeUrl && bridgeUrl.length > 0
? bridgeUrl
: envUrl && envUrl.length > 0
? envUrl
: `${window.location.protocol === "https:" ? "wss" : "ws"}://${window.location.hostname}:${window.location.port}`);
this.connect();
}
async request<T = unknown>(method: string, params?: unknown): Promise<T> {
if (typeof method !== "string" || method.length === 0) {
throw new Error("Request method is required");
}
const id = String(this.nextId++);
const body = params != null ? { ...params, _tag: method } : { _tag: method };
const message: WsRequestEnvelope = { id, body };
const encoded = JSON.stringify(message);
return new Promise<T>((resolve, reject) => {
const timeout = setTimeout(() => {
this.pending.delete(id);
reject(new Error(`Request timed out: ${method}`));
}, REQUEST_TIMEOUT_MS);
this.pending.set(id, {
resolve: resolve as (result: unknown) => void,
reject,
timeout,
});
this.send(encoded);
});
}
subscribe<C extends WsPushChannel>(
channel: C,
listener: PushListener<C>,
options?: SubscribeOptions,
): () => void {
let channelListeners = this.listeners.get(channel);
if (!channelListeners) {
channelListeners = new Set<(message: WsPush) => void>();
this.listeners.set(channel, channelListeners);
}
const wrappedListener = (message: WsPush) => {
listener(message as WsPushMessage<C>);
};
channelListeners.add(wrappedListener);
if (options?.replayLatest) {
const latest = this.latestPushByChannel.get(channel);
if (latest) {
wrappedListener(latest);
}
}
return () => {
channelListeners?.delete(wrappedListener);
if (channelListeners?.size === 0) {
this.listeners.delete(channel);
}
};
}
getLatestPush<C extends WsPushChannel>(channel: C): WsPushMessage<C> | null {
const latest = this.latestPushByChannel.get(channel);
return latest ? (latest as WsPushMessage<C>) : null;
}
getState(): TransportState {
return this.state;
}
dispose() {
this.disposed = true;
this.state = "disposed";
if (this.reconnectTimer !== null) {
clearTimeout(this.reconnectTimer);
this.reconnectTimer = null;
}
for (const pending of this.pending.values()) {
clearTimeout(pending.timeout);
pending.reject(new Error("Transport disposed"));
}
this.pending.clear();
this.outboundQueue.length = 0;
this.ws?.close();
this.ws = null;
}
private connect() {
if (this.disposed) {
return;
}
this.state = this.reconnectAttempt > 0 ? "reconnecting" : "connecting";
const ws = new WebSocket(this.url);
ws.addEventListener("open", () => {
this.ws = ws;
this.state = "open";
this.reconnectAttempt = 0;
this.flushQueue();
});
ws.addEventListener("message", (event) => {
this.handleMessage(event.data);
});
ws.addEventListener("close", () => {
if (this.ws === ws) {
this.ws = null;
}
if (this.disposed) {
this.state = "disposed";
return;
}
this.state = "closed";
this.scheduleReconnect();
});
ws.addEventListener("error", () => {
// close will follow
});
}
private handleMessage(raw: unknown) {
const decoded = decodeInboundMessage(raw);
if (!decoded.ok) {
console.warn("Dropped inbound WebSocket envelope", decoded.diagnostic);
return;
}
const message = decoded.message;
if (isWsPushMessage(message)) {
this.latestPushByChannel.set(message.channel, message);
const channelListeners = this.listeners.get(message.channel);
if (channelListeners) {
for (const listener of channelListeners) {
try {
listener(message);
} catch {
// Swallow listener errors
}
}
}
return;
}
if (!isWebSocketResponseEnvelope(message)) {
return;
}
const pending = this.pending.get(message.id);
if (!pending) {
return;
}
clearTimeout(pending.timeout);
this.pending.delete(message.id);
if (message.error) {
pending.reject(new Error(message.error.message));
return;
}
pending.resolve(message.result);
}
private send(encodedMessage: string) {
if (this.disposed) {
return;
}
this.outboundQueue.push(encodedMessage);
try {
this.flushQueue();
} catch {
// Swallow: flushQueue has queued the message for retry on reconnect
}
}
private flushQueue() {
if (this.ws?.readyState !== WebSocket.OPEN) {
return;
}
while (this.outboundQueue.length > 0) {
const message = this.outboundQueue.shift();
if (!message) {
continue;
}
try {
this.ws.send(message);
} catch (error) {
this.outboundQueue.unshift(message);
throw asError(error, "Failed to send WebSocket request.");
}
}
}
private scheduleReconnect() {
if (this.disposed || this.reconnectTimer !== null) {
return;
}
const delay =
RECONNECT_DELAYS_MS[Math.min(this.reconnectAttempt, RECONNECT_DELAYS_MS.length - 1)] ??
RECONNECT_DELAYS_MS[0]!;
this.reconnectAttempt += 1;
this.reconnectTimer = setTimeout(() => {
this.reconnectTimer = null;
this.connect();
}, delay);
}
}