feat(events): schedule, recurrence and the calendar (Phase 4)

The four closed recurrence shapes computed in the definition's own IANA zone,
a fourteen-day materialisation horizon with projections beyond it, series as a
managed thing, and the admin calendar that replaces the plugin this feature
exists to replace. An event now happens on its own.

No schema change: Phase 1 built every column this needed.

- events/recurrence.js is the ONE place an occurrence is computed, so the
  runner's expansion and the calendar's forecast cannot disagree. No date
  library added — Node ships the tzdata one would vendor, behind Intl.
- The runner's materialise leg is now two halves: expand, then sweep. The
  window starts at `now - grace`, so an occurrence nobody could have seen is
  never invented retroactively; the horizon is what makes the missed sweep
  mean anything for a recurrence.
- Publishing is the schedule switch and archiving turns it off, and publishing
  re-pins every occurrence that has not started.
- A projection is never drawn over an instant a run occupies, so a cancelled
  occurrence does not reappear as a forecast.

54 new tests, incl. the DST fixture set the plan asked for and three new
statements proved against a real MariaDB. Suite 1768/1711/56 skipped/1 fail
(pre-existing CRLF). Walked end to end on the local review stack.

Docs: RunicGateway/docs#PENDING

Co-Authored-By: Claude <noreply@anthropic.com>

server/test/eventRunnerSql.test.js

@@ -38,6 +38,22 @@
 //     unsettled seq instead, which is a different step whenever a phase carried
 //     on past an `on_failure: skip` failure.
 //
+// **Phase 4 added two reads**, and a read earns a place here when a stub cannot
+// tell it is wrong:
+//
+//   * **`findSchedulable`** - the query the runner runs on EVERY tick to decide
+//     what has a recurrence to expand. It joins a definition to its published
+//     version and left-joins the series, and every stub of it in
+//     `eventSchedule.test.js` is a hand-written object rather than that join. A
+//     syntax error or a wrong join direction here is a runner that materialises
+//     nothing, silently, for ever.
+//   * **`repinScheduled`** - the UPDATE publish runs over already-materialised
+//     occurrences. Its guard is the whole statement, and the two rows it must
+//     NOT touch are a run that has started and a run that is already terminal.
+//   * **`listInWindow`** - the calendar's real half, with a correlated subquery
+//     for `waiting_steps` and a LEFT JOIN that must not drop a definition with no
+//     series.
+//
 // Plus the two unique indexes that are load-bearing rather than tidy:
 // `uq_evrun_occurrence` (which, not the claim, is what stops two runs of one
 // occurrence existing) and `uq_evstep_slot` (which is what makes re-materialising
@@ -61,10 +77,29 @@ const mariadb = require('mariadb')
 // Trimmed to the columns these statements read or write. The ENUMs are verbatim,
 // because "is `missed` a legal value" is one of the things being proved.
 const SCHEMA = `
+CREATE TABLE event_series (
+  id       INT AUTO_INCREMENT PRIMARY KEY,
+  name     VARCHAR(160) NOT NULL,
+  slug     VARCHAR(160) NOT NULL,
+  ordering INT NOT NULL DEFAULT 0
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
 CREATE TABLE event_definitions (
   id            INT AUTO_INCREMENT PRIMARY KEY,
+  title         VARCHAR(200) NOT NULL DEFAULT 'x',
+  slug          VARCHAR(200) NOT NULL DEFAULT 'x',
+  state         ENUM('draft','ready','archived') NOT NULL DEFAULT 'draft',
+  current_version_id INT NULL,
+  series_id     INT NULL,
+  concurrency_key VARCHAR(190) NULL,
+  timezone      VARCHAR(64) NOT NULL DEFAULT 'UTC',
   grace_seconds INT NOT NULL DEFAULT 900
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+CREATE TABLE event_versions (
+  id            INT AUTO_INCREMENT PRIMARY KEY,
+  definition_id INT NOT NULL,
+  version       INT NOT NULL DEFAULT 1,
+  spec          JSON NOT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
 CREATE TABLE event_runs (
   id            BIGINT AUTO_INCREMENT PRIMARY KEY,
   definition_id INT      NOT NULL,
@@ -207,6 +242,41 @@ SELECT r.id FROM event_runs r
  WHERE r.status = 'scheduled'
    AND r.scheduled_for + INTERVAL d.grace_seconds SECOND < ?`
 
