import { DEFAULT_MAX_FRAME_LENGTH, encodeClientMessage, PROTOCOL_VERSION, ProtocolValidationError, type ServerHello, type ServerMessage, ServerMessageDecoder, } from "@earendil-works/pi-protocol"; import { DisconnectedError, ServerError, toDisconnectedError, toError } from "./errors.ts"; import { createPromiseResolvers, type PromiseResolvers } from "./promise.ts"; import type { ByteTransport, ByteTransportFactory, ByteTransportHandlers } from "./transport.ts"; import type { ConnectionState, ConnectionStateChange } from "./types.ts"; const MAX_UINT32 = 0xffff_ffff; type ActiveConnection = { id: number; decoder: ServerMessageDecoder; transport?: ByteTransport; }; type ConnectionLifecycle = | { state: "disconnected" } | ({ state: "connecting"; handshake: PromiseResolvers } & ActiveConnection) | ({ state: "connected"; transport: ByteTransport; handshake: PromiseResolvers | undefined; } & ActiveConnection); interface ConnectionOptions { transportFactory: ByteTransportFactory; serverId: string; maxFrameLength?: number; onHandshake(hello: ServerHello): void; onMessage(message: Exclude): void; onStateChange(change: ConnectionStateChange): void; } export class Connection { readonly #options: ConnectionOptions; readonly #maxFrameLength: number; #lifecycle: ConnectionLifecycle = { state: "disconnected" }; #sequence = 0; constructor(options: ConnectionOptions) { this.#options = options; this.#maxFrameLength = options.maxFrameLength ?? DEFAULT_MAX_FRAME_LENGTH; if ( !Number.isSafeInteger(this.#maxFrameLength) || this.#maxFrameLength <= 0 || this.#maxFrameLength > MAX_UINT32 ) { throw new TypeError(`Client maxFrameLength must be an integer between 1 and ${MAX_UINT32}`); } } get state(): ConnectionState { return this.#lifecycle.state; } get maxFrameLength(): number { return this.#maxFrameLength; } connect(): Promise { if (this.#lifecycle.state !== "disconnected") { return Promise.reject(new DisconnectedError(`Client is already ${this.#lifecycle.state}`)); } const id = ++this.#sequence; const handshake = createPromiseResolvers(); this.#lifecycle = { state: "connecting", id, decoder: new ServerMessageDecoder({ maxFrameLength: this.#maxFrameLength }), handshake, }; this.#options.onStateChange({ state: "connecting" }); const handlers = { onData: (chunk) => this.#handleData(id, chunk), onClose: () => { if (this.#isCurrent(id)) this.#handleClose(); }, onError: (error) => { if (this.#isCurrent(id)) this.#failAndClose(toDisconnectedError(error)); }, } satisfies ByteTransportHandlers; void this.#openTransport(id, handlers); return handshake.promise; } disconnect(reason: string | Error = "Client disconnected"): void { if (this.#lifecycle.state === "disconnected") return; this.#failAndClose(typeof reason === "string" ? new DisconnectedError(reason) : reason); } fail(error: Error): void { this.#failAndClose(error); } send(frame: Uint8Array): void { const lifecycle = this.#lifecycle; if (lifecycle.state !== "connected") throw new DisconnectedError(); let sending: Promise; try { sending = lifecycle.transport.send(frame); } catch (error) { this.#failAndClose(toDisconnectedError(error)); return; } void sending.catch((error: unknown) => { const current = this.#lifecycle; if (current.state !== "disconnected" && current.transport === lifecycle.transport) { this.#failAndClose(toDisconnectedError(error)); } }); } async #openTransport(id: number, handlers: ByteTransportHandlers): Promise { let transport: ByteTransport; try { transport = await this.#options.transportFactory(handlers); } catch (error) { if (this.#isCurrent(id)) this.#fail(toDisconnectedError(error)); return; } const lifecycle = this.#lifecycle; if (lifecycle.state !== "connecting" || lifecycle.id !== id) { transport.close(); return; } this.#lifecycle = { ...lifecycle, transport }; try { await transport.send( encodeClientMessage({ type: "hello", version: PROTOCOL_VERSION }, { maxFrameLength: this.#maxFrameLength }), ); } catch (error) { if (this.#isCurrent(id)) this.#failAndClose(toDisconnectedError(error)); } } #handleData(id: number, chunk: Uint8Array): void { const lifecycle = this.#lifecycle; if (lifecycle.state === "disconnected" || lifecycle.id !== id) return; if (lifecycle.state === "connecting" && !lifecycle.transport) { this.#failAndClose(new ProtocolValidationError("Received server data before the client hello was sent")); return; } let messages: ServerMessage[]; try { messages = lifecycle.decoder.push(chunk); } catch (error) { this.#failAndClose(toError(error)); return; } for (const message of messages) { if (this.#lifecycle.state === "disconnected") return; this.#handleMessage(message); } } #handleMessage(message: ServerMessage): void { const lifecycle = this.#lifecycle; if (lifecycle.state === "connecting") { if (message.type === "hello_error") { this.#failAndClose(new ServerError(message.error)); return; } if (message.type !== "hello") { this.#failAndClose(new ProtocolValidationError("Expected server hello as first message")); return; } if (message.serverId !== this.#options.serverId) { this.#failAndClose( new ProtocolValidationError( `Connected server ${JSON.stringify(message.serverId)} does not match ${JSON.stringify(this.#options.serverId)}`, ), ); return; } if (!lifecycle.transport) { this.#failAndClose(new ProtocolValidationError("Received server hello before the client hello was sent")); return; } const connected = { state: "connected", id: lifecycle.id, decoder: lifecycle.decoder, transport: lifecycle.transport, handshake: lifecycle.handshake, } satisfies Extract; this.#lifecycle = connected; try { this.#options.onHandshake(message); } catch (error) { if (this.#lifecycle === connected) this.#failAndClose(toError(error)); return; } if (this.#lifecycle !== connected) return; this.#options.onStateChange({ state: "connected" }); if (this.#lifecycle !== connected) return; this.#lifecycle = { ...connected, handshake: undefined }; lifecycle.handshake.resolve(message); return; } if (lifecycle.state !== "connected") return; if (message.type === "hello" || message.type === "hello_error") { this.#failAndClose(new ProtocolValidationError("Unexpected handshake message")); return; } this.#options.onMessage(message); } #handleClose(): void { const lifecycle = this.#lifecycle; if (lifecycle.state === "disconnected") return; let error: Error = new DisconnectedError("Byte transport closed"); try { lifecycle.decoder.end(); } catch (decoderError) { error = toError(decoderError); } this.#fail(error); } #failAndClose(error: Error): void { const lifecycle = this.#lifecycle; const transport = lifecycle.state === "disconnected" ? undefined : lifecycle.transport; this.#fail(error); transport?.close(); } #fail(error: Error): void { const lifecycle = this.#lifecycle; if (lifecycle.state === "disconnected") return; this.#lifecycle = { state: "disconnected" }; lifecycle.handshake?.reject(error); this.#options.onStateChange({ state: "disconnected", error }); } #isCurrent(id: number): boolean { return this.#lifecycle.state !== "disconnected" && this.#lifecycle.id === id; } }