// Runs PROTOCOL.md scenarios. Usage: node run.mjs <scenarioId|all> [repetitions]
import { fork } from 'node:child_process'
import { copyFileSync, mkdirSync, openSync, readFileSync, rmSync, writeFileSync, existsSync } from 'node:fs'
import { dirname, join, resolve } from 'node:path'
import { fileURLToPath } from 'node:url'
import { createRequire } from 'node:module'

const here = dirname(fileURLToPath(import.meta.url))
const root = resolve(here, '..')
const sleep = ms => new Promise(r => setTimeout(r, ms))
const LEASE_MS = 4000
const TERMINAL = new Set(['completed', 'failed', 'cancelled', 'timed_out'])

const variants = {
  current: { dir: here, label: '@mastra/core@1.72.0 + @mastra/libsql@1.24.0' },
  oldAdapter: { dir: join(root, 'experiment-old-adapter'), label: '@mastra/core@1.72.0 + @mastra/libsql@1.23.3' },
}

const base = { preSendMs: 300, postEffectMs: 2500, useIdempotencyKey: true }
const scenarios = {
  S1: { variant: 'current', workers: 1, args: { ...base, useIdempotencyKey: false } },
  S2: { variant: 'current', workers: 2, args: { ...base, postEffectMs: 9000 } },
  S3: { variant: 'current', workers: 2, args: { ...base, preSendMs: 1500 }, on: 'attempt.start', action: 'SIGKILL' },
  S4: { variant: 'current', workers: 2, args: { ...base, useIdempotencyKey: false }, on: 'send.response', action: 'SIGKILL' },
  S5: { variant: 'current', workers: 2, args: base, on: 'send.response', action: 'SIGKILL' },
  S6: { variant: 'current', workers: 2, args: base, on: 'send.response', action: 'FREEZE' },
  S7: { variant: 'current', workers: 2, args: { ...base, preSendMs: 1500 }, on: 'attempt.start', action: 'SIGTERM' },
  P1: { variant: 'oldAdapter', workers: 2, args: { ...base, postEffectMs: 9000 } },
  P2: { variant: 'oldAdapter', workers: 2, args: base, on: 'send.response', action: 'FREEZE' },
  P3: { variant: 'oldAdapter', workers: 2, args: { ...base, postEffectMs: 9000, useIdempotencyKey: false } },
}

