Типізація асинхронної конкурентності: AbortSignal, канали та черги

+-------------------------------------------------------------------------+
|                  АРХІТЕКТУРА КОНКУРЕНТНОГО КОНВЕЄРА                     |
+-------------------------------------------------------------------------+
| [Producer (SSE / WebSocket / LLM Stream)]                               |
|   |--> Породжує потік повідомлень: AsyncIterable<ServerPacket>          |
+-------------------------------------------------------------------------+
                                    |
                                    v (Backpressure Queue / Channel<T>)
+-------------------------------------------------------------------------+
| [Типобезпечний буферизований канал: Channel<ServerPacket>]              |
|   |-- push(packet: T): Promise<void>  (Блокується при переповненні)     |
|   |-- pop(signal?: AbortSignal): Promise<T> (Блокується при спустошенні)|
|   `-- close(): void                                                     |
+-------------------------------------------------------------------------+
                                    |
                                    v (Worker Pool: Concurrency Limit = 3)
+-------------------------------------------------------------------------+
| [Паралельний пул споживачів (Consumers / Workers)]                      |
|   |-- Worker 1: Обробка пакета A (AbortSignal.timeout(5000))            |
|   |-- Worker 2: Обробка пакета B (Переривання через abort())            |
|   `-- Worker 3: Обробка пакета C (Успішна фіналізація)                  |
+-------------------------------------------------------------------------+

Композиція сигналів скасування:
[Сигнал тайм-ауту: AbortSignal.timeout(5000)] -----\
                                                    +--> [AbortSignal.any([...])] --> fetch(url, { signal })
[Сигнал кнопки UI: userCancelController.signal] ---/

Запит переривається, якщо минуло 5 секунд АБО якщо користувач натиснув «Скасувати»!

export interface FetchWithTimeoutOptions extends RequestInit {
  readonly timeoutMs?: number;
  readonly externalSignal?: AbortSignal;
}

/**
 * Виконує стійкий мережевий запит із композитним скасуванням:
 * за таймаутом АБО за зовнішнім сигналом.
 */
export async function fetchWithResilience<TResponse>(
  url: string,
  options: FetchWithTimeoutOptions = {}
): Promise<TResponse> {
  const { timeoutMs = 8000, externalSignal, ...fetchOptions } = options;

  // 1. Створюємо нативний сигнал таймауту
  const timeoutSignal = AbortSignal.timeout(timeoutMs);

  // 2. Об'єднуємо таймаут і зовнішній сигнал через AbortSignal.any
  const combinedSignal = externalSignal
    ? AbortSignal.any([timeoutSignal, externalSignal])
    : timeoutSignal;

  try {
    const response = await fetch(url, {
      ...fetchOptions,
      signal: combinedSignal,
    });

    if (!response.ok) {
      throw new Error(`[HttpError] HTTP ${response.status}: ${response.statusText}`);
    }

    return (await response.json()) as TResponse;
  } catch (error: unknown) {
    if (combinedSignal.aborted) {
      // Розрізняємо причину переривання: таймаут чи дія користувача
      if (timeoutSignal.aborted) {
        throw new Error(`[TimeoutError] Запит до ${url} перевищив ліміт у ${timeoutMs}ms`);
      }
      throw new Error(`[AbortError] Запит до ${url} було скасовано викликаючою стороною`);
    }

    throw error instanceof Error ? error : new Error(String(error));
  }
}

interface Resolver<T> {
  resolve: (value: T) => void;
  reject: (reason?: unknown) => void;
}

/**
 * Типобезпечний асинхронний канал передачі даних (Communicating Sequential Processes).
 */
export class AsyncChannel<T> implements AsyncIterable<T> {
  private readonly buffer: T[] = [];
  private readonly receivers: Resolver<T>[] = [];
  private isClosed = false;

  constructor(private readonly maxCapacity: number = Infinity) {}

