Generic Temporal Nexus Operation Handler

src/Temporalio/Nexus/ITemporalNexusClient.cs

@@ -0,0 +1,76 @@
+using System;
+using System.Collections.Generic;
+using System.Linq.Expressions;
+using System.Threading.Tasks;
+using NexusRpc.Handlers;
+using Temporalio.Client;
+
+namespace Temporalio.Nexus
+{
+    /// <summary>
+    /// Nexus-aware client wrapping the Temporal client. Provides methods for starting workflows
+    /// from within a Nexus operation handler.
+    /// </summary>
+    /// <remarks>
+    /// <para>WARNING: Nexus support is experimental.</para>
+    /// <para>Obtained via the <see cref="TemporalOperationHandler.FromHandleFactory{TInput, TResult}"/>
+    /// start function parameter.</para>
+    /// <para>Example usage — starting a workflow from an operation handler:</para>
+    /// <code>
+    /// await client.StartWorkflowAsync&lt;MyWorkflow, MyResult&gt;(
+    ///     wf => wf.RunAsync(input),
+    ///     new(id: "my-workflow-id", taskQueue: "my-task-queue"));
+    /// </code>
+    /// <para>To perform a synchronous operation (e.g., sending a signal), use the underlying
+    /// <see cref="TemporalClient"/> and return a sync result:</para>
+    /// <code>
+    /// await client.TemporalClient
+    ///     .GetWorkflowHandle($"order-{input.OrderId}")
+    ///     .SignalAsync("requestCancellation", new[] { input });
+    /// return TemporalOperationResult&lt;NoValue&gt;.SyncResult(default);
+    /// </code>
+    /// </remarks>
+    public interface ITemporalNexusClient
+    {
+        /// <summary>
+        /// Gets the underlying Temporal client for advanced use cases such as sending signals
+        /// or queries.
+        /// </summary>
+        ITemporalClient TemporalClient { get; }
+
+        /// <summary>
+        /// Start a workflow via a lambda invoking the run method. Always returns an async result
+        /// with a workflow-run operation token.
+        /// </summary>
+        /// <typeparam name="TWorkflow">Workflow class type.</typeparam>
+        /// <typeparam name="TResult">Workflow result type.</typeparam>
+        /// <param name="workflowRunCall">Invocation of workflow run method with a result.</param>
+        /// <param name="options">Start workflow options. ID and TaskQueue are required.</param>
+        /// <returns>An async operation result containing the workflow-run token.</returns>
+        Task<TemporalOperationResult<TResult>> StartWorkflowAsync<TWorkflow, TResult>(
+            Expression<Func<TWorkflow, Task<TResult>>> workflowRunCall, WorkflowOptions options);
+
+        /// <summary>
+        /// Start a workflow via a lambda invoking the run method with no return value. Always
+        /// returns an async result with a workflow-run operation token.
+        /// </summary>
+        /// <typeparam name="TWorkflow">Workflow class type.</typeparam>
+        /// <param name="workflowRunCall">Invocation of workflow run method with no result.</param>
+        /// <param name="options">Start workflow options. ID and TaskQueue are required.</param>
+        /// <returns>An async operation result containing the workflow-run token.</returns>
+        Task<TemporalOperationResult<NoValue>> StartWorkflowAsync<TWorkflow>(
+            Expression<Func<TWorkflow, Task>> workflowRunCall, WorkflowOptions options);
+
+        /// <summary>
+        /// Start a workflow by name. Always returns an async result with a workflow-run operation
+        /// token.
+        /// </summary>
+        /// <typeparam name="TResult">Workflow result type.</typeparam>
+        /// <param name="workflow">Workflow type name.</param>
+        /// <param name="args">Arguments for the workflow.</param>
+        /// <param name="options">Start workflow options. ID and TaskQueue are required.</param>
+        /// <returns>An async operation result containing the workflow-run token.</returns>
+        Task<TemporalOperationResult<TResult>> StartWorkflowAsync<TResult>(
+            string workflow, IReadOnlyCollection<object?> args, WorkflowOptions options);
+    }
+}

