feat(infra): framework

packages/common/infra/src/livedata/__tests__/livedata.spec.ts

@@ -263,6 +263,13 @@ describe('livedata', () => {
       inner$.next(4);
       expect(flatten$.value).toEqual([4, 3]);
     }
+
+    {
+      const wrapped$ = new LiveData([] as LiveData<number>[]);
+      const flatten$ = wrapped$.flat();
+
+      expect(flatten$.value).toEqual([]);
+    }
   });
 
   test('computed', () => {

packages/common/infra/src/livedata/effect/index.ts

@@ -4,10 +4,43 @@ import { type OperatorFunction, Subject } from 'rxjs';
 
 const logger = new DebugLogger('effect');
 
-export interface Effect<T> {
-  (value: T): void;
-}
+export type Effect<T> = (T | undefined extends T // hack to detect if T is unknown
+  ? () => void
+  : (value: T) => void) & {
+  // unsubscribe effect, all ongoing effects will be cancelled.
+  unsubscribe: () => void;
+};
 
+/**
+ * Create an effect.
+ *
+ * `effect( op1, op2, op3, ... )`
+ *
+ * You can think of an effect as a pipeline. When the effect is called, argument will be sent to the pipeline,
+ * and the operators in the pipeline can be triggered.
+ *
+ *
+ *
+ * @example
+ * ```ts
+ * const loadUser = effect(
+ *   switchMap((id: number) =>
+ *     from(fetchUser(id)).pipe(
+ *       mapInto(user$),
+ *       catchErrorInto(error$),
+ *       onStart(() => isLoading$.next(true)),
+ *       onComplete(() => isLoading$.next(false))
+ *     )
+ *   )
+ * );
+ *
+ * // emit value to effect
+ * loadUser(1);
+ *
+ * // unsubscribe effect, will stop all ongoing processes
+ * loadUser.unsubscribe();
+ * ```
+ */
 export function effect<T, A>(op1: OperatorFunction<T, A>): Effect<T>;
 export function effect<T, A, B>(
   op1: OperatorFunction<T, A>,
@@ -42,23 +75,47 @@ export function effect<T, A, B, C, D, E, F>(
 export function effect(...args: any[]) {
   const subject$ = new Subject<any>();
 
+  const effectLocation = environment.isDebug
+    ? `(${new Error().stack?.split('\n')[2].trim()})`
+    : '';
+
+  class EffectError extends Unreachable {
+    constructor(message: string, value?: any) {
+      logger.error(`effect ${effectLocation} ${message}`, value);
+      super(
+        `effect ${effectLocation} ${message}` +
+          ` ${value ? (value instanceof Error ? value.stack ?? value.message : value + '') : ''}`
+      );
+    }
+  }
+
   // eslint-disable-next-line prefer-spread
-  subject$.pipe.apply(subject$, args as any).subscribe({
+  const subscription = subject$.pipe.apply(subject$, args as any).subscribe({
     next(value) {
-      logger.error('effect should not emit value', value);
-      throw new Unreachable('effect should not emit value');
+      const error = new EffectError('should not emit value', value);
+      setImmediate(() => {
+        throw error;
+      });
     },
     complete() {
-      logger.error('effect unexpected complete');
-      throw new Unreachable('effect unexpected complete');
+      const error = new EffectError('effect unexpected complete');
+      setImmediate(() => {
+        throw error;
+      });
     },
     error(error) {
-      logger.error('effect uncatched error', error);
-      throw new Unreachable('effect uncatched error');
+      const effectError = new EffectError('effect uncaught error', error);
+      setImmediate(() => {
+        throw effectError;
+      });
     },
   });
 
-  return ((value: unknown) => {
+  const fn = (value: unknown) => {
     subject$.next(value);
-  }) as never;
+  };
+
+  fn.unsubscribe = () => subscription.unsubscribe();
+
+  return fn as never;
 }

packages/common/infra/src/livedata/index.ts

@@ -1,4 +1,12 @@
 export { type Effect, effect } from './effect';
 export { LiveData, PoisonedError } from './livedata';
-export { catchErrorInto, mapInto, onComplete, onStart } from './ops';
+export {
+  backoffRetry,
+  catchErrorInto,
+  exhaustMapSwitchUntilChanged,
+  fromPromise,
+  mapInto,
+  onComplete,
+  onStart,
+} from './ops';
 export { useEnsureLiveData, useLiveData } from './react';

packages/common/infra/src/livedata/livedata.ts

@@ -428,6 +428,9 @@ export class LiveData<T = unknown>
           if (v instanceof LiveData) {
             return (v as LiveData<any>).flat();
           } else if (Array.isArray(v)) {
+            if (v.length === 0) {
+              return of([]);
+            }
             return combineLatest(
               v.map(v => {
                 if (v instanceof LiveData) {
@@ -446,6 +449,29 @@ export class LiveData<T = unknown>
     ) as any;
   }
 
+  waitFor(predicate: (v: T) => unknown, signal?: AbortSignal): Promise<T> {
+    return new Promise((resolve, reject) => {
+      const subscription = this.subscribe(v => {
+        if (predicate(v)) {
+          resolve(v as any);
+          setImmediate(() => {
+            subscription.unsubscribe();
+          });
+        }
+      });
+      signal?.addEventListener('abort', reason => {
+        subscription.unsubscribe();
+        reject(reason);
+      });
+    });
+  }
+
+  waitForNonNull(signal?: AbortSignal) {
+    return this.waitFor(v => v !== null && v !== undefined, signal) as Promise<
+      NonNullable<T>
+    >;
+  }
+
   reactSubscribe = (cb: () => void) => {
     if (this.isPoisoned) {
       throw this.poisonedError;

packages/common/infra/src/livedata/ops.ts

@@ -1,14 +1,28 @@
 import {
   catchError,
+  connect,
+  distinctUntilChanged,
   EMPTY,
+  exhaustMap,
+  merge,
   mergeMap,
   Observable,
+  type ObservableInput,
+  type ObservedValueOf,
+  of,
   type OperatorFunction,
   pipe,
+  retry,
+  switchMap,
+  throwError,
+  timer,
 } from 'rxjs';
 
 import type { LiveData } from './livedata';
 
+/**
+ * An operator that maps the value to the `LiveData`.
+ */
 export function mapInto<T>(l$: LiveData<T>) {
   return pipe(
     mergeMap((value: T) => {
@@ -18,15 +32,30 @@ export function mapInto<T>(l$: LiveData<T>) {
   );
 }
 
-export function catchErrorInto(l$: LiveData<any>) {
+/**
+ * An operator that catches the error and sends it to the `LiveData`.
+ *
+ * The `LiveData` will be set to `null` when the observable completes. This is useful for error state recovery.
+ *
+ * @param cb A callback that will be called when an error occurs.
+ */
+export function catchErrorInto<Error = any>(
+  l$: LiveData<Error | null>,
+  cb?: (error: Error) => void
+) {
   return pipe(
+    onComplete(() => l$.next(null)),
     catchError((error: any) => {
       l$.next(error);
+      cb?.(error);
       return EMPTY;
     })
   );
 }
 
+/**
+ * An operator that calls the callback when the observable starts.
+ */
 export function onStart<T>(cb: () => void): OperatorFunction<T, T> {
   return observable$ =>
     new Observable(subscribe => {
@@ -35,6 +64,9 @@ export function onStart<T>(cb: () => void): OperatorFunction<T, T> {
     });
 }
 
+/**
+ * An operator that calls the callback when the observable completes.
+ */
 export function onComplete<T>(cb: () => void): OperatorFunction<T, T> {
   return observable$ =>
     new Observable(subscribe => {
@@ -52,3 +84,95 @@ export function onComplete<T>(cb: () => void): OperatorFunction<T, T> {
       });
     });
 }
+
+/**
+ * Convert a promise to an observable.
+ *
+ * like `from` but support `AbortSignal`.
+ */
+export function fromPromise<T>(
+  promise: Promise<T> | ((signal: AbortSignal) => Promise<T>)
+): Observable<T> {
+  return new Observable(subscriber => {
+    const abortController = new AbortController();
+
+    const rawPromise =
+      promise instanceof Function ? promise(abortController.signal) : promise;
+
+    rawPromise
+      .then(value => {
+        subscriber.next(value);
+        subscriber.complete();
+      })
+      .catch(error => {
+        subscriber.error(error);
+      });
+
+    return () => abortController.abort('Aborted');
+  });
+}
+
+/**
+ * An operator that retries the source observable when an error occurs.
+ *
+ * https://en.wikipedia.org/wiki/Exponential_backoff
+ */
+export function backoffRetry<T>({
+  when,
+  count = 3,
+  delay = 200,
+  maxDelay = 15000,
+}: {
+  when?: (err: any) => boolean;
+  count?: number;
+  delay?: number;
+  maxDelay?: number;
+} = {}) {
+  return (obs$: Observable<T>) =>
+    obs$.pipe(
+      retry({
+        count,
+        delay: (err, retryIndex) => {
+          if (when && !when(err)) {
+            return throwError(() => err);
+          }
+          const d = Math.pow(2, retryIndex - 1) * delay;
+          return timer(Math.min(d, maxDelay));
+        },
+      })
+    );
+}
+
+/**
+ * An operator that combines `exhaustMap` and `switchMap`.
+ *
+ * This operator executes the `comparator` on each input, acting as an `exhaustMap` when the `comparator` returns `true`
+ * and acting as a `switchMap` when the comparator returns `false`.
+ *
+ * It is more useful for async processes that are relatively stable in results but sensitive to input.
+ * For example, when requesting the user's subscription status, `exhaustMap` is used because the user's subscription
+ * does not change often, but when switching users, the request should be made immediately like `switchMap`.
+ *
+ * @param onSwitch callback will be executed when `switchMap` occurs (including the first execution).
+ */
+export function exhaustMapSwitchUntilChanged<T, O extends ObservableInput<any>>(
+  comparator: (previous: T, current: T) => boolean,
+  project: (value: T, index: number) => O,
+  onSwitch?: (value: T) => void
+): OperatorFunction<T, ObservedValueOf<O>> {
+  return pipe(
+    connect(shared$ =>
+      shared$.pipe(
+        distinctUntilChanged(comparator),
+        switchMap(value => {
+          onSwitch?.(value);
+          return merge(of(value), shared$).pipe(
+            exhaustMap((value, index) => {
+              return project(value, index);
+            })
+          );
+        })
+      )
+    )
+  );
+}