feat(internals): handle async middlewares/observes

Signed-off-by: Tomas Dvorak <[email protected]>

src/context.ts

@@ -22,7 +22,7 @@ import { Callback } from "@/emitter/types.js";
 import { createNonOverridableObject } from "@/internals/helpers/object.js";
 import { registerSignals } from "@/internals/helpers/cancellation.js";
 import { Serializable } from "@/internals/serializable.js";
-import { LazyPromise } from "@/internals/helpers/promise.js";
+import { executeSequentially, LazyPromise } from "@/internals/helpers/promise.js";
 import { FrameworkError } from "@/errors.js";
 import { shallowCopy } from "@/serializer/utils.js";
 
@@ -41,6 +41,8 @@ export type GetRunContext<T, P = any> = T extends RunInstance ? RunContext<T, P>
 export type GetRunInstance<T> = T extends RunInstance<infer P> ? P : never;
 
 export class Run<R, I extends RunInstance, P = any> extends LazyPromise<R> {
+  protected readonly tasks: (() => Promise<void>)[] = [];
+
   constructor(
     handler: () => Promise<R>,
     protected readonly runContext: GetRunContext<I, P>,
@@ -51,20 +53,27 @@ export class Run<R, I extends RunInstance, P = any> extends LazyPromise<R> {
   readonly [Symbol.toStringTag] = "Promise";
 
   observe(fn: (emitter: Emitter<GetRunInstance<I>>) => void) {
-    fn(this.runContext.emitter);
+    this.tasks.push(async () => fn(this.runContext.emitter));
     return this;
   }
 
   context(value: object) {
-    Object.assign(this.runContext.context, value);
-    Object.assign(this.runContext.emitter.context, value);
+    this.tasks.push(async () => {
+      Object.assign(this.runContext.context, value);
+      Object.assign(this.runContext.emitter.context, value);
+    });
     return this;
   }
 
   middleware(fn: (context: GetRunContext<I, P>) => void) {
-    fn(this.runContext);
+    this.tasks.push(async () => fn(this.runContext));
     return this;
   }
+
+  protected async before(): Promise<void> {
+    await super.before();
+    await executeSequentially(this.tasks.splice(0, Infinity));
+  }
 }
 
 export interface RunContextInput<P> {

src/internals/helpers/promise.ts

@@ -112,21 +112,23 @@ export class LazyPromise<R> implements Promise<R> {
 
   readonly [Symbol.toStringTag] = "Promise";
 
+  protected async before(): Promise<void> {}
+
   then<TResult1 = R, TResult2 = never>(
     onfulfilled?: ((value: R) => PromiseLike<TResult1> | TResult1) | undefined | null,
     onrejected?: ((reason: any) => PromiseLike<TResult2> | TResult2) | undefined | null,
   ): Promise<TResult1 | TResult2> {
-    return this.handler().then(onfulfilled).catch(onrejected);
+    return this.before().then(this.handler).then(onfulfilled).catch(onrejected);
   }
 
   catch<TResult = never>(
     onrejected?: ((reason: any) => PromiseLike<TResult> | TResult) | undefined | null,
   ): Promise<R | TResult> {
-    return this.handler().then(undefined).catch(onrejected);
+    return this.before().then(this.handler).then(undefined).catch(onrejected);
   }
 
   finally(onfinally?: (() => void) | undefined | null): Promise<R> {
-    return this.handler().finally(onfinally);
+    return this.before().then(this.handler).finally(onfinally);
   }
 }
 
@@ -152,3 +154,9 @@ export async function signalRace<R>(
       .finally(() => signal?.removeEventListener?.("abort", signalFn));
   });
 }
+
+export async function executeSequentially(tasks: (() => Promise<any>)[]): Promise<void> {
+  for (const task of tasks) {
+    await task();
+  }
+}