async function runOnce(id, rep) {
  const sc = scenarios[id]
  const variant = variants[sc.variant]
  const out = join(root, 'runs', `${id}-${rep}`)
  rmSync(out, { recursive: true, force: true })
  mkdirSync(out, { recursive: true })
  if (variant.dir !== here) copyFileSync(join(here, 'worker.mjs'), join(variant.dir, 'worker.mjs'))
  const require = createRequire(join(variant.dir, 'package.json'))
  const { createClient } = await import(require.resolve('@libsql/client'))
  const versions = Object.fromEntries(['@mastra/core', '@mastra/libsql'].map(name =>
    [name, JSON.parse(readFileSync(join(variant.dir, 'node_modules', name, 'package.json'))).version]))

  const port = 47100 + Math.floor(Math.random() * 800)
  const dbUrl = `file:${join(out, 'mastra.db')}`
  const eventsPath = join(out, 'events.jsonl')
  const events = []
  const t0 = Date.now()
  const children = {}
  const marks = {}
  const spawnLogged = (name, file, args, env) => {
    const fd = openSync(join(out, `${name}.log`), 'a')
    const child = fork(file, args, { cwd: variant.dir, env: { ...process.env, ...env }, stdio: ['ignore', fd, fd, 'ipc'] })
    child.on('exit', (code, signal) => { events.push({ at: Date.now(), worker: name, event: 'process.exit', code, signal }) })
    children[name] = child
    return child
  }
  const waitFor = (pred, ms = 30_000) => new Promise((ok, fail) => {
    const started = Date.now()
    const tick = () => { const hit = events.find(pred); if (hit) ok(hit); else if (Date.now() - started > ms) fail(new Error('timeout')); else setTimeout(tick, 25) }
    tick()
  })
  const env = name => ({ WORKER: name, DB_URL: dbUrl, RECEIVER: `http://127.0.0.1:${port}`, EVENTS: eventsPath, LEASE_MS: String(LEASE_MS) })

  const receiver = spawnLogged('receiver', join(here, 'receiver.mjs'), [String(port), join(out, 'receiver.db')], {})
  receiver.on('message', m => events.push({ at: Date.now(), worker: 'receiver', ...m }))
  await waitFor(e => e.event === 'receiver.ready')

  const startWorker = name => {
    const child = spawnLogged(name, join(variant.dir, 'worker.mjs'), [], env(name))
    child.on('message', m => events.push(m))
    return waitFor(e => e.worker === name && e.event === 'worker.ready')
  }
  await startWorker('A')
  const operationId = `op-${id}-${rep}`
  children.A.send({ cmd: 'enqueue', args: { operationId, ...sc.args }, maxRetries: 1 })

  // Sample the persisted task row and keep every change of status / owner / lease.
  const db = createClient({ url: dbUrl })
  const COLUMNS = 'status, ownerId, leaseExpiresAt, retry_count AS retryCount, json(result) AS result, json(error) AS error'
  const readTask = async () => (await db.execute(`SELECT ${COLUMNS} FROM mastra_background_tasks`)).rows[0]
  const storage = []
  let sampling = true
  const sampler = (async () => {
    let last = ''
    while (sampling) {
      try {
        const row = await readTask()
        if (row) {
          const key = `${row.status}|${row.ownerId}|${row.leaseExpiresAt}|${row.retryCount}`
          if (key !== last) { last = key; storage.push({ at: Date.now(), ...row }) }
        }
      } catch {}
      await sleep(50)
    }
  })()

  let verdict = 'observed'
  try {
    await waitFor(e => e.worker === 'A' && e.event === 'attempt.start')
    if (sc.workers === 2) { marks.bStartRequested = Date.now(); startWorker('B') }
    if (sc.on) {
      await waitFor(e => e.worker === 'A' && e.event === sc.on)
      marks.interruptAt = Date.now()
      if (sc.action === 'FREEZE') children.A.kill('SIGSTOP')
      else children.A.kill(sc.action)
      events.push({ at: marks.interruptAt, worker: 'orchestrator', event: `interrupt.${sc.action}` })
    }
    const deadline = Date.now() + 45_000
    let task
    while (Date.now() < deadline) { task = await readTask(); if (task && TERMINAL.has(task.status)) break; await sleep(200) }
    if (!task || !TERMINAL.has(task.status)) verdict = 'inconclusive'
    if (sc.action === 'FREEZE') {
      marks.thawAt = Date.now()
      children.A.kill('SIGCONT')
      events.push({ at: marks.thawAt, worker: 'orchestrator', event: 'interrupt.SIGCONT' })
      await sleep(LEASE_MS + 5000)
    } else await sleep(3000)
    task = await readTask()
    const state = await (await fetch(`http://127.0.0.1:${port}/state`)).json()
    sampling = false
    await sampler
    db.close()
    const owners = Object.fromEntries(events.filter(e => e.event === 'worker.ready').map(e => [e.ownerId, e.worker]))
    const storageTimeline = storage.map(s => ({ t: s.at - t0, status: s.status, owner: s.ownerId ? owners[s.ownerId] ?? s.ownerId : null,
      leaseExpiresInMs: s.leaseExpiresAt ? Date.parse(s.leaseExpiresAt) - s.at : null, retryCount: s.retryCount }))
    return finish({ task, state, storageTimeline })
  } catch (error) {
    verdict = 'inconclusive'
    sampling = false
    return finish({ error: String(error) })
  }

  function finish(extra) {
    for (const child of Object.values(children)) { try { child.kill('SIGCONT'); child.kill('SIGKILL') } catch {} }
    const fileEvents = existsSync(eventsPath) ? readFileSync(eventsPath, 'utf8').trim().split('\n').filter(Boolean).map(JSON.parse) : []
    const timeline = [...fileEvents, ...events.filter(e => e.worker === 'orchestrator' || e.event === 'process.exit')]
      .sort((a, b) => a.at - b.at).map(e => ({ ...e, t: e.at - t0 }))
    const record = { scenario: id, rep, variant: variant.label, versions, leaseMs: LEASE_MS, args: { operationId, ...sc.args },
      action: sc.action ?? null, verdict, marks: Object.fromEntries(Object.entries(marks).map(([k, v]) => [k, v - t0])), timeline, ...extra }
    writeFileSync(join(out, 'record.json'), JSON.stringify(record, (k, v) => typeof v === 'bigint' ? Number(v) : v, 2))
    return record
  }
}

const [which = 'all', repsArg = '5'] = process.argv.slice(2)
const ids = which === 'all' ? Object.keys(scenarios) : which.split(',')
for (const id of ids) for (let rep = 1; rep <= Number(repsArg); rep++) {
  const record = await runOnce(id, rep)
  const effects = record.state?.effects?.length, requests = record.state?.requests?.length
  console.log(`${id}#${rep} ${record.verdict} status=${record.task?.status} owner=${record.task?.ownerId ?? record.task?.owner_id ?? '-'} requests=${requests} effects=${effects}`)
}