An (rxjs) observable cached promise

CachedPromise.ts

import {
  BehaviorSubject,
  filter,
  type Observable,
  type Subscription,
  tap,
} from 'rxjs';

const noop => () => {};

export const PromiseState = {
  /**
   * The promise has not been triggered or has been reset.
   */
  ready: 'ready',
  loading: 'loading',
  rejected: 'rejected',
  resolved: 'resolved',
} as const;
export type PromiseState = typeof PromiseState[keyof typeof PromiseState];

/**
 * This limits the possible value/error/state combinations from 8 to 4.
 * All other combinations are not possible.
 */
export type ValueErrorState<T, E = unknown> = Readonly<
  | [value: T, error: undefined, state: typeof PromiseState.loading]
  | [value: T, error: undefined, state: typeof PromiseState.ready]
  | [value: T, error: E, state: typeof PromiseState.rejected]
  | [value: T, error: undefined, state: typeof PromiseState.resolved]
>;
export type AsyncValueErrorState<T, E = unknown> = Promise<
  ValueErrorState<T, E>
>;

/**
 * Encapsulates an async function (`impl`) in a way that it will only be invoked once,
 * even when `request` is called multiple times (e.g. by multiple views).
 * `request()` resolves and rejects when `impl` does,
 * but it resolves/rejects with a tuple of `[value, error, state]` (aka `ValueErrorState`).
 * When `state` is `PromiseState.rejected`, `error` is the reason for the rejection,
 * otherwise `value` is either `initial` or the resolved value, and `error` is `undefined`.
 *
 * When `impl` rejects, the cache is cleared:
 * subsequent calls to `request` will make another call to `impl`.
 *
 * Additionally, the following methods invalidate the cache:
 * - `reset` clears the cache without triggering `request`
 *   (it currently doesn't cancel the promise,
 *   but the promise will resolve to `["ready", initial, undefined]`)
 * - `refresh` keeps the current `value` and triggers `request`
 *   (it is a noop when the current `state` is already `loading`)
 *
 * Last but not least, `getValueErrorState$` provides access to an RxJs Observable of
 * `ValueErrorState`, which triggers the function (in the same cached way as described above), as
 * soon as there is a subscription.
 */
export class CachedPromise<T, E = unknown> {
  /**
   * By storing all of it in a single field it provides a stable value within each `state`,
   * and can be observed with a single stable Observable.
   * @see getValueErrorState$
   * @private
   */
  #valueErrorState: ValueErrorState<T, E>;

  /**
   * The current `ValueErrorState` tuple, which is a stable within each state.
   */
  get valueErrorState(): ValueErrorState<T, E> {
    return this.#valueErrorState;
  }

  /**
   * The current value.
   * (Uses `initial` when no resolved value is available.)
   */
  get value(): T {
    return this.#valueErrorState[0];
  }

  /**
   * The current error (only set in `PromiseState.rejected`).
   */
  get error(): E | undefined {
    return this.#valueErrorState[1];
  }

  /**
   * The current PromiseState.
   */
  get state(): PromiseState {
    return this.#valueErrorState[2];
  }

  get isLoading() {
    return this.state === PromiseState.loading;
  }

  get isReady() {
    return this.state === PromiseState.ready;
  }

  get hasResolved() {
    return this.state === PromiseState.resolved;
  }

  get hasRejected() {
    return this.state === PromiseState.rejected;
  }

  get hasSettled() {
    return this.hasResolved || this.hasRejected;
  }

  constructor(
    private readonly impl: () => Promise<T>,
    readonly initial: T
  ) {
    this.#valueErrorState = [this.initial, undefined, PromiseState.ready];
  }

  #cache: Promise<ValueErrorState<T, E>> | undefined;

  /**
   * Provides a cached access to the promise returned by the wrapped async function.
   * Sets `valueErrorState` to `[this.value, undefined, PromiseState.loading]`.
   * Resolving/Rejecting will set the related state.
   *
   * The following actions invalidate the cache:
   * - the promise rejects
   * - calling `reset`
   * - calling `refresh` when not in state `PromiseState.loading`
   *
   * @returns the cached ValueErrorState promise
   */
  async request(): AsyncValueErrorState<T, E> {
    if (!this.#cache) {
      this.setValueErrorState([this.value, undefined, PromiseState.loading]);
      this.#cache = this.impl().then((value) => {
        if (this.state === PromiseState.loading) {
          this.setValueErrorState([value, undefined, PromiseState.resolved]);
        }
        return this.valueErrorState;
      });
      this.#cache.catch((reason: E) => {
        if (this.state === PromiseState.loading) {
          this.#cache = undefined;
          this.setValueErrorState([
            this.initial,
            reason,
            PromiseState.rejected,
          ]);
        }
      });
    }
    // since promises can only be resolved once, it needs to either resolve or reject.
    // we can not communicate the `loading` state this way,
    // but it can be accessed using `state` or `valueErrorState$`
    return this.#cache;
  }

  /**
   * Invalidates the cache without triggering `request`.
   * Sets `valueErrorState` to `[initial, undefined, PromiseState.ready]`.
   */
  reset = (): void => {
    this.#cache = undefined;
    this.setValueErrorState([this.initial, undefined, PromiseState.ready]);
  };

  /**
   * Invalidates the cache and triggers `request`, if `state` is not `PromiseState.loading`.
   * Sets `valueErrorState` to `[initial, undefined, PromiseState.loading]`.
   * @returns The cached promise.
   */
  async refresh(): AsyncValueErrorState<T, E> {
    if (this.state === PromiseState.resolved) {
      this.#cache = undefined;
    }
    return this.request();
  }

  #refreshSubscription: Subscription | undefined;

  /**
   * An observable where each `next` will trigger a call to `refresh`,
   * if the current `state` is either `resolved` or `rejected`.
   *
   * Only the last observable that was set is considered.
   * Setting multiple values will unsubscribe from all but the last one.
   * So to stop triggering refresh, set it to `undefined`.
   */
  set refresh$(trigger$: Observable<unknown> | undefined) {
    this.#refreshSubscription?.unsubscribe();
    this.#refreshSubscription = trigger$
      ?.pipe(
        filter(() => this.hasSettled),
        tap(() => {
          void this.refresh().catch(noop);
        })
      )
      .subscribe();
  }

  #subject: BehaviorSubject<ValueErrorState<T, E>> | undefined;

  /**
   * Provides an observable of ValueErrorState using a BehaviorSubject,
   * which always replays the most recent value when subscribing.
   *
   * The wrapped async function is invoked as part of calling this function,
   * when `state` is `PromiseState.ready`:
   * - if it has not already been triggered before.
   * - after calling `reset`
   */
  getValueErrorState$(): Observable<ValueErrorState<T, E>> {
    if (this.state === PromiseState.ready) {
      // to avoid uncaught promise rejection we have to "swallow" potential rejections
      // the error still ends up in the `valueErrorState` as is proven by test
      this.request().catch(noop);
    }
    if (!this.#subject) {
      this.#subject = new BehaviorSubject<ValueErrorState<T, E>>(
        this.valueErrorState
      );
    }
    return this.#subject;
  }

  private setValueErrorState(
    next: ValueErrorState<T, E>
  ): ValueErrorState<T, E> {
    this.#valueErrorState = next;
    this.#subject?.next(next);
    return next;
  }
}

This has grown into a full project: https://github.com/karfau/cached-promise