Typed search attributes

packages/client/src/helpers.ts

@@ -1,10 +1,10 @@
 import { ServiceError as GrpcServiceError, status as grpcStatus } from '@grpc/grpc-js';
 import {
+  decodeSearchAttributes,
+  decodeTypedSearchAttributes,
   LoadedDataConverter,
-  mapFromPayloads,
   NamespaceNotFoundError,
   searchAttributePayloadConverter,
-  SearchAttributes,
 } from '@temporalio/common';
 import { Replace } from '@temporalio/common/lib/type-helpers';
 import { optionalTsToDate, requiredTsToDate } from '@temporalio/common/lib/time';
@@ -71,11 +71,8 @@ export async function executionInfoFromRaw<T>(
     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
-    ),
+    searchAttributes: decodeSearchAttributes(raw.searchAttributes?.indexedFields),
+    typedSearchAttributes: decodeTypedSearchAttributes(raw.searchAttributes?.indexedFields),
     parentExecution: raw.parentExecution
       ? {
           workflowId: raw.parentExecution.workflowId!,

packages/client/src/schedule-client.ts

@@ -1,6 +1,11 @@
 import { status as grpcStatus } from '@grpc/grpc-js';
 import { v4 as uuid4 } from 'uuid';
-import { mapToPayloads, searchAttributePayloadConverter, Workflow } from '@temporalio/common';
+import {
+  decodeSearchAttributes,
+  decodeTypedSearchAttributes,
+  encodeUnifiedSearchAttributes,
+  Workflow,
+} from '@temporalio/common';
 import { composeInterceptors, Headers } from '@temporalio/common/lib/interceptors';
 import {
   encodeMapToPayloads,
@@ -39,7 +44,6 @@
   decodeScheduleRecentActions,
   decodeScheduleRunningActions,
   decodeScheduleSpec,
-  decodeSearchAttributes,
   encodeScheduleAction,
   encodeSchedulePolicies,
   encodeScheduleSpec,
@@ -238,11 +242,12 @@
         state: encodeScheduleState(opts.state),
       },
       memo: opts.memo ? { fields: await encodeMapToPayloads(this.dataConverter, opts.memo) } : undefined,
-      searchAttributes: opts.searchAttributes
-        ? {
-            indexedFields: mapToPayloads(searchAttributePayloadConverter, opts.searchAttributes),
-          }
-        : undefined,
+      searchAttributes:
+        opts.searchAttributes || opts.typedSearchAttributes
+          ? {
+              indexedFields: encodeUnifiedSearchAttributes(opts.searchAttributes, opts.typedSearchAttributes),
+            }
+          : undefined,
       initialPatch: {
         triggerImmediately: opts.state?.triggerImmediately
           ? { overlapPolicy: temporal.api.enums.v1.ScheduleOverlapPolicy.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL }
@@ -388,7 +393,8 @@
             workflowType: raw.info.workflowType.name,
           },
           memo: await decodeMapFromPayloads(this.dataConverter, raw.memo?.fields),
-          searchAttributes: decodeSearchAttributes(raw.searchAttributes),
+          searchAttributes: decodeSearchAttributes(raw.searchAttributes?.indexedFields),
+          typedSearchAttributes: decodeTypedSearchAttributes(raw.searchAttributes?.indexedFields),
           state: {
             paused: raw.info?.paused === true,
             note: raw.info?.notes ?? undefined,
@@ -425,7 +431,8 @@
           spec: decodeScheduleSpec(raw.schedule.spec),
           action: await decodeScheduleAction(this.client.dataConverter, raw.schedule.action),
           memo: await decodeMapFromPayloads(this.client.dataConverter, raw.memo?.fields),
-          searchAttributes: decodeSearchAttributes(raw.searchAttributes),
+          searchAttributes: decodeSearchAttributes(raw.searchAttributes?.indexedFields),
+          typedSearchAttributes: decodeTypedSearchAttributes(raw.searchAttributes?.indexedFields),
           policies: {
             // 'overlap' should never be missing on describe, as the server will replace UNSPECIFIED by an actual value
             overlap: decodeScheduleOverlapPolicy(raw.schedule.policies?.overlapPolicy) ?? ScheduleOverlapPolicy.SKIP,

packages/client/src/schedule-helpers.ts

@@ -1,13 +1,12 @@
 import Long from 'long'; // eslint-disable-line import/no-named-as-default
 import {
   compileRetryPolicy,
+  decodeSearchAttributes,
+  decodeTypedSearchAttributes,
   decompileRetryPolicy,
+  encodeUnifiedSearchAttributes,
   extractWorkflowType,
   LoadedDataConverter,
-  mapFromPayloads,
-  mapToPayloads,
-  searchAttributePayloadConverter,
-  SearchAttributes,
 } from '@temporalio/common';
 import { Headers } from '@temporalio/common/lib/interceptors';
 import {
@@ -260,11 +259,12 @@
       workflowTaskTimeout: msOptionalToTs(action.workflowTaskTimeout),
       retryPolicy: action.retry ? compileRetryPolicy(action.retry) : undefined,
       memo: action.memo ? { fields: await encodeMapToPayloads(dataConverter, action.memo) } : undefined,
-      searchAttributes: action.searchAttributes
-        ? {
-            indexedFields: mapToPayloads(searchAttributePayloadConverter, action.searchAttributes),
-          }
-        : undefined,
+      searchAttributes:
+        action.searchAttributes || action.typedSearchAttributes
+          ? {
+              indexedFields: encodeUnifiedSearchAttributes(action.searchAttributes, action.typedSearchAttributes),
+            }
+          : undefined,
       header: { fields: headers },
     },
   };
@@ -326,14 +326,8 @@
       args: await decodeArrayFromPayloads(dataConverter, pb.startWorkflow.input?.payloads),
       memo: await decodeMapFromPayloads(dataConverter, pb.startWorkflow.memo?.fields),
       retry: decompileRetryPolicy(pb.startWorkflow.retryPolicy),
-      searchAttributes: Object.fromEntries(
-        Object.entries(
-          mapFromPayloads(
-            searchAttributePayloadConverter,
-            pb.startWorkflow.searchAttributes?.indexedFields ?? {}
-          ) as SearchAttributes
-        )
-      ),
+      searchAttributes: decodeSearchAttributes(pb.startWorkflow.searchAttributes?.indexedFields),
+      typedSearchAttributes: decodeTypedSearchAttributes(pb.startWorkflow.searchAttributes?.indexedFields),
       workflowExecutionTimeout: optionalTsToMs(pb.startWorkflow.workflowExecutionTimeout),
       workflowRunTimeout: optionalTsToMs(pb.startWorkflow.workflowRunTimeout),
       workflowTaskTimeout: optionalTsToMs(pb.startWorkflow.workflowTaskTimeout),
@@ -342,17 +336,6 @@
   throw new TypeError('Unsupported schedule action');
 }
 
-export function decodeSearchAttributes(
-  pb: temporal.api.common.v1.ISearchAttributes | undefined | null
-): SearchAttributes {
-  if (!pb?.indexedFields) return {};
-  return Object.fromEntries(
-    Object.entries(mapFromPayloads(searchAttributePayloadConverter, pb.indexedFields) as SearchAttributes).filter(
-      ([_, v]) => v && v.length > 0
-    ) // Filter out empty arrays returned by pre 1.18 servers
-  );
-}
-
 export function decodeScheduleRunningActions(
   pb?: temporal.api.common.v1.IWorkflowExecution[] | null
 ): ScheduleExecutionStartWorkflowActionResult[] {

packages/client/src/schedule-types.ts

@@ -1,5 +1,11 @@
 import { checkExtends, Replace } from '@temporalio/common/lib/type-helpers';
-import { Duration, SearchAttributes, Workflow } from '@temporalio/common';
+import {
+  Duration,
+  SearchAttributes,
+  Workflow,
+  TypedSearchAttributePair,
+  TypedSearchAttributes,
+} from '@temporalio/common';
 import { makeProtoEnumConverters } from '@temporalio/common/lib/internal-workflow';
 import type { temporal } from '@temporalio/proto';
 import { WorkflowStartOptions } from './workflow-options';
@@ -70,9 +76,22 @@ export interface ScheduleOptions<A extends ScheduleOptionsAction = ScheduleOptio
    * https://docs.temporal.io/docs/typescript/search-attributes
    *
    * Values are always converted using {@link JsonPayloadConverter}, even when a custom Data Converter is provided.
+   *
+   * @deprecated Use {@link typedSearchAttributes} instead.
    */
   searchAttributes?: SearchAttributes;
 
+  /**
+   * Additional indexed information attached to the Schedule. More info:
+   * https://docs.temporal.io/docs/typescript/search-attributes
+   *
+   * Values are always converted using {@link JsonPayloadConverter}, even when a custom Data Converter is provided.
+   *
+   * If both {@link searchAttributes} and {@link typedSearchAttributes} are provided, conflicting keys will be overwritten
+   * by {@link typedSearchAttributes}.
+   */
+  typedSearchAttributes?: TypedSearchAttributePair[] | TypedSearchAttributes;
+
   /**
    * The initial state of the schedule, right after creation or update.
    */
@@ -128,8 +147,9 @@ export type CompiledScheduleOptions = Replace<
 /**
  * The specification of an updated Schedule, as expected by {@link ScheduleHandle.update}.
  */
+// TODO(thomas): support search attributes on update (ts issue #1475)
 export type ScheduleUpdateOptions<A extends ScheduleOptionsAction = ScheduleOptionsAction> = Replace<
-  Omit<ScheduleOptions, 'scheduleId' | 'memo' | 'searchAttributes'>,
+  Omit<ScheduleOptions, 'scheduleId' | 'memo' | 'searchAttributes' | 'typedSearchAttributes'>,
   {
     action: A;
     state: Omit<ScheduleOptions['state'], 'triggerImmediately' | 'backfill'>;
@@ -172,13 +192,23 @@ export interface ScheduleSummary {
   memo?: Record<string, unknown>;
 
   /**
-   * Additional indexed information attached to the Schedule.
-   * More info: https://docs.temporal.io/docs/typescript/search-attributes
+   * Additional indexed information attached to the Schedule. More info:
+   * https://docs.temporal.io/docs/typescript/search-attributes
    *
    * Values are always converted using {@link JsonPayloadConverter}, even when a custom Data Converter is provided.
+   *
+   * @deprecated Use {@link typedSearchAttributes} instead.
    */
   searchAttributes?: SearchAttributes;
 
+  /**
+   * Additional indexed information attached to the Schedule. More info:
+   * https://docs.temporal.io/docs/typescript/search-attributes
+   *
+   * Values are always converted using {@link JsonPayloadConverter}, even when a custom Data Converter is provided.
+   */
+  typedSearchAttributes?: TypedSearchAttributes;
+
   state: {
     /**
      * Whether Schedule is currently paused.
@@ -284,13 +314,23 @@ export type ScheduleDescription = {
   memo?: Record<string, unknown>;
 
   /**
-   * Additional indexed information attached to the Schedule.
-   * More info: https://docs.temporal.io/docs/typescript/search-attributes
+   * Additional indexed information attached to the Schedule. More info:
+   * https://docs.temporal.io/docs/typescript/search-attributes
    *
    * Values are always converted using {@link JsonPayloadConverter}, even when a custom Data Converter is provided.
+   *
+   * @deprecated Use {@link typedSearchAttributes} instead.
    */
   searchAttributes: SearchAttributes;
 
+  /**
+   * Additional indexed information attached to the Schedule. More info:
+   * https://docs.temporal.io/docs/typescript/search-attributes
+   *
+   * Values are always converted using {@link JsonPayloadConverter}, even when a custom Data Converter is provided.
+   */
+  typedSearchAttributes: TypedSearchAttributes;
+
   state: {
     /**
      * Whether Schedule is currently paused.
@@ -745,6 +785,7 @@ export type ScheduleOptionsStartWorkflowAction<W extends Workflow> = {
   | 'args'
   | 'memo'
   | 'searchAttributes'
+  | 'typedSearchAttributes'
   | 'retry'
   | 'workflowExecutionTimeout'
   | 'workflowRunTimeout'
@@ -776,6 +817,7 @@ export type ScheduleDescriptionStartWorkflowAction = ScheduleSummaryStartWorkflo
     | 'args'
     | 'memo'
     | 'searchAttributes'
+    | 'typedSearchAttributes'
     | 'retry'
     | 'workflowExecutionTimeout'
     | 'workflowRunTimeout'

packages/client/src/types.ts

@@ -1,5 +1,5 @@
 import type * as grpc from '@grpc/grpc-js';
-import type { SearchAttributes, SearchAttributeValue } from '@temporalio/common';
+import type { SearchAttributes, SearchAttributeValue, TypedSearchAttributes } 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';
@@ -47,7 +47,9 @@ export interface WorkflowExecutionInfo {
   executionTime?: Date;
   closeTime?: Date;
   memo?: Record<string, unknown>;
+  /** @deprecated Use {@link typedSearchAttributes} instead. */
   searchAttributes: SearchAttributes;
+  typedSearchAttributes: TypedSearchAttributes;
   parentExecution?: Required<proto.temporal.api.common.v1.IWorkflowExecution>;
   raw: RawWorkflowExecutionInfo;
 }

packages/client/src/workflow-client.ts

@@ -4,11 +4,9 @@
   BaseWorkflowHandle,
   CancelledFailure,
   compileRetryPolicy,
-  mapToPayloads,
   HistoryAndWorkflowId,
   QueryDefinition,
   RetryState,
-  searchAttributePayloadConverter,
   SignalDefinition,
   UpdateDefinition,
   TerminatedFailure,
@@ -24,6 +22,7 @@
   decodeRetryState,
   encodeWorkflowIdConflictPolicy,
   WorkflowIdConflictPolicy,
+  encodeUnifiedSearchAttributes,
 } from '@temporalio/common';
 import { composeInterceptors } from '@temporalio/common/lib/interceptors';
 import { History } from '@temporalio/common/lib/proto-utils';
@@ -1218,11 +1217,12 @@
       workflowStartDelay: options.startDelay,
       retryPolicy: options.retry ? compileRetryPolicy(options.retry) : undefined,
       memo: options.memo ? { fields: await encodeMapToPayloads(this.dataConverter, options.memo) } : undefined,
-      searchAttributes: options.searchAttributes
-        ? {
-            indexedFields: mapToPayloads(searchAttributePayloadConverter, options.searchAttributes),
-          }
-        : undefined,
+      searchAttributes:
+        options.searchAttributes || options.typedSearchAttributes
+          ? {
+              indexedFields: encodeUnifiedSearchAttributes(options.searchAttributes, options.typedSearchAttributes),
+            }
+          : undefined,
       cronSchedule: options.cronSchedule,
       header: { fields: headers },
     };
@@ -1265,6 +1265,7 @@
   protected async createStartWorkflowRequest(input: WorkflowStartInput): Promise<StartWorkflowExecutionRequest> {
     const { options: opts, workflowType, headers } = input;
     const { identity, namespace } = this.options;
+
     return {
       namespace,
       identity,
@@ -1284,11 +1285,12 @@
       workflowStartDelay: opts.startDelay,
       retryPolicy: opts.retry ? compileRetryPolicy(opts.retry) : undefined,
       memo: opts.memo ? { fields: await encodeMapToPayloads(this.dataConverter, opts.memo) } : undefined,
-      searchAttributes: opts.searchAttributes
-        ? {
-            indexedFields: mapToPayloads(searchAttributePayloadConverter, opts.searchAttributes),
-          }
-        : undefined,
+      searchAttributes:
+        opts.searchAttributes || opts.typedSearchAttributes
+          ? {
+              indexedFields: encodeUnifiedSearchAttributes(opts.searchAttributes, opts.typedSearchAttributes),
+            }
+          : undefined,
       cronSchedule: opts.cronSchedule,
       header: { fields: headers },
     };