{"version":3,"file":"client.d.ts","sourceRoot":"","sources":["../src/client.ts"],"names":[],"mappings":"AAAA,OAAO,EAKN,KAAK,SAAS,EAKd,KAAK,sBAAsB,EAC3B,KAAK,WAAW,EAChB,KAAK,qBAAqB,EAC1B,KAAK,WAAW,EAChB,KAAK,qBAAqB,EAG1B,MAAM,uBAAuB,CAAC;AAE/B,OAAO,EAMN,KAAK,SAAS,EACd,KAAK,WAAW,EAEhB,KAAK,aAAa,EAClB,MAAM,6BAA6B,CAAC;AAIrC,OAAO,KAAK,EACX,wBAAwB,EACxB,aAAa,EACb,eAAe,EACf,qBAAqB,EACrB,mBAAmB,EACnB,WAAW,EACX,MAAM,YAAY,CAAC;AAEpB,KAAK,aAAa,GAAG,SAAS,GAAG,SAAS,CAAC;AAmB3C,qBAAa,MAAM;;IAclB,YAAY,OAAO,EAAE,aAAa,EAejC;IAED,IAAI,QAAQ,IAAI,OAAO,CAEtB;IAED,IAAI,eAAe,IAAI,eAAe,CAErC;IAED,IAAI,SAAS,IAAI,OAAO,CAEvB;IAED,IAAI,QAAQ,IAAI,MAAM,CAErB;IAED,IAAI,KAAK,IAAI,WAAW,GAAG,SAAS,CAEnC;IAED,IAAI,UAAU,IAAI,aAAa,GAAG,SAAS,CAE1C;IAED,OAAa,OAAO,CAAC,OAAO,EAAE,aAAa,GAAG,OAAO,CAAC,MAAM,CAAC,CAS5D;IAED,OAAO,IAAI,OAAO,CAAC,WAAW,CAAC,CAI9B;IAED,SAAS,IAAI,OAAO,CAAC,WAAW,CAAC,CAEhC;IAED,UAAU,CAAC,MAAM,SAAwB,GAAG,IAAI,CAE/C;IAED,uBAAuB,CAAC,QAAQ,EAAE,CAAC,MAAM,EAAE,qBAAqB,KAAK,IAAI,GAAG,WAAW,CAItF;IAED,kBAAkB,CAAC,QAAQ,EAAE,wBAAwB,GAAG,WAAW,CAIlE;IAED,4EAA4E;IAC5E,OAAO,CAAC,MAAM,EAAE,SAAS,EAAE,IAAI,EAAE,WAAW,EAAE,MAAM,CAAC,EAAE,WAAW,GAAG,OAAO,CAAC,aAAa,CAAC,CAE1F;IAEK,gBAAgB,CAAC,MAAM,EAAE,SAAS,EAAE,MAAM,CAAC,EAAE,WAAW,GAAG,OAAO,CAAC,SAAS,qBAAqB,EAAE,CAAC,CAWzG;IAEK,gBAAgB,CACrB,MAAM,EAAE,SAAS,EACjB,SAAS,EAAE,MAAM,EACjB,IAAI,EAAE,WAAW,EACjB,QAAQ,EAAE,CAAC,MAAM,EAAE,qBAAqB,KAAK,IAAI,GAAG,OAAO,CAAC,IAAI,CAAC,EACjE,MAAM,CAAC,EAAE,WAAW,GAClB,OAAO,CAAC,mBAAmB,CAAC,CA0D9B;IA+ID,OAAO,IAAI,OAAO,CAAC,IAAI,CAAC,CAavB;IAED,CAAC,MAAM,CAAC,YAAY,CAAC,IAAI,OAAO,CAAC,IAAI,CAAC,CAErC;CAiDD;AAED,kFAAkF;AAClF,wBAAgB,4BAA4B,CAC3C,MAAM,EAAE,MAAM,EACd,SAAS,EAAE,MAAM,SAAS,GAAG,SAAS,GACpC,sBAAsB,CAuBxB","sourcesContent":["import {\n\tcreateServiceCatalogueCall,\n\tcreateServiceStateDecoder,\n\tcreateServiceSubscribeCall,\n\tcreateServiceUnsubscribeCall,\n\ttype JsonValue,\n\tparseServiceCall,\n\tparseServiceCatalogue,\n\tparseWireServiceProviderUpdate,\n\tparseWireServiceSubscriptionSnapshot,\n\ttype RemoteServiceTransport,\n\ttype ServiceCall,\n\ttype ServiceCatalogueEntry,\n\ttype ServiceMode,\n\ttype ServiceProviderUpdate,\n\ttype ServiceStateDecoder,\n\ttype ServiceSubscriptionSnapshot,\n} from \"@earendil-works/chord\";\nimport { BACKGROUND_CONTEXT } from \"@earendil-works/chord/context\";\nimport {\n\ttype AttachmentEnvelope,\n\tencodeClientMessage,\n\tisServerId,\n\tProtocolValidationError,\n\ttype ResponseEnvelope,\n\ttype RpcTarget,\n\ttype ServerHello,\n\ttype ServiceEventEnvelope,\n\ttype SessionTarget,\n} from \"@earendil-works/pi-protocol\";\nimport { Connection } from \"./connection.ts\";\nimport { ClientDisposedError, DisconnectedError, ServerError, toError } from \"./errors.ts\";\nimport { createPromiseResolvers } from \"./promise.ts\";\nimport type {\n\tAttachmentChangeListener,\n\tClientOptions,\n\tConnectionState,\n\tConnectionStateChange,\n\tServiceSubscription,\n\tUnsubscribe,\n} from \"./types.ts\";\n\ntype ServiceResult = JsonValue | undefined;\n\ninterface PendingRequest {\n\tresolve(result: ServiceResult): void;\n\treject(error: Error): void;\n\tcleanup(): void;\n}\n\ninterface ActiveServiceListener {\n\treadonly target: RpcTarget;\n\treadonly listener: (update: ServiceProviderUpdate) => void | Promise<void>;\n\treadonly decoder: ServiceStateDecoder;\n\treadonly queuedWireUpdates: JsonValue[];\n\treadonly queued: ServiceProviderUpdate[];\n\tdeliveryTail: Promise<void>;\n\thydrated: boolean;\n\tready: boolean;\n}\n\nexport class Client {\n\treadonly #options: ClientOptions;\n\treadonly #connection: Connection;\n\treadonly #pendingRequests = new Map<string, PendingRequest>();\n\treadonly #connectionStateListeners = new Set<(change: ConnectionStateChange) => void>();\n\treadonly #attachmentListeners = new Set<AttachmentChangeListener>();\n\treadonly #serviceListeners = new Map<string, ActiveServiceListener>();\n\t#requestSequence = 0;\n\t#serviceSubscriptionSequence = 0;\n\t#hello: ServerHello | undefined;\n\t#attachment: SessionTarget | undefined;\n\t#disposed = false;\n\t#disposePromise: Promise<void> | undefined;\n\n\tconstructor(options: ClientOptions) {\n\t\tif (!isServerId(options.serverId)) {\n\t\t\tthrow new TypeError(\"serverId must be a canonical lowercase UUIDv4\");\n\t\t}\n\t\tthis.#options = options;\n\t\tthis.#connection = new Connection({\n\t\t\ttransportFactory: options.transportFactory,\n\t\t\tserverId: options.serverId,\n\t\t\tmaxFrameLength: options.maxFrameLength,\n\t\t\tonHandshake: (hello) => {\n\t\t\t\tthis.#hello = hello;\n\t\t\t},\n\t\t\tonMessage: (message) => this.#handleMessage(message),\n\t\t\tonStateChange: (change) => this.#handleConnectionStateChange(change),\n\t\t});\n\t}\n\n\tget disposed(): boolean {\n\t\treturn this.#disposed;\n\t}\n\n\tget connectionState(): ConnectionState {\n\t\treturn this.#connection.state;\n\t}\n\n\tget connected(): boolean {\n\t\treturn this.#connection.state === \"connected\";\n\t}\n\n\tget serverId(): string {\n\t\treturn this.#options.serverId;\n\t}\n\n\tget hello(): ServerHello | undefined {\n\t\treturn this.#hello;\n\t}\n\n\tget attachment(): SessionTarget | undefined {\n\t\treturn this.#attachment;\n\t}\n\n\tstatic async connect(options: ClientOptions): Promise<Client> {\n\t\tconst client = new Client(options);\n\t\ttry {\n\t\t\tawait client.connect();\n\t\t\treturn client;\n\t\t} catch (error) {\n\t\t\tawait client.dispose();\n\t\t\tthrow error;\n\t\t}\n\t}\n\n\tconnect(): Promise<ServerHello> {\n\t\tif (this.#disposed) return Promise.reject(new ClientDisposedError());\n\t\tthis.#hello = undefined;\n\t\treturn this.#connection.connect();\n\t}\n\n\treconnect(): Promise<ServerHello> {\n\t\treturn this.connect();\n\t}\n\n\tdisconnect(reason = \"Client disconnected\"): void {\n\t\tthis.#connection.disconnect(reason);\n\t}\n\n\tonConnectionStateChange(listener: (change: ConnectionStateChange) => void): Unsubscribe {\n\t\tthis.#assertNotDisposed();\n\t\tthis.#connectionStateListeners.add(listener);\n\t\treturn () => this.#connectionStateListeners.delete(listener);\n\t}\n\n\tonAttachmentChange(listener: AttachmentChangeListener): Unsubscribe {\n\t\tthis.#assertNotDisposed();\n\t\tthis.#attachmentListeners.add(listener);\n\t\treturn () => this.#attachmentListeners.delete(listener);\n\t}\n\n\t/** Invoke one low-level protocol call against an explicit routed target. */\n\trequest(target: RpcTarget, call: ServiceCall, signal?: AbortSignal): Promise<ServiceResult> {\n\t\treturn this.#request(target, call, signal);\n\t}\n\n\tasync serviceCatalogue(target: RpcTarget, signal?: AbortSignal): Promise<readonly ServiceCatalogueEntry[]> {\n\t\tconst result = await this.#request(target, createServiceCatalogueCall(), signal);\n\t\ttry {\n\t\t\treturn parseServiceCatalogue(result);\n\t\t} catch (error) {\n\t\t\tconst validationError = new ProtocolValidationError(\n\t\t\t\terror instanceof Error ? error.message : \"Invalid service catalogue\",\n\t\t\t);\n\t\t\tthis.#connection.fail(validationError);\n\t\t\tthrow validationError;\n\t\t}\n\t}\n\n\tasync subscribeService(\n\t\ttarget: RpcTarget,\n\t\tserviceId: string,\n\t\tmode: ServiceMode,\n\t\tlistener: (update: ServiceProviderUpdate) => void | Promise<void>,\n\t\tsignal?: AbortSignal,\n\t): Promise<ServiceSubscription> {\n\t\tconst subscriptionId = `service-${++this.#serviceSubscriptionSequence}`;\n\t\tconst active: ActiveServiceListener = {\n\t\t\ttarget,\n\t\t\tlistener,\n\t\t\tdecoder: createServiceStateDecoder(),\n\t\t\tqueuedWireUpdates: [],\n\t\t\tqueued: [],\n\t\t\tdeliveryTail: Promise.resolve(),\n\t\t\thydrated: false,\n\t\t\tready: false,\n\t\t};\n\t\tthis.#serviceListeners.set(subscriptionId, active);\n\t\tlet snapshot: ServiceSubscriptionSnapshot;\n\t\ttry {\n\t\t\tsnapshot = await this.#request(\n\t\t\t\ttarget,\n\t\t\t\tcreateServiceSubscribeCall(subscriptionId, serviceId, mode),\n\t\t\t\tsignal,\n\t\t\t\t(result) => {\n\t\t\t\t\tconst decoded = active.decoder.decodeSnapshot(parseWireServiceSubscriptionSnapshot(result));\n\t\t\t\t\tactive.hydrated = true;\n\t\t\t\t\tfor (const update of active.queuedWireUpdates.splice(0)) {\n\t\t\t\t\t\tactive.queued.push(active.decoder.decodeUpdate(parseWireServiceProviderUpdate(update)));\n\t\t\t\t\t}\n\t\t\t\t\treturn decoded;\n\t\t\t\t},\n\t\t\t);\n\t\t} catch (error) {\n\t\t\tif (this.#serviceListeners.get(subscriptionId) === active) this.#serviceListeners.delete(subscriptionId);\n\t\t\tthrow error;\n\t\t}\n\t\tif (this.#serviceListeners.get(subscriptionId) !== active) throw new DisconnectedError();\n\t\tlet disposed = false;\n\t\treturn {\n\t\t\tid: subscriptionId,\n\t\t\ttarget,\n\t\t\tsnapshot,\n\t\t\tstart: () => {\n\t\t\t\tif (disposed || active.ready) return;\n\t\t\t\tactive.ready = true;\n\t\t\t\tfor (const update of active.queued.splice(0)) this.#deliverServiceUpdate(active, update);\n\t\t\t},\n\t\t\tdispose: async () => {\n\t\t\t\tif (disposed) return;\n\t\t\t\tdisposed = true;\n\t\t\t\tif (this.#serviceListeners.get(subscriptionId) === active) this.#serviceListeners.delete(subscriptionId);\n\t\t\t\ttry {\n\t\t\t\t\tif (this.connected && this.#targetIsCurrent(target)) {\n\t\t\t\t\t\tawait this.#request(target, createServiceUnsubscribeCall(subscriptionId));\n\t\t\t\t\t}\n\t\t\t\t\tawait active.deliveryTail;\n\t\t\t\t} finally {\n\t\t\t\t\tactive.queuedWireUpdates.length = 0;\n\t\t\t\t\tactive.queued.length = 0;\n\t\t\t\t}\n\t\t\t},\n\t\t};\n\t}\n\n\t#request<T = ServiceResult>(\n\t\ttarget: RpcTarget,\n\t\tcall: ServiceCall,\n\t\tsignal?: AbortSignal,\n\t\ttransform?: (result: ServiceResult) => T,\n\t): Promise<T> {\n\t\tif (this.#disposed) return Promise.reject(new ClientDisposedError());\n\t\tif (!this.connected) return Promise.reject(new DisconnectedError());\n\t\tif (signal?.aborted) return Promise.reject(abortError(signal));\n\t\tconst id = `request-${++this.#requestSequence}`;\n\t\tconst { promise, resolve, reject } = createPromiseResolvers<T>();\n\t\tlet sent = false;\n\t\tlet aborted = false;\n\t\tlet onAbort: (() => void) | undefined;\n\t\tconst sendCancel = (): void => {\n\t\t\tif (!sent || !this.connected) return;\n\t\t\ttry {\n\t\t\t\tthis.#connection.send(\n\t\t\t\t\tencodeClientMessage({ type: \"cancel\", id, target }, { maxFrameLength: this.#connection.maxFrameLength }),\n\t\t\t\t);\n\t\t\t} catch (error) {\n\t\t\t\tthis.#connection.fail(toError(error));\n\t\t\t}\n\t\t};\n\t\tif (signal !== undefined) {\n\t\t\tonAbort = () => {\n\t\t\t\tif (aborted) return;\n\t\t\t\taborted = true;\n\t\t\t\treject(abortError(signal));\n\t\t\t\tsendCancel();\n\t\t\t};\n\t\t\tsignal.addEventListener(\"abort\", onAbort, { once: true });\n\t\t}\n\t\tthis.#pendingRequests.set(id, {\n\t\t\tresolve: (result) => {\n\t\t\t\ttry {\n\t\t\t\t\tresolve(transform === undefined ? (result as T) : transform(result));\n\t\t\t\t} catch (error) {\n\t\t\t\t\tconst validationError = new ProtocolValidationError(\n\t\t\t\t\t\terror instanceof Error ? error.message : \"Invalid service operation stream\",\n\t\t\t\t\t);\n\t\t\t\t\tthis.#connection.fail(validationError);\n\t\t\t\t\treject(validationError);\n\t\t\t\t}\n\t\t\t},\n\t\t\treject,\n\t\t\tcleanup: () => {\n\t\t\t\tif (signal !== undefined && onAbort !== undefined) signal.removeEventListener(\"abort\", onAbort);\n\t\t\t},\n\t\t});\n\t\tlet frame: Uint8Array;\n\t\ttry {\n\t\t\tframe = encodeClientMessage(\n\t\t\t\t{ type: \"request\", id, target, call: parseServiceCall(call) as unknown as JsonValue },\n\t\t\t\t{ maxFrameLength: this.#connection.maxFrameLength },\n\t\t\t);\n\t\t} catch (error) {\n\t\t\tthis.#takePendingRequest(id)?.reject(toError(error));\n\t\t\treturn promise;\n\t\t}\n\t\tthis.#connection.send(frame);\n\t\tsent = true;\n\t\tif (aborted) sendCancel();\n\t\treturn promise;\n\t}\n\n\t#handleMessage(message: ResponseEnvelope | ServiceEventEnvelope | AttachmentEnvelope): void {\n\t\tif (message.type === \"attachment\") {\n\t\t\tif (message.attachment !== null && message.attachment.serverId !== this.#options.serverId) {\n\t\t\t\tthis.#connection.fail(new ProtocolValidationError(\"Attachment update belongs to another server\"));\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tthis.#setAttachment(message.attachment ?? undefined);\n\t\t\treturn;\n\t\t}\n\t\tif (message.type === \"service_update\") {\n\t\t\tconst active = this.#serviceListeners.get(message.subscriptionId);\n\t\t\tif (active === undefined) return;\n\t\t\tif (!active.hydrated) {\n\t\t\t\tactive.queuedWireUpdates.push(message.update);\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tlet update: ServiceProviderUpdate;\n\t\t\ttry {\n\t\t\t\tupdate = active.decoder.decodeUpdate(parseWireServiceProviderUpdate(message.update));\n\t\t\t} catch (error) {\n\t\t\t\tthis.#connection.fail(\n\t\t\t\t\tnew ProtocolValidationError(error instanceof Error ? error.message : \"Invalid service operation stream\"),\n\t\t\t\t);\n\t\t\t\treturn;\n\t\t\t}\n\t\t\tif (active.ready) this.#deliverServiceUpdate(active, update);\n\t\t\telse active.queued.push(update);\n\t\t\treturn;\n\t\t}\n\t\tconst pending = this.#takePendingRequest(message.id);\n\t\tif (!pending) {\n\t\t\tthis.#connection.fail(new ProtocolValidationError(\"Response has no matching request\"));\n\t\t\treturn;\n\t\t}\n\t\tif (!message.ok) {\n\t\t\tpending.reject(new ServerError(message.error));\n\t\t\treturn;\n\t\t}\n\t\tpending.resolve(message.result);\n\t}\n\n\t#handleConnectionStateChange(change: ConnectionStateChange): void {\n\t\tif (change.state === \"disconnected\") {\n\t\t\tthis.#hello = undefined;\n\t\t\tthis.#setAttachment(undefined);\n\t\t\tthis.#rejectPendingRequests(change.error ?? new DisconnectedError());\n\t\t\tthis.#serviceListeners.clear();\n\t\t}\n\t\tfor (const listener of this.#connectionStateListeners) {\n\t\t\ttry {\n\t\t\t\tlistener(change);\n\t\t\t} catch (error) {\n\t\t\t\tthis.#reportListenerError(error);\n\t\t\t}\n\t\t}\n\t}\n\n\t#takePendingRequest(id: string): PendingRequest | undefined {\n\t\tconst request = this.#pendingRequests.get(id);\n\t\tif (request) {\n\t\t\tthis.#pendingRequests.delete(id);\n\t\t\trequest.cleanup();\n\t\t}\n\t\treturn request;\n\t}\n\n\t#rejectPendingRequests(error: Error): void {\n\t\tconst requests = [...this.#pendingRequests.values()];\n\t\tthis.#pendingRequests.clear();\n\t\tfor (const request of requests) {\n\t\t\trequest.cleanup();\n\t\t\trequest.reject(error);\n\t\t}\n\t}\n\n\tdispose(): Promise<void> {\n\t\tif (this.#disposePromise) return this.#disposePromise;\n\t\tthis.#disposed = true;\n\t\tthis.#disposePromise = Promise.resolve();\n\t\tconst error = new ClientDisposedError();\n\t\tthis.#rejectPendingRequests(error);\n\t\tthis.#connection.disconnect(error);\n\t\tthis.#hello = undefined;\n\t\tthis.#setAttachment(undefined);\n\t\tthis.#connectionStateListeners.clear();\n\t\tthis.#attachmentListeners.clear();\n\t\tthis.#serviceListeners.clear();\n\t\treturn this.#disposePromise;\n\t}\n\n\t[Symbol.asyncDispose](): Promise<void> {\n\t\treturn this.dispose();\n\t}\n\n\t#setAttachment(attachment: SessionTarget | undefined): void {\n\t\tconst previous = this.#attachment;\n\t\tif (\n\t\t\tprevious?.serverId === attachment?.serverId &&\n\t\t\tprevious?.sessionId === attachment?.sessionId &&\n\t\t\tprevious?.attachmentId === attachment?.attachmentId\n\t\t) {\n\t\t\treturn;\n\t\t}\n\t\tthis.#attachment = attachment;\n\t\tfor (const listener of this.#attachmentListeners) {\n\t\t\ttry {\n\t\t\t\tlistener(attachment);\n\t\t\t} catch (error) {\n\t\t\t\tthis.#reportListenerError(error);\n\t\t\t}\n\t\t}\n\t}\n\n\t#deliverServiceUpdate(active: ActiveServiceListener, update: ServiceProviderUpdate): void {\n\t\tactive.deliveryTail = active.deliveryTail\n\t\t\t.then(() => active.listener(update))\n\t\t\t.catch((error: unknown) => this.#reportListenerError(error));\n\t}\n\n\t#targetIsCurrent(target: RpcTarget): boolean {\n\t\tif (!(\"sessionId\" in target)) return this.#hello?.serverId === target.serverId;\n\t\tconst attachment = this.#attachment;\n\t\treturn (\n\t\t\tattachment?.serverId === target.serverId &&\n\t\t\tattachment.sessionId === target.sessionId &&\n\t\t\tattachment.attachmentId === target.attachmentId\n\t\t);\n\t}\n\n\t#assertNotDisposed(): void {\n\t\tif (this.#disposed) throw new ClientDisposedError();\n\t}\n\n\t#reportListenerError(error: unknown): void {\n\t\tif (!this.#options.onListenerError) return;\n\t\ttry {\n\t\t\tthis.#options.onListenerError(toError(error));\n\t\t} catch {\n\t\t\t// Diagnostics cannot affect protocol or transport state.\n\t\t}\n\t}\n}\n\n/** Adapts a lazily resolved routed client target to a Chord service transport. */\nexport function createClientServiceTransport(\n\tclient: Client,\n\tgetTarget: () => RpcTarget | undefined,\n): RemoteServiceTransport {\n\tconst target = (): RpcTarget => {\n\t\tconst resolved = getTarget();\n\t\tif (resolved === undefined) throw new Error(\"Remote service target is unavailable\");\n\t\treturn resolved;\n\t};\n\treturn {\n\t\tinvoke: async (call, context) => client.request(target(), call, context.abortSignal),\n\t\tasync subscribe(serviceId, mode, listener, context) {\n\t\t\tconst subscription = await client.subscribeService(\n\t\t\t\ttarget(),\n\t\t\t\tserviceId,\n\t\t\t\tmode,\n\t\t\t\t(update) => listener(update, BACKGROUND_CONTEXT),\n\t\t\t\tcontext.abortSignal,\n\t\t\t);\n\t\t\treturn {\n\t\t\t\tsnapshot: subscription.snapshot,\n\t\t\t\tactivate: () => subscription.start(),\n\t\t\t\tclose: () => subscription.dispose(),\n\t\t\t};\n\t\t},\n\t};\n}\n\nfunction abortError(signal: AbortSignal): Error {\n\tconst reason: unknown = signal.reason;\n\treturn reason instanceof Error ? reason : new DOMException(\"The operation was aborted\", \"AbortError\");\n}\n"]}