xiaoshu-app/projects/shared/request-scheduler.ts

import { Observable, Subscription, defer, finalize, of, shareReplay, tap, throwError } from 'rxjs';

export interface RequestPolicy<T> {
  read: boolean;
  key?: string;
  heavy?: boolean;
  background?: boolean;
  valid?: () => boolean;
  cacheMs?: number;
  refresh?: boolean;
  cacheable?: (value: T) => boolean;
}
interface Ticket { read: boolean; heavy: boolean; background: boolean; start: () => void; }
export function stableRequestJson(value: unknown): string {
  if (Array.isArray(value)) return '[' + value.map(stableRequestJson).join(',') + ']';
  if (value && typeof value === 'object') return '{' + Object.entries(value).filter(([,v]) => v !== undefined).sort(([a],[b]) => a.localeCompare(b)).map(([k,v]) => JSON.stringify(k)+':'+stableRequestJson(v)).join(',') + '}';
  return JSON.stringify(value) ?? 'null';
}
/** One scheduler per app: all gateway calls share capacity, including retries. */
export class RequestScheduler {
  private active = 0;
  private reads = 0;
  private heavy = 0;
  private revision = 0;
  private queue: Ticket[] = [];
  private pending = new Map<string, Observable<unknown>>();
  private cache = new Map<string, { value: unknown; expires: number }>();
  private readonly now: () => number;
  private readonly queueTimeoutMs: number;
  private completed = 0;
  private cancelled = 0;
  private coalesced = 0;
  constructor(options: { now?: () => number; queueTimeoutMs?: number } = {}) {
    this.now = options.now ?? Date.now;
    this.queueTimeoutMs = options.queueTimeoutMs ?? 10_000;
  }
  invalidate(): void { this.revision++; this.cache.clear(); this.pending.clear(); }
  snapshot(): { active: number; queued: number; completed: number; cancelled: number; coalesced: number } {
    return { active: this.active, queued: this.queue.length, completed: this.completed, cancelled: this.cancelled, coalesced: this.coalesced };
  }
  schedule<T>(run: () => Observable<T>, policy: RequestPolicy<T>): Observable<T> {
    return defer(() => {
      if (policy.valid && !policy.valid()) return throwError(() => new Error('会话已变化，请重新查询'));
      const key = policy.read ? policy.key : undefined;
      const cached = key && !policy.refresh ? this.cache.get(key) : undefined;
      if (cached && cached.expires > this.now()) return of(cached.value as T);
      const pending = key ? this.pending.get(key) : undefined;
      if (pending) { this.coalesced++; return pending as Observable<T>; }
      const revision = this.revision;
      const request = new Observable<T>(subscriber => {
        let started = false, ended = false, sub: Subscription | undefined;
        let timer: ReturnType<typeof setTimeout> | undefined;
        const ticket: Ticket = { read: policy.read, heavy: !!policy.heavy, background: !!policy.background, start: () => {
          started = true; if (timer) clearTimeout(timer);
          this.active++; if (ticket.read) this.reads++; if (ticket.heavy) this.heavy++;
          if (policy.valid && !policy.valid()) { subscriber.error(new Error('会话已变化，请重新查询')); return; }
          sub = defer(run).subscribe({ next: value => {
            if (policy.valid && !policy.valid()) { subscriber.error(new Error('会话已变化，请重新查询')); return; }
            subscriber.next(value);
          }, error: error => subscriber.error(error), complete: () => { ended = true; this.completed++; subscriber.complete(); } });
        } };
        if (this.queue.length >= 64) subscriber.error(new Error('请求排队过多，请稍后重试'));
        else {
          this.queue.push(ticket);
          timer = setTimeout(() => subscriber.error(new Error('请求排队暂未完成，请稍后重试')), this.queueTimeoutMs);
          this.drain();
        }
        return () => {
          if (timer) clearTimeout(timer);
          this.queue = this.queue.filter(item => item !== ticket);
          sub?.unsubscribe();
          if (started) { this.active--; if (ticket.read) this.reads--; if (ticket.heavy) this.heavy--; }
          if (!ended) this.cancelled++;
          this.drain();
        };
      }).pipe(tap(value => {
        if (key && policy.cacheMs && policy.cacheable?.(value) && revision === this.revision) {
          // Bounded cache, timestamps originate at the response, never at a cache hit.
          if (this.cache.size >= 100) this.cache.delete(this.cache.keys().next().value!);
          this.cache.set(key, { value, expires: this.now() + Math.min(30_000, policy.cacheMs) });
        }
      }), finalize(() => { if (key && this.pending.get(key) === shared) this.pending.delete(key); }));
      const shared = request.pipe(shareReplay({ bufferSize: 1, refCount: true }));
      if (key) this.pending.set(key, shared);
      return shared;
    });
  }
  private drain(): void {
    while (this.active < 4) {
      const candidates = this.queue.filter(t => (!t.read || this.reads < 3) && (!t.heavy || this.heavy < 1));
      const next = candidates.find(t => !t.read) ?? candidates.find(t => !t.background) ?? candidates[0];
      if (!next) return;
      this.queue.splice(this.queue.indexOf(next), 1); next.start();
    }
  }
}