/**
 * Concurrency-pool scene renderer — the PARALLEL counterpart to
 * `runAutoChain` (`execute-autochain.ts`). Where auto-chain renders scenes
 * SEQUENTIALLY (each seeded from the previous scene's output), the pool keeps up
 * to N independent scene renders in-flight at once and auto-refills as each
 * completes. There is no chaining — scenes are treated as independent units.
 *
 * This productizes the hand-written `pool-driver.mjs` operators used for the
 * cartoon gens 3-6 (keep N in-flight, refill on completion, resumable). Mirrors
 * the auto-chain shape: a PURE scheduler (`runScenePool`) plus an injectable
 * per-scene runner (`PoolSceneRunner`), so the scheduler is fully offline-testable
 * with a simulated runner. The real runner + the CLI handler live in
 * `src/cli/handlers/execution.ts`.
 */

export interface PoolSceneRunner {
  (sceneIndex: number): Promise<{ sceneIndex: number; status: 'done' | 'failed'; error?: string }>;
}

export interface PoolSceneResult {
  sceneIndex: number;
  status: 'done' | 'failed' | 'skipped';
  error?: string;
}

export interface PoolResult {
  results: PoolSceneResult[];
  /**
   * The peak number of runner promises that were in-flight simultaneously.
   * INVARIANT: this must never exceed `maxConcurrent`. The scheduler enforces
   * the cap; this is the observable proof it held.
   */
  maxObservedInFlight: number;
}

export interface RunScenePoolOptions {
  /** Ordered scene indices to render. */
  scenes: number[];
  /** Maximum number of runner promises in-flight at once (>= 1). */
  maxConcurrent: number;
  /** Injectable per-scene runner — submit + poll + select for one scene. */
  runner: PoolSceneRunner;
  /**
   * Resume hook: when it returns true for a scene, that scene is recorded as
   * 'skipped' and never handed to the runner (it is already done/selected).
   */
  isAlreadyDone?: (sceneIndex: number) => boolean;
  /** Optional progress sink (a started/settled message per scene). */
  onProgress?: (msg: string) => void;
}

/**
 * Render `scenes` with a concurrency cap, auto-refilling as each settles.
 *
 * The cap is enforced by a fixed pool of `maxConcurrent` worker loops, each of
 * which pulls the next not-yet-started scene index from a shared cursor and
 * awaits its runner before pulling the next. Because there are at most
 * `maxConcurrent` workers and each holds exactly one scene at a time, at most
 * `maxConcurrent` runner promises are ever pending simultaneously — that is the
 * mechanism that bounds `maxObservedInFlight`. A worker that finishes a scene
 * immediately claims the next index (auto-refill); when the cursor is exhausted
 * the worker exits.
 *
 * Resumable: scenes for which `isAlreadyDone` returns true are recorded
 * 'skipped' and never run. Failure-isolated: a runner that rejects or returns
 * status:'failed' frees its worker and the pool keeps going, so one bad scene
 * never blocks (or aborts) the rest. Results are returned deterministically
 * ordered by sceneIndex.
 *
 * PURE: no fs / provider / timer access beyond awaiting the injected runner.
 */
export async function runScenePool(opts: RunScenePoolOptions): Promise<PoolResult> {
  const { scenes, maxConcurrent, runner, isAlreadyDone, onProgress } = opts;
  if (!Number.isInteger(maxConcurrent) || maxConcurrent < 1) {
    throw new Error(`runScenePool: maxConcurrent must be a positive integer, got ${maxConcurrent}`);
  }

  // Record results keyed by sceneIndex so the final order is deterministic
  // regardless of completion order.
  const resultsByScene = new Map<number, PoolSceneResult>();

  // Partition into the scenes that actually run (not already done) and the
  // skipped ones. Skips are recorded up front and never enter the work queue.
  const pending: number[] = [];
  for (const sceneIndex of scenes) {
    if (isAlreadyDone?.(sceneIndex)) {
      resultsByScene.set(sceneIndex, { sceneIndex, status: 'skipped' });
      onProgress?.(`scene ${sceneIndex}: skipped (already done)`);
    } else {
      pending.push(sceneIndex);
    }
  }

  // A shared cursor over `pending`. Each worker atomically (single-threaded JS
  // event loop — no await between read and increment) claims the next index.
  let cursor = 0;
  let inFlight = 0;
  let maxObservedInFlight = 0;

  async function worker(): Promise<void> {
    while (cursor < pending.length) {
      const sceneIndex = pending[cursor];
      cursor += 1;

      inFlight += 1;
      if (inFlight > maxObservedInFlight) maxObservedInFlight = inFlight;
      onProgress?.(`scene ${sceneIndex}: started (${inFlight} in-flight)`);

      try {
        const outcome = await runner(sceneIndex);
        resultsByScene.set(sceneIndex, {
          sceneIndex,
          status: outcome.status,
          ...(outcome.error ? { error: outcome.error } : {}),
        });
        onProgress?.(`scene ${sceneIndex}: ${outcome.status}`);
      } catch (error) {
        // A thrown/rejected runner is a per-scene failure, NOT a pool abort —
        // free the slot and let the worker claim the next scene.
        const message = error instanceof Error ? error.message : String(error);
        resultsByScene.set(sceneIndex, { sceneIndex, status: 'failed', error: message });
        onProgress?.(`scene ${sceneIndex}: failed (${message})`);
      } finally {
        inFlight -= 1;
      }
    }
  }

  // Spin up at most `maxConcurrent` workers (never more than there are scenes).
  const workerCount = Math.min(maxConcurrent, pending.length);
  const workers: Array<Promise<void>> = [];
  for (let i = 0; i < workerCount; i += 1) workers.push(worker());
  await Promise.all(workers);

  // Deterministic order by sceneIndex.
  const results = [...resultsByScene.values()].sort((a, b) => a.sceneIndex - b.sceneIndex);
  return { results, maxObservedInFlight };
}