Added a count method to the workflow client, a higher-level user-friendlier option to using the corresponding GRPC method directly.

Author: THardy98

packages/client/src/helpers.ts

@@ -10,7 +10,15 @@ import { Replace } from '@temporalio/common/lib/type-helpers';
 import { optionalTsToDate, requiredTsToDate } from '@temporalio/common/lib/time';
 import { decodeMapFromPayloads } from '@temporalio/common/lib/internal-non-workflow/codec-helpers';
 import { temporal, google } from '@temporalio/proto';
-import { RawWorkflowExecutionInfo, WorkflowExecutionInfo, WorkflowExecutionStatusName } from './types';
+import {
+  CountWorkflowExecution,
+  CountWorkflowExecutionsAggregationGroup,
+  RawCountWorkflowExecutions,
+  RawCountWorkflowExecutionsAggregationGroup,
+  RawWorkflowExecutionInfo,
+  WorkflowExecutionInfo,
+  WorkflowExecutionStatusName,
+} from './types';
 
 function workflowStatusCodeToName(code: temporal.api.enums.v1.WorkflowExecutionStatus): WorkflowExecutionStatusName {
   return workflowStatusCodeToNameInternal(code) ?? 'UNKNOWN';
@@ -81,6 +89,24 @@ export async function executionInfoFromRaw<T>(
   };
 }
 
+export function countWorkflowExecutionFromRaw(raw: RawCountWorkflowExecutions): CountWorkflowExecution {
+  return {
+    // Note: lossy conversion of Long to number
+    count: raw.count!.toNumber(),
+    groups: raw.groups!.map((group) => executionCountAggregationGroupFromRaw(group)),
+  };
+}
+
+export function executionCountAggregationGroupFromRaw(
+  raw: RawCountWorkflowExecutionsAggregationGroup
+): CountWorkflowExecutionsAggregationGroup {
+  return {
+    // Note: lossy conversion of Long to number
+    count: raw.count!.toNumber(),
+    group_values: raw.groupValues!.map((group_value) => searchAttributePayloadConverter.fromPayload(group_value)),
+  };
+}
+
 type ErrorDetailsName = `temporal.api.errordetails.v1.${keyof typeof temporal.api.errordetails.v1}`;
 
 /**

packages/client/src/types.ts

@@ -1,5 +1,5 @@
 import type * as grpc from '@grpc/grpc-js';
-import type { SearchAttributes } from '@temporalio/common';
+import type { SearchAttributes, SearchAttributeValue } from '@temporalio/common';
 import { makeProtoEnumConverters } from '@temporalio/common/lib/internal-workflow';
 import * as proto from '@temporalio/proto';
 import { Replace } from '@temporalio/common/lib/type-helpers';
@@ -14,6 +14,9 @@ export type GetWorkflowExecutionHistoryRequest =
 export type DescribeWorkflowExecutionResponse =
   proto.temporal.api.workflowservice.v1.IDescribeWorkflowExecutionResponse;
 export type RawWorkflowExecutionInfo = proto.temporal.api.workflow.v1.IWorkflowExecutionInfo;
+export type RawCountWorkflowExecutions = proto.temporal.api.workflowservice.v1.ICountWorkflowExecutionsResponse;
+export type RawCountWorkflowExecutionsAggregationGroup =
+  proto.temporal.api.workflowservice.v1.CountWorkflowExecutionsResponse.IAggregationGroup;
 export type TerminateWorkflowExecutionResponse =
   proto.temporal.api.workflowservice.v1.ITerminateWorkflowExecutionResponse;
 export type RequestCancelWorkflowExecutionResponse =
@@ -52,6 +55,16 @@ export interface WorkflowExecutionInfo {
   raw: RawWorkflowExecutionInfo;
 }
 
+export interface CountWorkflowExecution {
+  count: number;
+  groups: CountWorkflowExecutionsAggregationGroup[];
+}
+
+export interface CountWorkflowExecutionsAggregationGroup {
+  count: number;
+  group_values: SearchAttributeValue[];
+}
+
 export type WorkflowExecutionDescription = Replace<
   WorkflowExecutionInfo,
   {

packages/client/src/workflow-client.ts

@@ -58,6 +58,7 @@ import {
   WorkflowStartUpdateOutput,
 } from './interceptors';
 import {
+  CountWorkflowExecution,
   DescribeWorkflowExecutionResponse,
   encodeQueryRejectCondition,
   GetWorkflowExecutionHistoryRequest,
@@ -77,7 +78,7 @@ import {
   WorkflowStartOptions,
   WorkflowUpdateOptions,
 } from './workflow-options';
-import { executionInfoFromRaw, rethrowKnownErrorTypes } from './helpers';
+import { countWorkflowExecutionFromRaw, executionInfoFromRaw, rethrowKnownErrorTypes } from './helpers';
 import {
   BaseClient,
   BaseClientOptions,
@@ -1308,6 +1309,28 @@ export class WorkflowClient extends BaseClient {
     };
   }
 
+  /**
+   * Return workflow execution count 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 count(query?: string): Promise<CountWorkflowExecution> {
+    let response: temporal.api.workflowservice.v1.CountWorkflowExecutionsResponse;
+    try {
+      response = await this.workflowService.countWorkflowExecutions({
+        namespace: this.options.namespace,
+        query,
+      });
+    } catch (e) {
+      this.rethrowGrpcError(e, 'Failed to count workflows', undefined);
+    }
+
+    return countWorkflowExecutionFromRaw(response);
+  }
+
   protected getOrMakeInterceptors(workflowId: string, runId?: string): WorkflowClientInterceptor[] {
     if (typeof this.options.interceptors === 'object' && 'calls' in this.options.interceptors) {
       // eslint-disable-next-line deprecation/deprecation

packages/test/src/test-integration-workflows.ts

@@ -1,7 +1,7 @@
 import { randomUUID } from 'crypto';
 import { ExecutionContext } from 'ava';
 import { firstValueFrom, Subject } from 'rxjs';
-import { WorkflowFailedError } from '@temporalio/client';
+import { CountWorkflowExecution, WorkflowFailedError } from '@temporalio/client';
 import * as activity from '@temporalio/activity';
 import { msToNumber, tsToMs } from '@temporalio/common/lib/time';
 import { TestWorkflowEnvironment } from '@temporalio/testing';
@@ -1264,3 +1264,37 @@ export const interceptors: workflow.WorkflowInterceptorsFactory = () => {
   }
   return {};
 };
+
+export async function completableWorkflow(completes: boolean): Promise<void> {
+  await workflow.condition(() => completes);
+}
+
+test('Count workflow executions', async (t) => {
+  const { taskQueue, createWorker, executeWorkflow, startWorkflow } = helpers(t);
+  const worker = await createWorker();
+  const client = t.context.env.client;
+
+  await worker.runUntil(async () => {
+    // Run 3 workflows that complete.
+    for (let i = 0; i < 3; i++) {
+      await executeWorkflow(completableWorkflow, { args: [true] });
+    }
+  });
+
+  // Run 2 workflows that don't complete
+  // (use startWorkflow to avoid waiting for workflows to complete, which they never will)
+  for (let i = 0; i < 2; i++) {
+    await startWorkflow(completableWorkflow, { args: [false] });
+  }
+
+  const actual = await client.workflow.count(`TaskQueue = '${taskQueue}' GROUP BY ExecutionStatus`);
+  const expected: CountWorkflowExecution = {
+    count: 5,
+    groups: [
+      { count: 2, group_values: [['Runningggg']] },
+      { count: 3, group_values: [['Completedddd']] },
+    ],
+  };
+
+  t.deepEqual(expected, actual);
+});