WorkflowRunOperation

Author: bergundy

packages/client/src/internal.ts

@@ -0,0 +1,26 @@
+import { temporal } from '@temporalio/proto';
+
+// A key used internally to pass "hidden options to the WorkflowClient.start() call.
+export const InternalWorkflowStartOptionsKey = Symbol.for('__temporal_client_internal_workflow_start_options');
+export interface InternalWorkflowStartOptions {
+  requestId?: string;
+  /**
+   * Callbacks to be called by the server when this workflow reaches a terminal state.
+   * If the workflow continues-as-new, these callbacks will be carried over to the new execution.
+   * Callback addresses must be whitelisted in the server's dynamic configuration.
+   */
+  completionCallbacks?: temporal.api.common.v1.ICallback[];
+  /** Links to be associated with the workflow. */
+  links?: temporal.api.common.v1.ILink[];
+  /**
+   * Backlink copied by the client from the StartWorkflowExecutionResponse. Only populated in servers newer than 1.27.
+   */
+  backLink?: temporal.api.common.v1.ILink;
+
+  /**
+   * Conflict options for when USE_EXISTING is specified.
+   *
+   * Used by the nexus WorkflowRunOperations to attach to a callback to a running workflow.
+  */
+  onConflictOptions?: temporal.api.workflow.v1.IOnConflictOptions;
+}

packages/client/src/workflow-client.ts

@@ -91,6 +91,7 @@ import {
 } from './base-client';
 import { mapAsyncIterable } from './iterators-utils';
 import { WorkflowUpdateStage, encodeWorkflowUpdateStage } from './workflow-update-stage';
+import { InternalWorkflowStartOptions, InternalWorkflowStartOptionsKey } from './internal';
 
 const UpdateWorkflowExecutionLifecycleStage = temporal.api.enums.v1.UpdateWorkflowExecutionLifecycleStage;
 
