Send SANO links in requests (#2143)

Author: VegetarianOrc

CHANGELOG.md

@@ -30,3 +30,8 @@ to docs, or any other relevant information.
 ### Breaking Changes
 
 - `WorkflowHandle.runId` in `@temporalio/nexus` is now an optional property to support creating a handle using only a workflow ID.
+
+### Changed
+
+- Standalone Nexus operation links are now formatted to align with server side support.
+- Standalone Nexus operation links are now forwarded on start workflow and signal requests.

packages/client/src/workflow-client.ts

@@ -1212,8 +1212,6 @@ export class WorkflowClient extends BaseClient {
     const dataConverter = this.dataConverter;
     const context = this.workflowSerializationContext(input.workflowExecution.workflowId!);
     const internalOptions = (input as InternalWorkflowSignalInput)[InternalWorkflowSignalOptionsSymbol];
-    // Server currently only supports workflow_event and batch_job link types.
-    const links = internalOptions?.links?.filter((link) => link.workflowEvent != null || link.batchJob != null);
     const req: temporal.api.workflowservice.v1.ISignalWorkflowExecutionRequest = {
       identity: this.options.identity,
       namespace: this.options.namespace,
@@ -1223,7 +1221,7 @@ export class WorkflowClient extends BaseClient {
       signalName: input.signalName,
       header: { fields: input.headers },
       input: { payloads: await encodeToPayloadsWithContext(dataConverter, context, input.args) },
-      links,
+      links: internalOptions?.links,
     };
     try {
       const response = await this.workflowService.signalWorkflowExecution(req);
@@ -1248,8 +1246,6 @@ export class WorkflowClient extends BaseClient {
     const dataConverter = this.dataConverter;
     const context = this.workflowSerializationContext(options.workflowId);
     const internalOptions = (options as InternalWorkflowStartOptions)[InternalWorkflowStartOptionsSymbol];
-    // Server currently only supports workflow_event and batch_job link types.
-    const links = internalOptions?.links?.filter((link) => link.workflowEvent != null || link.batchJob != null);
     const req: temporal.api.workflowservice.v1.ISignalWithStartWorkflowExecutionRequest = {
       namespace: this.options.namespace,
       identity,
@@ -1282,7 +1278,7 @@ export class WorkflowClient extends BaseClient {
       userMetadata: await encodeUserMetadata(dataConverter, options.staticSummary, options.staticDetails, context),
       priority: options.priority ? compilePriority(options.priority) : undefined,
       versioningOverride: options.versioningOverride ?? undefined,
-      links,
+      links: internalOptions?.links,
     };
     try {
       const response = await this.workflowService.signalWithStartWorkflowExecution(req);
@@ -1350,19 +1346,6 @@ export class WorkflowClient extends BaseClient {
       );
     }
 
-    // Server currently only supports workflow_event and batch_job
-    // link types. This filter should be removed or adapted as
-    // server-side support comes online.
-    // See https://github.com/temporalio/temporal/issues/10345
-    const links = internalOptions?.links?.filter((link) => link.workflowEvent != null || link.batchJob != null);
-
-    const completionCallbacks = internalOptions?.completionCallbacks?.map((cb) => {
-      const links = cb?.links?.filter((link) => link.workflowEvent != null || link.batchJob != null);
-      return { ...cb, links };
-    });
-
-    const resolvedInternalOptions = { ...(internalOptions ?? {}), links, completionCallbacks };
-
     return {
       namespace,
       identity,
@@ -1394,7 +1377,7 @@ export class WorkflowClient extends BaseClient {
       priority: opts.priority ? compilePriority(opts.priority) : undefined,
       versioningOverride: opts.versioningOverride ?? undefined,
       requestEagerExecution: opts.requestEagerStart,
-      ...filterNullAndUndefined(resolvedInternalOptions),
+      ...filterNullAndUndefined(internalOptions ?? {}),
     };
   }
 

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

@@ -78,7 +78,7 @@ test('convertNexusOperationLinkToNexusLink and back with runId', (t) => {
 
   const nexusLink = convertNexusOperationLinkToNexusLink(opLink);
   t.is(nexusLink.type, NEXUS_OPERATION_TYPE);
-  t.is(nexusLink.url.toString(), 'temporal:///namespaces/ns/nexus-operations/op-123?runID=run-456');
+  t.is(nexusLink.url.toString(), 'temporal:///namespaces/ns/nexus-operations/op-123/run-456/details');
 
   const roundTrip = convertNexusLinkToTemporalLink(nexusLink);
   t.deepEqual(roundTrip, { nexusOperation: opLink });
@@ -88,17 +88,21 @@ test('convertNexusOperationLinkToNexusLink escapes URL path components', (t) =>
   const opLink = {
     namespace: 'name/space',
     operationId: 'operation id',
+    runId: 'run/id',
   };
 
   const nexusLink = convertNexusOperationLinkToNexusLink(opLink);
-  t.is(nexusLink.url.toString(), 'temporal:///namespaces/name%2Fspace/nexus-operations/operation%20id');
+  t.is(
+    nexusLink.url.toString(),
+    'temporal:///namespaces/name%2Fspace/nexus-operations/operation%20id/run%2Fid/details'
+  );
 
   const roundTrip = convertNexusLinkToTemporalLink(nexusLink);
   t.deepEqual(roundTrip, {
     nexusOperation: {
       namespace: 'name/space',
       operationId: 'operation id',
-      runId: null,
+      runId: 'run/id',
     },
   });
 });
@@ -181,6 +185,15 @@ test('throws on missing required fields', (t) => {
       }),
     { instanceOf: TypeError }
   );
+  t.throws(
+    () =>
+      convertNexusOperationLinkToNexusLink({
+        namespace: 'ns',
+        operationId: 'op-123',
+        runId: '',
+      }),
+    { instanceOf: TypeError }
+  );
 });
 
 test('throws on invalid URL scheme', (t) => {

packages/nexus/src/link-converter.ts

@@ -13,7 +13,6 @@ const LINK_EVENT_ID_PARAM = 'eventID';
 const LINK_EVENT_TYPE_PARAM = 'eventType';
 const LINK_REQUEST_ID_PARAM = 'requestID';
 const LINK_REFERENCE_TYPE_KEY = 'referenceType';
-const LINK_RUN_ID_KEY = 'runID';
 
 const EVENT_REFERENCE_TYPE = 'EventReference';
 const REQUEST_ID_REFERENCE_TYPE = 'RequestIdReference';
@@ -77,22 +76,16 @@ export function convertWorkflowEventLinkToNexusLink(we: WorkflowEventLink): Nexu
 }
 
 export function convertNexusOperationLinkToNexusLink(opLink: NexusOperationLink): NexusLink {
-  if (!opLink.namespace || !opLink.operationId) {
-    throw new TypeError('Missing required fields: namespace, or operationId');
+  if (!opLink.namespace || !opLink.operationId || !opLink.runId) {
+    throw new TypeError('Missing required fields: namespace, operationId, or runId');
   }
 
   const url = new URL(
     `temporal:///namespaces/${encodeURIComponent(opLink.namespace)}/nexus-operations/${encodeURIComponent(
       opLink.operationId
-    )}`
+    )}/${encodeURIComponent(opLink.runId)}/details`
   );
 
-  if (opLink.runId != null) {
-    const searchParams = new URLSearchParams();
-    searchParams.set(LINK_RUN_ID_KEY, opLink.runId);
-    url.search = searchParams.toString();
-  }
-
   return {
     url,
     type: NEXUS_OPERATION_TYPE,
@@ -132,16 +125,18 @@ export function convertNexusLinkToWorkflowEventLink(link: NexusLink): WorkflowEv
 }
 
 function convertNexusLinkToNexusOperationLink(link: NexusLink): NexusOperationLink {
-  // /namespaces/:namespace/nexus-operations/:operationId?runId=:runId
+  // /namespaces/:namespace/nexus-operations/:operationId/:runId/details
   const parts = link.url.pathname.split('/');
-  if (parts.length !== 5 || parts[1] !== 'namespaces' || parts[3] !== 'nexus-operations') {
+  if (parts.length !== 7 || parts[1] !== 'namespaces' || parts[3] !== 'nexus-operations' || parts[6] !== 'details') {
     throw new TypeError(`Invalid URL path: ${link.url}`);
   }
   const namespace = decodeURIComponent(parts[2]!);
   const operationId = decodeURIComponent(parts[4]!);
+  const runId = decodeURIComponent(parts[5]!);
 
-  const query = link.url.searchParams;
-  const runId = query.get(LINK_RUN_ID_KEY);
+  if (!namespace || !operationId || !runId) {
+    throw new TypeError('Missing required fields: namespace, operationId, or runId');
+  }
 
   return {
     namespace,

packages/test/src/test-nexus-standalone.ts

@@ -1,5 +1,7 @@
 import { randomUUID } from 'crypto';
+import type { ExecutionContext } from 'ava';
 import * as nexus from 'nexus-rpc';
+import { temporal } from '@temporalio/proto';
 import type { NexusOperationExecutionCount } from '@temporalio/client';
 import {
   Client,
@@ -20,9 +22,12 @@ import * as temporalnexus from '@temporalio/nexus';
 import * as workflow from '@temporalio/workflow';
 import { CancelledFailure, TerminatedFailure, ApplicationFailure, SearchAttributeType } from '@temporalio/common';
 import { generateWorkflowRunOperationToken } from '@temporalio/nexus/lib/token';
+import type { Context } from './helpers-integration';
 import { helpers, makeTestFunction } from './helpers-integration';
 import { waitUntil } from './helpers';
 
+const { EventType } = temporal.api.enums.v1;
+
 const test = makeTestFunction({
   workflowsPath: __filename,
   workflowEnvironmentOpts: {
@@ -97,6 +102,37 @@ function makeTestHandler() {
   };
 }
 
+function assertWorkflowCallbackLinksToNexusOperation(
+  t: ExecutionContext<Context>,
+  callbacks: temporal.api.workflow.v1.ICallbackInfo[] | null | undefined,
+  namespace: string,
+  operationId: string,
+  operationRunId: string
+): void {
+  const nexusOperationLink = callbacks
+    ?.flatMap((callbackInfo) => callbackInfo.callback?.links ?? [])
+    .find((link) => link.nexusOperation?.operationId === operationId)?.nexusOperation;
+  t.truthy(nexusOperationLink, 'expected workflow completion callback to link to the standalone Nexus operation');
+
+  t.is(nexusOperationLink?.namespace, namespace);
+  t.is(nexusOperationLink?.runId, operationRunId);
+}
+
+function assertOperationLinksToWorkflowStarted(
+  t: ExecutionContext<Context>,
+  links: temporal.api.common.v1.ILink[] | null | undefined,
+  namespace: string,
+  workflowId: string,
+  workflowRunId: string
+): void {
+  const workflowLink = links?.find((link) => link.workflowEvent?.workflowId === workflowId)?.workflowEvent;
+  t.truthy(workflowLink, 'expected Nexus operation to link to the started workflow');
+
+  t.is(workflowLink?.namespace, namespace);
+  t.is(workflowLink?.runId, workflowRunId);
+  t.is(workflowLink?.eventRef?.eventType, EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED);
+}
+
 test('start sync operation and get result', async (t) => {
   const { createWorker, registerNexusEndpoint } = helpers(t);
   const { endpointName } = await registerNexusEndpoint();
@@ -142,6 +178,22 @@ test('start async operation and poll result', async (t) => {
 
     const result = await handle.result();
     t.is(result, 'async-hello');
+
+    const [workflowDesc, operationDesc] = await Promise.all([wfHandle.describe(), handle.describe()]);
+    assertWorkflowCallbackLinksToNexusOperation(
+      t,
+      workflowDesc.raw.callbacks,
+      t.context.env.client.options.namespace,
+      operationDesc.operationId,
+      operationDesc.runId
+    );
+    assertOperationLinksToWorkflowStarted(
+      t,
+      operationDesc.raw.links,
+      t.context.env.client.options.namespace,
+      wfHandle.workflowId,
+      workflowDesc.runId
+    );
   });
 });