Support Workflow Update as a Nexus Operation

Author: Evanthx

CHANGELOG.md

@@ -21,6 +21,14 @@ to docs, or any other relevant information.
 
 ### Added
 
+- UpdateWorkflow-backed Nexus operations. A Temporal Nexus operation can now be backed by a Workflow
+  Update via `TemporalNexusClient.updateWorkflow`, in addition to a Workflow run. The Update request
+  carries the Nexus request ID (for deduplication), the request links, and a completion callback
+  bearing the operation token, so the Update's completion is delivered back to the Nexus caller. Only
+  asynchronous, `ACCEPTED`-stage updates are supported (a callback URL is required); an update that
+  has already completed is returned synchronously, and a completed-with-error update (e.g. a
+  validation rejection) surfaces as a failed Nexus operation. Cancellation is customizable via the
+  `cancelWorkflowUpdate` handler option; the default rejects with a `NOT_IMPLEMENTED` handler error.
 - Nexus operation link propagation for signals. When a Nexus operation handler signals a workflow
   (including signal-with-start), the inbound Nexus request links are now forwarded onto the signaled
   workflow so its history events link back to the caller, and the link the server returns for the

packages/client/src/internal.ts

@@ -49,6 +49,51 @@ export interface InternalWorkflowStartOptions extends WorkflowOptions {
   };
 }
 
