feat(core): Allow all CustomOrderProcess handlers to be async functions

packages/core/src/common/finite-state-machine/finite-state-machine.ts

@@ -1,5 +1,7 @@
 import { Observable } from 'rxjs';
 
+import { awaitPromiseOrObservable } from '../utils';
+
 /**
  * @description
  * A type which is used to define all valid transitions and transition callbacks
@@ -30,8 +32,8 @@ export type StateMachineConfig<T extends string, Data = undefined> = {
         toState: T,
         data: Data,
     ): boolean | string | void | Promise<boolean | string | void> | Observable<boolean | string | void>;
-    onTransitionEnd?(fromState: T, toState: T, data: Data): void | Promise<void>;
-    onError?(fromState: T, toState: T, message?: string): void;
+    onTransitionEnd?(fromState: T, toState: T, data: Data): void | Promise<void> | Observable<void>;
+    onError?(fromState: T, toState: T, message?: string): void | Promise<void> | Observable<void>;
 };
 
 /**
@@ -73,14 +75,13 @@ export class FSM<T extends string, Data = any> {
             // If the onTransitionStart callback is defined, invoke it. If it returns false,
             // then the transition will be cancelled.
             if (typeof this.config.onTransitionStart === 'function') {
-                const transitionResult = this.config.onTransitionStart(this._currentState, state, data);
-                const canTransition = await (transitionResult instanceof Observable
-                    ? transitionResult.toPromise()
-                    : transitionResult);
+                const canTransition = await awaitPromiseOrObservable(
+                    this.config.onTransitionStart(this._currentState, state, data),
+                );
                 if (canTransition === false) {
                     return;
                 } else if (typeof canTransition === 'string') {
-                    this.onError(this._currentState, state, canTransition);
+                    await this.onError(this._currentState, state, canTransition);
                     return;
                 }
             }
@@ -89,7 +90,7 @@ export class FSM<T extends string, Data = any> {
             this._currentState = state;
             // If the onTransitionEnd callback is defined, invoke it.
             if (typeof this.config.onTransitionEnd === 'function') {
-                await this.config.onTransitionEnd(fromState, state, data);
+                await awaitPromiseOrObservable(this.config.onTransitionEnd(fromState, state, data));
             }
         } else {
             return this.onError(this._currentState, state);
@@ -118,9 +119,9 @@ export class FSM<T extends string, Data = any> {
         return -1 < this.config.transitions[this._currentState].to.indexOf(state);
     }
 
-    private onError(fromState: T, toState: T, message?: string) {
+    private async onError(fromState: T, toState: T, message?: string) {
         if (typeof this.config.onError === 'function') {
-            return this.config.onError(fromState, toState, message);
+            await awaitPromiseOrObservable(this.config.onError(fromState, toState, message));
         }
     }
 }

packages/core/src/common/utils.ts

@@ -1,5 +1,6 @@
 import { AssetType } from '@vendure/common/lib/generated-types';
 import { ID } from '@vendure/common/lib/shared-types';
+import { Observable } from 'rxjs';
 
 /**
  * Takes a predicate function and returns a negated version.
@@ -13,7 +14,7 @@ export function not(predicate: (...args: any[]) => boolean) {
  * as determined by a === equality check on the given compareBy property.
  */
 export function foundIn<T>(set: T[], compareBy: keyof T) {
-    return (item: T) => set.some(t => t[compareBy] === item[compareBy]);
+    return (item: T) => set.some((t) => t[compareBy] === item[compareBy]);
 }
 
 /**
@@ -62,3 +63,15 @@ export function getAssetType(mimeType: string): AssetType {
 export function normalizeEmailAddress(input: string): string {
     return input.trim().toLowerCase();
 }
+
+/**
+ * Converts a value that may be wrapped into a Promise or Observable into a Promise-wrapped
+ * value.
+ */
+export async function awaitPromiseOrObservable<T>(value: T | Promise<T> | Observable<T>): Promise<T> {
+    let result = await value;
+    if (result instanceof Observable) {
+        result = await result.toPromise();
+    }
+    return result;
+}

packages/core/src/service/helpers/order-state-machine/order-state-machine.ts

@@ -1,6 +1,7 @@
 import { Injectable } from '@nestjs/common';
 import { InjectConnection } from '@nestjs/typeorm';
 import { HistoryEntryType } from '@vendure/common/lib/generated-types';
+import { Observable } from 'rxjs';
 import { Connection } from 'typeorm';
 
 import { RequestContext } from '../../../api/common/request-context';
@@ -12,6 +13,7 @@ import {
 } from '../../../common/finite-state-machine/finite-state-machine';
 import { mergeTransitionDefinitions } from '../../../common/finite-state-machine/merge-transition-definitions';
 import { validateTransitionDefinition } from '../../../common/finite-state-machine/validate-transition-definition';
+import { awaitPromiseOrObservable } from '../../../common/utils';
 import { ConfigService } from '../../../config/config.service';
 import { Order } from '../../../entity/order/order.entity';
 import { HistoryService } from '../../services/history.service';
@@ -142,7 +144,9 @@ export class OrderStateMachine {
             onTransitionStart: async (fromState, toState, data) => {
                 for (const process of customProcesses) {
                     if (typeof process.onTransitionStart === 'function') {
-                        const result = await process.onTransitionStart(fromState, toState, data);
+                        const result = await awaitPromiseOrObservable(
+                            process.onTransitionStart(fromState, toState, data),
+                        );
                         if (result === false || typeof result === 'string') {
                             return result;
                         }
@@ -153,15 +157,15 @@ export class OrderStateMachine {
             onTransitionEnd: async (fromState, toState, data) => {
                 for (const process of customProcesses) {
                     if (typeof process.onTransitionEnd === 'function') {
-                        await process.onTransitionEnd(fromState, toState, data);
+                        await awaitPromiseOrObservable(process.onTransitionEnd(fromState, toState, data));
                     }
                 }
                 await this.onTransitionEnd(fromState, toState, data);
             },
-            onError: (fromState, toState, message) => {
+            onError: async (fromState, toState, message) => {
                 for (const process of customProcesses) {
                     if (typeof process.onError === 'function') {
-                        process.onError(fromState, toState, message);
+                        await awaitPromiseOrObservable(process.onError(fromState, toState, message));
                     }
                 }
                 throw new IllegalOperationError(message || 'error.cannot-transition-order-from-to', {