Make Nexus work OOTB with token-based callback routing (temporalio#9513)

## Summary
- Use the system callback URL by default now that the previous minor
supports it (1.30)
- Extract system callback URL handling from `routeRequest` into a
dedicated `routeSystemCallbackRequest` function that routes based on
callback token namespace and active cluster
- Update both `components/callbacks` and `chasm/lib/callback` packages
to use namespace registry and callback token generator for routing
decisions
- Add unit tests for `routeRequest` and `routeSystemCallbackRequest`
covering external targets, local/unknown cluster routing, nil headers,
invalid tokens, namespace not found, and success paths

## Test plan
- [x] Unit tests added for both `components/callbacks` and
`chasm/lib/callback` packages
- [x] All 18 tests pass (9 per package)
- [x] `make lint` passes with no new issues

Author: bergundy

chasm/lib/callback/fx.go

@@ -9,6 +9,8 @@ import (
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/collection"
 	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/namespace"
+	commonnexus "go.temporal.io/server/common/nexus"
 	queuescommon "go.temporal.io/server/service/history/queues/common"
 	"go.uber.org/fx"
 )
@@ -23,6 +25,7 @@ func register(
 // httpCallerProviderProvider provides an HTTPCallerProvider for CHASM callbacks.
 func httpCallerProviderProvider(
 	clusterMetadata cluster.Metadata,
+	namespaceRegistry namespace.Registry,
 	rpcFactory common.RPCFactory,
 	httpClientCache *cluster.FrontendHTTPClientCache,
 	logger log.Logger,
@@ -32,12 +35,15 @@ func httpCallerProviderProvider(
 		return nil, fmt.Errorf("cannot create local frontend HTTP client: %w", err)
 	}
 	defaultClient := &http.Client{}
+	callbackTokenGenerator := commonnexus.NewCallbackTokenGenerator()
 
 	m := collection.NewOnceMap(func(queuescommon.NamespaceIDAndDestination) HTTPCaller {
 		return func(r *http.Request) (*http.Response, error) {
 			return routeRequest(r,
 				clusterMetadata,
+				namespaceRegistry,
 				httpClientCache,
+				callbackTokenGenerator,
 				defaultClient,
 				localClient,
 				logger,

chasm/lib/callback/request.go

@@ -1,29 +1,98 @@
 package callback
 
 import (
+	"errors"
 	"net/http"
 
+	"github.com/nexus-rpc/sdk-go/nexus"
+	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
-	"go.temporal.io/server/common/nexus"
+	"go.temporal.io/server/common/namespace"
+	commonnexus "go.temporal.io/server/common/nexus"
 )
 
 // Header key used to identify callbacks that originate from and target the same cluster.
 // Note: this is the nexusoperations.NexusCallbackSourceHeader stripped of Nexus-Callback-
 const callbackSourceHeader = "source"
 
+// routeSystemCallbackRequest routes a system callback request to the appropriate frontend client
+// based on the callback token's namespace and active cluster.
+func routeSystemCallbackRequest(
+	r *http.Request,
+	clusterMetadata cluster.Metadata,
+	namespaceRegistry namespace.Registry,
+	httpClientCache *cluster.FrontendHTTPClientCache,
+	callbackTokenGenerator *commonnexus.CallbackTokenGenerator,
+	localClient *common.FrontendHTTPClient,
+	logger log.Logger,
+) (*http.Response, error) {
+	var frontendClient *common.FrontendHTTPClient
+	if r.Header != nil {
+		token, err := commonnexus.DecodeCallbackToken(r.Header.Get(commonnexus.CallbackTokenHeader))
+		if err != nil {
+			logger.Error("failed to decode callback token", tag.Error(err))
+			return nil, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "invalid callback token")
+		}
+
+		completion, err := callbackTokenGenerator.DecodeCompletion(token)
+		if err != nil {
+			logger.Error("failed to decode completion from token", tag.Error(err))
+			return nil, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "invalid callback token")
+		}
+		ns, err := namespaceRegistry.GetNamespaceByID(namespace.ID(completion.NamespaceId))
+		if err != nil {
+			logger.Error("failed to get namespace for nexus completion request", tag.WorkflowNamespaceID(completion.NamespaceId), tag.Error(err))
+			var nfe *serviceerror.NamespaceNotFound
+			if errors.As(err, &nfe) {
+				return nil, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeNotFound, "namespace %q not found", completion.NamespaceId)
+			}
+			return nil, commonnexus.ConvertGRPCError(err, false)
+		}
+		clusterName := ns.ActiveClusterName(completion.GetWorkflowId())
+		if clusterMetadata.GetCurrentClusterName() == clusterName {
+			frontendClient = localClient
+		} else {
+			fec, err := httpClientCache.Get(clusterName)
+			if err != nil {
+				logger.Warn(
+					"HTTPCallerProvider unable to get FrontendHTTPClient for callback target cluster. Using local HTTP Client.",
+					tag.SourceCluster(clusterMetadata.GetCurrentClusterName()),
+					tag.TargetCluster(clusterName),
+					tag.Error(err),
+				)
+				frontendClient = localClient
+			} else {
+				frontendClient = fec
+			}
+		}
+	} else {
+		frontendClient = localClient
+	}
+	r.URL.Path = commonnexus.PathCompletionCallbackNoIdentifier
+	r.URL.Scheme = frontendClient.Scheme
+	r.URL.Host = frontendClient.Address
+	r.Host = frontendClient.Address
+	return frontendClient.Do(r)
+}
+
 func routeRequest(
 	r *http.Request,
 	clusterMetadata cluster.Metadata,
+	namespaceRegistry namespace.Registry,
 	httpClientCache *cluster.FrontendHTTPClientCache,
+	callbackTokenGenerator *commonnexus.CallbackTokenGenerator,
 	defaultClient *http.Client,
 	localClient *common.FrontendHTTPClient,
 	logger log.Logger,
 ) (*http.Response, error) {
+	if r.URL.String() == commonnexus.SystemCallbackURL {
+		return routeSystemCallbackRequest(r, clusterMetadata, namespaceRegistry, httpClientCache, callbackTokenGenerator, localClient, logger)
+	}
 	// This source header is populated in nexusoperations/executors (via the ClientProvider) for worker targets
-	// if this header is not populated then we assume it's and external target.
+	// if this header is not populated then we assume it's an external target.
 	if r.Header == nil || r.Header.Get(callbackSourceHeader) == "" {
 		return defaultClient.Do(r)
 	}
@@ -61,9 +130,6 @@ func routeRequest(
 		frontendClient = localClient
 	}
 
-	if r.URL.String() == nexus.SystemCallbackURL {
-		r.URL.Path = nexus.PathCompletionCallbackNoIdentifier
-	}
 	r.URL.Scheme = frontendClient.Scheme
 	r.URL.Host = frontendClient.Address
 	r.Host = frontendClient.Address