import { RATE_LIMIT_MS } from './constants';

/**
 * Concurrent pool that maintains exactly N active promises at all times.
 * Unlike batch processing (which waits for ALL items in a batch before starting the next),
 * this starts a new item as soon as any slot frees up.
 */
export async function asyncPool<T, R>(items: T[], concurrency: number, fn: (item: T) => Promise<R>): Promise<R[]> {
  if (!Number.isSafeInteger(concurrency) || concurrency <= 0) {
    throw new RangeError('concurrency must be a finite positive integer');
  }

  const results: R[] = [];
  const executing = new Set<Promise<void>>();
  const settled: Promise<void>[] = [];

  for (const [index, item] of items.entries()) {
    // Create the work promise that stores its result
    // Defer invocation by one microtask so a plain (non-async) callback that
    // throws synchronously follows the same rejection path as an async one.
    // Without this, the pool can reject before tracking the work already
    // started in earlier slots, leaving callers with detached promises.
    const p: Promise<void> = Promise.resolve()
      .then(() => fn(item))
      .then((result) => {
        results[index] = result;
      });
    settled.push(p);

    // Create a tracking promise that removes itself from executing on settle
    // This promise always resolves (even on error) so Promise.race works correctly
    const tracked: Promise<void> = p.then(
      () => {
        executing.delete(tracked);
      },
      () => {
        executing.delete(tracked);
      },
    );
    executing.add(tracked);
    if (executing.size >= concurrency) {
      await Promise.race(executing);
    }
  }
  // Wait for all work promises (not tracking promises) to propagate errors
  await Promise.all(settled);
  return results;
}

/**
 * Per-domain rate limiting to avoid hammering servers within this process.
 * The queue serializes request start times per host for concurrent callers in
 * one CLI/TUI process. Separate rss-ai processes do not share this state; a
 * cross-process guarantee requires a persistent coordinator with crash/stale
 * lease handling and is intentionally outside this in-memory utility's scope.
 */
const domainLastFetch = new Map<string, number>();
const domainStartQueues = new Map<string, Promise<void>>();
const domainExpiry = new Map<string, number>();
let cleanupTimer: ReturnType<typeof setTimeout> | undefined;
let cleanupAt = Number.POSITIVE_INFINITY;

function scheduleRateLimitCleanup(): void {
  let nextExpiry = Number.POSITIVE_INFINITY;
  for (const expiry of domainExpiry.values()) nextExpiry = Math.min(nextExpiry, expiry);
  if (!Number.isFinite(nextExpiry)) return;
  if (cleanupTimer !== undefined && cleanupAt <= nextExpiry) return;

  if (cleanupTimer !== undefined) clearTimeout(cleanupTimer);
  cleanupAt = nextExpiry;
  cleanupTimer = setTimeout(
    () => {
      cleanupTimer = undefined;
      cleanupAt = Number.POSITIVE_INFINITY;
      const now = Date.now();
      for (const [domain, expiry] of domainExpiry) {
        if (expiry <= now) {
          domainExpiry.delete(domain);
          domainLastFetch.delete(domain);
        }
      }
      scheduleRateLimitCleanup();
    },
    Math.max(0, nextExpiry - Date.now()),
  );

  // Rate-limit bookkeeping must not keep a short-lived CLI process alive.
  if (typeof cleanupTimer === 'object' && cleanupTimer !== null && 'unref' in cleanupTimer) {
    (cleanupTimer as { unref(): void }).unref();
  }
}

/**
 * Expose domainLastFetch for testing cleanup behavior
 */
export function getDomainLastFetch(): Map<string, number> {
  return domainLastFetch;
}

export function clearRateLimitState(): void {
  domainLastFetch.clear();
  domainStartQueues.clear();
  domainExpiry.clear();
  if (cleanupTimer !== undefined) clearTimeout(cleanupTimer);
  cleanupTimer = undefined;
  cleanupAt = Number.POSITIVE_INFINITY;
}

function getRateLimitMs(): number {
  const override = process.env.RSS_RATE_LIMIT_MS;
  if (override === undefined) return RATE_LIMIT_MS;

  const parsed = Number(override);
  return Number.isFinite(parsed) && parsed >= 0 ? parsed : RATE_LIMIT_MS;
}

/**
 * Wait for this caller's turn in the per-domain start queue, honoring the
 * rate-limit spacing. If `signal` aborts while waiting, the wait rejects
 * promptly so the caller doesn't keep holding the queue (and the next waiter
 * proceeds normally via the chained queue promise).
 */
