RxJS patterns lab

import {
  Observable,
  OperatorFunction,
  concatMap,
  debounceTime,
  distinctUntilChanged,
  exhaustMap,
  from,
  mergeMap,
  retry,
  switchMap,
  timer,
  toArray,
} from 'rxjs';

/*
 * The RxJS patterns that turn up over and over in real Angular work: a typeahead, a poller,
 * the three ways to flatten a stream of async tasks, and a retry with backoff. Each one is the
 * same lesson — the operators you choose encode a *policy* (cancel vs. queue vs. ignore, wait
 * vs. retry), and picking the wrong one is a class of bug you can't see in a quick read.
 *
 * They take their side effects as function arguments rather than reaching for `http` directly.
 * That keeps them pure and composable, and — because every delay runs through `timer` — it lets
 * a `TestScheduler` drive the whole thing in virtual time, so the tests are fast and deterministic
 * instead of littered with real `setTimeout`s and flaky waits.
 */

/*
 * A typeahead is a race, and switchMap is how you win it. Every keystroke starts a new search;
 * switchMap cancels the one still in flight, so a slow early request can never land after a
 * faster later one and overwrite the results you're looking at. debounceTime keeps us off the
 * backend on every keystroke; distinctUntilChanged drops no-op repeats (an arrow key, a paste of
 * the same text). The three together are the whole pattern — remove any one and you get either a
 * hammered backend, redundant requests, or results that flicker to a stale answer.
 *
 *   typeahead(term$, (q) => http.get(...)).subscribe(render);
 */
export function typeahead<T>(
  term$: Observable<string>,
  search: (term: string) => Observable<T>,
  debounceMs = 250,
): Observable<T> {
  return term$.pipe(
    debounceTime(debounceMs),
    distinctUntilChanged(),
    switchMap((term) => search(term)),
  );
}

/*
 * Polling, done so it can't trip over itself. `timer(0, intervalMs)` fires once right away
 * (no awkward blank first interval) and then on every tick. `switchMap` is the safety valve:
 * if a fetch is still running when the next tick arrives, it's abandoned rather than allowed to
 * stack up — so a backend slower than your interval gives you a backlog of one, not an
 * ever-growing pile of overlapping requests.
 *
 * There's no stop button by design. The stream polls exactly as long as something is subscribed,
 * so "stop" is just "unsubscribe." Pipe it through `takeUntilDestroyed()` in a component and the
 * polling ends when the component does, with no teardown code to remember.
 */
export function poll<T>(intervalMs: number, fetch: () => Observable<T>): Observable<T> {
  return timer(0, intervalMs).pipe(switchMap(() => fetch()));
}

/*
 * One unit of fake async work: resolves to its `id` after `delayMs`. The delay is the variable
 * that makes the strategies below visibly differ.
 */
export interface SimTask {
  readonly id: number;
  readonly delayMs: number;
}

export type ConcurrencyStrategy = 'merge' | 'concat' | 'exhaust';

/*
 * Turns a task into the observable that emits its id once the delay elapses — the async unit the
 * flattening operators below schedule against each other.
 */
function runTask(task: SimTask): Observable<number> {
  return timer(task.delayMs).pipe(switchMap(() => from([task.id])));
}

/*
 * The four "higher-order" operators — mergeMap, concatMap, exhaustMap (and switchMap above) —
 * all answer the same question, "a new inner observable arrived while the last one is still
 * going; what now?", and they answer it differently. People reach for mergeMap by reflex and
 * inherit bugs they never chose. This runs the *same* tasks through three of them and returns the
 * order results come back, because that arrival order IS the difference, made visible:
 *
 *   - `merge`   — start everything at once, let results race in; order follows each task's delay,
 *                 not the order they were kicked off. Maximum concurrency, no ordering guarantee.
 *   - `concat`  — queue them and run strictly one at a time; results always arrive in emission
 *                 order, however slow each is. Ordered, but a slow early task blocks the rest.
 *   - `exhaust` — the first one wins; anything emitted while it's still running is dropped on the
 *                 floor. This is the double-submit guard — ignore new clicks until the save lands.
 *
 * The choice is a policy decision: parallel-and-unordered, serialized-and-ordered, or
 * ignore-while-busy. Returns the id array as an observable so a marble test can assert the order.
 */
export function runWithStrategy(
  tasks: readonly SimTask[],
  strategy: ConcurrencyStrategy,
): Observable<number[]> {
  /*
   * Same `runTask`, same input list — only the flattening operator changes. That's deliberate:
   * hold everything else constant and the operator is the sole variable explaining the output.
   */
  const operator: OperatorFunction<SimTask, number> =
    strategy === 'merge'
      ? mergeMap((t) => runTask(t))
      : strategy === 'concat'
        ? concatMap((t) => runTask(t))
        : exhaustMap((t) => runTask(t));

  return from(tasks).pipe(operator, toArray());
}

/*
 * Retrying is easy; retrying *politely* is the part people skip. A flat retry loop hammers a
 * struggling backend at full speed and helps push it over. Exponential backoff spaces the
 * attempts out — wait `baseMs`, then 2x, then 4x — so a service having a bad moment gets room to
 * recover instead of a stampede. On the nth retry (1-based) it waits `baseMs * 2^(n-1)`.
 *
 * The second half that matters: once `maxRetries` is spent, `retry` lets the error through. You
 * want failure to eventually surface, not be swallowed into an infinite quiet retry loop the user
 * can never escape. Use it as a pipeable operator:
 *
 *   source$.pipe(retryWithBackoff(3, 200))
 *
 * Every delay is a `timer`, so a `TestScheduler` runs the entire backoff sequence in virtual time
 * — the test asserts the whole escalation in microseconds, with no real waiting and no flake.
 */
export function retryWithBackoff<T>(maxRetries = 3, baseMs = 300): OperatorFunction<T, T> {
  return retry<T>({
    count: maxRetries,
    delay: (_error, retryCount) => timer(baseMs * 2 ** (retryCount - 1)),
  });
}