  /**
   * Запис елемента в канал.
   */
  public push(item: T): boolean {
    if (this.isClosed) {
      throw new Error('[ChannelError] Спроба запису в закритий канал');
    }

    // Якщо є очікуючі споживачі — віддаємо першому в черзі
    if (this.receivers.length > 0) {
      const receiver = this.receivers.shift()!;
      receiver.resolve(item);
      return true;
    }

    // Перевірка ліміту буфера
    if (this.buffer.length >= this.maxCapacity) {
      // Буфер переповнено: витісняємо найстаріший елемент (Drop-oldest backpressure)
      this.buffer.shift();
    }

    this.buffer.push(item);
    return true;
  }

  /**
   * Асинхронне читання елемента з каналу з підтримкою скасування.
   */
  public async pop(signal?: AbortSignal): Promise<T | null> {
    if (this.buffer.length > 0) {
      return this.buffer.shift()!;
    }

    if (this.isClosed) {
      return null;
    }

    return new Promise<T | null>((resolve, reject) => {
      const resolver: Resolver<T> = {
        resolve: (val) => {
          cleanUp();
          resolve(val);
        },
        reject: (err) => {
          cleanUp();
          reject(err);
        },
      };

      const onAbort = () => {
        const index = this.receivers.indexOf(resolver);
        if (index > -1) {
          this.receivers.splice(index, 1);
        }
        cleanUp();
        reject(new Error('[AbortError] Очікування елемента з каналу перервано сигналом'));
      };

      const cleanUp = () => {
        signal?.removeEventListener('abort', onAbort);
      };

      if (signal?.aborted) {
        onAbort();
        return;
      }

      signal?.addEventListener('abort', onAbort);
      this.receivers.push(resolver);
    });
  }

  public close(): void {
    this.isClosed = true;
    // Сповіщаємо всіх очікуючих читачів про закриття каналу
    while (this.receivers.length > 0) {
      const receiver = this.receivers.shift()!;
      receiver.resolve(null as unknown as T);
    }
  }

  /**
   * Реалізація протоколу AsyncIterable для ітерації через цикл for await (... of ...)
   */
  public async *[Symbol.asyncIterator](): AsyncIterator<T> {
    while (true) {
      const value = await this.pop();
      if (value === null && this.isClosed) {
        break;
      }
      if (value !== null) {
        yield value;
      }
    }
  }
}

export type AsyncTask<T> = (signal: AbortSignal) => Promise<T>;

/**
 * Обмежує кількість одночасно виконуваних асинхронних завдань.
 */
export class ConcurrencyPool {
  private activeCount = 0;
  private readonly queue: Array<() => void> = [];

  constructor(public readonly concurrencyLimit: number) {
    if (concurrencyLimit < 1) {
      throw new RangeError('Ліміт паралелізму має бути >= 1');
    }
  }

  /**
   * Запускає завдання з дотриманням пулу паралелізму та контролем скасування.
   */
  public run<T>(task: AsyncTask<T>, signal?: AbortSignal): Promise<T> {
    return new Promise<T>((resolve, reject) => {
      const execute = async () => {
        if (signal?.aborted) {
          this.next();
          reject(new Error('[AbortError] Завдання скасовано до початку виконання'));
          return;
        }

        this.activeCount++;
        try {
          // Створюємо локальний контролер, пов'язаний із зовнішнім сигналом
          const ctrl = new AbortController();
          const onAbort = () => ctrl.abort();
          signal?.addEventListener('abort', onAbort);

          const result = await task(ctrl.signal);
          signal?.removeEventListener('abort', onAbort);
          resolve(result);
        } catch (err: unknown) {
          reject(err);
        } finally {
          this.activeCount--;
          this.next();
        }
      };

      if (this.activeCount < this.concurrencyLimit) {
        void execute();
      } else {
        this.queue.push(execute);
      }
    });
  }

  private next(): void {
    if (this.queue.length > 0 && this.activeCount < this.concurrencyLimit) {
      const nextTask = this.queue.shift();
      nextTask?.();
    }
  }
}