website/server/src/model/events/eventRuns.db.js

// ── event_runs — SQL only ──────────────────────────────────────────────────
//
// EVENTS.md §D and §E. Phase 1 writes exactly one kind of row — a `scheduled`
// occurrence — and reads them back for the admin surface. **The claim, the CAS
// transitions and the lease reclaim are Phase 2's** and are deliberately not
// stubbed here: a half-written claim is worse than no claim, because it reads as
// protection.
//
// What Phase 1 does own is the INSERT, and it owns the important half of it:
// materialisation is `INSERT IGNORE` against `UNIQUE (definition_id, scope,
// scheduled_for)`, so a second attempt at one occurrence writes nothing and
// answers honestly rather than raising a duplicate-key error a caller has to
// interpret.

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

const hydrate = (row) => row && { ...row, params: parseJson(row.params, null), rehearsal: Boolean(row.rehearsal) }

// The statuses a run never leaves. A transition INTO one of these stamps
// `ended_at` and drops the claim, and only rows in one of them are eligible for
// the log retention sweep -- Engagement Phase 14's rule, which is a bound at all
// only if every path a run can take reaches one of them.
const TERMINAL = ['completed', 'cancelled', 'failed', 'missed']

// §E's health values, worst last. Health is a HIGH-WATER MARK in this system —
// nothing has ever cleared `degraded`, because a run whose announcement landed
// on the second attempt did have trouble and that stays true for the rest of its
// life — and `setHealth` enforces that rather than leaving it to every caller to
// remember. `FIELD()` gives the same order inside the WHERE clause, 1-indexed,
// which is what makes the guard one statement rather than a read and a write.
const HEALTH_ORDER = ['ok', 'degraded', 'stalled']
const HEALTH_RANK = Object.fromEntries(HEALTH_ORDER.map((h, i) => [h, i + 1]))
const HEALTH_SQL_ORDER = HEALTH_ORDER.map((h) => `'${h}'`).join(', ')

// `waiting_steps` is the count of PARKED steps: `running` with a NULL lease, the
// pair `park()` alone produces, which means a cue waiting on a human. It is a
// correlated subquery on an admin list bounded at 500 rows rather than a column,
// because it is derived from the steps and a column would be a second writer's
// opinion of them. It earns its cost on the list screen: a cue nobody notices is
// a run that never advances, and the run itself looks perfectly healthy until
// somebody opens it.
const SELECT_LIST = `
  SELECT r.*, d.title AS definition_title, d.slug AS definition_slug, v.version AS version_number,
         (SELECT COUNT(*) FROM event_run_steps s
           WHERE s.run_id = r.id AND s.status = 'running' AND s.claim_expires_at IS NULL) AS waiting_steps
    FROM event_runs r
    JOIN event_definitions d ON d.id = r.definition_id
    JOIN event_versions    v ON v.id = r.version_id
`

/**
 * The admin run list. Newest occurrence first, across every definition.
 *
 * `limit` is interpolated after an integer coercion rather than bound, because
 * MariaDB will not take a placeholder in LIMIT on a prepared statement. It never
 * reaches SQL as anything but a number.
 */
const list = async ({ definitionId = null, status = null, limit = 100 } = {}) => {
  const where = []
  const args = []
  if (definitionId) {
    where.push('r.definition_id = ?')
    args.push(definitionId)
  }
  if (status) {
    where.push('r.status = ?')
    args.push(status)
  }
  const clause = where.length ? `WHERE ${where.join(' AND ')}` : ''
  const n = Math.min(Math.max(Number(limit) || 100, 1), 500)
  const rows = await query(
    `${SELECT_LIST} ${clause} ORDER BY r.scheduled_for DESC, r.id DESC LIMIT ${n}`,
    args,
  )
  return rows.map(hydrate)
}

const getById = async (id) => {
  const [row] = await query(`${SELECT_LIST} WHERE r.id = ?`, [id])
  return hydrate(row)
}

/**
 * Materialise one occurrence. Answers the row id, or `null` when one already
 * existed — which is not an error and is the ordinary answer under a tick that
 * overran into the next one.
 */