+// Verbatim `eventDefinitions.db#findSchedulable`.
+const FIND_SCHEDULABLE = `
+SELECT d.id, d.title, d.slug, d.timezone, d.grace_seconds, d.concurrency_key,
+       d.current_version_id, d.series_id, v.spec AS version_spec,
+       s.name AS series_name, s.slug AS series_slug
+  FROM event_definitions d
+  JOIN event_versions v ON v.id = d.current_version_id
+  LEFT JOIN event_series s ON s.id = d.series_id
+ WHERE d.state = 'ready'
+ ORDER BY d.id`
+
+// Verbatim `eventRuns.db#listInWindow`, with no optional filter applied.
+const LIST_IN_WINDOW = `
+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 r.scheduled_for >= ? AND r.scheduled_for < ?
+ ORDER BY r.scheduled_for, r.id
+ LIMIT 500`
+
+// Verbatim `eventRuns.db#repinScheduled`.
+const REPIN_SCHEDULED = `
+UPDATE event_runs
+   SET version_id = ?
+ WHERE definition_id = ?
+   AND status = 'scheduled'
+   AND started_at IS NULL
+   AND version_id <> ?`
+
 const DB = `rg_events_test_${process.pid}`
 let pool = null
 let available = false
@@ -274,6 +344,8 @@ beforeEach(async () => {
   await pool.query('DELETE FROM event_run_steps')
   await pool.query('DELETE FROM event_runs')
   await pool.query('DELETE FROM event_definitions')
+  await pool.query('DELETE FROM event_versions')
+  await pool.query('DELETE FROM event_series')
 })
 
 async function seedRun(over = {}) {
@@ -659,3 +731,163 @@ test('a guarded transition refuses a run that was cancelled underneath it', asyn
   assert.equal(rows(await pool.query(TRANSITION, ['running', 'two', runId, 'running'])), 0)
   assert.equal((await runById(runId)).status, 'cancelled')
 })
