/**
 * audio/parallel — jobs made at once, each on a thread of its own, one file, two faces.
 *
 * On the calling side, parallel(mod, shared, jobs, solve, each): each job made by `solve(shared, job)`, module `mod`'s
 * export of that name (a promise of it as well), the results in the jobs' order, and each as it comes, to `each(result,
 * index)` where given, which returning false gives out no more of them (those out come back, the rest stay undone).
 * In Node on worker threads (this file again, as their entry), one fewer than the cores, eight at
 * most; in a page or a worker, on the workers the host makes (audio.worker, a () → Worker whose entry calls serve());
 * with neither, here in turn, by the `solve` given. The jobs go out a few at a time as threads come free, `shared` once to each thread a
 * call; a view of part of a buffer goes as a copy of its part. A kernel whose jobs depend on `shared` alone comes out
 * the same, bit for bit, however they are spread (@audio/denoise-declip's declipAsync).
 *
 * Inside a thread, serve(load): `load(mod)` gives the module (an import() in Node; a page's entry, which has no import
 * maps, gives its own static imports):
 *   import { serve } from 'audio/parallel'
 *   import * as declip from '@audio/denoise/declip'
 *   serve(mod => ({ '@audio/denoise/declip': declip })[mod])
 */

const node = typeof process !== 'undefined' && !!process.versions?.node
const nodeWT = node ? await import('node:worker_threads').catch(() => null) : null
// the cores there are
const cores = node ? (await import('node:os')).availableParallelism?.() ?? 4 : globalThis.navigator?.hardwareConcurrency ?? 4

// a port's event heard, Node's or the web's, till the returned function is called
function on(port, type, fn) {
  if (port.on) { port.on(type, fn); return () => port.off(type, fn) }
  const f = type === 'message' ? e => fn(e.data) : fn
  port.addEventListener(type, f)
  return () => port.removeEventListener(type, f)
}

// the threads, one fewer than the cores, eight at most, made as calls need them (as many as a call has jobs) and kept;
// none where there is no way to make one. The host's maker is core's, read only here: a thread serving jobs loads no core
const pool = []
let make = null
async function threads(n) {
  if (!make && nodeWT?.isMainThread) {
    // the process's flags but those that make it a script of its own (node --input-type=module -e …), a thread's entry
    // being this file, and a profiler's: a thread profiling as it is let go of holds the process at its exit (Node 25)
    const execArgv = process.execArgv.filter((a, i, all) => !/^--(input-type|eval|print|cpu-prof|heap-prof)\b|^-[ep]$/.test(a) && !/^-[ep]$|^--(eval|print|cpu-prof-dir|cpu-prof-name|cpu-prof-interval|heap-prof-dir|heap-prof-name|heap-prof-interval)$/.test(all[i - 1] ?? ''))
    make = () => { const w = new nodeWT.Worker(new URL(import.meta.url), { workerData: { parallel: true }, execArgv }); w.unref(); return w }
  } else if (!make) make = (await import('./core.js')).default.worker
  if (!make) return null
  while (pool.length < Math.min(n, Math.max(1, Math.min(8, cores - 1)))) pool.push(make())
  return pool
}

let ids = 0
export default async function parallel(mod, shared, jobs, solve, each) {
  const all = jobs.length > 1 && await threads(jobs.length)
  if (!all) {
    const out = []
    for (const [i, j] of jobs.entries()) { out.push(await solve(shared, j)); if (each?.(out[i], i) === false) break }
    return out
  }
  shared = Object.fromEntries(Object.entries(shared).map(([k, v]) => [k, ArrayBuffer.isView(v) && v.byteLength !== v.buffer.byteLength ? v.slice() : v]))
  const id = ++ids, out = new Array(jobs.length), size = Math.max(1, Math.floor(jobs.length / (all.length * 8)))
  let next = 0
  await Promise.all(all.slice(0, Math.ceil(jobs.length / size)).map(w => new Promise((resolve, reject) => {
    const feed = () => {
      if (next >= jobs.length) { stop(); w.postMessage({ id, done: true }); return resolve() }
      const at = next
      next += size
      w.postMessage({ id, at, jobs: jobs.slice(at, next) })
    }
    const got = m => {
      if (m?.id !== id) return
      // a job that throws: the call rejects, every thread let go of it as its batch comes back
      if (m.error) { stop(); w.postMessage({ id, done: true }); next = jobs.length; return reject(new Error(m.error)) }
      m.out.forEach((v, i) => { out[m.at + i] = v; if (each?.(v, m.at + i) === false) next = jobs.length })
      feed()
    }
    const offs = [on(w, 'message', got), on(w, 'error', e => { stop(); reject(e) })], stop = () => offs.forEach(off => off())
    w.postMessage({ id, mod, shared })
    feed()
  })))
  return out
}

// Inside a thread: each call's `shared` kept till it is done, each batch of its jobs made and sent back, their buffers
// (and those of the arrays of them a job makes) moved, not copied
export function serve(load, port = nodeWT?.parentPort ?? globalThis) {
  const held = new Map()
  on(port, 'message', async m => {
    if (m.done) return held.delete(m.id)
    if (m.shared) return held.set(m.id, Promise.resolve(load(m.mod)).then(k => ({ solve: k.solve, shared: m.shared })))
    try {
      const { solve, shared } = await held.get(m.id), out = await Promise.all(m.jobs.map(j => solve(shared, j)))
      port.postMessage({ id: m.id, at: m.at, out }, [...new Set(out.flatMap(v => Array.isArray(v) ? v : [v]).map(v => v?.buffer).filter(b => b instanceof ArrayBuffer))])
    } catch (e) { port.postMessage({ id: m.id, error: e?.message ?? String(e) }) }
  })
}

if (nodeWT && !nodeWT.isMainThread && nodeWT.workerData?.parallel) serve(mod => import(mod))