+/**
+ * A symbol used to attach extra, SDK-internal options to a `WorkflowHandle.startUpdate()` /
+ * `executeUpdate()` call.
+ *
+ * Used by the Temporal Nexus helpers to attach a completion callback, request ID, and request links
+ * onto the UpdateWorkflowExecution request, and to capture the response link the server returns.
+ *
+ * @internal
+ * @hidden
+ */
+export const InternalWorkflowUpdateOptionsSymbol = Symbol.for('__temporal_internal_client_workflow_update_options');
+export interface InternalWorkflowUpdateOptions {
+  [InternalWorkflowUpdateOptionsSymbol]?: {
+    /**
+     * Request ID used for server-side deduplication of the Update request.
+     */
+    requestId?: string;
+
+    /**
+     * Callbacks to be invoked by the server when the Update reaches a terminal state.
+     * Callback addresses must be whitelisted in the server's dynamic configuration.
+     */
+    completionCallbacks?: temporal.api.common.v1.ICallback[];
+
+    /**
+     * Links to be associated with the Update.
+     */
+    links?: temporal.api.common.v1.ILink[];
+
+    /**
+     * Response link copied by the client from the UpdateWorkflowExecutionResponse. On success it
+     * points at the Update Accepted event; on validation failure it may be a plain Workflow link.
+     * Only populated by servers that return an update response link; left unset otherwise.
+     */
+    responseLink?: temporal.api.common.v1.ILink;
+
+    /**
+     * Terminal outcome of the Update, copied by the client from the UpdateWorkflowExecutionResponse
+     * when the Update already reached a completed state (e.g. a retried request with the same update
+     * ID, or an Update that completes immediately). Left unset while the Update is only accepted.
+     */
+    outcome?: temporal.api.update.v1.IOutcome;
+  };
+}
+
 /**
  * A symbol used to attach extra, SDK-internal options to a `WorkflowHandle.signal()` call.
  *

packages/client/src/workflow-client.ts

@@ -94,7 +94,12 @@ import { BaseClient, defaultBaseClientOptions } from './base-client';
 import { mapAsyncIterable } from './iterators-utils';
 import { WorkflowUpdateStage, encodeWorkflowUpdateStage } from './workflow-update-stage';
 import type { InternalWorkflowHandle, InternalWorkflowSignalInput, InternalWorkflowStartOptions } from './internal';
-import { InternalWorkflowSignalOptionsSymbol, InternalWorkflowStartOptionsSymbol } from './internal';
+import {
+  InternalWorkflowSignalOptionsSymbol,
+  InternalWorkflowStartOptionsSymbol,
+  type InternalWorkflowUpdateOptions,
+  InternalWorkflowUpdateOptionsSymbol,
+} from './internal';
 import { adaptWorkflowClientInterceptor } from './interceptor-adapters';
 
 const UpdateWorkflowExecutionLifecycleStage = temporal.api.enums.v1.UpdateWorkflowExecutionLifecycleStage;
@@ -994,6 +999,9 @@ export class WorkflowClient extends BaseClient {
     const dataConverter = this.dataConverter;
     const context = this.workflowSerializationContext(input.workflowExecution.workflowId!);
     const updateId = input.options?.updateId ?? randomUUID();
+    const internalOptions = (input.options as InternalWorkflowUpdateOptions | undefined)?.[
+      InternalWorkflowUpdateOptionsSymbol
+    ];
     return {
       namespace: this.options.namespace,
       workflowExecution: input.workflowExecution,
@@ -1011,6 +1019,9 @@ export class WorkflowClient extends BaseClient {
           name: input.updateName,
           args: { payloads: await encodeToPayloadsWithContext(dataConverter, context, input.args) },
         },
+        requestId: internalOptions?.requestId,
+        completionCallbacks: internalOptions?.completionCallbacks,
+        links: internalOptions?.links,
       },
     };
   }
@@ -1048,6 +1059,17 @@ export class WorkflowClient extends BaseClient {
     } catch (err) {
       this.rethrowUpdateGrpcError(err, 'Workflow Update failed', input.workflowExecution);
     }
+    const internalOptions = (input.options as InternalWorkflowUpdateOptions | undefined)?.[
+      InternalWorkflowUpdateOptionsSymbol
+    ];
+    if (internalOptions != null) {
+      // Capture the link the server attached to the Update response so the Nexus helper can add it
+      // as a handler link. Older servers leave it unset.
+      internalOptions.responseLink = response.link ?? undefined;
+      // Capture the terminal outcome (if any) so the Nexus helper can distinguish an already-completed
+      // Update (return a synchronous result) from one that is merely accepted (return async).
+      internalOptions.outcome = response.outcome ?? undefined;
+    }
     return {
       updateId: request.request!.meta!.updateId!,
 

packages/nexus/src/__tests__/test-nexus-link-converter.ts

@@ -2,16 +2,19 @@ import test from 'ava';
 import Long from 'long';
 import { temporal } from '@temporalio/proto';
 import {
+  convertCommonLinkToNexusLink,
   convertNexusLinkToTemporalLink,
   convertNexusLinkToWorkflowEventLink,
   convertNexusOperationLinkToNexusLink,
   convertTemporalLinkToNexusLink,
   convertWorkflowEventLinkToNexusLink,
+  convertWorkflowLinkToNexusLink,
 } from '../link-converter';
 
 const { EventType } = temporal.api.enums.v1;
 const WORKFLOW_EVENT_TYPE = (temporal.api.common.v1.Link.WorkflowEvent as any).fullName.slice(1);
 const NEXUS_OPERATION_TYPE = (temporal.api.common.v1.Link.NexusOperation as any).fullName.slice(1);
+const WORKFLOW_TYPE = (temporal.api.common.v1.Link.Workflow as any).fullName.slice(1);
 
 function makeEventRef(eventId: number, eventType: keyof typeof EventType) {
   return {
@@ -107,6 +110,47 @@ test('convertNexusOperationLinkToNexusLink escapes URL path components', (t) =>
   });
 });
 
+test('convertWorkflowLinkToNexusLink produces a history URL with the Workflow link type', (t) => {
+  const nexusLink = convertWorkflowLinkToNexusLink({
+    namespace: 'ns',
+    workflowId: 'wid',
+    runId: 'rid',
+  });
+  t.is(nexusLink.type, WORKFLOW_TYPE);
+  t.is(nexusLink.url.toString(), 'temporal:///namespaces/ns/workflows/wid/rid/history');
+});
+
+test('convertWorkflowLinkToNexusLink escapes URL path components', (t) => {
+  const nexusLink = convertWorkflowLinkToNexusLink({
+    namespace: 'name/space',
+    workflowId: 'work id',
+    runId: 'run/id',
+  });
+  t.is(nexusLink.url.toString(), 'temporal:///namespaces/name%2Fspace/workflows/work%20id/run%2Fid/history');
+});
+
+test('convertWorkflowLinkToNexusLink throws on missing required fields', (t) => {
+  t.throws(() => convertWorkflowLinkToNexusLink({ namespace: '', workflowId: 'wid', runId: 'rid' }), {
+    instanceOf: TypeError,
+  });
+  t.throws(() => convertWorkflowLinkToNexusLink({ namespace: 'ns', workflowId: '', runId: 'rid' }), {
+    instanceOf: TypeError,
+  });
+});
+
+test('convertCommonLinkToNexusLink dispatches by variant, preferring workflowEvent', (t) => {
+  const workflowEvent = {
+    namespace: 'ns',
+    workflowId: 'wid',
+    runId: 'rid',
+    eventRef: makeEventRef(42, 'EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ACCEPTED'),
+  };
+  t.is(convertCommonLinkToNexusLink({ workflowEvent }).type, WORKFLOW_EVENT_TYPE);
+
+  const workflowLink = { namespace: 'ns', workflowId: 'wid', runId: 'rid' };
+  t.is(convertCommonLinkToNexusLink({ workflow: workflowLink }).type, WORKFLOW_TYPE);
+});
+
 test('convertTemporalLinkToNexusLink dispatches by Temporal link variant', (t) => {
   const workflowEvent = {
     namespace: 'ns',

packages/nexus/src/__tests__/test-nexus-token-helpers.ts

@@ -1,8 +1,10 @@
 import test from 'ava';
 import {
   base64URLEncodeNoPadding,
+  generateUpdateWorkflowOperationToken,
   generateWorkflowRunOperationToken,
   loadOperationToken,
+  loadUpdateWorkflowOperationToken,
   loadWorkflowRunOperationToken,
 } from '../token';
 
@@ -47,3 +49,69 @@ test('decode workflow run Operation token errors', (t) => {
     message: /invalid workflow run token: missing workflow ID \(wid\)/,
   });
 });
+
+test('encode and decode update workflow Operation token', (t) => {
+  const expected = {
+    t: 3,
+    ns: 'ns',
+    wid: 'w',
+    rid: 'r',
+    uid: 'u',
+  };
+  const token = generateUpdateWorkflowOperationToken('ns', 'w', 'r', 'u');
+  const decoded = loadUpdateWorkflowOperationToken(token);
+  t.deepEqual(decoded, expected);
+});
+
+test('update workflow Operation token matches Go wire format byte-for-byte', (t) => {
+  // Field order and names must match the Go SDK's json.Marshal output: t, ns, wid, rid, uid.
+  // The version field ("v") must be absent.
+  const token = generateUpdateWorkflowOperationToken('ns', 'w', 'r', 'u');
+  const decodedJSON = Buffer.from(token, 'base64url').toString('utf-8');
+  t.is(decodedJSON, '{"t":3,"ns":"ns","wid":"w","rid":"r","uid":"u"}');
+});
+
+test('encode update workflow Operation token with empty runId omits rid', (t) => {
+  const token = generateUpdateWorkflowOperationToken('ns', 'w', '', 'u');
+  const decodedJSON = Buffer.from(token, 'base64url').toString('utf-8');
+  // `rid` is optional: it must be omitted entirely rather than emitted as an empty string.
+  t.is(decodedJSON, '{"t":3,"ns":"ns","wid":"w","uid":"u"}');
+
+  const parsed = JSON.parse(decodedJSON);
+  t.false('rid' in parsed);
+
+  // Round-trips: the load side tolerates an absent rid, yielding no pinned run.
+  const decoded = loadUpdateWorkflowOperationToken(token);
+  t.deepEqual(decoded, { t: 3, ns: 'ns', wid: 'w', uid: 'u' });
+  t.is(decoded.rid, undefined);
+});
+
+test('load update workflow Operation token tolerates legacy empty rid', (t) => {
+  // Legacy tokens may carry rid:"" explicitly; it must be accepted and treated as no pinned run.
+  const legacyToken = base64URLEncodeNoPadding('{"t":3,"ns":"ns","wid":"w","rid":"","uid":"u"}');
+  const decoded = loadUpdateWorkflowOperationToken(legacyToken);
+  t.deepEqual(decoded, { t: 3, ns: 'ns', wid: 'w', rid: '', uid: 'u' });
+});
+
+test('generate update workflow Operation token validates required params', (t) => {
+  t.throws(() => generateUpdateWorkflowOperationToken('', 'w', 'r', 'u'), { message: /namespace/ });
+  t.throws(() => generateUpdateWorkflowOperationToken('ns', '', 'r', 'u'), { message: /workflow ID/ });
+  t.throws(() => generateUpdateWorkflowOperationToken('ns', 'w', 'r', ''), { message: /update ID/ });
+});
+
+test('decode update workflow Operation token errors', (t) => {
+  const wrongTypeToken = base64URLEncodeNoPadding('{"t":1,"ns":"ns","wid":"w"}');
+  t.throws(() => loadUpdateWorkflowOperationToken(wrongTypeToken), {
+    message: /invalid update workflow token type: 1, expected: 3/,
+  });
+
+  const missingWIDToken = base64URLEncodeNoPadding('{"t":3,"ns":"ns","uid":"u"}');
+  t.throws(() => loadUpdateWorkflowOperationToken(missingWIDToken), {
+    message: /invalid update workflow token: missing workflow ID \(wid\)/,
+  });
+
+  const missingUIDToken = base64URLEncodeNoPadding('{"t":3,"ns":"ns","wid":"w"}');
+  t.throws(() => loadUpdateWorkflowOperationToken(missingUIDToken), {
+    message: /invalid update workflow token: missing update ID \(uid\)/,
+  });
+});

packages/nexus/src/__tests__/test-nexus-update-cancel.ts

@@ -0,0 +1,61 @@
+import test from 'ava';
+import * as nexus from 'nexus-rpc';
+import { TemporalOperationHandler, TemporalOperationResult } from '../workflow-helpers';
+import { generateUpdateWorkflowOperationToken } from '../token';
+
+const cancelContext: nexus.CancelOperationContext = {
+  abortSignal: new AbortController().signal,
+  headers: {},
+  operation: 'operation',
+  service: 'service',
+};
+
+test('cancel of an UpdateWorkflow operation without a custom handler throws NOT_IMPLEMENTED', async (t) => {
+  const handler = new TemporalOperationHandler({
+    async start() {
+      return TemporalOperationResult.sync(undefined);
+    },
+  });
+  const token = generateUpdateWorkflowOperationToken('ns', 'wid', 'rid', 'uid');
+
+  const err = await t.throwsAsync(() => handler.cancel(cancelContext, token));
+  t.true(err instanceof nexus.HandlerError);
+  t.is((err as nexus.HandlerError).type, 'NOT_IMPLEMENTED');
+  t.regex(err?.message ?? '', /cannot cancel an UpdateWorkflow operation/);
+});
+
+test('cancel of an UpdateWorkflow operation routes to the custom cancelWorkflowUpdate handler', async (t) => {
+  let received: { workflowId: string; runId: string; updateId: string } | undefined;
+  const handler = new TemporalOperationHandler({
+    async start() {
+      return TemporalOperationResult.sync(undefined);
+    },
+    async cancelWorkflowUpdate(_ctx, options) {
+      received = options;
+    },
+    async cancelWorkflowRun() {
+      throw new Error('cancelWorkflowRun should not be called for an UpdateWorkflow token');
+    },
+  });
+  const token = generateUpdateWorkflowOperationToken('ns', 'wid', 'rid', 'uid');
+
+  await handler.cancel(cancelContext, token);
+  t.deepEqual(received, { workflowId: 'wid', runId: 'rid', updateId: 'uid' });
+});
+
+test('cancel rejects a malformed UpdateWorkflow operation token', async (t) => {
+  const handler = new TemporalOperationHandler({
+    async start() {
+      return TemporalOperationResult.sync(undefined);
+    },
+    async cancelWorkflowUpdate() {
+      throw new Error('cancelWorkflowUpdate should not be called for an invalid token');
+    },
+  });
+  // Type 3 (UpdateWorkflow) but missing the required update ID.
+  const token = Buffer.from(JSON.stringify({ t: 3, ns: 'ns', wid: 'wid' })).toString('base64url');
+
+  const err = await t.throwsAsync(() => handler.cancel(cancelContext, token));
+  t.true(err instanceof nexus.HandlerError);
+  t.is((err as nexus.HandlerError).type, 'BAD_REQUEST');
+});

packages/nexus/src/__tests__/test-nexus-update-workflow.ts

@@ -0,0 +1,60 @@
+import test from 'ava';
+import * as nexus from 'nexus-rpc';
+import { WorkflowUpdateStage } from '@temporalio/client';
+import { TemporalOperationHandler } from '../workflow-helpers';
+
+/**
+ * Builds a minimal {@link nexus.StartOperationContext} for driving a {@link TemporalOperationHandler}
+ * start handler directly. The `updateWorkflow` input-validation guards run before any Temporal Client
+ * or handler context is touched, so these paths can be exercised without a live worker.
+ */
+function makeStartContext(overrides: Partial<nexus.StartOperationContext> = {}): nexus.StartOperationContext {
+  return {
+    service: 'service',
+    operation: 'operation',
+    headers: {},
+    abortSignal: new AbortController().signal,
+    requestId: 'request-id',
+    inboundLinks: [],
+    outboundLinks: [],
+    ...overrides,
+  };
+}
+
+test('updateWorkflow rejects a non-ACCEPTED waitForStage as a failed operation', async (t) => {
+  const handler = new TemporalOperationHandler<undefined, number>({
+    async start(_ctx, client) {
+      return await client.updateWorkflow<number>({
+        workflowId: 'wid',
+        update: 'someUpdate',
+        // Only ACCEPTED is supported for async Nexus Update operations.
+        waitForStage: WorkflowUpdateStage.COMPLETED,
+      });
+    },
+  });
+
+  // A callback URL is present so the failure is attributable to the wait stage, not the missing URL.
+  const err = await t.throwsAsync(() =>
+    handler.start(makeStartContext({ callbackUrl: 'http://callback' }), undefined)
+  );
+  t.true(err instanceof nexus.OperationError);
+  t.is((err as nexus.OperationError).state, 'failed');
+  t.regex(err?.message ?? '', /ACCEPTED wait stage/);
+});
+
+test('updateWorkflow without a callback URL fails with a BAD_REQUEST handler error', async (t) => {
+  const handler = new TemporalOperationHandler<undefined, number>({
+    async start(_ctx, client) {
+      return await client.updateWorkflow<number>({
+        workflowId: 'wid',
+        update: 'someUpdate',
+        // waitForStage defaults to ACCEPTED, so the missing callback URL is what fails the call.
+      });
+    },
+  });
+
+  const err = await t.throwsAsync(() => handler.start(makeStartContext(), undefined));
+  t.true(err instanceof nexus.HandlerError);
+  t.is((err as nexus.HandlerError).type, 'BAD_REQUEST');
+  t.regex(err?.message ?? '', /callback URL is required/);
+});

packages/nexus/src/index.ts

@@ -19,6 +19,8 @@ export {
   startWorkflow,
   signalWithStartWorkflow,
   type CancelWorkflowRunOptions,
+  type CancelWorkflowUpdateOptions,
+  type NexusUpdateWorkflowOptions,
   type TemporalOperationHandlerOptions,
   TemporalOperationHandler,
   TemporalOperationResult,