Nexus handler near complete implementation

- This PR is blocked on publishing an initial version of the `nexus-rpc` package.
- There is a TODO to figure out HandlerError and OperationError message rehydration, pending discussion.
- Interceptors not yet implemented.
- `WorkflowRunOperation` and `getClient()` not implemented for the `@temporalio/nexus` package.
- Tests use the HTTP API directly in lieu of a workflow caller or strongly typed client, we can refactor those later.

Author: bergundy

package-lock.json

@@ -43,6 +43,7 @@
     "@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",

package.json

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

packages/common/package.json

@@ -1,3 +1,6 @@
+import * as nexus from 'nexus-rpc';
+import Long from 'long';
+import type { temporal } from '@temporalio/proto';
 import {
   ActivityFailure,
   ApplicationFailure,
@@ -8,16 +11,43 @@ import {
   encodeRetryState,
   encodeTimeoutType,
   FAILURE_SOURCE,
+  NexusOperationFailure,
   ProtoFailure,
   ServerFailure,
   TemporalFailure,
   TerminatedFailure,
   TimeoutFailure,
 } from '../failure';
+import { makeProtoEnumConverters } from '../internal-workflow';
 import { isError } from '../type-helpers';
 import { msOptionalToTs } from '../time';
 import { arrayFromPayloads, fromPayloadsAtIndex, PayloadConverter, toPayloads } from './payload-converter';
 
+// Can't import enums into the workflow sandbox, use this helper type and enum converter instead.
+const NexusHandlerErrorRetryBehavior = {
+  RETRYABLE: 'RETRYABLE',
+  NON_RETRYABLE: 'NON_RETRYABLE',
+} as const;
+
+type NexusHandlerErrorRetryBehavior =
+  (typeof NexusHandlerErrorRetryBehavior)[keyof typeof NexusHandlerErrorRetryBehavior];
+
+const [encodeNexusHandlerErrorRetryBehavior, decodeNexusHandlerErrorRetryBehavior] =
+  makeProtoEnumConverters<
+    temporal.api.enums.v1.NexusHandlerErrorRetryBehavior,
+    typeof temporal.api.enums.v1.NexusHandlerErrorRetryBehavior,
+    keyof typeof temporal.api.enums.v1.NexusHandlerErrorRetryBehavior,
+    typeof NexusHandlerErrorRetryBehavior,
+    'NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_'
+  >(
+    {
+      UNSPECIFIED: 0,
+      [NexusHandlerErrorRetryBehavior.RETRYABLE]: 1,
+      [NexusHandlerErrorRetryBehavior.NON_RETRYABLE]: 2,
+    } as const,
+    'NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_'
+  );
+
 function combineRegExp(...regexps: RegExp[]): RegExp {
   return new RegExp(regexps.map((x) => `(?:${x.source})`).join('|'));
 }
@@ -28,6 +58,8 @@ function combineRegExp(...regexps: RegExp[]): RegExp {
 const CUTOFF_STACK_PATTERNS = combineRegExp(
   /** Activity execution */
   /\s+at Activity\.execute \(.*[\\/]worker[\\/](?:src|lib)[\\/]activity\.[jt]s:\d+:\d+\)/,
+  /** Nexus execution */
+  /\s+at NexusHandler\.invokeUserCode \(.*[\\/]worker[\\/](?:src|lib)[\\/]nexus\.[jt]s:\d+:\d+\)/,
   /** Workflow activation */
   /\s+at Activator\.\S+NextHandler \(.*[\\/]workflow[\\/](?:src|lib)[\\/]internals\.[jt]s:\d+:\d+\)/,
   /** Workflow run anything in context */
@@ -120,7 +152,7 @@ export class DefaultFailureConverter implements FailureConverter {
    *
    * Does not set common properties, that is done in {@link failureToError}.
    */
-  failureToErrorInner(failure: ProtoFailure, payloadConverter: PayloadConverter): TemporalFailure {
+  failureToErrorInner(failure: ProtoFailure, payloadConverter: PayloadConverter): Error {
     if (failure.applicationFailureInfo) {
       return new ApplicationFailure(
         failure.message ?? undefined,
@@ -192,6 +224,38 @@ export class DefaultFailureConverter implements FailureConverter {
         this.optionalFailureToOptionalError(failure.cause, payloadConverter)
       );
     }
+    if (failure.nexusHandlerFailureInfo) {
+      if (failure.cause == null) {
+        throw new TypeError('Missing failure cause on nexusHandlerFailureInfo');
+      }
+      let retryable: boolean | undefined = undefined;
+      const retryBehavior = decodeNexusHandlerErrorRetryBehavior(failure.nexusHandlerFailureInfo.retryBehavior);
+      switch (retryBehavior) {
+        case 'RETRYABLE':
+          retryable = true;
+          break;
+        case 'NON_RETRYABLE':
+          retryable = false;
+          break;
+      }
+
+      return new nexus.HandlerError({
+        type: (failure.nexusHandlerFailureInfo.type as nexus.HandlerErrorType) ?? 'INTERNAL',
+        cause: this.failureToError(failure.cause, payloadConverter),
+        retryable,
+      });
+    }
+    if (failure.nexusOperationExecutionFailureInfo) {
+      return new NexusOperationFailure(
+        failure.nexusOperationExecutionFailureInfo.scheduledEventId?.toNumber(),
+        // We assume these will always be set or gracefully set to empty strings.
+        failure.nexusOperationExecutionFailureInfo.endpoint ?? '',
+        failure.nexusOperationExecutionFailureInfo.service ?? '',
+        failure.nexusOperationExecutionFailureInfo.operation ?? '',
+        failure.nexusOperationExecutionFailureInfo.operationToken ?? undefined,
+        this.optionalFailureToOptionalError(failure.cause, payloadConverter)
+      );
+    }
     return new TemporalFailure(
       failure.message ?? undefined,
       this.optionalFailureToOptionalError(failure.cause, payloadConverter)
@@ -216,7 +280,9 @@ export class DefaultFailureConverter implements FailureConverter {
     }
     const err = this.failureToErrorInner(failure, payloadConverter);
     err.stack = failure.stackTrace ?? '';
-    err.failure = failure;
+    if (err instanceof TemporalFailure) {
+      err.failure = failure;
+    }
     return err;
   }
 
@@ -232,8 +298,8 @@ export class DefaultFailureConverter implements FailureConverter {
   }
 
   errorToFailureInner(err: unknown, payloadConverter: PayloadConverter): ProtoFailure {
-    if (err instanceof TemporalFailure) {
-      if (err.failure) return err.failure;
+    if (err instanceof TemporalFailure || err instanceof nexus.HandlerError) {
+      if (err instanceof TemporalFailure && err.failure) return err.failure;
       const base = {
         message: err.message,
         stackTrace: cutoffStackTrace(err.stack),
@@ -310,6 +376,34 @@ export class DefaultFailureConverter implements FailureConverter {
           terminatedFailureInfo: {},
         };
       }
+      if (err instanceof nexus.HandlerError) {
+        let retryBehavior: temporal.api.enums.v1.NexusHandlerErrorRetryBehavior | undefined = undefined;
+        if (err.retryable === true) {
+          retryBehavior = encodeNexusHandlerErrorRetryBehavior("RETRYABLE");
+        } else if (err.retryable === false) {
+          retryBehavior = encodeNexusHandlerErrorRetryBehavior("NON_RETRYABLE");
+        }
+
+        return {
+          ...base,
+          nexusHandlerFailureInfo: {
+            type: err.type,
+            retryBehavior,
+          },
+        };
+      }
+      if (err instanceof NexusOperationFailure) {
+        return {
+          ...base,
+          nexusOperationExecutionFailureInfo: {
+            scheduledEventId: err.scheduledEventId ? Long.fromNumber(err.scheduledEventId) : undefined,
+            endpoint: err.endpoint,
+            service: err.service,
+            operation: err.operation,
+            operationToken: err.operationToken,
+          },
+        };
+      }
       // Just a TemporalFailure
       return base;
     }