Language interoperability

schemas/system_db_schema.ts

@@ -1,3 +1,5 @@
+export type SysDBSerializationFormat = string;
+
 export interface workflow_status {
   workflow_uuid: string;
   status: string;
@@ -26,18 +28,21 @@ export interface workflow_status {
   queue_partition_key?: string; // Partition key for partitioned queues.
   forked_from?: string;
   owner_xid?: string;
+  serialization: SysDBSerializationFormat | null;
 }
 
 export interface notifications {
   destination_uuid: string;
   topic: string;
   message: string;
+  serialization: SysDBSerializationFormat | null;
 }
 
 export interface workflow_events {
   workflow_uuid: string;
   key: string;
   value: string;
+  serialization: SysDBSerializationFormat | null;
 }
 
 export interface operation_outputs {
@@ -49,6 +54,7 @@ export interface operation_outputs {
   function_name?: string;
   started_at_epoch_ms?: number;
   completed_at_epoch_ms?: number;
+  serialization: SysDBSerializationFormat | null; // Relevant only to getEvent / recv / etc.
 }
 
 export interface event_dispatch_kv {
@@ -69,13 +75,15 @@ export interface streams {
   value: string;
   offset: number;
   function_id: number;
+  serialization: SysDBSerializationFormat | null;
 }
 
 export interface workflow_events_history {
   workflow_uuid: string;
   function_id: number;
   key: string;
   value: string;
+  serialization: SysDBSerializationFormat | null;
 }
 
 // This is the deserialized version of operation_outputs
@@ -88,3 +96,42 @@ export interface step_info {
   started_at_epoch_ms?: number;
   completed_at_epoch_ms?: number;
 }
+
+// This is system DB schema for portable inputs / outputs / messages / events / errors
+
+// ---------- Canonical JSON value space ----------
+// Note the absensce of "Date", etc.
+// Canonical Date = RFC 3339 / ISO-8601 UTC string: YYYY-MM-DDTHH:mm:ss(.sss)Z
+// This can be fixed with AJV (applied later)
+export type JsonPrimitive = null | boolean | number | string;
+export type JsonValue = JsonPrimitive | JsonObject | JsonArray;
+export type JsonObject = { [k: string]: JsonValue };
+export type JsonArray = JsonValue[];
+
+// ---------- Workflow args + result ----------
+export type JsonWorkflowArgs = {
+  positionalArgs?: JsonArray;
+  namedArgs?: JsonObject;
+};
+export type JsonWorkflowResult = JsonValue;
+export interface JsonWorkflowErrorData {
+  name: string; // Error name
+  message: string; // Human-readable ;-) string
+  code?: number | string;
+  data?: JsonValue; // structured details (retryable, origin, etc.)
+}
+
+export class PortableWorkflowError extends Error {
+  constructor(
+    message: string,
+    readonly name: string,
+    readonly code?: number | string,
+    readonly data?: JsonValue,
+  ) {
+    super(message);
+  }
+}
+
+// --------- Notification(Message) and WF event
+export type JsonMessage = JsonValue;
+export type JsonEvent = JsonValue;

src/client.ts

@@ -14,10 +14,18 @@ import {
   StatusString,
   type StepInfo,
   type WorkflowHandle,
+  WorkflowSerializationFormat,
   type WorkflowStatus,
 } from './workflow';
 import { sleepms } from './utils';
-import { DBOSJSON, DBOSSerializer } from './serialization';
+import {
+  DBOSJSON,
+  DBOSSerializer,
+  deserializePositionalArgs,
+  deserializeValue,
+  serializeArgs,
+  serializeValue,
+} from './serialization';
 import {
   forkWorkflow,
   getWorkflow,
@@ -75,6 +83,14 @@ interface ClientEnqueueOptions {
    */
   deduplicationID?: string;
 
+  /**
+   * Serialization to use for enqueued request
+   *   Default is to use the serialization for JS/TS, as this is the most flexible
+   *   If `portable_json` is specified, a more limited JSON serialization is used,
+   *    allowing cross-language enqueues of workflows with simple semantics
+   */
+  serializationType?: WorkflowSerializationFormat;
+
   /**
    * An optional priority for the workflow.
    * Workflows with higher priority will be dequeued first.
@@ -87,6 +103,19 @@ interface ClientEnqueueOptions {
   queuePartitionKey?: string;
 }
 
+/**
+ * Options for client send
+ */
+interface ClientSendOptions {
+  /**
+   * Serialization to use for sent message
+   *   Default is to use the serialization for TS/JS, as this is the most flexible
+   *   If `portable_json` is specified, a more limited JSON serialization is used,
+   *     allowing cross-language message sends
+   */
+  serializationType?: WorkflowSerializationFormat;
+}
+
 export class ClientHandle<R> implements WorkflowHandle<R> {
   constructor(
     readonly systemDatabase: SystemDatabase,
@@ -116,7 +145,7 @@ export class ClientHandle<R> implements WorkflowHandle<R> {
 
   async getWorkflowInputs<T extends unknown[]>(): Promise<T> {
     const status = (await this.systemDatabase.getWorkflowStatus(this.workflowUUID)) as WorkflowStatusInternal;
-    return this.systemDatabase.getSerializer().parse(status.input) as T;
+    return deserializePositionalArgs(status.input, status.serialization, this.systemDatabase.getSerializer()) as T;
   }
 }
 
@@ -184,6 +213,7 @@ export class DBOSClient {
     const { workflowName, workflowClassName, workflowConfigName, queueName, appVersion } = options;
     const workflowUUID = options.workflowID ?? randomUUID();
 
+    const serparam = serializeArgs(args, undefined, this.serializer, options?.serializationType);
     const internalStatus: WorkflowStatusInternal = {
       workflowUUID: workflowUUID,
       status: StatusString.ENQUEUED,
@@ -203,10 +233,11 @@ export class DBOSClient {
       createdAt: Date.now(),
       timeoutMS: options.workflowTimeoutMS,
       deadlineEpochMS: undefined,
-      input: this.serializer.stringify(args),
+      input: serparam.serializedValue,
       deduplicationID: options.deduplicationID,
       priority: options.priority ?? 0,
       queuePartitionKey: options.queuePartitionKey,
+      serialization: serparam.serialization,
     };
 
     await this.systemDatabase.initWorkflowStatus(internalStatus, null);
@@ -222,8 +253,21 @@ export class DBOSClient {
    * @param idempotencyKey - An optional idempotency key to ensure that the message is only sent once.
    * @returns A Promise that resolves when the message has been sent.
    */
-  async send<T>(destinationID: string, message: T, topic?: string, idempotencyKey?: string): Promise<void> {
+  async send<T>(
+    destinationID: string,
+    message: T,
+    topic?: string,
+    idempotencyKey?: string,
+    options?: ClientSendOptions,
+  ): Promise<void> {
     idempotencyKey ??= randomUUID();
+    const sermsg = serializeValue(message, this.serializer, options?.serializationType);
+    const srwfp = serializeArgs(
+      [destinationID, message, topic, options?.serializationType],
+      undefined,
+      this.serializer,
+      options?.serializationType,
+    );
     const internalStatus: WorkflowStatusInternal = {
       workflowUUID: `${destinationID}-${idempotencyKey}`,
       status: StatusString.SUCCESS,
@@ -239,18 +283,20 @@ export class DBOSClient {
       executorId: '',
       applicationID: '',
       createdAt: Date.now(),
-      input: this.serializer.stringify([destinationID, message, topic]),
+      input: srwfp.serializedValue,
       deduplicationID: undefined,
       priority: 0,
       queuePartitionKey: undefined,
+      serialization: srwfp.serialization,
     };
     await this.systemDatabase.initWorkflowStatus(internalStatus, null);
     await this.systemDatabase.send(
       internalStatus.workflowUUID,
       0,
       destinationID,
-      this.serializer.stringify(message),
+      sermsg.serializedValue,
       topic,
+      sermsg.serialization,
     );
   }
 
@@ -262,7 +308,8 @@ export class DBOSClient {
    * @returns A Promise that resolves with the event payload.
    */
   async getEvent<T>(workflowID: string, key: string, timeoutSeconds?: number): Promise<T | null> {
-    return this.serializer.parse(await this.systemDatabase.getEvent(workflowID, key, timeoutSeconds ?? 60)) as T;
+    const evt = await this.systemDatabase.getEvent(workflowID, key, timeoutSeconds ?? 60);
+    return deserializeValue(evt.serializedValue, evt.serialization, this.serializer) as T;
   }
 
   /**
@@ -320,10 +367,10 @@ export class DBOSClient {
     while (true) {
       try {
         const value = await this.systemDatabase.readStream(workflowID, key, offset);
-        if (value === DBOS_STREAM_CLOSED_SENTINEL) {
+        if (value.serializedValue === DBOS_STREAM_CLOSED_SENTINEL) {
           break;
         }
-        yield value as T;
+        yield deserializeValue(value.serializedValue, value.serialization, this.serializer) as T;
         offset += 1;
       } catch (error: unknown) {
         if (error instanceof Error && error.message.includes('No value found')) {

src/context.ts

@@ -5,6 +5,7 @@ import { AsyncLocalStorage } from 'async_hooks';
 import { DBOSInvalidWorkflowTransitionError } from './error';
 import Koa from 'koa';
 import { DBOSExecutor } from './dbos-executor';
+import { WorkflowSerializationFormat } from './workflow';
 
 export interface StepStatus {
   stepID: number;
@@ -23,6 +24,7 @@ export interface DBOSContextOptions {
   operationType?: string; // A custom helper for users to set a operation type of their choice. Intended for functions setting a pctx to run DBOS operations from.
   operationCaller?: string; // This is made to pass through the operationName to DBOS contexts, and potentially the caller span name.
   workflowTimeoutMS?: number | null;
+  serializationType?: WorkflowSerializationFormat;
 }
 
 export interface DBOSLocalCtx extends DBOSContextOptions {