Học từ Facebook #57: Hàng đợi lệnh cho video và stream sắp lại chunk

/**
 * Hàng đợi lệnh cho <video> có nén lệnh (ý tưởng từ VideoPlayerImplementationStateMachineHostCallQueue
 * của Facebook).
 *
 * Khi element video chưa sẵn sàng (chưa mount, đang đổi nguồn…), các lệnh play/pause/seek/volume
 * được xếp hàng. Lúc áp, không cần chạy lại cả chuỗi: chỉ trạng thái CUỐI mới quan trọng.
 * - mỗi loại lệnh (seek, volume, muted, rate) chỉ giữ lệnh cuối cùng;
 * - play và pause xoá lẫn nhau: chỉ còn lệnh cuối trong hai loại;
 * - thứ tự tương đối của các lệnh còn lại được giữ.
 */

export type VideoCommand =
  | { type: 'play' }
  | { type: 'pause' }
  | { type: 'seek'; time: number }
  | { type: 'volume'; volume: number }
  | { type: 'muted'; muted: boolean }
  | { type: 'rate'; rate: number };

/** Nén hàng đợi: giữ lệnh cuối mỗi loại, play/pause tính chung một loại. */
export function compactCommands(queue: VideoCommand[]): VideoCommand[] {
  let out: VideoCommand[] = [];
  for (const cmd of queue) {
    const group = cmd.type === 'play' || cmd.type === 'pause' ? 'playback' : cmd.type;
    out = out.filter((c) => (c.type === 'play' || c.type === 'pause' ? 'playback' : c.type) !== group);
    out.push(cmd);
  }
  return out;
}

export interface DesiredState {
  volume: number;
  muted: boolean;
}

export type PlayEvent =
  | { type: 'play_promise_created'; id: number }
  | { type: 'play_promise_resolved'; id: number }
  | { type: 'play_promise_rejected'; id: number; reason: string }
  | { type: 'command_failed'; command: VideoCommand; error: string; readyState: number; networkState: number }
  | { type: 'applied'; command: VideoCommand };

function apply(video: HTMLVideoElement, cmd: VideoCommand, id: number, emit: (e: PlayEvent) => void) {
  switch (cmd.type) {
    case 'play': {
      const p = video.play();
      // play() trả Promise: bị chặn autoplay hay bị pause() chen ngang thì reject
      emit({ type: 'play_promise_created', id });
      p.then(
        () => emit({ type: 'play_promise_resolved', id }),
        (err: unknown) => emit({ type: 'play_promise_rejected', id, reason: err instanceof Error ? err.name : String(err) }),
      );
      break;
    }
    case 'pause':
      video.pause();
      break;
    case 'seek':
      video.currentTime = cmd.time;
      break;
    case 'volume':
      video.volume = cmd.volume;
      break;
    case 'muted':
      video.muted = cmd.muted;
      break;
    case 'rate':
      video.playbackRate = cmd.rate;
      break;
  }
}

export interface VideoCommandQueue {
  /** áp ngay nếu có element, không thì xếp hàng */
  send(cmd: VideoCommand): void;
  /** gắn element (hoặc null khi gỡ); có element thì xả hàng đợi */
  attach(video: HTMLVideoElement | null): void;
  /** các lệnh đang chờ */
  pending(): VideoCommand[];
}

export function createVideoCommandQueue({
  desired,
  compact = true,
  onEvent = () => {},
}: {
  /** trạng thái người dùng muốn (volume/muted): luôn được áp sau cùng khi xả, thắng trạng thái DOM */
  desired: () => DesiredState;
  compact?: boolean;
  onEvent?: (e: PlayEvent) => void;
}): VideoCommandQueue {
  let video: HTMLVideoElement | null = null;
  let queue: VideoCommand[] = [];
  let nextId = 1;

  const run = (cmd: VideoCommand) => {
    if (!video) return;
    try {
      apply(video, cmd, nextId++, onEvent);
      onEvent({ type: 'applied', command: cmd });
    } catch (e) {
      // ghi kèm readyState/networkState để biết element đang ở trạng thái nào khi lỗi
      onEvent({
        type: 'command_failed',
        command: cmd,
        error: e instanceof Error ? e.message : String(e),
        readyState: video.readyState,
        networkState: video.networkState,
      });
    }
  };

  return {
    send(cmd) {
      if (video) run(cmd);
      else queue.push(cmd);
    },
    attach(el) {
      video = el;
      if (!el) return;
      const { volume, muted } = desired();
      const all = queue.concat([{ type: 'volume', volume }, { type: 'muted', muted }]);
      queue = [];
      for (const cmd of compact ? compactCommands(all) : all) run(cmd);
    },
    pending: () => queue.slice(),
  };
}

