Forward Nexus requests using same dispatch type as original request (#8199)

## What changed?
When forwarding Nexus requests that were originally sent to the
`DispatchByEndpoint` URL, the forwarding URL will also be constructed to
send the request to the `DispatchByEndpoint` URL on the remote cluster.
Previously, we were always sending forwarding requests using
`DispatchByNamespaceAndTaskQueue`

## Why?
bug fix

## How did you test it?
existing tests

Author: pdoerner

service/frontend/nexus_handler.go

@@ -48,6 +48,7 @@ type nexusContext struct {
 	namespaceName                        string
 	taskQueue                            string
 	endpointName                         string
+	endpointID                           string
 	claims                               *authorization.Claims
 	namespaceValidationInterceptor       *interceptor.NamespaceValidatorInterceptor
 	namespaceRateLimitInterceptor        interceptor.NamespaceRateLimitInterceptor
@@ -668,12 +669,21 @@ func (h *nexusHandler) nexusClientForActiveCluster(oc *operationContext, service
 		return response, nil
 	}
 
-	baseURL, err := url.JoinPath(
-		httpClient.BaseURL(),
-		commonnexus.RouteDispatchNexusTaskByNamespaceAndTaskQueue.Path(commonnexus.NamespaceAndTaskQueue{
-			Namespace: oc.namespaceName,
-			TaskQueue: oc.taskQueue,
-		}))
+	var baseURL string
+	if oc.endpointID != "" {
+		// If the request was originally dispatched by endpoint, forward by endpoint as well.
+		baseURL, err = url.JoinPath(httpClient.BaseURL(),
+			commonnexus.RouteDispatchNexusTaskByEndpoint.Path(oc.endpointID))
+	} else {
+		// Fallback to dispatch by namespace and task queue since those have already been resolved by this point.
+		baseURL, err = url.JoinPath(
+			httpClient.BaseURL(),
+			commonnexus.RouteDispatchNexusTaskByNamespaceAndTaskQueue.Path(commonnexus.NamespaceAndTaskQueue{
+				Namespace: oc.namespaceName,
+				TaskQueue: oc.taskQueue,
+			}))
+	}
+
 	if err != nil {
 		oc.logger.Error("failed to forward Nexus request. error constructing ServiceBaseURL",
 			tag.URL(httpClient.BaseURL()),

service/frontend/nexus_http_handler.go

@@ -269,6 +269,7 @@ func (h *NexusHTTPHandler) nexusContextFromEndpoint(entry *persistencespb.NexusE
 		nc.namespaceName = nsName.String()
 		nc.taskQueue = v.Worker.GetTaskQueue()
 		nc.endpointName = entry.Endpoint.Spec.Name
+		nc.endpointID = entry.Id
 		return nc, true
 	default:
 		h.writeNexusFailure(w, http.StatusBadRequest, &nexus.Failure{Message: "invalid endpoint target"})