@@ -1251,7 +1252,12 @@ export class WorkflowClient extends BaseClient {
     const req = await this.createStartWorkflowRequest(input);
     const { options: opts, workflowType } = input;
     try {
-      return (await this.workflowService.startWorkflowExecution(req)).runId;
+      const response = await this.workflowService.startWorkflowExecution(req);
+      const internalOptions = (opts as any)[InternalWorkflowStartOptionsKey] as InternalWorkflowStartOptions | undefined;
+      if (internalOptions != null) {
+        internalOptions.backLink = response.link ?? undefined;
+      }
+      return response.runId;
     } catch (err: any) {
       if (err.code === grpcStatus.ALREADY_EXISTS) {
         throw new WorkflowExecutionAlreadyStartedError(
@@ -1267,11 +1273,13 @@ export class WorkflowClient extends BaseClient {
   protected async createStartWorkflowRequest(input: WorkflowStartInput): Promise<StartWorkflowExecutionRequest> {
     const { options: opts, workflowType, headers } = input;
     const { identity, namespace } = this.options;
+    const internalOptions = (opts as any)[InternalWorkflowStartOptionsKey] as InternalWorkflowStartOptions | undefined;
 
+    console.log(internalOptions);
     return {
       namespace,
       identity,
-      requestId: uuid4(),
+      requestId: internalOptions?.requestId ?? uuid4(),
       workflowId: opts.workflowId,
       workflowIdReusePolicy: encodeWorkflowIdReusePolicy(opts.workflowIdReusePolicy),
       workflowIdConflictPolicy: encodeWorkflowIdConflictPolicy(opts.workflowIdConflictPolicy),
@@ -1296,6 +1304,7 @@ export class WorkflowClient extends BaseClient {
       cronSchedule: opts.cronSchedule,
       header: { fields: headers },
       priority: opts.priority ? compilePriority(opts.priority) : undefined,
+      ...internalOptions,
     };
   }
 

packages/nexus/src/context.ts

@@ -1,10 +1,16 @@
-import { HandlerContext as BaseHandlerContext, getHandlerContext } from 'nexus-rpc/lib/handler';
-import { Logger, LogLevel, LogMetadata } from '@temporalio/common';
-import { Client } from '@temporalio/client';
+import * as nexus from 'nexus-rpc';
+import { HandlerContext as BaseHandlerContext, getHandlerContext, handlerLinks } from 'nexus-rpc/lib/handler';
+import { Logger, LogLevel, LogMetadata, Workflow } from '@temporalio/common';
+import { Client, WorkflowStartOptions } from '@temporalio/client';
+import { temporal } from '@temporalio/proto';
+import { InternalWorkflowStartOptionsKey, InternalWorkflowStartOptions } from '@temporalio/client/lib/internal';
+import { generateWorkflowRunOperationToken, loadWorkflowRunOperationToken } from './token';
+import { convertNexusLinkToWorkflowEventLink, convertWorkflowEventLinkToNexusLink } from './link-converter';
 
 export interface HandlerContext extends BaseHandlerContext {
   log: Logger;
   client: Client;
+  namespace: string;
 }
 
 function getLogger() {
@@ -35,6 +41,8 @@ export const log: Logger = {
   },
 };
 
+// TODO: also support getting a metrics handler.
+
 /**
  * Returns a client to be used in a Nexus Operation's context, this Client is powered by the same Connection that the
  * worker was created with.
@@ -43,4 +51,74 @@ export function getClient(): Client {
   return getHandlerContext<HandlerContext>().client;
 }
 
-// TODO: also support getting a metrics handler.
+export interface WorkflowHandle<_T> {
+  readonly workflowId: string;
+  readonly runId: string;
+}
+
+export async function startWorkflow<T extends Workflow>(workflowTypeOrFunc: string | T, workflowOptions: WorkflowStartOptions<T>, nexusOptions: nexus.StartOperationOptions): Promise<WorkflowHandle<T>> {
+  const links = Array<temporal.api.common.v1.ILink>();
+  if (nexusOptions.links?.length > 0) {
+    for (const l of nexusOptions.links) {
+      try {
+        links.push({
+          workflowEvent: convertNexusLinkToWorkflowEventLink(l),
+        });
+      } catch (error) {
+        log.warn('failed to convert Nexus link to Workflow event link', { error });
+      }
+    }
+  }
+  const internalOptions: InternalWorkflowStartOptions = { links, requestId: nexusOptions.requestId };
+
+  if (workflowOptions.workflowIdConflictPolicy === 'USE_EXISTING') {
+    internalOptions.onConflictOptions = {
+      attachLinks: true,
+      attachCompletionCallbacks: true,
+      attachRequestId: true,
+    };
+  }
+
+  if (nexusOptions.callbackURL) {
+    internalOptions.completionCallbacks = [
+      {
+        nexus: {url: nexusOptions.callbackURL, header: nexusOptions.callbackHeaders},
+        links, // pass in links here as well, the server dedupes them.
+      },
+    ];
+  }
+  (workflowOptions as any)[InternalWorkflowStartOptionsKey] = internalOptions;
+  const handle = await getClient().workflow.start<T>(workflowTypeOrFunc, workflowOptions);
+  if (internalOptions.backLink?.workflowEvent != null) {
+    try {
+      handlerLinks().push(convertWorkflowEventLinkToNexusLink(internalOptions.backLink.workflowEvent));
+    } catch (error) {
+        log.warn('failed to convert Workflow event link to Nexus link', { error });
+    }
+  }
+  return { workflowId: handle.workflowId, runId: handle.firstExecutionRunId };
+}
+
+export type WorkflowRunOperationHandler<I, O> = (input: I, options: nexus.StartOperationOptions) => Promise<WorkflowHandle<O>>;
+
+export class WorkflowRunOperation<I, O> implements nexus.OperationHandler<I, O> {
+  constructor(readonly handler: WorkflowRunOperationHandler<I, O>) { }
+
+  async start(input: I, options: nexus.StartOperationOptions): Promise<nexus.HandlerStartOperationResult<O>> {
+    const { namespace } = getHandlerContext<HandlerContext>();
+    const handle = await this.handler(input, options);
+    return { token: generateWorkflowRunOperationToken(namespace, handle.workflowId) };
+  }
+  getResult(_token: string, _options: nexus.GetOperationResultOptions): Promise<O> {
+    // Not implemented in Temporal yet.
+    throw new nexus.HandlerError({ type: 'NOT_IMPLEMENTED', message: 'Method not implemented' });
+  }
+  getInfo(_token: string, _options: nexus.GetOperationInfoOptions): Promise<nexus.OperationInfo> {
+    // Not implemented in Temporal yet.
+    throw new nexus.HandlerError({ type: 'NOT_IMPLEMENTED', message: 'Method not implemented' });
+  }
+  async cancel(token: string, _options: nexus.CancelOperationOptions): Promise<void> {
+    const decoded = loadWorkflowRunOperationToken(token);
+    await getClient().workflow.getHandle(decoded.wid).cancel();
+  }
+}

packages/nexus/src/index.ts

@@ -1 +1,8 @@
-export { log, getClient } from './context';
+export {
+  log,
+  getClient,
+  startWorkflow,
+  WorkflowHandle,
+  WorkflowRunOperation,
+  WorkflowRunOperationHandler,
+} from './context';

packages/worker/src/nexus/link-converter.ts renamed to packages/nexus/src/link-converter.ts

@@ -0,0 +1,78 @@
+const OPERATION_TOKEN_TYPE_WORKFLOW_RUN = 1;
+
+// OperationTokenType is used to identify the type of operation token.
+// Currently, we only have one type of operation token: WorkflowRun.
+type OperationTokenType = typeof OPERATION_TOKEN_TYPE_WORKFLOW_RUN;
+
+interface WorkflowRunOperationToken {
+  // Version of the token, by default we assume we're on version 1, this field is not emitted as part of the output,
+  // it's only used to reject newer token versions on load.
+  v?: number;
+  // Type of the operation. Must be OPERATION_TOKEN_TYPE_WORKFLOW_RUN.
+  t: OperationTokenType;
+  ns: string;
+  wid: string;
+}
+
+/**
+ * Generate a workflow run operation token.
+ */
+export function generateWorkflowRunOperationToken(namespace: string, workflowID: string): string {
+  const token: WorkflowRunOperationToken = {
+    t: OPERATION_TOKEN_TYPE_WORKFLOW_RUN,
+    ns: namespace,
+    wid: workflowID,
+  };
+  return base64URLEncodeNoPadding(JSON.stringify(token));
+}
+
+/**
+ * Load and validate a workflow run operation token.
+ */
+export function loadWorkflowRunOperationToken(data: string): WorkflowRunOperationToken {
+  if (!data) {
+    throw new TypeError('invalid workflow run token: token is empty');
+  }
+  let decoded: string;
+  try {
+    decoded = base64URLDecodeNoPadding(data);
+  } catch (err) {
+    throw new TypeError('failed to decode token', {cause: err});
+  }
+  let token: WorkflowRunOperationToken;
+  try {
+    token = JSON.parse(decoded);
+  } catch (err) {
+    throw new TypeError('failed to unmarshal workflow run operation token', {cause: err});
+  }
+  if (token.t !== OPERATION_TOKEN_TYPE_WORKFLOW_RUN) {
+    throw new TypeError(`invalid workflow token type: ${token.t}, expected: ${OPERATION_TOKEN_TYPE_WORKFLOW_RUN}`);
+  }
+  if (token.v !== undefined && token.v !== 0) {
+    throw new TypeError('invalid workflow run token: "v" field should not be present');
+  }
+  if (!token.wid) {
+    throw new TypeError('invalid workflow run token: missing workflow ID (wid)');
+  }
+  return token;
+}
+
+// Exported for use in tests.
+export function base64URLEncodeNoPadding(str: string): string {
+  const base64 = Buffer.from(str).toString('base64url');
+  return base64.replace(/=+$/, '');
+}
+
+function base64URLDecodeNoPadding(str: string): string {
+  // Validate the string contains only valid base64URL characters
+  if (!/^[-A-Za-z0-9_]*$/.test(str)) {
+    throw new TypeError('invalid base64URL encoded string: contains invalid characters');
+  }
+  
+  const paddingLength = str.length % 4;
+  if (paddingLength > 0) {
+    str += '='.repeat(4 - paddingLength);
+  }
+  
+  return Buffer.from(str, 'base64url').toString('utf-8');
+}