// ── event_run_resources — SQL only ─────────────────────────────────────────
//
// EVENTS.md §D and §L ("The ledger's two rules"), and Phase 8 of EVENTS_PLAN.md.
// Everything one run created or leased, and what became of it.
//
// **Rule 1 lives in `reserve()`.** A resource is recorded BEFORE it is
// confirmed, so the placeholder this writes is the row that exists while the
// dispatch is in flight — and the row that SURVIVES when the acknowledgement is
// lost. Recording on the answer instead would make every object whose ack went
// missing invisible to cleanup for ever.
//
// **Rule 2 lives in the status column and in `failRevert()`.** A revert that
// never succeeds leaves its row unreverted, with the error on it, and the run
// completes with `cleanup_status = 'incomplete'` rather than being held open.
// Loud and sticky.
//
// **The unique key is enforced by the database, not by a read.** `reserve()`
// answers `{ ok: false, code: 'held' }` on a duplicate key rather than checking
// first and then inserting — two runs entering the same tick would both pass the
// check. It is the argument `event_run_budget.spend()` makes about the cap and
// `runsDb.transition` makes about a status, in the third place it applies.

const { query } = require('../../utils/db')
const { parseJson } = require('./eventJson')

// The one `kind` core owns. A module's kinds are opaque and stored verbatim; this
// one is core's own, and `registries` refuses a module resource that claims it.
const STEP_KIND = '@step'

// The statuses that mean "core still believes this resource is this run's". They
// are exactly the ones the `live_marker` generated column keeps non-NULL, so the
// unique target key holds while a row is in one of them and releases when it
// leaves. Duplicated here as a JavaScript list because the sweeps read by it too,
// and a second copy that can drift is better than a query that cannot express it.
const HELD = ['pending', 'confirmed', 'reverting']

// Every status that still wants a human or a retry: `HELD` plus the two that mean
// "we let go, and not cleanly". This is what "unreverted" means everywhere in
// this feature — the console's list, `cleanup_status`, and the manual retry.
const UNRESOLVED = [...HELD, 'orphaned', 'drifted']

const COLUMNS = `id, run_id, step_id, owner_module, kind, ref, payload, lease_until,
                 status, revert_attempts, last_error, member_key, created_at, updated_at`

// `payload` is opaque to core and stored verbatim, but it comes back as a string
// from the driver and every caller wants the object — the cleanup sweep reads a
// lease's baseline out of it, and the console renders it. Hydrated here for the
// same reason a step's params are: one place rather than at each read.
const hydrate = (row) => row && { ...row, payload: parseJson(row.payload, null) }

/**
 * Record a resource that does not exist yet.
 *
 * Answers `{ ok: true, id }`, or `{ ok: false, code: 'held', holder }` when the
 * target is already someone's — which is the lease conflict, surfaced as a
 * refusal rather than a failure because nothing is wrong with the system: another
 * run has the thing.
 *
 * **`ER_DUP_ENTRY` is the check.** The holder is looked up only to name it in the
 * refusal, and only after the insert has already lost the race.
 */
async function reserve({ runId, stepId = null, owner, kind, ref, payload = null, leaseUntil = null, memberKey = null }) {
  try {
    const result = await query(
      `INSERT INTO event_run_resources
         (run_id, step_id, owner_module, kind, ref, payload, lease_until, member_key, status)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'pending')`,
      [runId, stepId, owner, kind, ref, payload === null ? null : JSON.stringify(payload), leaseUntil, memberKey],
    )
    return { ok: true, id: Number(result.insertId) }
  } catch (err) {
    if (err && (err.code === 'ER_DUP_ENTRY' || err.errno === 1062)) {
      const [holder] = await query(
        `SELECT run_id, status FROM event_run_resources
          WHERE owner_module = ? AND kind = ? AND ref = ? AND status IN (?, ?, ?)
          LIMIT 1`,
        [owner, kind, ref, ...HELD],
      )
      return { ok: false, code: 'held', holder: holder || null }
    }
    throw err
  }
}