/**
 * Bộ đệm sắp lại chunk của stream theo số thứ tự và phát lại cho người đăng ký muộn
 * (ý tưởng từ RelayAsyncStreamPool của Facebook: dữ liệu @defer/@stream của GraphQL
 * có thể về lệch thứ tự, UI vẫn phải nhận đúng thứ tự).
 *
 * Khác bản gốc (xem bài viết):
 * - lỗi/complete đúng lượt vẫn được xử lý khi đã có complete "đến sớm" đang chờ
 *   (bản gốc bỏ qua, stream treo mãi);
 * - gọi start() TRƯỚC khi phát lại sự kiện cũ cho observer;
 * - dọn pending sau khi xả.
 */

export interface StreamObserver<T> {
  start?(sub: { unsubscribe(): void }): void;
  next(value: T): void;
  error(err: unknown): void;
  complete(): void;
}

type Event<T> = { kind: 'next'; data: T } | { kind: 'error'; error: unknown } | { kind: 'complete' };

interface Stream<T> {
  done: boolean;
  events: Event<T>[];
  pending: Map<number, Event<T>>;
  observers: Map<number, StreamObserver<T>>;
}

export class StreamPool<T> {
  private streams = new Map<string, Stream<T>>();
  private nextObserverId = 0;
  /** mô phỏng hành vi bản gốc để so sánh trong demo */
  constructor(private readonly options: { originalBehavior?: boolean } = {}) {}

  private get(id: string): Stream<T> {
    let s = this.streams.get(id);
    if (!s) {
      s = { done: false, events: [], pending: new Map(), observers: new Map() };
      this.streams.set(id, s);
    }
    return s;
  }

  private deliver(s: Stream<T>, ev: Event<T>) {
    s.events.push(ev);
    const observers = [...s.observers.values()];
    if (ev.kind !== 'next') {
      s.done = true;
      s.observers.clear();
      s.pending.clear();
    }
    for (const o of observers) {
      if (ev.kind === 'next') o.next(ev.data);
      else if (ev.kind === 'error') o.error(ev.error);
      else o.complete();
    }
  }

  /** Nhận sự kiện có số thứ tự `seq`. Đúng lượt thì phát và xả tiếp các sự kiện chờ liền sau. */
  push(id: string, seq: number, ev: Event<T>) {
    if (!Number.isInteger(seq) || seq < 0) throw new Error(`seq không hợp lệ: ${seq}`);
    const s = this.get(id);
    if (s.done) return;
    // Bản gốc: error()/complete() return ngay khi status = "có complete đang chờ".
    if (this.options.originalBehavior && ev.kind !== 'next' && s.pending.size > 0 &&
        [...s.pending.values()].some((p) => p.kind !== 'next')) return;
    const expected = s.events.length;
    if (seq < expected || s.pending.has(seq)) throw new Error(`seq ${seq} bị trùng`);
    if (seq > expected) {
      s.pending.set(seq, ev);
      return;
    }
    this.deliver(s, ev);
    for (let k = seq + 1; !s.done && s.pending.has(k); k++) {
      const next = s.pending.get(k)!;
      s.pending.delete(k);
      this.deliver(s, next);
    }
  }

  next(id: string, seq: number, data: T) {
    this.push(id, seq, { kind: 'next', data });
  }
  error(id: string, seq: number, error: unknown) {
    this.push(id, seq, { kind: 'error', error });
  }
  complete(id: string, seq: number) {
    this.push(id, seq, { kind: 'complete' });
  }

  /** Đăng ký: nhận lại mọi sự kiện đã phát (đúng thứ tự), rồi nhận tiếp sự kiện mới. */
  subscribe(id: string, observer: StreamObserver<T>) {
    const s = this.get(id);
    const key = this.nextObserverId++;
    const sub = { unsubscribe: () => s.observers.delete(key) };
    const replay = () => {
      for (const ev of s.events) {
        if (ev.kind === 'next') observer.next(ev.data);
        else if (ev.kind === 'error') observer.error(ev.error);
        else observer.complete();
      }
    };
    if (this.options.originalBehavior) {
      // bản gốc: phát lại trước rồi mới start(), và vẫn thêm observer vào stream đã đóng
      replay();
      s.observers.set(key, observer);
      observer.start?.(sub);
    } else {
      observer.start?.(sub);
      replay();
      if (!s.done) s.observers.set(key, observer);
    }
    return sub;
  }

  /** Số sự kiện đang chờ (để demo thấy pending được dọn). */
  pendingCount(id: string) {
    return this.streams.get(id)?.pending.size ?? 0;
  }
  isDone(id: string) {
    return this.streams.get(id)?.done ?? false;
  }
  clear(id: string) {
    this.streams.delete(id);
  }
}