// Measures channel() fan-out distribution and throughput under various
// load profiles:
//   - No backpressure: consumers are trivial, producer races ahead
//   - Backpressure: consumers burn CPU per item, producer stalls on caps
//
// Skip-based RR skips consumers at cap, so distribution skew is highest
// when the producer outpaces consumers AND initial warm-up timing varies.
// As N grows the transient fades and counts even out.
//
// Requires Node v24+.
// Run: node examples/benchmark-dispatch/fanout.ts

import { availableParallelism } from 'node:os';
import { workers, channel, assign } from '../../src/index.ts';
import { noopConsume, noopConsumeBusy } from './noop.ts';

const maxWorkers = availableParallelism();

interface FanoutResult {
  counts: number[];
  throughput: number;
  elapsedMs: number;
}

async function fanout(n: number, numWorkers: number): Promise<FanoutResult> {
  using run = workers(numWorkers);
  async function* source() {
    for (let i = 0; i < n; i++) yield i;
  }
  const ch = channel(source());
  const tasks = run.workers.map((w) => assign(w, noopConsume(ch)));
  const t0 = performance.now();
  const counts = (await run(tasks)) as number[];
  const elapsedMs = performance.now() - t0;
  const total = counts.reduce((a, b) => a + b, 0);
  if (total !== n) throw new Error(`expected ${n} items, got ${total}`);
  return { counts, throughput: n / (elapsedMs / 1000), elapsedMs };
}

async function fanoutBusy(n: number, numWorkers: number, busyUs: number): Promise<FanoutResult> {
  using run = workers(numWorkers);
  async function* source() {
    for (let i = 0; i < n; i++) yield i;
  }
  const ch = channel(source());
  const tasks = run.workers.map((w) => assign(w, noopConsumeBusy(ch, busyUs)));
  const t0 = performance.now();
  const counts = (await run(tasks)) as number[];
  const elapsedMs = performance.now() - t0;
  const total = counts.reduce((a, b) => a + b, 0);
  if (total !== n) throw new Error(`expected ${n} items, got ${total}`);
  return { counts, throughput: n / (elapsedMs / 1000), elapsedMs };
}

function fmtCounts(counts: number[]): string {
  const mean = counts.reduce((a, b) => a + b, 0) / counts.length;
  const min = Math.min(...counts);
  const max = Math.max(...counts);
  const spread = ((max - min) / mean) * 100;
  return `[${counts.map((c) => c.toString().padStart(6)).join(', ')}]  spread ${spread.toFixed(1)}% (min ${min}, max ${max}, mean ${Math.round(mean)})`;
}

function fmtThroughput(r: FanoutResult): string {
  return `${Math.round(r.throughput).toLocaleString().padStart(10)} items/s  (${r.elapsedMs.toFixed(1)}ms)`;
}

const W = 4;
console.log(`workers: ${W}  (system has ${maxWorkers} cpus)\n`);

console.log('no-backpressure fan-out (trivial consumer, producer races ahead):');
for (const n of [400, 4_000, 40_000, 400_000]) {
  const r = await fanout(n, W);
  console.log(`  n=${n.toString().padStart(7)}  ${fmtThroughput(r)}  ${fmtCounts(r.counts)}`);
}

console.log('\nbackpressure fan-out (consumer burns CPU per item):');
for (const busyUs of [10, 50, 200]) {
  const n = busyUs >= 200 ? 4_000 : 20_000;
  const r = await fanoutBusy(n, W, busyUs);
  console.log(
    `  busy=${String(busyUs).padStart(3)}µs n=${n.toString().padStart(6)}  ${fmtThroughput(r)}  ${fmtCounts(r.counts)}`,
  );
}