/**
 * Promote a reserved row to `confirmed`, optionally attaching what the module
 * finally said about it.
 *
 * Guarded on `pending` so a late answer cannot un-revert a row cleanup has
 * already dealt with — the same reason every other write in this feature is a
 * compare-and-set rather than a read-then-write.
 */
async function confirm(id, { payload, leaseUntil, memberKey } = {}) {
  const sets = ["status = 'confirmed'"]
  const params = []
  if (payload !== undefined) {
    sets.push('payload = ?')
    params.push(payload === null ? null : JSON.stringify(payload))
  }
  if (leaseUntil !== undefined) {
    sets.push('lease_until = ?')
    params.push(leaseUntil)
  }
  if (memberKey !== undefined) {
    sets.push('member_key = ?')
    params.push(memberKey)
  }
  const result = await query(
    `UPDATE event_run_resources SET ${sets.join(', ')} WHERE id = ? AND status = 'pending'`,
    [...params, id],
  )
  return (result.affectedRows || 0) > 0
}

/**
 * Resolve a step placeholder once the module has named what it actually made.
 *
 * The placeholder's whole job is over at this point: the real rows exist, so the
 * `@step` row must stop being one of the things cleanup will try to revert.
 * `reverted` is the honest terminal state for it — there is nothing left to undo
 * that the rows it stood in for do not now cover — and it releases the
 * idempotency key for a later run, which matters because keys are per step and a
 * re-materialised step reuses its own.
 */
async function resolvePlaceholder(id) {
  const result = await query(
    `UPDATE event_run_resources
        SET status = 'reverted', last_error = NULL
      WHERE id = ? AND kind = ? AND status IN ('pending', 'confirmed')`,
    [id, STEP_KIND],
  )
  return (result.affectedRows || 0) > 0
}

/**
 * One row by its target, live or not — how a caller that lost the insert race
 * finds the row it meant to write. Newest first, so a target that has been held
 * and released several times answers with the current holder.
 */
async function findByTarget(owner, kind, ref) {
  const [row] = await query(
    `SELECT ${COLUMNS} FROM event_run_resources
      WHERE owner_module = ? AND kind = ? AND ref = ?
      ORDER BY id DESC LIMIT 1`,
    [owner, kind, ref],
  )
  return hydrate(row) || null
}

/** One run's whole ledger, oldest first — the console's read. */
async function forRun(runId) {
  const rows = await query(
    `SELECT ${COLUMNS} FROM event_run_resources WHERE run_id = ? ORDER BY id`,
    [runId],
  )
  return rows.map(hydrate)
}

/** The rows of one run that still want something: the cleanup sweep's input. */
async function unresolvedForRun(runId, { maxAttempts = null } = {}) {
  const params = [runId, ...UNRESOLVED]
  const attemptClause = maxAttempts === null ? '' : ' AND revert_attempts < ?'
  if (maxAttempts !== null) params.push(maxAttempts)
  const rows = await query(
    `SELECT ${COLUMNS} FROM event_run_resources
      WHERE run_id = ? AND status IN (?, ?, ?, ?, ?)${attemptClause}
      ORDER BY id`,
    params,
  )
  return rows.map(hydrate)
}

/** How many of one run's rows are still unresolved — what `cleanup_status` is derived from. */
async function unresolvedCount(runId) {
  const [row] = await query(
    `SELECT COUNT(*) AS n FROM event_run_resources
      WHERE run_id = ? AND status IN (?, ?, ?, ?, ?)`,
    [runId, ...UNRESOLVED],
  )
  return Number(row?.n || 0)
}

/** Unresolved counts for several runs at once, keyed by run id — the run LIST's read. */
async function unresolvedCounts(runIds) {
  const ids = [...new Set(runIds || [])].filter(Boolean)
  if (!ids.length) return new Map()
  const rows = await query(
    `SELECT run_id, COUNT(*) AS n FROM event_run_resources
      WHERE run_id IN (${ids.map(() => '?').join(',')}) AND status IN (?, ?, ?, ?, ?)
      GROUP BY run_id`,
    [...ids, ...UNRESOLVED],
  )
  return new Map(rows.map((r) => [r.run_id, Number(r.n)]))
}