+
+
+// ── Phase 4: the two reads ────────────────────────────────────────
+
+/** A definition with a published version, and optionally a series. */
+async function seedDefinition({ state = 'ready', spec = { schedule: { kind: 'manual' } }, series = null } = {}) {
+  let seriesId = null
+  if (series) {
+    const s = await pool.query('INSERT INTO event_series (name, slug) VALUES (?, ?)', [series, series])
+    seriesId = s.insertId
+  }
+  const d = await pool.query(
+    'INSERT INTO event_definitions (state, series_id, timezone) VALUES (?, ?, ?)',
+    [state, seriesId, 'Europe/Berlin'],
+  )
+  const v = await pool.query(
+    'INSERT INTO event_versions (definition_id, version, spec) VALUES (?, 1, ?)',
+    [d.insertId, JSON.stringify(spec)],
+  )
+  await pool.query('UPDATE event_definitions SET current_version_id = ? WHERE id = ?', [
+    v.insertId,
+    d.insertId,
+  ])
+  return { definitionId: d.insertId, versionId: v.insertId, seriesId }
+}
+
+test('findSchedulable returns ready definitions with their PUBLISHED spec, series or not', async (t) => {
+  if (needDb(t)) return
+  const withSeries = await seedDefinition({
+    spec: { schedule: { kind: 'weekly', days: ['friday'], time: '20:00' } },
+    series: 'royal-spy',
+  })
+  const withoutSeries = await seedDefinition({ spec: { schedule: { kind: 'manual' } } })
+
+  const found = await pool.query(FIND_SCHEDULABLE)
+  const ids = found.map((r) => r.id).sort((a, b) => a - b)
+  assert.deepEqual(ids, [withSeries.definitionId, withoutSeries.definitionId].sort((a, b) => a - b))
+
+  // The LEFT JOIN must not drop the definition that belongs to no arc — an
+  // inner join here would make every event outside a series unschedulable, and
+  // most events are outside one.
+  const plain = found.find((r) => r.id === withoutSeries.definitionId)
+  assert.equal(plain.series_name, null)
+
+  const arced = found.find((r) => r.id === withSeries.definitionId)
+  assert.equal(arced.series_name, 'royal-spy')
+  assert.equal(arced.timezone, 'Europe/Berlin')
+
+  // The spec really came back, and really came back parseable.
+  const spec = typeof arced.version_spec === 'string' ? JSON.parse(arced.version_spec) : arced.version_spec
+  assert.equal(spec.schedule.kind, 'weekly')
+})
+
+test('findSchedulable skips a draft, an archived one, and one with no published version', async (t) => {
+  if (needDb(t)) return
+  await seedDefinition({ state: 'draft' })
+  await seedDefinition({ state: 'archived' })
+  // `ready` with a dangling version pointer: the JOIN is what must drop it, and
+  // a definition whose version row went missing must not become a runner crash.
+  const orphan = await seedDefinition({ state: 'ready' })
+  await pool.query('DELETE FROM event_versions WHERE id = ?', [orphan.versionId])
+
+  assert.equal((await pool.query(FIND_SCHEDULABLE)).length, 0)
+})
+
+test('listInWindow is half-open on the window, and counts only PARKED steps as waiting', async (t) => {
+  if (needDb(t)) return
+  const def = await seedDefinition()
+  const at = async (when) => {
+    const r = await pool.query(
+      'INSERT INTO event_runs (definition_id, version_id, scope, scheduled_for) VALUES (?, ?, ?, ?)',
+      [def.definitionId, def.versionId, '', when],
+    )
+    return r.insertId
+  }
+  const before = await at(new Date('2026-09-01T00:00:00Z'))
+  const onFrom = await at(new Date('2026-09-02T00:00:00Z'))
+  const inside = await at(new Date('2026-09-05T00:00:00Z'))
+  const onTo = await at(new Date('2026-09-09T00:00:00Z'))
+
+  // `>= from AND < to` — the instant ON the upper bound belongs to the NEXT
+  // window. A closed interval would draw the last day of one month and the first
+  // of the next as the same occurrence twice.
+  const found = await pool.query(LIST_IN_WINDOW, [
+    new Date('2026-09-02T00:00:00Z'),
+    new Date('2026-09-09T00:00:00Z'),
+  ])
+  assert.deepEqual(found.map((r) => r.id), [onFrom, inside])
+  assert.ok(!found.some((r) => r.id === before || r.id === onTo))
+
+  // A parked step is `running` with a NULL lease; a leased one is the runner
+  // mid-dispatch and is not waiting on anybody.
+  await seedStep(inside, { status: 'running', claimExpiresAt: null, key: 'a'.repeat(40) })
+  await seedStep(inside, {
+    seq: 1,
+    status: 'running',
+    claimedBy: 'host',
+    claimExpiresAt: later(60_000),
+    key: 'b'.repeat(40),
+  })
+  const again = await pool.query(LIST_IN_WINDOW, [
+    new Date('2026-09-02T00:00:00Z'),
+    new Date('2026-09-09T00:00:00Z'),
+  ])
+  assert.equal(Number(again.find((r) => r.id === inside).waiting_steps), 1)
+})
+
+
+test('repinScheduled moves the occurrences that have not begun, and only those', async (t) => {
+  if (needDb(t)) return
+  // The case: an editor fixes a typo on a weekly event on Wednesday. Two Fridays
+  // are already materialised on v3, last Friday's run is finished, and one is in
+  // flight right now.
+  const def = await pool.query('INSERT INTO event_definitions (grace_seconds) VALUES (900)')
+  const mk = async (status, versionId, startedAt, when) =>
+    (
+      await pool.query(
+        `INSERT INTO event_runs (definition_id, version_id, scope, status, scheduled_for, started_at)
+         VALUES (?, ?, ?, ?, ?, ?)`,
+        [def.insertId, versionId, `s${when}`, status, T0, startedAt],
+      )
+    ).insertId
+
+  const ahead1 = await mk('scheduled', 3, null, 1)
+  const ahead2 = await mk('scheduled', 3, null, 2)
+  const running = await mk('running', 3, T0, 3)
+  const done = await mk('completed', 3, T0, 4)
+  // Already on the new version: excluded by `version_id <> ?`, so a second
+  // publish of an unchanged definition is not a fleet of pointless writes.
+  const already = await mk('scheduled', 4, null, 5)
+
+  const moved = rows(await pool.query(REPIN_SCHEDULED, [4, def.insertId, 4]))
+  assert.equal(moved, 2)
+
+  const versionOf = async (id) =>
+    Number((await pool.query('SELECT version_id FROM event_runs WHERE id = ?', [id]))[0].version_id)
+  assert.equal(await versionOf(ahead1), 4)
+  assert.equal(await versionOf(ahead2), 4)
+  // A run that has begun keeps the version it pinned, for ever: that pin is what
+  // makes it explicable afterwards.
+  assert.equal(await versionOf(running), 3)
+  assert.equal(await versionOf(done), 3)
+  assert.equal(await versionOf(already), 4)
+})
+
+test('a scheduled run whose started_at is somehow set is left alone', async (t) => {
+  if (needDb(t)) return
+  // Belt and braces on the guard: `status = 'scheduled'` and `started_at IS NULL`
+  // are two conditions rather than one because a row that has both is the only
+  // row that is provably untouched.
+  const def = await pool.query('INSERT INTO event_definitions (grace_seconds) VALUES (900)')
+  const r = await pool.query(
+    `INSERT INTO event_runs (definition_id, version_id, scope, status, scheduled_for, started_at)
+     VALUES (?, 3, '', 'scheduled', ?, ?)`,
+    [def.insertId, T0, T0],
+  )
+  assert.equal(rows(await pool.query(REPIN_SCHEDULED, [4, def.insertId, 4])), 0)
+  const after = (await pool.query('SELECT version_id FROM event_runs WHERE id = ?', [r.insertId]))[0]
+  assert.equal(Number(after.version_id), 3)
+})