Workflow authoring support

package.json

@@ -8,7 +8,7 @@
     "test:load": "jest --runInBand --detectOpenHandles",
     "test:load:http": "TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 dapr run --app-id test-suite --app-protocol http --app-port 50001 --dapr-http-port 50000 --components-path ./test/components -- npm run test:load 'test/load'",
     "test:e2e": "jest --runInBand --detectOpenHandles",
-    "test:e2e:all": "npm run test:e2e:http; npm run test:e2e:grpc; npm run test:e2e:common",
+    "test:e2e:all": "npm run test:e2e:http; npm run test:e2e:grpc; npm run test:e2e:common; npm run test:e2e:workflow",
     "test:e2e:grpc": "npm run test:e2e:grpc:client && npm run test:e2e:grpc:server && npm run test:e2e:grpc:clientWithApiToken",
     "test:e2e:grpc:client": "npm run prebuild && TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 dapr run --app-id test-suite --app-protocol grpc --app-port 50001 --dapr-grpc-port 50000 --components-path ./test/components -- jest --runInBand --detectOpenHandles --testMatch [ '**/test/e2e/grpc/*client.test.ts' ]",
     "test:e2e:grpc:clientWithApiToken": "npm run prebuild && TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 DAPR_API_TOKEN=test dapr run --app-id test-suite --app-protocol grpc --app-port 50001 --dapr-grpc-port 50000 --components-path ./test/components -- jest --runInBand --detectOpenHandles --testMatch [ '**/test/e2e/grpc/clientWithApiToken.test.ts' ]",
@@ -20,6 +20,9 @@
     "test:e2e:common": "npm run test:e2e:common:client && npm run test:e2e:common:server",
     "test:e2e:common:client": "./scripts/test-e2e-common.sh client",
     "test:e2e:common:server": "./scripts/test-e2e-common.sh server",
+    "test:e2e:workflow": "npm run prebuild && TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 dapr run --app-id workflow-test-suite --app-protocol grpc --dapr-grpc-port 4001 --components-path ./test/components/workflow -- jest --runInBand --detectOpenHandles --testMatch [ '**/test/e2e/workflow/workflow.test.ts' ]",
+    "test:e2e:workflow:internal": "jest test/e2e/workflow --runInBand --detectOpenHandles",
+    "test:e2e:workflow:durabletask": "./scripts/test-e2e-workflow.sh",
     "test:unit": "jest --runInBand --detectOpenHandles",
     "test:unit:all": "npm run test:unit:main && npm run test:unit:http && npm run test:unit:grpc && npm run test:unit:common && npm run test:unit:actors && npm run test:unit:logger && npm run test:unit:utils && npm run test:unit:errors",
     "test:unit:main": "NODE_ENV=test npm run test:unit 'test/unit/main/.*\\.test\\.ts'",
@@ -30,13 +33,15 @@
     "test:unit:logger": "NODE_ENV=test npm run test:unit 'test/unit/logger/.*\\.test\\.ts'",
     "test:unit:utils": "NODE_ENV=test npm run test:unit 'test/unit/utils/.*\\.test\\.ts'",
     "test:unit:errors": "NODE_ENV=test npm run test:unit 'test/unit/errors/.*\\.test\\.ts'",
+    "test:unit:workflow": "NODE_ENV=test npm run test:unit 'test/unit/workflow/.*\\.test\\.ts'",
     "lint": "eslint . --ext .js,.jsx,.ts,.tsx",
     "prebuild": "./scripts/prebuild.sh",
     "build-ci": "npm run prebuild && ./scripts/build.sh",
     "build": "npm install && npm run lint && npm run pretty && ./scripts/build.sh",
     "start:dev": "npm run build && nodemon --ext \".ts,.js\" --watch \"./src\" --exec \"npm run build\"",
     "pretty": "prettier --list-different \"**/*.{ts,tsx,js,jsx,json,md}\"",
-    "pretty-fix": "prettier --write \"**/*.{ts,tsx,js,jsx,json,md}\""
+    "pretty-fix": "prettier --write \"**/*.{ts,tsx,js,jsx,json,md}\"",
+    "example": "ts-node --swc"
   },
   "keywords": [],
   "author": "Xavier Geerinck",
@@ -50,13 +55,16 @@
     "express": "^4.18.2",
     "google-protobuf": "^3.18.0",
     "http-terminator": "^3.2.0",