/**
 * Claim one row for a revert: `pending | confirmed | orphaned | drifted → reverting`,
 * and `reverting` again once the claim on it has gone stale.
 *
 * The compare-and-set that keeps the cleanup leg and the manual cleanup route off
 * each other's rows. A row another pass is mid-revert on is left alone, exactly as
 * a step with a live claim is.
 *
 * **"Exactly as a step" has to include the expiry, and it did not until the Phase
 * 16 acceptance walk.** A step's claim carries `claim_expires_at`, so a step whose
 * process died is reclaimed once the lease lapses — that reclaim is the whole
 * reason §E's CAS survives §N4's single instance. A `reverting` row had no such
 * bound and nothing released it, so a process killed mid-teardown stranded the row
 * for good: the sweep skipped it every 15s forever, `cleanup_status` never left
 * `pending`, and `POST …/cleanup` — the recourse §I names — answered 200 and did
 * nothing, because it claims through this same function. Observed with a lease,
 * which then blocked the NEXT run of the same event from taking the value.
 *
 * The stale test is `updated_at`, not a new column: the row is stamped exactly
 * when it enters `reverting` and is not written again until the revert resolves,
 * so for a `reverting` row `updated_at` IS "when this claim was taken". The bound
 * is the run lease's, for the run lease's reason — it has to outlast a whole
 * tick's work on one run, and every revert in a sweep is bounded by its action's
 * own `budgetMs` long before this.
 *
 * `revert_attempts` is deliberately NOT incremented by reclaiming. A stale claim
 * is a process that died, not an attempt that failed, and counting it would burn
 * the retry budget on crashes — Engagement Phase 14's rule, one table over.
 *
 * **`updated_at` is re-stamped explicitly, and that is what keeps this a CAS.**
 * This connector sends `CLIENT_FOUND_ROWS`, so `affectedRows` counts rows MATCHED
 * rather than changed. For the four fresh statuses that is harmless — the winner
 * moves the row to `reverting` and the loser's `status IN (…)` no longer matches.
 * A stale `reverting` row has no such natural change: without re-stamping, the
 * row would still satisfy `status = 'reverting' AND updated_at < …` and a second
 * claimer would match it too. Writing the column is what makes the second one
 * miss.
 */
const REVERT_CLAIM_TTL_MS = Number(process.env.EVENT_REVERT_CLAIM_TTL_MS) || 15 * 60 * 1000

async function claimRevert(id) {
  const result = await query(
    `UPDATE event_run_resources
        SET status = 'reverting', updated_at = NOW()
      WHERE id = ?
        AND (status IN ('pending', 'confirmed', 'orphaned', 'drifted')
             OR (status = 'reverting'
                 AND updated_at < (NOW() - INTERVAL ? MICROSECOND)))`,
    [id, REVERT_CLAIM_TTL_MS * 1000],
  )
  return (result.affectedRows || 0) > 0
}

/** The revert worked. `reverted` is terminal and releases the target. */
async function markReverted(id) {
  await query(
    `UPDATE event_run_resources SET status = 'reverted', last_error = NULL WHERE id = ?`,
    [id],
  )
}

/**
 * The revert did not work, and the row goes back to being unresolved.
 *
 * `revert_attempts` is incremented here and NOWHERE else, and it is never reset by
 * a sweep — Engagement Phase 14's rule, whose defect was a reclaim that returned
 * every stale row to its start state and made the attempt ceiling unreachable, so
 * the row cycled for ever and was never eligible for any retention sweep. The one
 * thing that may reset it is a human pressing cleanup, which is the same licence
 * a human's step retry has.
 *
 * `restoreTo` is where the row lands: `drifted` when the module says somebody else
 * moved the value, `orphaned` when it says the thing is gone, and `confirmed`
 * otherwise — still ours, still out there, try again.
 */
async function failRevert(id, error, restoreTo = 'confirmed') {
  await query(
    `UPDATE event_run_resources
        SET status = ?, revert_attempts = revert_attempts + 1, last_error = ?
      WHERE id = ?`,
    [restoreTo, String(error || 'the revert did not answer').slice(0, 500), id],
  )
}