src/Temporalio/Nexus/NexusWorkflowRunHandle.cs

@@ -14,6 +14,11 @@ namespace Temporalio.Nexus
     /// <remarks>WARNING: Nexus support is experimental.</remarks>
     public class NexusWorkflowRunHandle
     {
+        /// <summary>
+        /// Token-type value identifying a workflow-run operation token.
+        /// </summary>
+        internal const int WorkflowRunOperationTokenType = 1;
+
         private static readonly JsonSerializerOptions TokenSerializerOptions = new()
         {
 #pragma warning disable SYSLIB0020 // Need to use obsolete form, alternative not in all our versions
@@ -86,6 +91,19 @@ internal static byte[] Base64UrlDecode(string s)
         /// <returns>Created handle.</returns>
         /// <exception cref="ArgumentException">If the token is invalid.</exception>
         internal static NexusWorkflowRunHandle FromToken(string token)
+        {
+            var data = ParseToken(token);
+            return new(data.Namespace, data.WorkflowId, data.Version ?? 0);
+        }
+
+        /// <summary>
+        /// Parse an operation token to its underlying fields. Validates encoding, JSON shape, and
+        /// version (but not type — callers decide which token types they support).
+        /// </summary>
+        /// <param name="token">Base64url-encoded token string.</param>
+        /// <returns>Parsed token fields.</returns>
+        /// <exception cref="ArgumentException">If the token is invalid.</exception>
+        internal static OperationToken ParseToken(string token)
         {
             byte[] bytes;
             try
@@ -96,10 +114,10 @@ internal static NexusWorkflowRunHandle FromToken(string token)
             {
                 throw new ArgumentException("Token invalid");
             }
-            Token? tokenObj;
+            OperationToken? tokenObj;
             try
             {
-                tokenObj = JsonSerializer.Deserialize<Token>(bytes, TokenSerializerOptions);
+                tokenObj = JsonSerializer.Deserialize<OperationToken>(bytes, TokenSerializerOptions);
             }
             catch (JsonException e)
             {
@@ -113,26 +131,29 @@ internal static NexusWorkflowRunHandle FromToken(string token)
             {
                 throw new ArgumentException($"Unsupported token version: {tokenObj.Version}");
             }
-            return new(tokenObj.Namespace, tokenObj.WorkflowId, tokenObj.Version ?? 0);
+            return tokenObj;
         }
 
         /// <summary>
         /// Create a string token based on this handle.
         /// </summary>
         /// <returns>Operation token.</returns>
         internal string ToToken() => Base64UrlEncode(JsonSerializer.SerializeToUtf8Bytes(
-            new Token(Namespace, WorkflowId, Version == 0 ? null : Version),
+            new OperationToken(Namespace, WorkflowId, Version == 0 ? null : Version),
             TokenSerializerOptions));
 
-        private record Token(
+        /// <summary>
+        /// Represents the fields of a Nexus operation token.
+        /// </summary>
+        internal record OperationToken(
             [property: JsonPropertyName("ns")]
             string Namespace,
             [property: JsonPropertyName("wid")]
             string WorkflowId,
             [property: JsonPropertyName("v")]
             int? Version,
             [property: JsonPropertyName("t")]
-            int Type = 1);
+            int Type = WorkflowRunOperationTokenType);
     }
 
     /// <inheritdoc />

src/Temporalio/Nexus/NexusWorkflowStartHelper.cs

@@ -0,0 +1,122 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using NexusRpc.Handlers;
+using Temporalio.Api.Common.V1;
+using Temporalio.Api.Enums.V1;
+using Temporalio.Client;
+
+namespace Temporalio.Nexus
+{
+    /// <summary>
+    /// Internal helper for starting workflows from Nexus operations and managing operation tokens.
+    /// Shared by both <see cref="WorkflowRunOperationContext"/> and <see cref="TemporalNexusClient"/>.
+    /// </summary>
+    internal static class NexusWorkflowStartHelper
+    {
+        private const string NexusOperationTokenHeader = "Nexus-Operation-Token";
+
+        /// <summary>
+        /// Start a workflow and return the workflow-run handle. This handles all Nexus plumbing:
+        /// cloning options, setting task queue, processing links, injecting callbacks, and
+        /// adding outbound links.
+        /// </summary>
+        /// <param name="nexusStartContext">Nexus start context for callbacks and links.</param>
+        /// <param name="temporalContext">Temporal operation context for client, info, and logging.</param>
+        /// <param name="workflow">Workflow type name.</param>
+        /// <param name="args">Workflow arguments.</param>
+        /// <param name="options">Workflow start options. ID and TaskQueue are required.</param>
+        /// <returns>Workflow-run handle for the started workflow.</returns>
+        internal static async Task<NexusWorkflowRunHandle> StartWorkflowAsync(
+            OperationStartContext nexusStartContext,
+            NexusOperationExecutionContext temporalContext,
+            string workflow,
+            IReadOnlyCollection<object?> args,
+            WorkflowOptions options)
+        {
+            var client = temporalContext.TemporalClient;
+            var namespace_ = client.Options.Namespace;
+            var workflowId = options.Id ?? string.Empty;
+
+            // Generate the handle and token before starting the workflow (token is needed for the
+            // callback header).
+            var handle = new NexusWorkflowRunHandle(namespace_, workflowId, 0);
+            var token = handle.ToToken();
+
+            // Shallow clone the options so we can mutate them. We just overwrite any of these
+            // internal options since they cannot be user set at this time.
+            options = (WorkflowOptions)options.Clone();
+            options.TaskQueue ??= temporalContext.Info.TaskQueue;
+            if (options.IdConflictPolicy == WorkflowIdConflictPolicy.UseExisting)
+            {
+                options.OnConflictOptions = new()
+                {
+                    AttachLinks = true,
+                    AttachCompletionCallbacks = true,
+                    AttachRequestId = true,
+                };
+            }
+            if (nexusStartContext.InboundLinks.Count > 0)
+            {
+                options.Links = nexusStartContext.InboundLinks.Select(link =>
+                {
+                    try
+                    {
+                        return new Link { WorkflowEvent = link.ToWorkflowEvent() };
+                    }
+                    catch (ArgumentException e)
+                    {
+                        temporalContext.Logger.LogWarning(e, "Invalid Nexus link: {Url}", link.Uri);
+                        return null;
+                    }
+                }).OfType<Link>().ToList();
+            }
+            if (nexusStartContext.CallbackUrl is { } callbackUrl)
+            {
+                var callback = new Callback() { Nexus = new() { Url = callbackUrl } };
+                var callbackHeadersHasToken = false;
+                if (nexusStartContext.CallbackHeaders is { } callbackHeaders)
+                {
+                    foreach (var kv in callbackHeaders)
+                    {
+                        callback.Nexus.Header.Add(kv.Key, kv.Value);
+                        if (string.Equals(
+                                kv.Key, NexusOperationTokenHeader, StringComparison.OrdinalIgnoreCase))
+                        {
+                            callbackHeadersHasToken = true;
+                        }
+                    }
+                }
+                // Set operation token if not already present (header is case-insensitive)
+                if (!callbackHeadersHasToken)
+                {
+                    callback.Nexus.Header[NexusOperationTokenHeader] = token;
+                }
+                if (options.Links is { } links)
+                {
+                    callback.Links.AddRange(links);
+                }
+                options.CompletionCallbacks = new[] { callback };
+            }
+            options.RequestId = nexusStartContext.RequestId;
+
+            // Do the start call
+            var wfHandle = await client.StartWorkflowAsync(
+                workflow, args, options).ConfigureAwait(false);
+
+            // Add the outbound link
+            nexusStartContext.OutboundLinks.Add(new Link.Types.WorkflowEvent
+            {
+                Namespace = namespace_,
+                WorkflowId = workflowId,
+                RunId = wfHandle.FirstExecutionRunId ??
+                    throw new InvalidOperationException("Handle unexpectedly missing run ID"),
+                EventRef = new() { EventId = 1, EventType = EventType.WorkflowExecutionStarted },
+            }.ToNexusLink());
+
+            return handle;
+        }
+    }
+}

src/Temporalio/Nexus/TemporalNexusClient.cs

@@ -0,0 +1,79 @@
+using System;
+using System.Collections.Generic;
+using System.Linq.Expressions;
+using System.Threading.Tasks;
+using NexusRpc.Handlers;
+using Temporalio.Client;
+
+namespace Temporalio.Nexus
+{
+    /// <summary>
+    /// Nexus-aware client wrapping the Temporal client. Provides methods for starting workflows
+    /// from within Nexus operation handlers, handling all Nexus plumbing (links, callbacks, token
+    /// generation) internally.
+    /// </summary>
+    /// <remarks>
+    /// WARNING: Nexus support is experimental.
+    /// This client is created by <see cref="TemporalOperationHandler"/> and passed to the
+    /// user's start function. It should not be instantiated directly.
+    /// </remarks>
+    public class TemporalNexusClient : ITemporalNexusClient
+    {
+        private readonly OperationStartContext nexusStartContext;
+        private readonly NexusOperationExecutionContext temporalContext;
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="TemporalNexusClient"/> class.
+        /// </summary>
+        /// <param name="nexusStartContext">Nexus start context for callbacks and links.</param>
+        /// <param name="temporalContext">Temporal operation context. </param>
+        internal TemporalNexusClient(
+            OperationStartContext nexusStartContext,
+            NexusOperationExecutionContext temporalContext)
+        {
+            this.nexusStartContext = nexusStartContext;
+            this.temporalContext = temporalContext;
+        }
+
+        /// <inheritdoc/>
+        public ITemporalClient TemporalClient => temporalContext.TemporalClient;
+
+        /// <inheritdoc/>
+        public Task<TemporalOperationResult<TResult>> StartWorkflowAsync<TWorkflow, TResult>(
+            Expression<Func<TWorkflow, Task<TResult>>> workflowRunCall, WorkflowOptions options)
+        {
+            var (runMethod, args) = Common.ExpressionUtil.ExtractCall(workflowRunCall);
+            return StartWorkflowAsync<TResult>(
+                Workflows.WorkflowDefinition.NameFromRunMethodForCall(runMethod),
+                args,
+                options);
+        }
+
+        /// <inheritdoc/>
+        public async Task<TemporalOperationResult<NoValue>> StartWorkflowAsync<TWorkflow>(
+            Expression<Func<TWorkflow, Task>> workflowRunCall, WorkflowOptions options)
+        {
+            var (runMethod, args) = Common.ExpressionUtil.ExtractCall(workflowRunCall);
+            var handle = await NexusWorkflowStartHelper.StartWorkflowAsync(
+                nexusStartContext,
+                temporalContext,
+                Workflows.WorkflowDefinition.NameFromRunMethodForCall(runMethod),
+                args,
+                options).ConfigureAwait(false);
+            return TemporalOperationResult<NoValue>.AsyncResult(handle.ToToken());
+        }
+
+        /// <inheritdoc/>
+        public async Task<TemporalOperationResult<TResult>> StartWorkflowAsync<TResult>(
+            string workflow, IReadOnlyCollection<object?> args, WorkflowOptions options)
+        {
+            var handle = await NexusWorkflowStartHelper.StartWorkflowAsync(
+                nexusStartContext,
+                temporalContext,
+                workflow,
+                args,
+                options).ConfigureAwait(false);
+            return TemporalOperationResult<TResult>.AsyncResult(handle.ToToken());
+        }
+    }
+}