Merge branch 'main' of https://github.com/microsoft/durabletask-js

Author: YunchuWang

packages/durabletask-js/src/client/client.ts

@@ -21,7 +21,12 @@ import { callWithMetadata, MetadataGenerator } from "../utils/grpc-helper.util";
 import { OrchestrationQuery, ListInstanceIdsOptions, DEFAULT_PAGE_SIZE } from "../orchestration/orchestration-query";
 import { Page, AsyncPageable, createAsyncPageable } from "../orchestration/page";
 import { FailureDetails } from "../task/failure-details";
+import { HistoryEvent } from "../orchestration/history-event";
+import { convertProtoHistoryEvent } from "../utils/history-event-converter";
 import { Logger, ConsoleLogger } from "../types/logger.type";
+import { StartOrchestrationOptions } from "../task/options";
+import { mapToRecord } from "../utils/tags.util";
+import { populateTagsMap } from "../utils/pb-helper.util";
 
 // Re-export MetadataGenerator for backward compatibility
 export { MetadataGenerator } from "../utils/grpc-helper.util";
@@ -133,13 +138,46 @@ export class TaskHubGrpcClient {
     input?: TInput,
     instanceId?: string,
     startAt?: Date,
+  ): Promise<string>;
+  /**
+   * Schedules a new orchestrator using the DurableTask client.
+   *
+   * @param {TOrchestrator | string} orchestrator - The orchestrator or the name of the orchestrator to be scheduled.
+   * @param {TInput} input - Optional input for the orchestrator.
+   * @param {StartOrchestrationOptions} options - Options for instance ID, start time, and tags.
+   * @return {Promise<string>} A Promise resolving to the unique ID of the scheduled orchestrator instance.
+   */
+  async scheduleNewOrchestration(
+    orchestrator: TOrchestrator | string,
+    input?: TInput,
+    options?: StartOrchestrationOptions,
+  ): Promise<string>;
+  async scheduleNewOrchestration(
+    orchestrator: TOrchestrator | string,
+    input?: TInput,
+    instanceIdOrOptions?: string | StartOrchestrationOptions,
+    startAt?: Date,
   ): Promise<string> {
     let name;
     if (typeof orchestrator === "string") {
       name = orchestrator;
     } else {
       name = getName(orchestrator);
     }
+
+    const instanceId =
+      typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined
+        ? instanceIdOrOptions
+        : instanceIdOrOptions.instanceId;
+    const scheduledStartAt =
+      typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined
+        ? startAt
+        : instanceIdOrOptions.startAt;
+    const tags =
+      typeof instanceIdOrOptions === "string" || instanceIdOrOptions === undefined
+        ? undefined
+        : instanceIdOrOptions.tags;
+
     const req = new pb.CreateInstanceRequest();
     req.setName(name);
     req.setInstanceid(instanceId ?? randomUUID());
@@ -148,11 +186,13 @@ export class TaskHubGrpcClient {
     i.setValue(JSON.stringify(input));
 
     const ts = new Timestamp();
-    ts.fromDate(new Date(startAt?.getTime() ?? 0));
+    ts.fromDate(new Date(scheduledStartAt?.getTime() ?? 0));
 
     req.setInput(i);
     req.setScheduledstarttimestamp(ts);
 
+    populateTagsMap(req.getTagsMap(), tags);
+
     this._logger.info(`Starting new ${name} instance with ID = ${req.getInstanceid()}`);
 
     const res = await callWithMetadata<pb.CreateInstanceRequest, pb.CreateInstanceResponse>(
@@ -567,14 +607,15 @@ export class TaskHubGrpcClient {
    * @example
    * ```typescript
    * // Iterate over all matching instances
+   * const logger = new ConsoleLogger();
    * const pageable = client.getAllInstances({ statuses: [OrchestrationStatus.COMPLETED] });
    * for await (const instance of pageable) {
-   *   console.log(instance.instanceId);
+   *   logger.info(instance.instanceId);
    * }
    *
    * // Iterate over pages
    * for await (const page of pageable.asPages()) {
-   *   console.log(`Page has ${page.values.length} items`);
+   *   logger.info(`Page has ${page.values.length} items`);
    * }
    * ```
    *
@@ -735,6 +776,77 @@ export class TaskHubGrpcClient {
     return new Page(instanceIds, lastInstanceKey);
   }
 
+  /**
+   * Retrieves the history of the specified orchestration instance as a list of HistoryEvent objects.
+   *
+   * This method streams the history events from the backend and returns them as an array.
+   * The history includes all events that occurred during the orchestration execution,
+   * such as task scheduling, completion, failure, timer events, and more.
+   *
+   * If the orchestration instance does not exist, an empty array is returned.
+   *
+   * @param instanceId - The unique identifier of the orchestration instance.
+   * @returns A Promise that resolves to an array of HistoryEvent objects representing
+   *          the orchestration's history. Returns an empty array if the instance is not found.
+   * @throws {Error} If the instanceId is null or empty.
+   * @throws {Error} If the operation is canceled.
+   * @throws {Error} If an internal error occurs while retrieving the history.
+   *
+   * @example
+   * ```typescript
+   * const history = await client.getOrchestrationHistory(instanceId);
+   * for (const event of history) {
+   *   console.log(`Event ${event.eventId}: ${event.type} at ${event.timestamp}`);
+   * }
+   * ```
+   */
+  async getOrchestrationHistory(instanceId: string): Promise<HistoryEvent[]> {
+    if (!instanceId) {
+      throw new Error("instanceId is required");
+    }
+
+    const req = new pb.StreamInstanceHistoryRequest();
+    req.setInstanceid(instanceId);
+    req.setForworkitemprocessing(false);
+
+    const metadata = this._metadataGenerator ? await this._metadataGenerator() : new grpc.Metadata();
+    const stream = this._stub.streamInstanceHistory(req, metadata);
+
+    return new Promise<HistoryEvent[]>((resolve, reject) => {
+      const historyEvents: HistoryEvent[] = [];
+
+      stream.on("data", (chunk: pb.HistoryChunk) => {
+        const protoEvents = chunk.getEventsList();
+        for (const protoEvent of protoEvents) {
+          const event = convertProtoHistoryEvent(protoEvent);
+          if (event) {
+            historyEvents.push(event);
+          }
+        }
+      });
+
+      stream.on("end", () => {
+        stream.removeAllListeners();
+        resolve(historyEvents);
+      });
+
+      stream.on("error", (err: grpc.ServiceError) => {
+        stream.removeAllListeners();
+        // Return empty array for NOT_FOUND to be consistent with DTS behavior
+        // (DTS returns empty stream for non-existent instances) and user-friendly
+        if (err.code === grpc.status.NOT_FOUND) {
+          resolve([]);
+        } else if (err.code === grpc.status.CANCELLED) {
+          reject(new Error(`The getOrchestrationHistory operation was canceled.`));
+        } else if (err.code === grpc.status.INTERNAL) {
+          reject(new Error(`An error occurred while retrieving the history for orchestration with instanceId '${instanceId}'.`));
+        } else {
+          reject(err);
+        }
+      });
+    });
+  }
+
   /**
    * Helper method to create an OrchestrationState from a protobuf OrchestrationState.
    */
@@ -779,6 +891,8 @@ export class TaskHubGrpcClient {
       );
     }
 
+    const tags = mapToRecord(protoState.getTagsMap());
+
     return new OrchestrationState(
       instanceId,
       name ?? "",
@@ -789,6 +903,7 @@ export class TaskHubGrpcClient {
       serializedOutput,
       serializedCustomStatus,
       failureDetails,
+      tags,
     );
   }
 }

packages/durabletask-js/src/index.ts

@@ -18,6 +18,44 @@ export { OrchestrationState } from "./orchestration/orchestration-state";
 export { OrchestrationQuery, ListInstanceIdsOptions, DEFAULT_PAGE_SIZE } from "./orchestration/orchestration-query";
 export { Page, AsyncPageable, createAsyncPageable } from "./orchestration/page";
 
+// History event types
+export {
+  HistoryEvent,
+  HistoryEventType,
+  HistoryEventBase,
+  ExecutionStartedEvent,
+  ExecutionCompletedEvent,
+  ExecutionTerminatedEvent,
+  ExecutionSuspendedEvent,
+  ExecutionResumedEvent,
+  ExecutionRewoundEvent,
+  TaskScheduledEvent,
+  TaskCompletedEvent,
+  TaskFailedEvent,
+  SubOrchestrationInstanceCreatedEvent,
+  SubOrchestrationInstanceCompletedEvent,
+  SubOrchestrationInstanceFailedEvent,
+  TimerCreatedEvent,
+  TimerFiredEvent,
+  OrchestratorStartedEvent,
+  OrchestratorCompletedEvent,
+  EventSentEvent,
+  EventRaisedEvent,
+  GenericEvent,
+  HistoryStateEvent,
+  ContinueAsNewEvent,
+  OrchestrationInstance,
+  ParentInstanceInfo,
+  TraceContext,
+  EntityOperationSignaledEvent,
+  EntityOperationCalledEvent,
+  EntityOperationCompletedEvent,
+  EntityOperationFailedEvent,
+  EntityLockRequestedEvent,
+  EntityLockGrantedEvent,
+  EntityUnlockSentEvent,
+} from "./orchestration/history-event";
+
 // Proto types (for advanced usage)
 export { OrchestrationStatus as ProtoOrchestrationStatus } from "./proto/orchestrator_service_pb";
 
@@ -27,7 +65,13 @@ export { Task } from "./task/task";
 
 // Retry policies and task options
 export { RetryPolicy, RetryPolicyOptions } from "./task/retry";
-export { TaskOptions, SubOrchestrationOptions, taskOptionsFromRetryPolicy, subOrchestrationOptionsFromRetryPolicy } from "./task/options";
+export {
+  TaskOptions,
+  SubOrchestrationOptions,
+  StartOrchestrationOptions,
+  taskOptionsFromRetryPolicy,
+  subOrchestrationOptionsFromRetryPolicy,
+} from "./task/options";
 
 // Types
 export { TOrchestrator } from "./types/orchestrator.type";