async function waitForDomainSlot(domain: string, signal?: AbortSignal): Promise<void> {
  const previous = domainStartQueues.get(domain) ?? Promise.resolve();

  // `turn` remains in the queue even if its caller aborts while an earlier
  // request is still waiting. That preserves FIFO spacing for later callers;
  // the caller itself races it against the abort signal and therefore rejects
  // immediately rather than waiting for every earlier queue entry to settle.
  const turn = previous.then(async () => {
    if (signal?.aborted) {
      throw signal.reason instanceof Error ? signal.reason : new Error('Aborted');
    }

    const rateLimitMs = getRateLimitMs();
    const lastFetch = domainLastFetch.get(domain) || 0;
    const elapsed = Date.now() - lastFetch;

    if (rateLimitMs > 0 && elapsed < rateLimitMs) {
      await new Promise<void>((resolve, reject) => {
        const timer = setTimeout(() => {
          signal?.removeEventListener('abort', onAbort);
          resolve();
        }, rateLimitMs - elapsed);

        const onAbort = () => {
          clearTimeout(timer);
          reject(signal?.reason instanceof Error ? signal.reason : new Error('Aborted'));
        };

        signal?.addEventListener('abort', onAbort, { once: true });
      });
    }

    const startedAt = Date.now();
    domainLastFetch.set(domain, startedAt);
    domainExpiry.set(domain, startedAt + rateLimitMs);
    scheduleRateLimitCleanup();
  });
  const current = turn.catch(() => {
    // An aborted waiter has no start time to record, but must not poison the
    // serialized queue behind it.
  });
  current.finally(() => {
    if (domainStartQueues.get(domain) === current) {
      domainStartQueues.delete(domain);
    }
  });

  domainStartQueues.set(domain, current);

  if (!signal) {
    await turn;
    return;
  }
  if (signal.aborted) {
    throw signal.reason instanceof Error ? signal.reason : new Error('Aborted');
  }

  await new Promise<void>((resolve, reject) => {
    const onAbort = () => reject(signal.reason instanceof Error ? signal.reason : new Error('Aborted'));
    signal.addEventListener('abort', onAbort, { once: true });
    turn.then(
      () => {
        signal.removeEventListener('abort', onAbort);
        resolve();
      },
      (error) => {
        signal.removeEventListener('abort', onAbort);
        reject(error);
      },
    );
  });
}

export interface RateLimitedFetchOptions {
  /**
   * Abort the HTTP request after this many ms. The timeout timer is started
   * AFTER the domain slot is acquired, so time spent queueing behind the
   * per-domain rate limit is never counted against the request timeout.
   */
  timeoutMs?: number;
  /** Optional caller-provided signal; combined with the internal timeout. */
  signal?: AbortSignal;
  /**
   * Logical hostname used for rate limiting when the transport URL is an
   * already-resolved IP literal (for example, the protected Article client).
   */
  rateLimitHostname?: string;
}

interface ResponseLifetime {
  cleanup: () => void;
}

// A fetch promise resolves when headers arrive, but its abort signal still has
// to outlive headers while consumers read the body. A WeakMap keeps that
// lifetime attached to the Response without changing the standard Response
// surface. Consumers that consume/cancel a body should call
// `completeRateLimitedResponse`; bounded-response does this automatically.
const responseLifetimes = new WeakMap<Response, ResponseLifetime>();

export function completeRateLimitedResponse(response: Response): void {
  responseLifetimes.get(response)?.cleanup();
}

export async function cancelRateLimitedResponse(response: Response): Promise<void> {
  try {
    await response.body?.cancel();
  } catch {
    // The original HTTP/body error is more useful than a failed connection
    // teardown. The controller is still cleaned up below.
  } finally {
    completeRateLimitedResponse(response);
  }
}

/**
 * Fetch with per-domain rate limiting and an optional request timeout.
 *
 * The domain slot is acquired first; only then is the timeout AbortController
 * created and started. This means a large same-host batch that queues for
 * minutes does not spuriously abort at the request timeout — the timeout
 * measures only the actual HTTP request and response body consumption. A
 * caller-provided `signal` is honored both while waiting in the queue and
 * until the returned response is consumed or cancelled.
 */
export async function rateLimitedFetch(
  url: string,
  options: RequestInit = {},
  fetchOptions: RateLimitedFetchOptions = {},
): Promise<Response> {
  const domain = fetchOptions.rateLimitHostname ?? new URL(url).hostname;
  const { timeoutMs } = fetchOptions;
  const callerSignals = [fetchOptions.signal, options.signal ?? undefined].filter(
    (signal): signal is AbortSignal => signal !== undefined,
  );
  const callerSignal =
    callerSignals.length === 0
      ? undefined
      : callerSignals.length === 1
        ? callerSignals[0]
        : AbortSignal.any(callerSignals);
  if (timeoutMs !== undefined && (!Number.isSafeInteger(timeoutMs) || timeoutMs < 0)) {
    throw new RangeError('timeoutMs must be a non-negative safe integer');
  }

  // Acquire the domain slot first, observing the caller's signal so a cancelled
  // caller releases the queue instead of blocking later waiters.
  await waitForDomainSlot(domain, callerSignal);

  // Start the timeout only now, so queue time is not counted against it.
  const controller = new AbortController();
  const timeoutId = timeoutMs !== undefined ? setTimeout(() => controller.abort(), timeoutMs) : undefined;

  const onCallerAbort = () => controller.abort(callerSignal?.reason);
  if (callerSignal) {
    if (callerSignal.aborted) {
      controller.abort(callerSignal.reason);
    } else {
      callerSignal.addEventListener('abort', onCallerAbort, { once: true });
    }
  }

  try {
    const response = await fetch(url, { ...options, signal: controller.signal });
    let cleanedUp = false;
    responseLifetimes.set(response, {
      cleanup: () => {
        if (cleanedUp) return;
        cleanedUp = true;
        if (timeoutId !== undefined) clearTimeout(timeoutId);
        callerSignal?.removeEventListener('abort', onCallerAbort);
      },
    });
    return response;
  } catch (error) {
    if (timeoutId !== undefined) clearTimeout(timeoutId);
    callerSignal?.removeEventListener('abort', onCallerAbort);
    throw error;
  }
}