feat(nexus): Initial Nexus support (#1708)

Co-authored-by: James Watkins-Harvey <[email protected]>

Author: bergundy

package-lock.json

@@ -43,12 +43,14 @@
     "@temporalio/common": "file:packages/common",
     "@temporalio/create": "file:packages/create-project",
     "@temporalio/interceptors-opentelemetry": "file:packages/interceptors-opentelemetry",
+    "@temporalio/nexus": "file:packages/nexus",
     "@temporalio/nyc-test-coverage": "file:packages/nyc-test-coverage",
     "@temporalio/proto": "file:packages/proto",
     "@temporalio/test": "file:packages/test",
     "@temporalio/testing": "file:packages/testing",
     "@temporalio/worker": "file:packages/worker",
     "@temporalio/workflow": "file:packages/workflow",
+    "nexus-rpc": "^0.0.1",
     "temporalio": "file:packages/meta"
   },
   "devDependencies": {
@@ -86,6 +88,7 @@
     "packages/core-bridge",
     "packages/create-project",
     "packages/interceptors-opentelemetry",
+    "packages/nexus",
     "packages/nyc-test-coverage",
     "packages/proto",
     "packages/test",

package.json

@@ -0,0 +1,45 @@
+import { temporal } from '@temporalio/proto';
+import { WorkflowOptions } from './workflow-options';
+
+/**
+ * A symbol used to attach extra, SDK-internal options to the `WorkflowClient.start()` call.
+ *
+ * These are notably used by the Temporal Nexus helpers.
+ *
+ * @internal
+ * @hidden
+ */
+export const InternalWorkflowStartOptionsSymbol = Symbol.for('__temporal_internal_client_workflow_start_options');
+export interface InternalWorkflowStartOptions extends WorkflowOptions {
+  [InternalWorkflowStartOptionsSymbol]?: {
+    /**
+     * Request ID to be used for the workflow.
+     */
+    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/internal.ts

@@ -74,15 +74,15 @@ export type WorkflowExecutionDescription = Replace<
    * General fixed details for this workflow execution that may appear in UI/CLI.
    * This can be in Temporal markdown format and can span multiple lines.
    *
-   * @experimental User metadata is a new API and suspectible to change.
+   * @experimental User metadata is a new API and susceptible to change.
    */
   staticDetails: () => Promise<string | undefined>;
 
   /**
    * A single-line fixed summary for this workflow execution that may appear in the UI/CLI.
    * This can be in single-line Temporal markdown format.
    *
-   * @experimental User metadata is a new API and suspectible to change.
+   * @experimental User metadata is a new API and susceptible to change.
    */
   staticSummary: () => Promise<string | undefined>;
 };

packages/client/src/types.ts

@@ -93,6 +93,7 @@ import {
 } from './base-client';
 import { mapAsyncIterable } from './iterators-utils';
 import { WorkflowUpdateStage, encodeWorkflowUpdateStage } from './workflow-update-stage';
+import { InternalWorkflowStartOptionsSymbol, InternalWorkflowStartOptions } from './internal';
 
 const UpdateWorkflowExecutionLifecycleStage = temporal.api.enums.v1.UpdateWorkflowExecutionLifecycleStage;
 
@@ -1248,8 +1249,13 @@ export class WorkflowClient extends BaseClient {
   protected async _startWorkflowHandler(input: WorkflowStartInput): Promise<string> {
     const req = await this.createStartWorkflowRequest(input);
     const { options: opts, workflowType } = input;
+    const internalOptions = (opts as InternalWorkflowStartOptions)[InternalWorkflowStartOptionsSymbol];
     try {
-      return (await this.workflowService.startWorkflowExecution(req)).runId;
+      const response = await this.workflowService.startWorkflowExecution(req);
+      if (internalOptions != null) {
+        internalOptions.backLink = response.link ?? undefined;
+      }
+      return response.runId;
     } catch (err: any) {
       if (err.code === grpcStatus.ALREADY_EXISTS) {
         throw new WorkflowExecutionAlreadyStartedError(
@@ -1265,10 +1271,12 @@ 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 InternalWorkflowStartOptions)[InternalWorkflowStartOptionsSymbol];
+
     return {
       namespace,
       identity,
-      requestId: uuid4(),
+      requestId: internalOptions?.requestId ?? uuid4(),
       workflowId: opts.workflowId,
       workflowIdReusePolicy: encodeWorkflowIdReusePolicy(opts.workflowIdReusePolicy),
       workflowIdConflictPolicy: encodeWorkflowIdConflictPolicy(opts.workflowIdConflictPolicy),
@@ -1295,6 +1303,7 @@ export class WorkflowClient extends BaseClient {
       userMetadata: await encodeUserMetadata(this.dataConverter, opts.staticSummary, opts.staticDetails),
       priority: opts.priority ? compilePriority(opts.priority) : undefined,
       versioningOverride: opts.versioningOverride ?? undefined,
+      ...filterNullAndUndefined(internalOptions ?? {}),
     };
   }
 

packages/client/src/workflow-client.ts

@@ -15,6 +15,7 @@
     "@temporalio/proto": "file:../proto",
     "long": "^5.2.3",
     "ms": "^3.0.0-canary.1",
+    "nexus-rpc": "^0.0.1",
     "proto3-json-serializer": "^2.0.0"
   },
   "devDependencies": {

packages/common/package.json

@@ -129,7 +129,7 @@ export interface ActivityOptions {
    * A fixed, single-line summary for this workflow execution that may appear in the UI/CLI.
    * This can be in single-line Temporal markdown format.
    *
-   * @experimental User metadata is a new API and suspectible to change.
+   * @experimental User metadata is a new API and susceptible to change.
    */
   summary?: string;
 
@@ -206,7 +206,7 @@ export interface LocalActivityOptions {
    * A fixed, single-line summary for this workflow execution that may appear in the UI/CLI.
    * This can be in single-line Temporal markdown format.
    *
-   * @experimental User metadata is a new API and suspectible to change.
+   * @experimental User metadata is a new API and susceptible to change.
    */
   summary?: string;
 }