import type { temporal } from '@temporalio/proto'; import type { Payload } from '../interfaces'; import { arrayFromPayloads, convertOptionalToPayload, fromPayloadsAtIndex, toPayloadWithTypeInfo, toPayloadsWithContext, } from '../converter/payload-converter'; import { PayloadConverterError } from '../errors'; import type { PayloadCodec } from '../converter/payload-codec'; import type { ProtoFailure } from '../failure'; import type { LoadedDataConverter } from '../converter/data-converter'; import type { UserMetadata } from '../user-metadata'; import type { TypeInfo } from '../type-info'; import type { SerializationContext } from '../converter/serialization-context'; import type { DecodedPayload, DecodedProtoFailure, EncodedPayload, EncodedProtoFailure } from './codec-types'; /** * Decode through each codec, starting with the last codec. */ export async function decode( codecs: PayloadCodec[], payloads: Payload[], context?: SerializationContext ): Promise { for (let i = codecs.length - 1; i >= 0; i--) { payloads = await codecs[i]!.decode(payloads, context); } return payloads as DecodedPayload[]; } /** * Encode through each codec, starting with the first codec. */ export async function encode( codecs: PayloadCodec[], payloads: Payload[], context?: SerializationContext ): Promise { for (let i = 0; i < codecs.length; i++) { payloads = await codecs[i]!.encode(payloads, context); } return payloads as EncodedPayload[]; } /** Run {@link PayloadCodec.decode} on `payloads` */ async function decodeOptional( codecs: PayloadCodec[], payloads: Payload[] | null | undefined, context?: SerializationContext ): Promise { if (payloads == null) return payloads; return await decode(codecs, payloads, context); } async function encodeSingle( codecs: PayloadCodec[], payload: Payload, context?: SerializationContext ): Promise { const encodedPayloads = await encode(codecs, [payload], context); return encodedPayloads[0] as EncodedPayload; } async function decodeSingle( codecs: PayloadCodec[], payload: Payload, context?: SerializationContext ): Promise { const [decodedPayload] = await decode(codecs, [payload], context); return decodedPayload!; } /** Run {@link PayloadCodec.encode} on a single Payload */ async function encodeOptionalSingle( codecs: PayloadCodec[], payload: Payload | null | undefined, context?: SerializationContext ): Promise { if (payload == null) return payload; return await encodeSingle(codecs, payload, context); } /** Run {@link PayloadCodec.decode} on a single Payload */ export async function decodeOptionalSingle( codecs: PayloadCodec[], payload: Payload | null | undefined, context?: SerializationContext ): Promise { if (payload == null) return payload; return await decodeSingle(codecs, payload, context); } /** Run {@link PayloadCodec.decode} and convert from a single Payload */ export async function decodeOptionalSinglePayload( dataConverter: LoadedDataConverter, payload?: Payload | null | undefined, context?: SerializationContext ): Promise { const { payloadConverter, payloadCodecs } = dataConverter; const decoded = await decodeOptionalSingle(payloadCodecs, payload, context); if (decoded == null) return decoded; return payloadConverter.fromPayload(decoded, context); } /** Apply optional TypeInfo, run payload conversion, and then encode the resulting Payload. */ export async function encodeToPayload( converter: LoadedDataConverter, value: T, context?: SerializationContext, typeInfo?: TypeInfo ): Promise { const { payloadConverter, payloadCodecs } = converter; return await encodeSingle(payloadCodecs, toPayloadWithTypeInfo(payloadConverter, value, context, typeInfo), context); } /** * Decode `payloads` and then return {@link arrayFromPayloads}`. */ export async function decodeArrayFromPayloads( converter: LoadedDataConverter, payloads?: Payload[] | null, context?: SerializationContext, typeInfo?: readonly TypeInfo[] ): Promise { const { payloadConverter, payloadCodecs } = converter; return arrayFromPayloads(payloadConverter, await decodeOptional(payloadCodecs, payloads, context), context, typeInfo); } /** * Decode `payloads` and then return {@link fromPayloadsAtIndex}. */ export async function decodeFromPayloadsAtIndex( converter: LoadedDataConverter, index: number, payloads?: Payload[] | null, context?: SerializationContext, typeInfo?: TypeInfo ): Promise { const { payloadConverter, payloadCodecs } = converter; return await fromPayloadsAtIndex( payloadConverter, index, await decodeOptional(payloadCodecs, payloads, context), context, typeInfo ); } /** * Run {@link decodeFailure} and then return {@link failureToError}. */ export async function decodeOptionalFailureToOptionalError( converter: LoadedDataConverter, failure: ProtoFailure | undefined | null, context?: SerializationContext ): Promise { const { failureConverter, payloadConverter, payloadCodecs } = converter; return failure ? failureConverter.failureToError(await decodeFailure(payloadCodecs, failure, context), payloadConverter, context) : undefined; } /** * Run {@link PayloadConverter.toPayload} on values, and then encode them. */ export async function encodeToPayloads( converter: LoadedDataConverter, ...values: unknown[] ): Promise { return encodeToPayloadsWithContext(converter, undefined, values); } /** * Run {@link PayloadConverter.toPayload} on values with an optional serialization context, and then encode them. */ export async function encodeToPayloadsWithContext( converter: LoadedDataConverter, context: SerializationContext | undefined, values: unknown[], typeInfo?: readonly TypeInfo[] ): Promise { const { payloadConverter, payloadCodecs } = converter; if (values.length === 0) { return undefined; } const payloads = toPayloadsWithContext(payloadConverter, context, values, typeInfo); return payloads ? await encode(payloadCodecs, payloads, context) : undefined; } /** * Run {@link PayloadCodec.decode} and then {@link PayloadConverter.fromPayload} on values in `map`. */ export async function decodeMapFromPayloads( converter: LoadedDataConverter, map: Record | null | undefined, context?: SerializationContext ): Promise | undefined> { if (!map) return undefined; const { payloadConverter, payloadCodecs } = converter; return Object.fromEntries( await Promise.all( Object.entries(map).map(async ([k, payload]): Promise<[K, unknown]> => { const [decodedPayload] = await decode(payloadCodecs, [payload as Payload], context); const value = payloadConverter.fromPayload(decodedPayload!, context); return [k as K, value]; }) ) ) as Record; } /** * Run {@link PayloadConverter.toPayload} and then {@link PayloadCodec.encode} on values in `map`. */ export async function encodeMapToPayloads( converter: LoadedDataConverter, map: Record, context?: SerializationContext ): Promise> { const { payloadConverter, payloadCodecs } = converter; return Object.fromEntries( await Promise.all( Object.entries(map).map(async ([k, v]): Promise<[K, Payload]> => { const payload = payloadConverter.toPayload(v, context); if (payload === undefined) throw new PayloadConverterError(`Failed to encode entry: ${k}: ${v}`); const [encodedPayload] = await encode(payloadCodecs, [payload], context); return [k as K, encodedPayload!]; }) ) ) as Record; } /** * Run {@link errorToFailure} on `error`, and then {@link encodeFailure}. */ export async function encodeErrorToFailure( dataConverter: LoadedDataConverter, error: unknown, context?: SerializationContext ): Promise { const { failureConverter, payloadConverter, payloadCodecs } = dataConverter; return await encodeFailure(payloadCodecs, failureConverter.errorToFailure(error, payloadConverter, context), context); } /** * Return a new {@link ProtoFailure} with `codec.encode()` run on all the {@link Payload}s. */ export async function encodeFailure( codecs: PayloadCodec[], failure: ProtoFailure, context?: SerializationContext ): Promise { return { ...failure, encodedAttributes: failure.encodedAttributes ? (await encode(codecs, [failure.encodedAttributes], context))[0] : undefined, cause: failure.cause ? await encodeFailure(codecs, failure.cause, context) : null, applicationFailureInfo: failure.applicationFailureInfo ? { ...failure.applicationFailureInfo, details: failure.applicationFailureInfo.details ? { payloads: await encode(codecs, failure.applicationFailureInfo.details.payloads ?? [], context), } : undefined, } : undefined, timeoutFailureInfo: failure.timeoutFailureInfo ? { ...failure.timeoutFailureInfo, lastHeartbeatDetails: failure.timeoutFailureInfo.lastHeartbeatDetails ? { payloads: await encode(codecs, failure.timeoutFailureInfo.lastHeartbeatDetails.payloads ?? [], context), } : undefined, } : undefined, canceledFailureInfo: failure.canceledFailureInfo ? { ...failure.canceledFailureInfo, details: failure.canceledFailureInfo.details ? { payloads: await encode(codecs, failure.canceledFailureInfo.details.payloads ?? [], context), } : undefined, } : undefined, resetWorkflowFailureInfo: failure.resetWorkflowFailureInfo ? { ...failure.resetWorkflowFailureInfo, lastHeartbeatDetails: failure.resetWorkflowFailureInfo.lastHeartbeatDetails ? { payloads: await encode( codecs, failure.resetWorkflowFailureInfo.lastHeartbeatDetails.payloads ?? [], context ), } : undefined, } : undefined, }; } /** * Return a new {@link ProtoFailure} with `codec.decode()` run on all the {@link Payload}s. */ export async function decodeFailure( codecs: PayloadCodec[], failure: ProtoFailure, context?: SerializationContext ): Promise { return { ...failure, encodedAttributes: failure.encodedAttributes ? (await decode(codecs, [failure.encodedAttributes], context))[0] : undefined, cause: failure.cause ? await decodeFailure(codecs, failure.cause, context) : null, applicationFailureInfo: failure.applicationFailureInfo ? { ...failure.applicationFailureInfo, details: failure.applicationFailureInfo.details ? { payloads: await decode(codecs, failure.applicationFailureInfo.details.payloads ?? [], context), } : undefined, } : undefined, timeoutFailureInfo: failure.timeoutFailureInfo ? { ...failure.timeoutFailureInfo, lastHeartbeatDetails: failure.timeoutFailureInfo.lastHeartbeatDetails ? { payloads: await decode(codecs, failure.timeoutFailureInfo.lastHeartbeatDetails.payloads ?? [], context), } : undefined, } : undefined, canceledFailureInfo: failure.canceledFailureInfo ? { ...failure.canceledFailureInfo, details: failure.canceledFailureInfo.details ? { payloads: await decode(codecs, failure.canceledFailureInfo.details.payloads ?? [], context), } : undefined, } : undefined, resetWorkflowFailureInfo: failure.resetWorkflowFailureInfo ? { ...failure.resetWorkflowFailureInfo, lastHeartbeatDetails: failure.resetWorkflowFailureInfo.lastHeartbeatDetails ? { payloads: await decode( codecs, failure.resetWorkflowFailureInfo.lastHeartbeatDetails.payloads ?? [], context ), } : undefined, } : undefined, }; } export async function encodeUserMetadata( dataConverter: LoadedDataConverter, staticSummary: string | undefined, staticDetails: string | undefined, context?: SerializationContext ): Promise { if (staticSummary == null && staticDetails == null) return undefined; const { payloadConverter, payloadCodecs } = dataConverter; const summary = await encodeOptionalSingle( payloadCodecs, convertOptionalToPayload(payloadConverter, staticSummary, context), context ); const details = await encodeOptionalSingle( payloadCodecs, convertOptionalToPayload(payloadConverter, staticDetails, context), context ); if (summary == null && details == null) return undefined; return { summary, details }; } export async function decodeUserMetadata( dataConverter: LoadedDataConverter, metadata: temporal.api.sdk.v1.IUserMetadata | undefined | null, context?: SerializationContext ): Promise { const res = { staticSummary: undefined, staticDetails: undefined }; if (metadata == null) return res; const staticSummary = (await decodeOptionalSinglePayload(dataConverter, metadata.summary, context)) ?? undefined; const staticDetails = (await decodeOptionalSinglePayload(dataConverter, metadata.details, context)) ?? undefined; return { staticSummary, staticDetails }; }