+    "kaibocai-durabletask-js": "^0.0.5-beta.3",
     "node-fetch": "^2.6.7"
   },
   "devDependencies": {
+    "@swc/core": "^1.3.55",
     "@types/body-parser": "^1.19.1",
     "@types/express": "^4.17.15",
     "@types/jest": "^27.0.1",
     "@types/node": "^16.9.1",
+    "@types/readline-sync": "^1.4.8",
     "@types/uuid": "^8.3.1",
     "@typescript-eslint/eslint-plugin": "^5.1.0",
     "@typescript-eslint/parser": "^5.1.0",
@@ -70,7 +78,9 @@
     "nodemon": "^2.0.20",
     "prettier": "^2.4.0",
     "pretty-quick": "^3.1.3",
+    "readline-sync": "^1.4.10",
     "ts-jest": "^27.0.5",
+    "ts-node": "^10.9.1",
     "typescript": "^4.5.5"
   },
   "repository": {

scripts/test-e2e-workflow.sh

@@ -0,0 +1,29 @@
+#!/bin/bash
+# Start the sidecar if it is not running yet
+if [ ! "$(docker ps -q -f name=durabletask-sidecar)" ]; then
+    if [ "$(docker ps -aq -f status=exited -f name=durabletask-sidecar)" ]; then
+        # cleanup
+        docker rm durabletask-sidecar
+    fi
+
+    # run your container
+    echo "Starting Sidecar"
+    docker run \
+        --name durabletask-sidecar -d --rm \
+        -p 4001:4001 \
+        --env 'DURABLETASK_SIDECAR_LOGLEVEL=Debug' \
+        cgillum/durabletask-sidecar:latest start \
+        --backend Emulator
+fi
+
+echo "Running workflow E2E tests"
+npm run test:e2e:workflow:internal
+
+# It should fail if the npm run fails
+if [ $? -ne 0 ]; then
+    echo "E2E tests failed"
+    exit 1
+fi
+
+echo "Stopping Sidecar"
+docker stop durabletask-sidecar

src/workflow/client/WorkflowClient.ts

@@ -0,0 +1,180 @@
+/*
+Copyright 2022 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+import { TaskHubGrpcClient } from "kaibocai-durabletask-js";
+import * as grpc from "@grpc/grpc-js";
+import { WorkflowState } from "./WorkflowState";
+import { generateInterceptors } from "../internal/ApiTokenClientInterceptor";
+import { TWorkflow } from "../types/Workflow.type";
+import { getFunctionName } from "../internal";
+
+export default class WorkflowClient {
+  private readonly _innerClient: TaskHubGrpcClient;
+
+  /**
+   * Initializes a new instance of the DaprWorkflowClient.
+   * @param {string | undefined} hostAddress - The address of the Dapr runtime hosting the workflow services.
+   * @param {grpc.ChannelOptions | undefined} options - Additional options for configuring the gRPC channel.
+   */
+  constructor(hostAddress?: string, options?: grpc.ChannelOptions) {
+    this._innerClient = this._buildInnerClient(hostAddress, options);
+  }
+
+  _buildInnerClient(hostAddress = "127.0.0.1:50001", options: grpc.ChannelOptions = {}): TaskHubGrpcClient {
+    const innerOptions = {
+      ...options,
+      interceptors: [generateInterceptors(), ...(options?.interceptors ?? [])],
+    };
+    return new TaskHubGrpcClient(hostAddress, innerOptions);
+  }
+
+  /**
+   * Schedules a new workflow using the DurableTask client.
+   *
+   * @param {TWorkflow | string} workflow - The Workflow or the name of the workflow to be scheduled.
+   * @return {Promise<string>} A Promise resolving to the unique ID of the scheduled workflow instance.
+   */
+  public async scheduleNewWorkflow(
+    workflow: TWorkflow | string,
+    input?: any,
+    instanceId?: string,
+    startAt?: Date,
+  ): Promise<string> {
+    if (typeof workflow === "string") {
+      return await this._innerClient.scheduleNewOrchestration(workflow, input, instanceId, startAt);
+    }
+    return await this._innerClient.scheduleNewOrchestration(getFunctionName(workflow), input, instanceId, startAt);
+  }
+
+  /**
+   * Terminates the workflow associated with the provided instance id.
+   *
+   * @param {string} workflowInstanceId - Workflow instance id to terminate.
+   * @param {any} output - The optional output to set for the terminated workflow instance.
+   */
+  public async terminateWorkflow(workflowInstanceId: string, output: any) {
+    await this._innerClient.terminateOrchestration(workflowInstanceId, output);
+  }
+
+  /**
+   * Fetches workflow instance metadata from the configured durable store.
+   *
+   * @param {string} workflowInstanceId - The unique identifier of the workflow instance to fetch.
+   * @param {boolean} getInputsAndOutputs - Indicates whether to fetch the workflow instance's
+   *                                       inputs, outputs, and custom status (true) or omit them (false).
+   * @returns {Promise<WorkflowState | undefined>} A Promise that resolves to a metadata record describing
+   *                                              the workflow instance and its execution status, or undefined
+   *                                              if the instance is not found.
+   */
+  public async getWorkflowState(
+    workflowInstanceId: string,
+    getInputsAndOutputs: boolean,
+  ): Promise<WorkflowState | undefined> {
+    const state = await this._innerClient.getOrchestrationState(workflowInstanceId, getInputsAndOutputs);
+    if (state !== undefined) {
+      return new WorkflowState(state);
+    }
+  }
+
+  /**
+   * Waits for a workflow to start running and returns a {@link WorkflowState} object
+   * containing metadata about the started instance, and optionally, its input, output,
+   * and custom status payloads.
+   *
+   * A "started" workflow instance refers to any instance not in the Pending state.
+   *
+   * If a workflow instance is already running when this method is called, it returns immediately.
+   *
+   * @param {string} workflowInstanceId - The unique identifier of the workflow instance to wait for.
+   * @param {boolean} fetchPayloads - Indicates whether to fetch the workflow instance's
+   *                                  inputs, outputs (true) or omit them (false).
+   * @param {number} timeout - The amount of time, in seconds, to wait for the workflow instance to start.
+   * @returns {Promise<WorkflowState | undefined>} A Promise that resolves to the workflow instance metadata
+   *                                               or undefined if no such instance is found.
+   */
+  public async waitForWorkflowStart(
+    workflowInstanceId: string,
+    fetchPayloads?: boolean,
+    timeout?: number,
+  ): Promise<WorkflowState | undefined> {
+    const state = await this._innerClient.waitForOrchestrationStart(workflowInstanceId, fetchPayloads, timeout);
+    if (state !== undefined) {
+      return new WorkflowState(state);
+    }
+  }
+
+  /**
+   * Waits for a workflow to complete running and returns a {@link WorkflowState} object
+   * containing metadata about the completed instance, and optionally, its input, output,
+   * and custom status payloads.
+   *
+   * A "completed" workflow instance refers to any instance in one of the terminal states.
+   * For example, the Completed, Failed, or Terminated states.
+   *
+   * If a workflow instance is already running when this method is called, it returns immediately.
+   *
+   * @param {string} workflowInstanceId - The unique identifier of the workflow instance to wait for.
+   * @param {boolean} fetchPayloads - Indicates whether to fetch the workflow instance's
+   *                                  inputs, outputs (true) or omit them (false).
+   * @param {number} timeout - The amount of time, in seconds, to wait for the workflow instance to start.
+   * @returns {Promise<WorkflowState | undefined>} A Promise that resolves to the workflow instance metadata
+   *                                               or undefined if no such instance is found.
+   */
+  public async waitForWorkflowCompletion(
+    workflowInstanceId: string,
+    fetchPayloads = true,
+    timeout: number,
+  ): Promise<WorkflowState | undefined> {
+    const state = await this._innerClient.waitForOrchestrationCompletion(workflowInstanceId, fetchPayloads, timeout);
+    if (state != undefined) {
+      return new WorkflowState(state);
+    }
+  }
+
+  /**
+   * Sends an event notification message to an awaiting workflow instance.
+   *
+   * This method triggers the specified event in a running workflow instance,
+   * allowing the workflow to respond to the event if it has defined event handlers.
+   *
+   * @param {string} workflowInstanceId - The unique identifier of the workflow instance that will handle the event.
+   * @param {string} eventName - The name of the event. Event names are case-insensitive.
+   * @param {any} [eventPayload] - An optional serializable data payload to include with the event.
+   */
+  public async raiseEvent(workflowInstanceId: string, eventName: string, eventPayload?: any) {
+    this._innerClient.raiseOrchestrationEvent(workflowInstanceId, eventName, eventPayload);
+  }
+
+  /**
+   * Purges the workflow instance state from the workflow state store.
+   *
+   * This method removes the persisted state associated with a workflow instance from the state store.
+   *
+   * @param {string} workflowInstanceId - The unique identifier of the workflow instance to purge.
+   * @return {Promise<boolean>} A Promise that resolves to true if the workflow state was found and purged successfully, otherwise false.
+   */
+  public async purgeWorkflow(workflowInstanceId: string): Promise<boolean> {
+    const purgeResult = await this._innerClient.purgeOrchestration(workflowInstanceId);
+    if (purgeResult !== undefined) {
+      return purgeResult.deletedInstanceCount > 0;
+    }
+    return false;
+  }
+
+  /**
+   * Closes the inner DurableTask client and shutdown the GRPC channel.
+   */
+  public async stop() {
+    await this._innerClient.stop();
+  }
+}

src/workflow/client/WorkflowFailureDetails.ts

@@ -0,0 +1,46 @@
+/*
+Copyright 2022 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+import { FailureDetails } from "kaibocai-durabletask-js/task/failure-details";
+
+export class WorkflowFailureDetails {
+  private readonly failureDetails: FailureDetails;
+
+  constructor(failureDetails: FailureDetails) {
+    this.failureDetails = failureDetails;
+  }
+
+  /**
+   * Gets the error type, which is the namespace-qualified exception type name.
+   * @return {string} The error type.
+   */
+  public getErrorType(): string {
+    return this.failureDetails.errorType;
+  }
+
+  /**
+   * Gets the error message.
+   * @return {string} The error message.
+   */
+  public getErrorMessage(): string {
+    return this.failureDetails.message;
+  }
+
+  /**
+   * Gets the stack trace.
+   * @return {string | undefined} The stack trace, or undefined if not available.
+   */
+  public getStackTrace(): string | undefined {
+    return this.failureDetails.stackTrace;
+  }
+}