feat(client): Implement basic list API (#942)

bergundy · web-flow · commit adf872ea4b17 · 2022-11-01T15:12:37.000-07:00
diff --git a/packages/client/src/helpers.ts b/packages/client/src/helpers.ts
@@ -0,0 +1,75 @@
+import {
+  LoadedDataConverter,
+  mapFromPayloads,
+  searchAttributePayloadConverter,
+  SearchAttributes,
+} from '@temporalio/common';
+import { optionalTsToDate, tsToDate } from '@temporalio/common/lib/time';
+import { decodeMapFromPayloads } from '@temporalio/common/lib/internal-non-workflow/codec-helpers';
+import { RawWorkflowExecutionInfo, WorkflowExecutionInfo, WorkflowExecutionStatusName } from './types';
+import { temporal } from '@temporalio/proto';
+
+function workflowStatusCodeToName(code: temporal.api.enums.v1.WorkflowExecutionStatus): WorkflowExecutionStatusName {
+  return workflowStatusCodeToNameInternal(code) ?? 'UNKNOWN';
+}
+
+/**
+ * Intentionally leave out `default` branch to get compilation errors when new values are added
+ */
+function workflowStatusCodeToNameInternal(
+  code: temporal.api.enums.v1.WorkflowExecutionStatus
+): WorkflowExecutionStatusName {
+  switch (code) {
+    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_UNSPECIFIED:
+      return 'UNSPECIFIED';
+    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_RUNNING:
+      return 'RUNNING';
+    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_FAILED:
+      return 'FAILED';
+    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_TIMED_OUT:
+      return 'TIMED_OUT';
+    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_CANCELED:
+      return 'CANCELLED';
+    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_TERMINATED:
+      return 'TERMINATED';
+    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_COMPLETED:
+      return 'COMPLETED';
+    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW:
+      return 'CONTINUED_AS_NEW';
+  }
+}
+
+export async function executionInfoFromRaw(
+  raw: RawWorkflowExecutionInfo,
+  dataConverter: LoadedDataConverter
+): Promise<WorkflowExecutionInfo> {
+  return {
+    /* eslint-disable @typescript-eslint/no-non-null-assertion */
+    type: raw.type!.name!,
+    workflowId: raw.execution!.workflowId!,
+    runId: raw.execution!.runId!,
+    taskQueue: raw.taskQueue!,
+    status: {
+      code: raw.status!,
+      name: workflowStatusCodeToName(raw.status!),
+    },
+    // Safe to convert to number, max history length is 50k, which is much less than Number.MAX_SAFE_INTEGER
+    historyLength: raw.historyLength!.toNumber(),
+    startTime: tsToDate(raw.startTime!),
+    executionTime: optionalTsToDate(raw.executionTime),
+    closeTime: optionalTsToDate(raw.closeTime),
+    memo: await decodeMapFromPayloads(dataConverter, raw.memo?.fields),
+    searchAttributes: Object.fromEntries(
+      Object.entries(
+        mapFromPayloads(searchAttributePayloadConverter, raw.searchAttributes?.indexedFields ?? {}) as SearchAttributes
+      ).filter(([_, v]) => v && v.length > 0) // Filter out empty arrays returned by pre 1.18 servers
+    ),
+    parentExecution: raw.parentExecution
+      ? {
+          workflowId: raw.parentExecution.workflowId!,
+          runId: raw.parentExecution.runId!,
+        }
+      : undefined,
+    raw,
+  };
+}
diff --git a/packages/client/src/types.ts b/packages/client/src/types.ts
@@ -1,6 +1,7 @@
 import type { SearchAttributes } from '@temporalio/common';
 import * as proto from '@temporalio/proto';
 import type * as grpc from '@grpc/grpc-js';
+import { Replace } from '@temporalio/common/lib/type-helpers';
 
 export interface WorkflowExecution {
   workflowId: string;
@@ -11,6 +12,7 @@ export type GetWorkflowExecutionHistoryRequest =
   proto.temporal.api.workflowservice.v1.IGetWorkflowExecutionHistoryRequest;
 export type DescribeWorkflowExecutionResponse =
   proto.temporal.api.workflowservice.v1.IDescribeWorkflowExecutionResponse;
+export type RawWorkflowExecutionInfo = proto.temporal.api.workflow.v1.IWorkflowExecutionInfo;
 export type TerminateWorkflowExecutionResponse =
   proto.temporal.api.workflowservice.v1.ITerminateWorkflowExecutionResponse;
 export type RequestCancelWorkflowExecutionResponse =
@@ -27,7 +29,7 @@ export type WorkflowExecutionStatusName =
   | 'TIMED_OUT'
   | 'UNKNOWN'; // UNKNOWN is reserved for future enum values
 
-export interface WorkflowExecutionDescription {
+export interface WorkflowExecutionInfo {
   type: string;
   workflowId: string;
   runId: string;
@@ -40,9 +42,16 @@ export interface WorkflowExecutionDescription {
   memo?: Record<string, unknown>;
   searchAttributes: SearchAttributes;
   parentExecution?: Required<proto.temporal.api.common.v1.IWorkflowExecution>;
-  raw: DescribeWorkflowExecutionResponse;
+  raw: RawWorkflowExecutionInfo;
 }
 
+export type WorkflowExecutionDescription = Replace<
+  WorkflowExecutionInfo,
+  {
+    raw: DescribeWorkflowExecutionResponse;
+  }
+>;
+
 export type WorkflowService = proto.temporal.api.workflowservice.v1.WorkflowService;
 export const { WorkflowService } = proto.temporal.api.workflowservice.v1;
 export type OperatorService = proto.temporal.api.operatorservice.v1.OperatorService;
diff --git a/packages/client/src/workflow-client.ts b/packages/client/src/workflow-client.ts
@@ -5,12 +5,10 @@ import {
   compileRetryPolicy,
   DataConverter,
   LoadedDataConverter,
-  mapFromPayloads,
   mapToPayloads,
   QueryDefinition,
   RetryState,
   searchAttributePayloadConverter,
-  SearchAttributes,
   SignalDefinition,
   TerminatedFailure,
   TimeoutFailure,
@@ -21,12 +19,10 @@ import {
   WorkflowNotFoundError,
   WorkflowResultType,
 } from '@temporalio/common';
-import { tsToDate, optionalTsToDate } from '@temporalio/common/lib/time';
 import { composeInterceptors } from '@temporalio/common/lib/interceptors';
 import {
   decodeArrayFromPayloads,
   decodeFromPayloadsAtIndex,
-  decodeMapFromPayloads,
   decodeOptionalFailureToOptionalError,
   encodeMapToPayloads,
   encodeToPayloads,
@@ -61,10 +57,11 @@ import {
   TerminateWorkflowExecutionResponse,
   WorkflowExecution,
   WorkflowExecutionDescription,
-  WorkflowExecutionStatusName,
+  WorkflowExecutionInfo,
   WorkflowService,
 } from './types';
 import { compileWorkflowOptions, WorkflowOptions, WorkflowSignalWithStartOptions } from './workflow-options';
+import { executionInfoFromRaw } from './helpers';
 
 /**
  * A client side handle to a single Workflow instance.
@@ -288,6 +285,22 @@ interface WorkflowHandleOptions extends GetWorkflowHandleOptions {
  */
 export type WorkflowStartOptions<T extends Workflow = Workflow> = WithWorkflowArgs<T, WorkflowOptions>;
 
+/**
+ * Options for {@link WorkflowClient.list}
+ */
+export interface ListOptions {
+  /**
+   * Maximum number of results to fetch per page.
+   *
+   * @default depends on server config, typically 1000
+   */
+  pageSize?: number;
+  /**
+   * Query string for matching and ordering the results
+   */
+  query?: string;
+}
+
 /**
  * Client for starting Workflow executions and creating Workflow handles.
  *
@@ -848,38 +861,9 @@ export class WorkflowClient {
         const raw = await fn({
           workflowExecution: { workflowId, runId },
         });
-        return {
-          /* eslint-disable @typescript-eslint/no-non-null-assertion */
-          type: raw.workflowExecutionInfo!.type!.name!,
-          workflowId: raw.workflowExecutionInfo!.execution!.workflowId!,
-          runId: raw.workflowExecutionInfo!.execution!.runId!,
-          taskQueue: raw.workflowExecutionInfo!.taskQueue!,
-          status: {
-            code: raw.workflowExecutionInfo!.status!,
-            name: workflowStatusCodeToName(raw.workflowExecutionInfo!.status!),
-          },
-          // Safe to convert to number, max history length is 50k, which is much less than Number.MAX_SAFE_INTEGER
-          historyLength: raw.workflowExecutionInfo!.historyLength!.toNumber(),
-          startTime: tsToDate(raw.workflowExecutionInfo!.startTime!),
-          executionTime: optionalTsToDate(raw.workflowExecutionInfo!.executionTime),
-          closeTime: optionalTsToDate(raw.workflowExecutionInfo!.closeTime),
-          memo: await decodeMapFromPayloads(this.client.dataConverter, raw.workflowExecutionInfo!.memo?.fields),
-          searchAttributes: Object.fromEntries(
-            Object.entries(
-              mapFromPayloads(
-                searchAttributePayloadConverter,
-                raw.workflowExecutionInfo!.searchAttributes?.indexedFields ?? {}
-              ) as SearchAttributes
-            ).filter(([_, v]) => v && v.length > 0) // Filter out empty arrays returned by pre 1.18 servers
-          ),
-          parentExecution: raw.workflowExecutionInfo?.parentExecution
-            ? {
-                workflowId: raw.workflowExecutionInfo.parentExecution.workflowId!,
-                runId: raw.workflowExecutionInfo.parentExecution.runId!,
-              }
-            : undefined,
-          raw,
-        };
+        const info = await executionInfoFromRaw(raw.workflowExecutionInfo ?? {}, this.client.dataConverter);
+        (info as unknown as WorkflowExecutionDescription).raw = raw;
+        return info;
       },
       async signal<Args extends any[]>(def: SignalDefinition<Args> | string, ...args: Args): Promise<void> {
         const next = this.client._signalWorkflowHandler.bind(this.client);
@@ -939,6 +923,34 @@ export class WorkflowClient {
       followRuns: options?.followRuns ?? true,
     });
   }
+
+  /**
+   * List workflows by given `query`.
+   *
+   * ⚠️ To use advanced query functionality, as of the 1.18 server release, you must use Elasticsearch based visibility.
+   *
+   * More info on the concept of "visibility" and the query syntax on the Temporal documentation site:
+   * https://docs.temporal.io/visibility
+   */
+  public async *list(options?: ListOptions): AsyncIterable<WorkflowExecutionInfo> {
+    let nextPageToken: Uint8Array = Buffer.alloc(0);
+    for (;;) {
+      const response = await this.workflowService.listWorkflowExecutions({
+        namespace: this.options.namespace,
+        query: options?.query,
+        nextPageToken,
+        pageSize: options?.pageSize,
+      });
+      // Not decoding memo payloads concurrently even though we could have to keep the lazy nature of this iterator.
+      // Decoding is done for `memo` fields which tend to be small.
+      // We might decide to change that based on user feedback.
+      for (const raw of response.executions) {
+        yield await executionInfoFromRaw(raw, this.dataConverter);
+      }
+      nextPageToken = response.nextPageToken;
+      if (nextPageToken == null || nextPageToken.length == 0) break;
+    }
+  }
 }
 
 export class QueryRejectedError extends Error {
@@ -954,33 +966,3 @@ export class QueryNotRegisteredError extends Error {
     super(message);
   }
 }
-
-function workflowStatusCodeToName(code: temporal.api.enums.v1.WorkflowExecutionStatus): WorkflowExecutionStatusName {
-  return workflowStatusCodeToNameInternal(code) ?? 'UNKNOWN';
-}
-
-/**
- * Intentionally leave out `default` branch to get compilation errors when new values are added
- */
-function workflowStatusCodeToNameInternal(
-  code: temporal.api.enums.v1.WorkflowExecutionStatus
-): WorkflowExecutionStatusName {
-  switch (code) {
-    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_UNSPECIFIED:
-      return 'UNSPECIFIED';
-    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_RUNNING:
-      return 'RUNNING';
-    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_FAILED:
-      return 'FAILED';
-    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_TIMED_OUT:
-      return 'TIMED_OUT';
-    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_CANCELED:
-      return 'CANCELLED';
-    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_TERMINATED:
-      return 'TERMINATED';
-    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_COMPLETED:
-      return 'COMPLETED';
-    case temporal.api.enums.v1.WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW:
-      return 'CONTINUED_AS_NEW';
-  }
-}
diff --git a/packages/test/src/integration-tests.ts b/packages/test/src/integration-tests.ts
@@ -1325,33 +1325,32 @@ export function runIntegrationTests(codec?: PayloadCodec): void {
     t.pass();
   });
 
-  test('Download and replay multiple executions', async (t) => {
+  /**
+   * NOTE: this test uses the `list` API which requires advanced visibility as of server 1.18.
+   * Run with docker-compose
+   */
+  test('Download and replay multiple executions with client list method', async (t) => {
     const { metaClient: client } = t.context;
     const taskQueue = 'test';
-    const e1 = await client.workflow.start(workflows.argsAndReturn, {
-      taskQueue,
-      workflowId: uuid4(),
-      args: ['Hello', undefined, u8('world!')],
-    });
-    const e2 = await client.workflow.start(workflows.cancelFakeProgress, {
-      taskQueue,
-      workflowId: uuid4(),
-    });
-    const e3 = await client.workflow.start(workflows.childWorkflowInvoke, {
-      taskQueue,
-      workflowId: uuid4(),
-    });
-    const e4 = await client.workflow.start(workflows.activityFailures, {
-      taskQueue,
-      workflowId: uuid4(),
-    });
-    const handles = [e1, e2, e3, e4];
+    const fns = [
+      workflows.http,
+      workflows.cancelFakeProgress,
+      workflows.childWorkflowInvoke,
+      workflows.activityFailures,
+    ];
+    const handles = await Promise.all(
+      fns.map((fn) =>
+        client.workflow.start(fn, {
+          taskQueue,
+          workflowId: uuid4(),
+        })
+      )
+    );
+    // Wait for the workflows to complete first
     await Promise.all(handles.map((h) => h.result()));
-    const executions = (async function* () {
-      for (const { workflowId, firstExecutionRunId } of handles) {
-        yield { workflowId, runId: firstExecutionRunId };
-      }
-    })();
+    // Test the list API too while we're at it
+    const workflowIds = handles.map(({ workflowId }) => `'${workflowId}'`);
+    const executions = client.workflow.list({ query: `WorkflowId IN (${workflowIds.join(', ')})` });
 
     await Worker.runReplayHistories(
       {
