/**
 * Bounded async FIFO. The producer awaits when the queue is full; the
 * consumer awaits when the queue is empty. Used by the upload orchestrator
 * to pace per-lane chunk dispatch without unbounded memory growth.
 */
export class BoundedAsyncQueue<T> {
  private items: T[] = [];
  private waiters: Array<(value: IteratorResult<T>) => void> = [];
  private spaceWaiters: Array<() => void> = [];
  private closed = false;
  private aborted = false;
  private abortReason: unknown = null;

  constructor(private readonly capacity: number) {
    if (capacity < 1) throw new Error(`capacity must be >= 1`);
  }

  /** Push an item. Awaits if the queue is at capacity. */
  async push(item: T): Promise<void> {
    if (this.aborted) throw this.abortReason;
    if (this.closed) throw new Error('queue closed');
    while (this.items.length >= this.capacity) {
      await new Promise<void>((resolve) => this.spaceWaiters.push(resolve));
      if (this.aborted) throw this.abortReason;
      if (this.closed) throw new Error('queue closed mid-push');
    }
    this.items.push(item);
    const waiter = this.waiters.shift();
    if (waiter !== undefined) {
      const next = this.items.shift()!;
      waiter({ value: next, done: false });
      this.notifySpace();
    }
  }

  /** Returns next item or `{done: true}` when closed and drained. */
  async next(): Promise<IteratorResult<T>> {
    if (this.aborted) throw this.abortReason;
    if (this.items.length > 0) {
      const value = this.items.shift()!;
      this.notifySpace();
      return { value, done: false };
    }
    if (this.closed) return { value: undefined as never, done: true };
    return new Promise<IteratorResult<T>>((resolve) => {
      this.waiters.push(resolve);
    });
  }

  /** Mark the queue as closed; subsequent `next()` calls drain remaining items. */
  close(): void {
    if (this.closed) return;
    this.closed = true;
    for (const waiter of this.waiters) {
      waiter({ value: undefined as never, done: true });
    }
    this.waiters = [];
  }

  abort(reason: unknown): void {
    if (this.aborted) return;
    this.aborted = true;
    this.abortReason = reason;
    this.closed = true;
    for (const w of this.spaceWaiters) w();
    this.spaceWaiters = [];
    for (const w of this.waiters) {
      // Reject pending consumers via a thrown abort; consumers handle.
      // We can't reject here because the next() returns IteratorResult, not a rejected promise.
      // So instead, re-resolve as `done`; consumers check `aborted` separately.
      w({ value: undefined as never, done: true });
    }
    this.waiters = [];
  }

  get size(): number {
    return this.items.length;
  }

  get isClosed(): boolean {
    return this.closed;
  }

  private notifySpace(): void {
    const waiter = this.spaceWaiters.shift();
    if (waiter !== undefined) waiter();
  }
}
