// How the merge queue is doing, by day: how long pulls take from queued
// to landed, how often they are ejected and why, how long batches take,
// what cancels them, and which tests fail or pass only on a retry.
//
//   node harness/queue-stats.mjs [<handle>/<name>] [--days <n>] [--json]
//
// It reads what the queue already writes: the queue records in the
// owner's PDS and its runners' (a pull is queued by its owner and ejected
// by a runner), the batch checks the runners file (through `knotless
// checks --queue`, so the CLI must be logged in or the repository
// public), their logs, and main's log in this checkout for when each pull
// landed (`git fetch` first). The runner prunes batch checks it no longer
// needs, so each run keeps what it read in target/queue-stats/, and a
// later run counts it too: run it now and again after a change, and
// compare the days either side.

import { execFileSync } from 'node:child_process'
import { existsSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs'
import { dirname, join, resolve } from 'node:path'
import { fileURLToPath } from 'node:url'

const root = resolve(dirname(fileURLToPath(import.meta.url)), '..')
const args = process.argv.slice(2)
const flag = (name, fallback) => {
  const i = args.indexOf(name)
  return i < 0 ? fallback : args[i + 1]
}
const repo =
  args.find((a, i) => !a.startsWith('--') && !args[i - 1]?.startsWith('--')) ??
  'chadtmiller.com/knotless'
const days = Number(flag('--days', 7))
const json = args.includes('--json')
const since = Date.now() - days * 86_400_000
const store = join(root, 'target/queue-stats')
const kept = join(store, `${repo.replace('/', '_')}.json`)

const run = (cmd, argv) =>
  execFileSync(cmd, argv, { cwd: root, encoding: 'utf8', maxBuffer: 1 << 28 })
const get = async (url) => {
  const res = await fetch(url)
  if (!res.ok) throw new Error(`${url}: ${res.status}`)
  return res
}

/// An account's PDS, from its DID document.
const pdsOf = new Map()
const pds = async (did) => {
  if (!pdsOf.has(did)) {
    const doc = did.startsWith('did:web:')
      ? await (await get(`https://${did.slice(8)}/.well-known/did.json`)).json()
      : await (await get(`https://plc.directory/${did}`)).json()
    pdsOf.set(did, doc.service.find((s) => s.id === '#atproto_pds').serviceEndpoint)
  }
  return pdsOf.get(did)
}

/// Every record of `collection` in `did`'s repository, newest first,
/// down to `since`.
const records = async (did, collection) => {
  const base = await pds(did)
  const out = []
  let cursor
  do {
    const q = new URLSearchParams({
      repo: did,
      collection,
      limit: '100',
      ...(cursor && { cursor }),
    })
    const page = await (await get(`${base}/xrpc/com.atproto.repo.listRecords?${q}`)).json()
    out.push(...page.records)
    cursor = page.cursor
    const last = page.records.at(-1)?.value.createdAt
    if (last && Date.parse(last) < since) break
  } while (cursor)
  return out
}

/// What a batch's log says that its record does not: why it was
/// cancelled, the tests that failed, and those that passed on a retry.
const facts = (log) => ({
  cancelled: log.match(/^==> cancelled (.*)$/m)?.[1] ?? null,
  flaky: [...log.matchAll(/^FLAKY: (\S+) (.+?) \((\w+)\) (\S+)$/gm)].map(
    (m) => `${m[1]} ${m[2]} ${m[4]}`,
  ),
  failed: [...new Set([...log.matchAll(/^test (\S+) \.\.\. FAILED$/gm)].map((m) => m[1]))],
  uncovered: /^Uncovered Lines:/m.test(log),
})

const saved = existsSync(kept) ? JSON.parse(readFileSync(kept, 'utf8')) : { batches: {} }

// Batch checks, newest first, with what their logs say: those read
// before are kept, so pruned ones still count.
const listed = JSON.parse(
  run('knotless', ['run', 'list', `knotless://${repo}`, '--queue', 'main', '--json']),
)
for (const c of listed) {
  const known = saved.batches[c.uri]
  if (known && known.status === c.status) continue
  let read = { cancelled: null, flaky: [], failed: [], uncovered: false }
  if (c.status !== 'running' && c.status !== 'queued' && c.log?.ref?.$link) {
    const q = new URLSearchParams({ did: c.runner, cid: c.log.ref.$link })
    read = facts(
      await (await get(`${await pds(c.runner)}/xrpc/com.atproto.sync.getBlob?${q}`)).text(),
    )
  }
  saved.batches[c.uri] = {
    workflow: c.workflow,
    status: c.status,
    commit: c.commit,
    base: c.baseCommit,
    pulls: (c.batch ?? []).map((b) => b.pull),
    startedAt: c.startedAt ?? null,
    finishedAt: c.finishedAt ?? null,
    ...read,
  }
}

// Queue records: when each pull was queued, and each ejection and why.
const owner = listed[0]?.repo.uri.split('/')[2]
const name = repo.split('/')[1]
const repoUri = owner && `at://${owner}/dev.knotless.repo/${name}`
const repoRecord = async () => {
  const q = new URLSearchParams({ repo: owner, collection: 'dev.knotless.repo', rkey: name })
  const res = await get(`${await pds(owner)}/xrpc/com.atproto.repo.getRecord?${q}`)
  return (await res.json()).value
}
const runners = owner ? ((await repoRecord()).runners ?? []) : []
for (const did of owner ? [owner, ...runners.map((r) => r.did)] : []) {
  for (const r of await records(did, 'dev.knotless.queue')) {
    saved.queue = { ...saved.queue, [r.uri]: r.value }
  }
}

// Landed: the squashed commits on main, each naming its pull.
const landed = {}
const log = run('git', [
  'log',
  'origin/main',
  `--since=${new Date(since).toISOString()}`,
  '--format=%H%x00%cI%x00%B%x1e',
])
for (const entry of log.split('\x1e')) {
  const [sha, at, body] = entry.trim().split('\x00')
  const pull = body?.match(/at:\/\/\S+\/dev\.knotless\.pull\/[a-z0-9]+/g)?.at(-1)
  if (pull) landed[pull] = { sha, at }
}

mkdirSync(store, { recursive: true })
writeFileSync(kept, JSON.stringify(saved))

const day = (t) => t.slice(0, 10)
const minutes = (a, b) => (Date.parse(b) - Date.parse(a)) / 60_000
const median = (xs) => {
  if (!xs.length) return null
  const s = [...xs].sort((a, b) => a - b)
  return s[Math.floor((s.length - 1) / 2)]
}
const pct = (q) => (xs) =>
  xs.length ? [...xs].sort((a, b) => a - b)[Math.ceil(xs.length * q) - 1] : null
const p95 = pct(0.95)
const kind = (reason = '') =>
  reason.includes('pre-flight')
    ? 'failed pre-flight'
    : reason.includes('not reviewed') || reason.includes('changes asked')
      ? 'review'
      : reason.includes('conflicts')
        ? 'conflict'
        : reason.includes('failed its checks')
          ? 'failed batch'
          : reason.includes('has no branch')
            ? 'branch gone'
            : 'other'
const cancel = (why = '') =>
  why.includes('runner stopped')
    ? 'runner restart'
    : why.includes('dropped its batch')
      ? 'batch dropped'
      : why.includes('no longer at that head')
        ? 'moved on'
        : why || 'unknown'
const tally = (xs) =>
  Object.entries(xs.reduce((n, x) => ({ ...n, [x]: (n[x] ?? 0) + 1 }), {})).sort(
    (a, b) => b[1] - a[1],
  )

const byDay = {}
const at = (d) =>
  (byDay[d] ??= {
    landed: [],
    ejected: [],
    batches: [],
    tests: [],
    cancelled: [],
    flaky: [],
    failed: [],
  })

// Each landed pull: from the first time it was queued for this landing,
// past any ejections, to the commit on main.
// A runner serves other repositories too, and the owner has others.
const queueRecords = Object.values(saved.queue ?? {}).filter(
  (v) => Date.parse(v.createdAt) >= since && (!repoUri || v.repo?.uri === repoUri),
)
for (const [pull, { at: when }] of Object.entries(landed)) {
  const mine = queueRecords
    .filter((v) => v.subject === pull && v.createdAt <= when)
    .sort((a, b) => a.createdAt.localeCompare(b.createdAt))
  const first = mine.find((v) => v.state === 'queued')
  if (!first) continue
  at(day(when)).landed.push({
    pull,
    minutes: minutes(first.createdAt, when),
    queued: mine.filter((v) => v.state === 'queued').length,
    ejected: mine.filter((v) => v.state === 'ejected').length,
  })
}
for (const v of queueRecords) {
  if (v.state === 'queued') at(day(v.createdAt)).entered = (at(day(v.createdAt)).entered ?? 0) + 1
  if (v.state === 'ejected') at(day(v.createdAt)).ejected.push(kind(v.reason))
}

// A batch is its six checks at one commit and base; its time is its
// `test` run's, the longest.
const batches = Object.values(saved.batches).filter(
  (b) => b.startedAt && Date.parse(b.startedAt) >= since,
)
const runs = {}
for (const b of batches) (runs[`${b.commit} ${b.base}`] ??= []).push(b)
for (const checks of Object.values(runs)) {
  const d = at(day(checks[0].startedAt))
  const status = checks.some((c) => c.status === 'failure')
    ? 'failure'
    : checks.some((c) => c.status === 'cancelled')
      ? 'cancelled'
      : checks.some((c) => c.status === 'running' || c.status === 'queued')
        ? 'running'
        : 'success'
  d.batches.push({ status, size: checks[0].pulls.length })
  for (const c of checks) {
    if (c.status === 'cancelled') d.cancelled.push(cancel(c.cancelled))
    d.flaky.push(...c.flaky)
    d.failed.push(...c.failed)
  }
  const test = checks.find((c) => c.workflow === 'test' && c.status === 'success' && c.finishedAt)
  if (test) d.tests.push(minutes(test.startedAt, test.finishedAt))
}

const report = Object.keys(byDay)
  .sort()
  .map((d) => {
    const x = byDay[d]
    const waits = x.landed.map((l) => l.minutes)
    const done = x.batches.filter((b) => b.status === 'success' || b.status === 'failure')
    return {
      day: d,
      queued: x.entered ?? 0,
      landed: x.landed.length,
      queuedToLanded: { median: median(waits), p95: p95(waits) },
      ejectionRate: x.entered ? x.ejected.length / x.entered : null,
      queuedAgain: x.landed.filter((l) => l.queued > 1).length,
      ejections: Object.fromEntries(tally(x.ejected)),
      batches: Object.fromEntries(tally(x.batches.map((b) => b.status))),
      batchSize: median(x.batches.map((b) => b.size)),
      batchSuccess: done.length
        ? done.filter((b) => b.status === 'success').length / done.length
        : null,
      batchTest: { median: median(x.tests), p95: p95(x.tests) },
      cancelledChecks: Object.fromEntries(tally(x.cancelled)),
      flaky: Object.fromEntries(tally(x.flaky)),
      failedTests: Object.fromEntries(tally(x.failed)),
    }
  })

if (json) {
  console.log(JSON.stringify(report, null, 2))
} else {
  const n = (v) => (v == null ? '-' : Math.round(v))
  const list = (o) =>
    Object.entries(o)
      .map(([k, v]) => `${k} ${v}`)
      .join(', ') || '-'
  console.log(`${repo}, the last ${days} days (UTC)\n`)
  const pc = (v) => (v == null ? '-' : `${Math.round(v * 100)}%`)
  console.log(
    'day         queued  landed  queued→landed min p50/p95  ejected  batches passed  batch test min p50/p95  batches',
  )
  for (const r of report) {
    console.log(
      `${r.day}  ${String(r.queued).padStart(6)}  ${String(r.landed).padStart(6)}  ${`${n(r.queuedToLanded.median)}/${n(r.queuedToLanded.p95)}`.padStart(25)}  ${pc(r.ejectionRate).padStart(7)}  ${pc(r.batchSuccess).padStart(14)}  ${`${n(r.batchTest.median)}/${n(r.batchTest.p95)}`.padStart(22)}  ${list(r.batches)}`,
    )
  }
  console.log(
    '\nHealthy, as merge queues are commonly judged: queued→landed p50 under 30 min and p95 under 60; a few percent ejected; over 80% of batches passing.',
  )
  for (const r of report) {
    console.log(`\n${r.day}`)
    console.log(`  ejections: ${list(r.ejections)}`)
    console.log(`  cancelled checks: ${list(r.cancelledChecks)}`)
    console.log(`  passed on a retry: ${list(r.flaky)}`)
    console.log(`  failed, on any try: ${list(r.failedTests)}`)
  }
}