Add invoke support for web pubsub client (Azure#36750)

### Packages impacted by this PR
@azure/web-pubsub-client

### Issues associated with this PR


### Describe the problem that is addressed by this PR


### What are the possible designs available to address the problem? If
there are more than one possible design, why was the one in this PR
chosen?


### Are there test cases added in this PR? _(If not, why?)_


### Provide a list of related PRs _(if any)_


### Command used to generate this PR:**_(Applicable only to SDK release
request PRs)_

### Checklists
- [ ] Added impacted package name to the issue description
- [ ] Does this PR needs any fixes in the SDK Generator?** _(If so,
create an Issue in the
[Autorest/typescript](https://github.com/Azure/autorest.typescript)
repository and link it here)_
- [ ] Added a changelog (if necessary)

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: vicancy

sdk/web-pubsub/web-pubsub-client/README.md

@@ -91,6 +91,21 @@ await client.joinGroup(groupName);
 await client.sendToGroup(groupName, "hello world", "text");
 ```
 
+### 5. Invoke upstream events (preview)
+
+```ts snippet:ReadmeSampleInvokeEvent
+import { WebPubSubClient } from "@azure/web-pubsub-client";
+
+const client = new WebPubSubClient("<client-access-url>");
+await client.start();
+
+const result = await client.invokeEvent("processOrder", { orderId: 1 }, "json");
+console.log(`Invocation result: ${JSON.stringify(result.data)}`);
+```
+
+`invokeEvent` sends an `invoke` request to the service, awaits the correlated `invokeResponse`, and returns the payload. You can abort the invocation by passing `{ abortSignal }`.
+_Streaming and service-initiated invocations are not yet supported._
+
 ---
 
 ## Examples

sdk/web-pubsub/web-pubsub-client/review/web-pubsub-client-node.api.md

@@ -20,6 +20,12 @@ export interface AckMessageError {
     name: string;
 }
 
+// @public
+export interface CancelInvocationMessage extends WebPubSubMessageBase {
+    invocationId: string;
+    readonly kind: "cancelInvocation";
+}
+
 // @public
 export interface ConnectedMessage extends WebPubSubMessageBase {
     connectionId: string;
@@ -59,7 +65,11 @@ export type DownstreamMessageType =
 /**
 * Type for ServerDataMessage
 */
-| "serverData";
+| "serverData"
+/**
+* Type for InvokeResponseMessage
+*/
+| "invokeResponse";
 
 // @public
 export interface GetClientAccessUrlOptions {
@@ -76,6 +86,58 @@ export interface GroupDataMessage extends WebPubSubMessageBase {
     sequenceId?: number;
 }
 
+// @public
+export class InvocationError extends Error {
+    constructor(message: string, options: InvocationErrorOptions);
+    errorDetail?: InvokeResponseError;
+    invocationId: string;
+}
+
+// @public (undocumented)
+export interface InvocationErrorOptions {
+    errorDetail?: InvokeResponseError;
+    invocationId: string;
+}
+
+// @public
+export interface InvokeEventOptions {
+    abortSignal?: AbortSignalLike;
+    invocationId?: string;
+}
+
+// @public
+export interface InvokeEventResult {
+    data?: JSONTypes | ArrayBuffer;
+    dataType?: WebPubSubDataType;
+    invocationId: string;
+}
+
+// @public
+export interface InvokeMessage extends WebPubSubMessageBase {
+    data?: JSONTypes | ArrayBuffer;
+    dataType?: WebPubSubDataType;
+    event?: string;
+    invocationId: string;
+    readonly kind: "invoke";
+    target?: "event";
+}
+
+// @public
+export interface InvokeResponseError {
+    message: string;
+    name: string;
+}
+
+// @public
+export interface InvokeResponseMessage extends WebPubSubMessageBase {
+    data?: JSONTypes | ArrayBuffer;
+    dataType?: WebPubSubDataType;
+    error?: InvokeResponseError;
+    invocationId: string;
+    readonly kind: "invokeResponse";
+    success?: boolean;
+}
+
 // @public
 export interface JoinGroupMessage extends WebPubSubMessageBase {
     ackId?: number;
@@ -242,12 +304,21 @@ export type UpstreamMessageType =
 /**
 * Type for PingMessage
 */
-| "ping";
+| "ping"
+/**
+* Type for InvokeMessage
+*/
+| "invoke"
+/**
+* Type for CancelInvocationMessage
+*/
+| "cancelInvocation";
 
 // @public
 export class WebPubSubClient {
     constructor(clientAccessUrl: string, options?: WebPubSubClientOptions);
     constructor(credential: WebPubSubClientCredential, options?: WebPubSubClientOptions);
+    invokeEvent(eventName: string, content: JSONTypes | ArrayBuffer, dataType: WebPubSubDataType, options?: InvokeEventOptions): Promise<InvokeEventResult>;
     joinGroup(groupName: string, options?: JoinGroupOptions): Promise<WebPubSubResult>;
     leaveGroup(groupName: string, options?: LeaveGroupOptions): Promise<WebPubSubResult>;
     off(event: "connected", listener: (e: OnConnectedArgs) => void): void;
@@ -318,7 +389,7 @@ export const WebPubSubJsonProtocol: () => WebPubSubClientProtocol;
 export const WebPubSubJsonReliableProtocol: () => WebPubSubClientProtocol;
 
 // @public
-export type WebPubSubMessage = GroupDataMessage | ServerDataMessage | JoinGroupMessage | LeaveGroupMessage | ConnectedMessage | DisconnectedMessage | SendToGroupMessage | SendEventMessage | SequenceAckMessage | PingMessage | AckMessage | PongMessage;
+export type WebPubSubMessage = GroupDataMessage | ServerDataMessage | JoinGroupMessage | LeaveGroupMessage | ConnectedMessage | DisconnectedMessage | SendToGroupMessage | SendEventMessage | SequenceAckMessage | PingMessage | AckMessage | InvokeMessage | InvokeResponseMessage | CancelInvocationMessage | PongMessage;
 
 // @public
 export interface WebPubSubMessageBase {

sdk/web-pubsub/web-pubsub-client/src/errors/index.ts

@@ -1,7 +1,7 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
-import type { AckMessageError } from "../models/messages.js";
+import type { AckMessageError, InvokeResponseError } from "../models/messages.js";
 
 /**
  * Error when sending message failed
@@ -42,3 +42,35 @@ export interface SendMessageErrorOptions {
    */
   errorDetail?: AckMessageError;
 }
+
+export interface InvocationErrorOptions {
+  /**
+   * The invocation id of the request.
+   */
+  invocationId: string;
+  /**
+   * Error details from the service if available.
+   */
+  errorDetail?: InvokeResponseError;
+}
+
+/**
+ * Error thrown when an invocation fails or is cancelled.
+ */
+export class InvocationError extends Error {
+  /**
+   * The invocation id of the request.
+   */
+  public invocationId: string;
+  /**
+   * Error details from the service if available.
+   */
+  public errorDetail?: InvokeResponseError;
+
+  constructor(message: string, options: InvocationErrorOptions) {
+    super(message);
+    this.name = "InvocationError";
+    this.invocationId = options.invocationId;
+    this.errorDetail = options.errorDetail;
+  }
+}

sdk/web-pubsub/web-pubsub-client/src/invocationManager.ts

@@ -0,0 +1,161 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+import type { AbortSignalLike } from "@azure/abort-controller";
+import type { InvokeResponseMessage } from "./models/messages.js";
+import { InvocationError } from "./errors/index.js";
+
+export interface InvocationWaitOptions {
+  abortSignal?: AbortSignalLike;
+}
+
+export interface InvocationRegistration {
+  invocationId: string;
+  wait(options?: InvocationWaitOptions): Promise<InvokeResponseMessage>;
+}
+
+/**
+ * Manages pending invocations awaiting invokeResponse frames.
+ */
+export class InvocationManager {
+  private readonly _entries = new Map<string, InvocationEntity>();
+  private _nextId = 0;
+
+  public registerInvocation(invocationId?: string): InvocationRegistration {
+    const resolvedId = invocationId ?? this._generateInvocationId();
+    if (this._entries.has(resolvedId)) {
+      throw new InvocationError("Invocation id is already registered.", {
+        invocationId: resolvedId,
+      });
+    }
+
+    const entity = new InvocationEntity(resolvedId);
+    this._entries.set(resolvedId, entity);
+    return {
+      invocationId: resolvedId,
+      wait: (options?: InvocationWaitOptions) => this._waitForEntry(entity, options),
+    };
+  }
+
+  public resolveInvocation(message: InvokeResponseMessage): boolean {
+    const entry = this._entries.get(message.invocationId);
+    if (!entry) {
+      return false;
+    }
+    this._entries.delete(message.invocationId);
+    entry.resolve(message);
+    return true;
+  }
+
+  public rejectInvocation(invocationId: string, reason: unknown): boolean {
+    const entry = this._entries.get(invocationId);
+    if (!entry) {
+      return false;
+    }
+    this._entries.delete(invocationId);
+    entry.reject(reason);
+    return true;
+  }
+
+  public discard(invocationId: string): void {
+    this._entries.delete(invocationId);
+  }
+
+  public rejectAll(createReason: (invocationId: string) => unknown): void {
+    this._entries.forEach((entry, invocationId) => {
+      if (this._entries.delete(invocationId)) {
+        entry.reject(createReason(invocationId));
+      }
+    });
+  }
+
+  private _waitForEntry(
+    entry: InvocationEntity,
+    options?: InvocationWaitOptions,
+  ): Promise<InvokeResponseMessage> {
+    const waitPromise = entry.promise();
+    const abortSignal = options?.abortSignal;
+
+    if (!abortSignal) {
+      return waitPromise;
+    }
+
+    if (abortSignal.aborted) {
+      if (this._entries.delete(entry.invocationId)) {
+        entry.reject(this._createAbortError(entry.invocationId));
+      }
+      return waitPromise;
+    }
+
+    return new Promise<InvokeResponseMessage>((resolve, reject) => {
+      const onAbort = (): void => {
+        abortSignal.removeEventListener("abort", onAbort);
+        if (this._entries.delete(entry.invocationId)) {
+          entry.reject(this._createAbortError(entry.invocationId));
+        }
+      };
+
+      abortSignal.addEventListener("abort", onAbort);
+
+      waitPromise
+        .then((result) => {
+          abortSignal.removeEventListener("abort", onAbort);
+          return resolve(result);
+        })
+        .catch((err) => {
+          abortSignal.removeEventListener("abort", onAbort);
+          return reject(err);
+        });
+    });
+  }
+
+  private _generateInvocationId(): string {
+    this._nextId += 1;
+    return this._nextId.toString();
+  }
+
+  private _createAbortError(invocationId: string): InvocationError {
+    return new InvocationError("Invocation cancelled by abortSignal.", {
+      invocationId,
+    });
+  }
+}
+
+class InvocationEntity {
+  private readonly _promise: Promise<InvokeResponseMessage>;
+  private _resolve:
+    | ((value: InvokeResponseMessage | PromiseLike<InvokeResponseMessage>) => void)
+    | undefined;
+  private _reject: ((reason?: unknown) => void) | undefined;
+
+  constructor(public readonly invocationId: string) {
+    this._promise = new Promise<InvokeResponseMessage>((resolve, reject) => {
+      this._resolve = resolve;
+      this._reject = reject;
+    });
+  }
+
+  public promise(): Promise<InvokeResponseMessage> {
+    return this._promise;
+  }
+
+  public resolve(value: InvokeResponseMessage | PromiseLike<InvokeResponseMessage>): void {
+    const callback = this._resolve;
+    if (!callback) {
+      return;
+    }
+    this._resolve = undefined;
+    this._reject = undefined;
+    callback(value);
+  }
+
+  public reject(reason?: unknown): void {
+    const callback = this._reject;
+    if (!callback) {
+      return;
+    }
+    this._resolve = undefined;
+    this._reject = undefined;
+    callback(reason);
+  }
+}