const materialise = async (run) => {
  const result = await query(
    `INSERT IGNORE INTO event_runs
       (definition_id, version_id, scope, scheduled_for, timezone, concurrency_key,
        params, rehearsal, started_by)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
    [
      run.definition_id,
      run.version_id,
      run.scope || '',
      run.scheduled_for,
      run.timezone || 'UTC',
      run.concurrency_key,
      run.params === null || run.params === undefined ? null : JSON.stringify(run.params),
      run.rehearsal ? 1 : 0,
      run.started_by,
    ],
  )
  return Number(result?.affectedRows || 0) === 1 ? result.insertId : null
}

/**
 * Every run whose OCCUPIED INTERVAL overlaps a window — the calendar's real half.
 *
 * Ascending, unlike the admin run list: a calendar is read forwards. The join
 * reaches the series so a month can be filtered to one arc without a second
 * round trip, and `d.timezone` is NOT what comes back — `r.timezone` is, because
 * a run records the zone it was COMPUTED in and a definition's zone can be
 * edited afterwards.
 *
 * **A run OVERLAPS the window; it does not merely START in it.** This asked
 * `scheduled_for >= from` alone until the Phase 16 acceptance walk, and a run is
 * not an instant — it is an interval, and a multi-phase event's whole point is
 * that the interval is long. A run that began before `from` and has not ended is
 * happening DURING the window and belongs in it. With the instant test, the
 * public calendar answered `entries: []` while that same event's own page said
 * `live: true`, so the site disagreed with itself about whether something was on
 * — and `EVENTS.md` §I promises this route serves "upcoming, **live** and
 * recent". The admin calendar had the same hole for the same reason: a run that
 * started last Sunday and is still going was missing from "this week".
 *
 * A finished run needs no clause: it is `recent` only if its instant is in the
 * window, which is what the window's own `from` decides (see
 * `eventPublic.model.calendar`, which backdates its default `from` so that
 * "recent" has somewhere to live).
 */
const LIVE_STATUSES = ['starting', 'running', 'paused', 'ending']

const listInWindow = async ({
  from,
  to,
  status = null,
  scope = null,
  seriesId = null,
  limit = 500,
  publicOnly = false,
} = {}) => {
  const where = [
    `((r.scheduled_for >= ? AND r.scheduled_for < ?)
       OR (r.scheduled_for < ? AND r.status IN (${LIVE_STATUSES.map(() => '?').join(',')})))`,
  ]
  const args = [from, to, to, ...LIVE_STATUSES]
  if (status) {
    where.push('r.status = ?')
    args.push(status)
  }
  // The public calendar's two exclusions, in SQL rather than in the model that
  // maps the rows. A rehearsal "is excluded from the public calendar and from
  // participation history" by §D's own column comment, and an unlisted
  // definition is one an operator chose not to announce. Both belong in the
  // query because a filter applied after the read is a filter somebody can
  // forget in the next caller.
  if (publicOnly) {
    where.push('r.rehearsal = 0', 'd.listed = 1', "d.state <> 'archived'")
  }
  if (scope !== null && scope !== undefined) {
    where.push('r.scope = ?')
    args.push(scope)
  }
  if (seriesId) {
    where.push('d.series_id = ?')
    args.push(seriesId)
  }
  const n = Math.min(Math.max(Number(limit) || 500, 1), 1000)
  const rows = await query(
    `SELECT r.*, d.title AS definition_title, d.slug AS definition_slug,
            d.series_id AS series_id, se.name AS series_name, se.slug AS series_slug,
            v.version AS version_number,
            (SELECT COUNT(*) FROM event_run_steps s
              WHERE s.run_id = r.id AND s.status = 'running' AND s.claim_expires_at IS NULL) AS waiting_steps
       FROM event_runs r
       JOIN event_definitions d ON d.id = r.definition_id
       JOIN event_versions    v ON v.id = r.version_id
       LEFT JOIN event_series se ON se.id = d.series_id
      WHERE ${where.join(' AND ')}
      ORDER BY r.scheduled_for, r.id
      LIMIT ${n}`,
    args,
  )
  return rows.map(hydrate)
}

/**
 * Point every not-yet-started occurrence of a definition at a new version.
 *
 * Publishing calls this, and the guard is the whole statement: `status =
 * 'scheduled'` and `started_at IS NULL`. A run that has begun keeps the version
 * it pinned, for ever, because that pin is what makes it explicable afterwards
 * -- and a run that has NOT begun has nothing to explain yet.
 *
 * **Why re-pinning is the right answer and doing nothing is not** (org lead,
 * 2026-09-02): occurrences are materialised a fortnight ahead, so on the day an
 * editor fixes a typo there are already fourteen days of rows carrying the old
 * spec. Left alone, the fix reaches none of them, and the operator's only
 * recourse -- cancelling each one -- is worse: a cancelled row still holds its
 * slot in `uq_evrun_occurrence`, so the occurrence does not come back on the new
 * version, it disappears.
 *
 * Answers how many were moved, so publish can say so rather than leaving it to
 * be noticed.
 */
const repinScheduled = async (definitionId, versionId) => {
  const result = await query(
    `UPDATE event_runs
        SET version_id = ?
      WHERE definition_id = ?
        AND status = 'scheduled'
        AND started_at IS NULL
        AND version_id <> ?`,
    [versionId, definitionId, versionId],
  )
  return Number(result?.affectedRows || 0)
}

/** The scheduled, not-yet-started occurrences a re-pin would move. */
const listScheduledFor = async (definitionId) =>
  (
    await query(
      `SELECT id, version_id, scheduled_for FROM event_runs
        WHERE definition_id = ? AND status = 'scheduled' AND started_at IS NULL
        ORDER BY scheduled_for`,
      [definitionId],
    )
  ).map(hydrate)

/** The occurrence the unique key names, whether or not this call created it. */
const findOccurrence = async (definitionId, scope, scheduledFor) => {
  const [row] = await query(
    `${SELECT_LIST} WHERE r.definition_id = ? AND r.scope = ? AND r.scheduled_for = ?`,
    [definitionId, scope || '', scheduledFor],
  )
  return hydrate(row)
}

/** Is anything of this definition not yet terminal? The archive pre-check. */
const countActiveForDefinition = async (definitionId) => {
  const [row] = await query(
    `SELECT COUNT(*) AS n FROM event_runs
      WHERE definition_id = ?
        AND status IN ('scheduled','starting','running','paused','ending')`,
    [definitionId],
  )
  return Number(row?.n || 0)
}

// ── Phase 2: the claim, the transitions and the reclaim ────────────────────
//
// Everything below is the runner's, and none of it existed in Phase 1 for a
// stated reason: a half-written claim is worse than no claim, because it reads
// as protection. It is written here now, in full.
//
// **The division of labour with the unique index has not changed.** The index one
// section up is what makes "one run per occurrence per scope" TRUE; the CAS below
// decides only WHO advances an occurrence that already exists. Neither substitutes
// for the other, and this deployment being single-instance (§N4) changes the test
// rather than the design — the same two protections are what keep a tick that
// overran into the next one from advancing a run twice.

/**
 * Runs the runner should look at this tick: due, and not yet terminal.
 *
 * It selects rather than claims — `claimStart` and `claimTick` below are one row
 * at a time — so two sweepers see the same candidates and then disagree,
 * harmlessly, about which of them owns each. `idx_evrun_due (status,
 * scheduled_for)` is this query.
 *
 * `paused` is absent from the status list on purpose. A paused run is waiting on
 * a human and must not be advanced by a tick; the only thing that moves it is
 * Phase 3's resume control.
 */
const findDue = async (now, limit = 50) => {
  const n = Math.min(Math.max(Number(limit) || 50, 1), 500)
  return (
    await query(
      `SELECT * FROM event_runs
        WHERE status IN ('scheduled','starting','running','ending')
          AND scheduled_for <= ?
        ORDER BY scheduled_for, id
        LIMIT ${n}`,
      [now],
    )
  ).map(hydrate)
}

/**
 * Take ownership of a run that has not started: the CAS `scheduled -> starting`.
 *
 * Verbatim the outbox claim the org lead settled over `SELECT ... FOR UPDATE
 * SKIP LOCKED` — the instance the server reports `affectedRows = 1` to owns the
 * row, every other sweeper gets 0 and moves on. No transaction to hold open and
 * no MariaDB version floor.
 */
async function claimStart(id, owner, leaseUntil) {
  const result = await query(
    `UPDATE event_runs
        SET status = 'starting', claimed_by = ?, claim_expires_at = ?,
            started_at = COALESCE(started_at, NOW())
      WHERE id = ? AND status = 'scheduled'`,
    [owner, leaseUntil, id],
  )
  return Number(result?.affectedRows || 0) === 1
}

/**
 * Take a lease on a run already in flight, so one tick works on it at a time.
 *
 * Unlike `claimStart` this does not change `status` — the run is already
 * `starting`, `running` or `ending`, and what is being claimed is the right to
 * advance it.
 *
 * **A live lease is not re-enterable, not even by the process that took it**, and
 * that is the whole point rather than an oversight. `setInterval` fires the next
 * tick whether or not the last one has returned, so an owner-matches escape
 * clause here would let one process advance one run twice at once — which is
 * precisely the overrun the plan says the CAS is meant to protect against. A run
 * this process still holds is a run this process is still working on; the tick
 * skips it, and `releaseClaim` below is what ends that in the ordinary case.
 */
async function claimTick(id, owner, leaseUntil, now) {
  const result = await query(
    `UPDATE event_runs
        SET claimed_by = ?, claim_expires_at = ?
      WHERE id = ?
        AND status IN ('starting','running','ending')
        AND (claim_expires_at IS NULL OR claim_expires_at < ?)`,
    [owner, leaseUntil, id, now],
  )
  return Number(result?.affectedRows || 0) === 1
}

/**
 * Give a still-in-flight run back, so the next tick can pick it up at once.
 *
 * A run left parked on a GM cue, or waiting out a `core.wait`, is not finished
 * and must not carry a lease: without this the run would be unadvanceable until
 * the lease expired, which would turn every wait into `max(wait, leaseMs)`.
 * Scoped to `claimed_by = ?` so a process can only release its own claim.
 */
async function releaseClaim(id, owner) {
  const result = await query(
    'UPDATE event_runs SET claimed_by = NULL, claim_expires_at = NULL WHERE id = ? AND claimed_by = ?',
    [id, owner],
  )
  return Number(result?.affectedRows || 0) === 1
}

/**
 * A guarded status transition: `from -> to`, and only from `from`.
 *
 * Every move the runner makes goes through here rather than through a bare
 * UPDATE, so "did this transition actually happen" is answerable at each call
 * site. A `false` is not an error — it is another worker, or this run having been
 * cancelled from the admin surface between the read and the write, which is a
 * race Phase 3's live controls make ordinary.
 */
async function transition(id, from, to, { phase, error, clearClaim = false } = {}) {
  const sets = ['status = ?']
  const args = [to]
  if (phase !== undefined) {
    sets.push('current_phase = ?')
    args.push(phase)
  }
  if (error !== undefined) {
    sets.push('last_error = ?')
    args.push(error === null ? null : String(error).slice(0, 500))
  }
  if (TERMINAL.includes(to)) sets.push('ended_at = COALESCE(ended_at, NOW())')
  if (clearClaim || TERMINAL.includes(to)) sets.push('claimed_by = NULL', 'claim_expires_at = NULL')

  const froms = Array.isArray(from) ? from : [from]
  const result = await query(
    `UPDATE event_runs SET ${sets.join(', ')}
      WHERE id = ? AND status IN (${froms.map(() => '?').join(',')})`,
    [...args, id, ...froms],
  )
  return Number(result?.affectedRows || 0) === 1
}

/**
 * Just this run's status, for a caller that must not act on a stale read.
 *
 * The runner drains a bounded batch of steps from one run inside a single tick,
 * and Phase 3 put a pause and a cancel button in a human's hand — so between two
 * steps of that batch the run may have stopped. A loop that only re-checked at
 * the top of the tick would answer a pause by dispatching another two dozen
 * steps, which is not a pause. One column, by primary key.
 */
const statusOf = async (id) => {
  const [row] = await query('SELECT status FROM event_runs WHERE id = ?', [id])
  return row?.status || null
}

/**
 * Set health without touching status (§E).
 *
 * The two columns are separate because a run can be genuinely running and
 * degraded at once — announcements landing, world writes parked — and one column
 * cannot say both. Guarded on the current value so a tick that re-observes the
 * same degradation does not restamp `updated_at`.
 */
async function setHealth(id, health) {
  // **Escalation only, and this is the guard rather than a convention.** Health
  // has always been a high-water mark here — `degraded` is never cleared,
  // because a run whose announcement landed on the second attempt DID have
  // trouble and that stays true — and Phase 5 gave the column a second writer
  // for `stalled`. Without a rank, a step that retried after a stall would
  // quietly demote `stalled` to `degraded` and a run that waited ninety minutes
  // on a boss that never came would end its life claiming it merely wobbled.
  const rank = HEALTH_RANK[health]
  if (!rank) return false
  const result = await query(
    `UPDATE event_runs SET health = ?
      WHERE id = ? AND FIELD(health, ${HEALTH_SQL_ORDER}) < ?`,
    [health, id, rank],
  )
  return Number(result?.affectedRows || 0) === 1
}

/**
 * Set `cleanup_status`, optionally guarded on where it is now (Phase 8).
 *
 * Four values and three writers, which is why the guard is a parameter rather
 * than baked in. The ledger stamps `pending` the first time a run records
 * anything, and it must do so only over `not_required` — a run already marked
 * `complete` must not be walked back to `pending` by a late resource, and a
 * `incomplete` one must not be silently tidied. The cleanup sweep sets `complete`
 * or `incomplete` from what it found, unguarded, because the sweep IS the
 * authority on that. A human's cleanup route re-opens `pending` deliberately, and
 * says so in the log with the actor.
 *
 * **`pending` on a run that is still running is not a bug and reads correctly**:
 * there is something to clean up and it has not happened yet. The alternative -
 * a fifth value meaning "there will be something later" - is a state nothing
 * would ever branch on.
 */
async function setCleanupStatus(id, to, from = null) {
  const guard = from === null ? '' : ` AND cleanup_status IN (${from.map(() => '?').join(',')})`
  const result = await query(
    `UPDATE event_runs SET cleanup_status = ? WHERE id = ?${guard}`,
    from === null ? [to, id] : [to, id, ...from],
  )
  return Number(result?.affectedRows || 0) === 1
}

/**
 * Stamp this run's results table as published (EVENTS.md §J, Phase 10).
 *
 * **Unguarded, and re-stampable.** `core.results.publish` is an ordinary step
 * that an author may place more than once — before an announcement and again
 * after a late correction — and each publication is a real one whose moment is
 * worth recording. Guarding it on `IS NULL` would make the second silently do
 * nothing while the ranking beside it did move, which is the worst of both.
 */
async function markResultsPublished(id, at = new Date()) {
  const result = await query('UPDATE event_runs SET results_published_at = ? WHERE id = ?', [at, id])
  return Number(result?.affectedRows || 0) === 1
}

/**
 * Runs whose start instant passed more than their own grace window ago (§E, §L).
 *
 * The window is per definition, so the comparison is against `grace_seconds` on
 * the joined row rather than against a constant here: an event whose announcement
 * gives a fifteen-minute window and one that must start on the second are the
 * same query with different data.
 *
 * Only `scheduled` runs qualify. A run that reached `starting` has begun, and
 * "began and then stalled" is a different fact from "never began" — conflating
 * them would let `missed` describe a run that had already announced itself.
 */
const findMissed = async (now, limit = 100) => {
  const n = Math.min(Math.max(Number(limit) || 100, 1), 500)
  return (
    await query(
      `SELECT r.* FROM event_runs r
         JOIN event_definitions d ON d.id = r.definition_id
        WHERE r.status = 'scheduled'
          AND r.scheduled_for + INTERVAL d.grace_seconds SECOND < ?
        ORDER BY r.scheduled_for
        LIMIT ${n}`,
      [now],
    )
  ).map(hydrate)
}

/**
 * Is another run holding this concurrency key?
 *
 * The org lead's answer for a held key (2026-09-02) is to leave the run
 * `scheduled` and let the grace window decide, so this is a READ rather than a
 * claim: the caller holds off, logs which run holds the key, and tries again next
 * tick. `idx_evrun_concurrency (concurrency_key, status)` is this query, and a
 * NULL key is skipped by that index — which is right, because a definition with
 * no key never contends.
 */
const concurrencyHolder = async (key, exceptRunId) => {
  if (!key) return null
  const [row] = await query(
    `SELECT id, status, definition_id FROM event_runs
      WHERE concurrency_key = ?
        AND id <> ?
        AND status IN ('starting','running','paused','ending')
      ORDER BY id LIMIT 1`,
    [key, exceptRunId],
  )
  return row || null
}

/**
 * Recover runs whose claim outlived the process that took it.
 *
 * **It does not change status and it touches no counter.** All it releases is the
 * lease; the run stays exactly where it was and the next tick picks it up through
 * `findDue`. This is Engagement Phase 14's lesson applied one table over: a sweep
 * that returned a stale row to its start state made the attempt ceiling
 * unreachable, so the row cycled forever, never terminal, and therefore never
 * eligible for any retention sweep.
 */
const reclaimStale = async (now) => {
  const result = await query(
    `UPDATE event_runs SET claimed_by = NULL, claim_expires_at = NULL
      WHERE status IN ('starting','running','ending')
        AND claim_expires_at IS NOT NULL
        AND claim_expires_at < ?`,
    [now],
  )
  return Number(result?.affectedRows || 0)
}

/**
 * One definition's public occurrences, newest first (Phase 14a).
 *
 * Rehearsals are excluded here rather than by the caller, for `listInWindow`'s
 * reason. The definition's own `listed`/`state` are NOT re-checked: the only
 * caller has already resolved the definition through `getPublicBySlug`, and a
 * second copy of that rule is a second thing to keep in step with the first.
 *
 * `scheduled` runs come back too — an upcoming occurrence is exactly what a
 * visitor came to the page for — and the caller splits past from future on the
 * instant rather than on the status, because a `missed` run is in the past
 * whatever its status says.
 */
const listPublicForDefinition = async (definitionId, limit = 50) => {
  const n = Math.min(Math.max(Number(limit) || 50, 1), 200)
  const rows = await query(
    `SELECT r.*, v.version AS version_number
       FROM event_runs r
       JOIN event_versions v ON v.id = r.version_id
      WHERE r.definition_id = ? AND r.rehearsal = 0
      ORDER BY r.scheduled_for DESC, r.id DESC
      LIMIT ${n}`,
    [definitionId],
  )
  return rows.map(hydrate)
}

/** Terminal runs that ended before `before` — what the log retention sweep walks. */
const terminalBefore = async (before, limit = 500) => {
  const n = Math.min(Math.max(Number(limit) || 500, 1), 5000)
  return (
    await query(
      `SELECT id FROM event_runs
        WHERE status IN (${TERMINAL.map(() => '?').join(',')})
          AND COALESCE(ended_at, updated_at) < ?
        ORDER BY id LIMIT ${n}`,
      [...TERMINAL, before],
    )
  ).map((r) => Number(r.id))
}

module.exports = {
  list,
  getById,
  materialise,
  listInWindow,
  listPublicForDefinition,
  repinScheduled,
  listScheduledFor,
  findOccurrence,
  countActiveForDefinition,
  findDue,
  findMissed,
  claimStart,
  claimTick,
  releaseClaim,
  statusOf,
  transition,
  setHealth,
  setCleanupStatus,
  markResultsPublished,
  concurrencyHolder,
  reclaimStale,
  terminalBefore,
  TERMINAL,
}