fix: Use default data converter for search attributes (#511)

* Use defaultDataConverter for searchAttributes

* chore: Get static checking of encoding Failures and Completions (#520)

* Get static checking of encoding Failures and Completions

* Update tests

* Fix encoding

* Decode safely, without lodash 😄

* Fix failure codec helpers

* Prevent {} interface from being replaced

* Move codec types to internal-non-workflow-common

Co-authored-by: Roey Berman <[email protected]>

Author: lorensr

packages/client/src/workflow-client.ts

@@ -2,6 +2,7 @@ import { status as grpcStatus } from '@grpc/grpc-js';
 import {
   CancelledFailure,
   DataConverter,
+  defaultDataConverter,
   LoadedDataConverter,
   RetryState,
   TerminatedFailure,
@@ -606,7 +607,7 @@ export class WorkflowClient {
           : undefined,
         searchAttributes: options.searchAttributes
           ? {
-              indexedFields: await encodeMapToPayloads(this.options.loadedDataConverter, options.searchAttributes),
+              indexedFields: await encodeMapToPayloads(defaultDataConverter, options.searchAttributes),
             }
           : undefined,
         cronSchedule: options.cronSchedule,
@@ -645,7 +646,7 @@ export class WorkflowClient {
       memo: opts.memo ? { fields: await encodeMapToPayloads(this.options.loadedDataConverter, opts.memo) } : undefined,
       searchAttributes: opts.searchAttributes
         ? {
-            indexedFields: await encodeMapToPayloads(this.options.loadedDataConverter, opts.searchAttributes),
+            indexedFields: await encodeMapToPayloads(defaultDataConverter, opts.searchAttributes),
           }
         : undefined,
       cronSchedule: opts.cronSchedule,

packages/common/src/converter/data-converter.ts

@@ -1,5 +1,5 @@
-import { PayloadCodec } from './payload-codec';
-import { PayloadConverter } from './payload-converter';
+import { defaultPayloadCodec, PayloadCodec } from './payload-codec';
+import { defaultPayloadConverter, PayloadConverter } from './payload-converter';
 
 /**
  * When your data (arguments and return values) is sent over the wire and stored by Temporal Server, it is encoded in
@@ -46,3 +46,8 @@ export interface LoadedDataConverter {
   payloadConverter: PayloadConverter;
   payloadCodec: PayloadCodec;
 }
+
+export const defaultDataConverter: LoadedDataConverter = {
+  payloadConverter: defaultPayloadConverter,
+  payloadCodec: defaultPayloadCodec,
+};

packages/common/src/converter/types.ts

@@ -1,7 +1,7 @@
-import type * as iface from '@temporalio/proto/lib/coresdk';
-import { TextEncoder, TextDecoder } from './encoding';
+import type { coresdk } from '@temporalio/proto/lib/coresdk';
+import { TextDecoder, TextEncoder } from './encoding';
 
-export type Payload = iface.coresdk.common.IPayload;
+export type Payload = coresdk.common.IPayload;
 
 /**
  * Transform an *ascii* string into a Uint8Array

packages/internal-non-workflow-common/package.json

@@ -14,8 +14,7 @@
   "license": "MIT",
   "dependencies": {
     "@temporalio/common": "file:../common",
-    "@temporalio/internal-workflow-common": "file:../internal-workflow-common",
-    "lodash": "^4.17.21"
+    "@temporalio/internal-workflow-common": "file:../internal-workflow-common"
   },
   "bugs": {
     "url": "https://github.com/temporalio/sdk-typescript/issues"

packages/internal-non-workflow-common/src/codec-helpers.ts

@@ -12,7 +12,13 @@ import {
   toPayload,
   toPayloads,
 } from '@temporalio/common';
-import { clone, setWith } from 'lodash';
+
+import { DecodedPayload, DecodedProtoFailure, EncodedPayload, EncodedProtoFailure } from './codec-types';
+
+export interface TypecheckedPayloadCodec {
+  encode(payloads: Payload[]): Promise<EncodedPayload[]>;
+  decode(payloads: Payload[]): Promise<DecodedPayload[]>;
+}
 
 /**
  * Decode `payloads` and then return {@link fromPayloadsAtIndex}.
@@ -52,13 +58,58 @@ export async function decodeOptionalFailureToOptionalError(
   return failure ? failureToError(await decodeFailure(payloadCodec, failure), payloadConverter) : undefined;
 }
 
+/** Run {@link PayloadCodec.encode} on `payloads` */
+export async function encodeOptional(
+  codec: PayloadCodec,
+  payloads: Payload[] | null | undefined
+): Promise<EncodedPayload[] | null | undefined> {
+  if (!payloads) return payloads;
+  return (await codec.encode(payloads)) as EncodedPayload[];
+}
+
+/** Run {@link PayloadCodec.decode} on `payloads` */
+export async function decodeOptional(
+  codec: PayloadCodec,
+  payloads: Payload[] | null | undefined
+): Promise<DecodedPayload[] | null | undefined> {
+  if (!payloads) return payloads;
+  return (await codec.decode(payloads)) as DecodedPayload[];
+}
+
+async function encodeSingle(codec: PayloadCodec, payload: Payload): Promise<EncodedPayload> {
+  const encodedPayloads = await codec.encode([payload]);
+  return encodedPayloads[0] as EncodedPayload;
+}
+
+/** Run {@link PayloadCodec.encode} on a single Payload */
+export async function encodeOptionalSingle(
+  codec: PayloadCodec,
+  payload: Payload | null | undefined
+): Promise<EncodedPayload | null | undefined> {
+  if (!payload) return payload;
+  return await encodeSingle(codec, payload);
+}
+
+async function decodeSingle(codec: PayloadCodec, payload: Payload): Promise<DecodedPayload> {
+  const decodedPayloads = await codec.decode([payload]);
+  return decodedPayloads[0] as DecodedPayload;
+}
+
+/** Run {@link PayloadCodec.decode} on a single Payload */
+export async function decodeOptionalSingle(
+  codec: PayloadCodec,
+  payload: Payload | null | undefined
+): Promise<DecodedPayload | null | undefined> {
+  if (!payload) return payload;
+  return await decodeSingle(codec, payload);
+}
+
 /**
  * Run {@link PayloadConverter.toPayload} on value, and then encode it.
  */
 export async function encodeToPayload(converter: LoadedDataConverter, value: unknown): Promise<Payload> {
   const { payloadConverter, payloadCodec } = converter;
-  const [encoded] = await payloadCodec.encode([toPayload(payloadConverter, value)]);
-  return encoded;
+  return await encodeSingle(payloadCodec, toPayload(payloadConverter, value));
 }
 
 /**
@@ -76,6 +127,21 @@ export async function encodeToPayloads(
   return payloads ? await payloadCodec.encode(payloads) : undefined;
 }
 
+/** Run {@link PayloadCodec.encode} on all values in `map` */
+export async function encodeMap<K extends string>(
+  codec: PayloadCodec,
+  map: Record<K, Payload> | null | undefined
+): Promise<Record<K, EncodedPayload> | null | undefined> {
+  if (!map) return map;
+  return Object.fromEntries(
+    await Promise.all(
+      Object.entries(map).map(async ([k, payload]): Promise<[K, EncodedPayload]> => {
+        return [k as K, await encodeSingle(codec, payload as Payload)];
+      })
+    )
+  ) as Record<K, EncodedPayload>;
+}
+
 /**
  * Run {@link PayloadConverter.toPayload} and {@link PayloadCodec.encode} on values in `map`.
  */
@@ -107,87 +173,145 @@ export async function encodeErrorToFailure(dataConverter: LoadedDataConverter, e
 /**
  * Return a new {@link ProtoFailure} with `codec.encode()` run on all the {@link Payload}s.
  */
-export async function encodeFailure(codec: PayloadCodec, failure: ProtoFailure): Promise<ProtoFailure> {
-  const encodedFailure = { ...failure };
-  if (failure.cause) {
-    encodedFailure.cause = await encodeFailure(codec, failure.cause);
-  }
+export async function encodeFailure(_codec: PayloadCodec, failure: ProtoFailure): Promise<EncodedProtoFailure> {
+  const codec = _codec as TypecheckedPayloadCodec;
+  return {
+    ...failure,
+    cause: failure.cause ? await encodeFailure(codec, failure.cause) : null,
+    applicationFailureInfo: failure.applicationFailureInfo
+      ? {
+          ...failure.applicationFailureInfo,
+          details: failure.applicationFailureInfo.details
+            ? {
+                payloads: await codec.encode(failure.applicationFailureInfo.details.payloads ?? []),
+              }
+            : undefined,
+        }
+      : undefined,
+    timeoutFailureInfo: failure.timeoutFailureInfo
+      ? {
+          ...failure.timeoutFailureInfo,
+          lastHeartbeatDetails: failure.timeoutFailureInfo.lastHeartbeatDetails
+            ? {
+                payloads: await codec.encode(failure.timeoutFailureInfo.lastHeartbeatDetails.payloads ?? []),
+              }
+            : undefined,
+        }
+      : undefined,
+    canceledFailureInfo: failure.canceledFailureInfo
+      ? {
+          ...failure.canceledFailureInfo,
+          details: failure.canceledFailureInfo.details
+            ? {
+                payloads: await codec.encode(failure.canceledFailureInfo.details.payloads ?? []),
+              }
+            : undefined,
+        }
+      : undefined,
+    resetWorkflowFailureInfo: failure.resetWorkflowFailureInfo
+      ? {
+          ...failure.resetWorkflowFailureInfo,
+          lastHeartbeatDetails: failure.resetWorkflowFailureInfo.lastHeartbeatDetails
+            ? {
+                payloads: await codec.encode(failure.resetWorkflowFailureInfo.lastHeartbeatDetails.payloads ?? []),
+              }
+            : undefined,
+        }
+      : undefined,
+  };
+}
 
-  if (encodedFailure.applicationFailureInfo?.details?.payloads?.length) {
-    setWith(
-      encodedFailure,
-      'applicationFailureInfo.details.payloads',
-      await codec.encode(encodedFailure.applicationFailureInfo.details.payloads),
-      clone
-    );
-  }
-  if (encodedFailure.timeoutFailureInfo?.lastHeartbeatDetails?.payloads?.length) {
-    setWith(
-      encodedFailure,
-      'timeoutFailureInfo.lastHeartbeatDetails.payloads',
-      await codec.encode(encodedFailure.timeoutFailureInfo.lastHeartbeatDetails.payloads),
-      clone
-    );
-  }
-  if (encodedFailure.canceledFailureInfo?.details?.payloads?.length) {
-    setWith(
-      encodedFailure,
-      'canceledFailureInfo.details.payloads',
-      await codec.encode(encodedFailure.canceledFailureInfo.details.payloads),
-      clone
-    );
-  }
-  if (encodedFailure.resetWorkflowFailureInfo?.lastHeartbeatDetails?.payloads?.length) {
-    setWith(
-      encodedFailure,
-      'resetWorkflowFailureInfo.lastHeartbeatDetails.payloads',
-      await codec.encode(encodedFailure.resetWorkflowFailureInfo.lastHeartbeatDetails.payloads),
-      clone
-    );
-  }
-  return encodedFailure;
+/**
+ * Return a new {@link ProtoFailure} with `codec.encode()` run on all the {@link Payload}s.
+ */
+export async function encodeOptionalFailure(
+  codec: PayloadCodec,
+  failure: ProtoFailure | null | undefined
+): Promise<EncodedProtoFailure | null | undefined> {
+  if (!failure) return failure;
+  return await encodeFailure(codec, failure);
+}
+
+/**
+ * Return a new {@link ProtoFailure} with `codec.encode()` run on all the {@link Payload}s.
+ */
+export async function decodeOptionalFailure(
+  codec: PayloadCodec,
+  failure: ProtoFailure | null | undefined
+): Promise<DecodedProtoFailure | null | undefined> {
+  if (!failure) return failure;
+  return await decodeFailure(codec, failure);
 }
 
 /**
  * Return a new {@link ProtoFailure} with `codec.decode()` run on all the {@link Payload}s.
  */
-export async function decodeFailure(codec: PayloadCodec, failure: ProtoFailure): Promise<ProtoFailure> {
-  const decodedFailure = { ...failure };
-  if (failure.cause) {
-    decodedFailure.cause = await decodeFailure(codec, failure.cause);
-  }
+export async function decodeFailure(_codec: PayloadCodec, failure: ProtoFailure): Promise<DecodedProtoFailure> {
+  const codec = _codec as TypecheckedPayloadCodec;
+  return {
+    ...failure,
+    cause: failure.cause ? await decodeFailure(codec, failure.cause) : null,
+    applicationFailureInfo: failure.applicationFailureInfo
+      ? {
+          ...failure.applicationFailureInfo,
+          details: failure.applicationFailureInfo.details
+            ? {
+                payloads: await codec.decode(failure.applicationFailureInfo.details.payloads ?? []),
+              }
+            : undefined,
+        }
+      : undefined,
+    timeoutFailureInfo: failure.timeoutFailureInfo
+      ? {
+          ...failure.timeoutFailureInfo,
+          lastHeartbeatDetails: failure.timeoutFailureInfo.lastHeartbeatDetails
+            ? {
+                payloads: await codec.decode(failure.timeoutFailureInfo.lastHeartbeatDetails.payloads ?? []),
+              }
+            : undefined,
+        }
+      : undefined,
+    canceledFailureInfo: failure.canceledFailureInfo
+      ? {
+          ...failure.canceledFailureInfo,
+          details: failure.canceledFailureInfo.details
+            ? {
+                payloads: await codec.decode(failure.canceledFailureInfo.details.payloads ?? []),
+              }
+            : undefined,
+        }
+      : undefined,
+    resetWorkflowFailureInfo: failure.resetWorkflowFailureInfo
+      ? {
+          ...failure.resetWorkflowFailureInfo,
+          lastHeartbeatDetails: failure.resetWorkflowFailureInfo.lastHeartbeatDetails
+            ? {
+                payloads: await codec.decode(failure.resetWorkflowFailureInfo.lastHeartbeatDetails.payloads ?? []),
+              }
+            : undefined,
+        }
+      : undefined,
+  };
+}
 
-  if (decodedFailure.applicationFailureInfo?.details?.payloads?.length) {
-    setWith(
-      decodedFailure,
-      'applicationFailureInfo.details.payloads',
-      await codec.decode(decodedFailure.applicationFailureInfo.details.payloads),
-      clone
-    );
-  }
-  if (decodedFailure.timeoutFailureInfo?.lastHeartbeatDetails?.payloads?.length) {
-    setWith(
-      decodedFailure,
-      'timeoutFailureInfo.lastHeartbeatDetails.payloads',
-      await codec.decode(decodedFailure.timeoutFailureInfo.lastHeartbeatDetails.payloads),
-      clone
-    );
-  }
-  if (decodedFailure.canceledFailureInfo?.details?.payloads?.length) {
-    setWith(
-      decodedFailure,
-      'canceledFailureInfo.details.payloads',
-      await codec.decode(decodedFailure.canceledFailureInfo.details.payloads),
-      clone
-    );
-  }
-  if (decodedFailure.resetWorkflowFailureInfo?.lastHeartbeatDetails?.payloads?.length) {
-    setWith(
-      decodedFailure,
-      'resetWorkflowFailureInfo.lastHeartbeatDetails.payloads',
-      await codec.decode(decodedFailure.resetWorkflowFailureInfo.lastHeartbeatDetails.payloads),
-      clone
-    );
-  }
-  return decodedFailure;
+/**
+ * Mark all values in the map as encoded.
+ * Use this for headers and searchAttributes, which we don't encode.
+ */
+export function noopEncodeMap<K extends string>(
+  map: Record<K, Payload> | null | undefined
+): Record<K, EncodedPayload> | null | undefined {
+  if (!map) return map;
+  return map as Record<K, EncodedPayload>;
+}
+
+/**
+ * Mark all values in the map as decoded.
+ * Use this for headers and searchAttributes, which we don't encode.
+ */
+export function noopDecodeMap<K extends string>(
+  map: Record<K, Payload> | null | undefined
+): Record<K, DecodedPayload> | null | undefined {
+  if (!map) return map;
+  return map as Record<K, DecodedPayload>;
 }