/**
 * A human is trying again: clear the attempt counter on one run's unresolved rows.
 *
 * Only ever called from the cleanup route with an actor behind it. The automatic
 * leg must never do this (see `failRevert`).
 */
async function resetAttempts(runId) {
  const result = await query(
    `UPDATE event_run_resources
        SET revert_attempts = 0
      WHERE run_id = ? AND status IN (?, ?, ?, ?, ?)`,
    [runId, ...UNRESOLVED],
  )
  return result.affectedRows || 0
}

/** Every live row one module owns, for the reconcile sweep. */
async function liveForModule(owner, { limit = 500 } = {}) {
  const rows = await query(
    `SELECT ${COLUMNS} FROM event_run_resources
      WHERE owner_module = ? AND status IN ('pending', 'confirmed')
      ORDER BY id LIMIT ?`,
    [owner, Number(limit)],
  )
  return rows.map(hydrate)
}

/** Every module that currently owns a live row — who the reconcile sweep asks. */
async function modulesWithLiveRows() {
  const rows = await query(
    `SELECT DISTINCT owner_module FROM event_run_resources
      WHERE status IN ('pending', 'confirmed')`,
  )
  return rows.map((r) => r.owner_module)
}

/**
 * The game no longer has it. Never reached by a revert — a revert that finds
 * nothing there is a SUCCESS (§L, and it is what a Rust wipe needs) — only by
 * reconcile, which is a different question: nobody asked for this to go.
 */
async function markOrphaned(id, detail = null) {
  await query(
    `UPDATE event_run_resources
        SET status = 'orphaned', last_error = ?
      WHERE id = ? AND status IN ('pending', 'confirmed', 'reverting')`,
    [detail === null ? null : String(detail).slice(0, 500), id],
  )
}

/**
 * Terminal runs that still owe the world something — the cleanup leg's scan.
 *
 * **Both halves of the WHERE were live-walk findings, and they are opposite
 * mistakes.**
 *
 * `cleanup_status = 'pending'` alone missed a run whose only resource was a
 * LEASE: `core.lease` reserves its own row and never goes through the ledger's
 * `markRunDirty`, so the flag stayed `not_required` and the lease was never given
 * back at all. Hence `not_required` is in the list — a terminal run with an
 * unresolved row has something to do whatever any summary column says, and
 * treating that combination as work is the fail-safe direction.
 *
 * And the run status filter alone made `MAX_REVERT_ATTEMPTS` mean ONE attempt,
 * because the first failing sweep set `incomplete` and nothing looked at the run
 * again. That is fixed in `cleanupRun`, which now only writes `incomplete` once
 * there is nothing left it will try — so `incomplete` genuinely means "finished
 * with, and not finished", which is exactly what excludes both a run whose
 * retries are spent and a run an admin cancelled without cleanup.
 *
 * The attempt bound is in the join for a different reason: without it a run whose
 * rows are all spent would be selected, worked over and found to have nothing to
 * do on every tick for the rest of its life.
 */
async function runsNeedingCleanup(limit = 25, maxAttempts = 3) {
  return query(
    `SELECT DISTINCT r.id, r.status, r.cleanup_status, r.version_id, r.definition_id, r.scope
       FROM event_runs r
       JOIN event_run_resources res ON res.run_id = r.id
      WHERE r.status IN ('completed', 'cancelled', 'failed', 'missed')
        AND r.cleanup_status IN ('pending', 'not_required')
        AND res.status IN (?, ?, ?, ?, ?)
        AND res.revert_attempts < ?
      ORDER BY r.id
      LIMIT ?`,
    [...UNRESOLVED, Number(maxAttempts), Number(limit)],
  )
}

module.exports = {
  STEP_KIND,
  HELD,
  UNRESOLVED,
  reserve,
  confirm,
  resolvePlaceholder,
  findByTarget,
  forRun,
  unresolvedForRun,
  unresolvedCount,
  unresolvedCounts,
  claimRevert,
  markReverted,
  failRevert,
  resetAttempts,
  liveForModule,
  modulesWithLiveRows,
  markOrphaned,
  runsNeedingCleanup,
}