// ── The engagement engine (ENGAGEMENT.md Phase 4a) ─────────────────────────
//
// The phase's acceptance criteria, one test apiece:
//
//   • a trigger fired twice inside `cooldown_seconds` for the same
//     (rule, user, subject) sends once
//   • the same trigger for a DIFFERENT subject sends again — the multi-house
//     case §4.1 names, which is the one a per-user cooldown gets wrong
//   • a scheduled row is cancelled by a `cancel_on` trigger and never sends
//   • a restart mid-window still sends exactly once
//   • a duplicate `dedupe_key` is a successful no-op
//
// …plus the two properties that are security boundaries rather than behaviour:
// the G24 ceiling is re-checked at SEND time and not only at save, and a composed
// segment takes the NARROWEST ceiling in its tree.
//
// **The five tables are stubbed at the `.db` layer** and the engine's own logic
// runs for real against them, the shape `notificationChannelPrefs.test.js` uses.
// The one place that is not enough is the raw SQL whose correctness IS a server
// contract - the cooldown claim, the outbox compare-and-set, and the scoped
// dedupe key. Those run against a real MariaDB in `engagementEngineSql.test.js`,
// which skips when there is none, and the first time it ran it disproved the
// cooldown statement this file's stub had been agreeing with.
//
// Point the DB at a closed port before requiring anything: the registries reach
// utils/discordAnnounce, which builds the pool at require time.
process.env.DB_HOST = '127.0.0.1'
process.env.DB_PORT = '59999'

const { test, beforeEach, afterEach, after } = require('node:test')
const assert = require('node:assert/strict')

const registries = require('../src/modules/registries')
const channels = require('../src/engagement/channels')
const engine = require('../src/engagement/engine')
const conditions = require('../src/engagement/conditions')
const segments = require('../src/engagement/segments')
const worker = require('../src/utils/engagementWorker')
const rules = require('../src/model/engagement/engagementRules.model')
const rulesDb = require('../src/model/engagement/engagementRules.db')
const outboxDb = require('../src/model/engagement/engagementOutbox.db')
const cooldownsDb = require('../src/model/engagement/engagementCooldowns.db')
const sendsDb = require('../src/model/engagement/engagementSends.db')
const segmentsDb = require('../src/model/engagement/engagementSegments.db')
const recipients = require('../src/model/engagement/engagementRecipients.db')
const db = require('../src/utils/db')

after(() => db.close())

const T0 = new Date('2026-08-29T12:00:00Z')
const later = (ms) => new Date(T0.getTime() + ms)

// ── In-memory stand-ins for the five tables ────────────────────────────────

let store
const originals = {}

function snapshotOriginals() {
  for (const [name, mod] of [
    ['rulesDb', rulesDb], ['outboxDb', outboxDb], ['cooldownsDb', cooldownsDb],
    ['sendsDb', sendsDb], ['segmentsDb', segmentsDb], ['recipients', recipients],
  ]) {
    originals[name] = { mod, fns: { ...mod } }
  }
}
snapshotOriginals()

function restoreOriginals() {
  for (const { mod, fns } of Object.values(originals)) Object.assign(mod, fns)
}

function installStubs() {
  store = {
    rules: new Map(),
    segments: new Map(),
    cooldowns: new Map(),
    outbox: new Map(),
    sends: [],
    users: new Map(), // id -> { id, role, status }
    prefs: new Map(), // "<user> <id> <channel>" -> mode
    nextOutboxId: 1,
  }

  rulesDb.enabledForTrigger = async (triggerId) =>
    [...store.rules.values()].filter((r) => r.enabled && r.trigger_id === triggerId)
  rulesDb.enabledCancelledBy = async (triggerId) =>
    [...store.rules.values()].filter((r) => r.enabled && (r.cancel_on || []).includes(triggerId))
  rulesDb.getById = async (id) => store.rules.get(id) || null
  rulesDb.list = async () => [...store.rules.values()]
  rulesDb.countUsingSegment = async (segmentId) =>
    [...store.rules.values()].filter((r) => r.audience_segment_id === segmentId).length

  segmentsDb.getById = async (id) => store.segments.get(id) || null
  segmentsDb.list = async () => [...store.segments.values()]

  // The two statements' semantics, reproduced: a guarded UPDATE that matches
  // claims the fire; otherwise an INSERT IGNORE claims a first fire; otherwise
  // the pair is still cooling. `engagementEngineSql.test.js` is what proves the
  // SQL itself - a stub can only ever agree with whoever wrote it, and in this
  // case the first version of both was wrong together.
  cooldownsDb.claim = async (ruleId, userId, subjectKey, channel, cooldownSeconds, now) => {
    // `channel` is in the key, exactly as the PRIMARY KEY is: a rule naming two
    // channels must deliver on both, and a stub that dropped the channel would
    // agree with the engine bug Phase 11b's live walk found.
    const key = `${ruleId}|${userId}|${subjectKey}|${channel}`
    const row = store.cooldowns.get(key)
    if (!row) {
      store.cooldowns.set(key, { last_fired_at: now, fire_count: 1 })
      return true
    }
    if (row.last_fired_at.getTime() <= now.getTime() - cooldownSeconds * 1000) {
      row.fire_count += 1
      row.last_fired_at = now
      return true
    }
    return false
  }

  outboxDb.enqueue = async (row) => {
    if (row.dedupe_key) {
      const clash = [...store.outbox.values()].find(
        (r) =>
          r.dedupe_key === row.dedupe_key &&
          r.rule_id === row.rule_id &&
          r.user_id === row.user_id &&
          r.channel === row.channel,
      )
      if (clash) return null
    }
    const id = store.nextOutboxId++
    store.outbox.set(id, { id, status: 'scheduled', attempts: 0, subject_key: '', ...row })
    return id
  }
  // Copies, not the live objects: a SQL SELECT hands back a snapshot, and
  // `processRow` reads `row.attempts` as the value BEFORE its own claim
  // incremented it. Returning references here made the retry budget off by one
  // in the stub only, which is exactly the class of thing a stub must not invent.
  outboxDb.findDue = async (now, limit = 100) =>
    [...store.outbox.values()]
      .filter((r) => r.status === 'scheduled' && r.due_at <= now)
      .sort((a, b) => a.due_at - b.due_at || a.id - b.id)
      .slice(0, limit)
      .map((r) => ({ ...r }))
  outboxDb.claim = async (id) => {
    const row = store.outbox.get(id)
    if (!row || row.status !== 'scheduled') return false
    row.status = 'sending'
    row.attempts += 1
    return true
  }
  outboxDb.reschedule = async (id, dueAt, error) => {
    const row = store.outbox.get(id)
    if (row && row.status === 'sending') Object.assign(row, { status: 'scheduled', due_at: dueAt, last_error: error })
  }
  outboxDb.finish = async (id, status, error) => {
    const row = store.outbox.get(id)
    if (row) Object.assign(row, { status, last_error: error })
  }
  outboxDb.cancel = async (ruleId, subjectKey, userId = null) => {
    let n = 0
    for (const row of store.outbox.values()) {
      if (row.rule_id !== ruleId || row.subject_key !== subjectKey || row.status !== 'scheduled') continue
      if (userId !== null && userId !== undefined && row.user_id !== userId) continue
      row.status = 'cancelled'
      n += 1
    }
    return n
  }
  outboxDb.reclaimStale = async () => {}
  outboxDb.getById = async (id) => (store.outbox.has(id) ? { ...store.outbox.get(id) } : null)

  sendsDb.record = async (entry) => {
    store.sends.push({ id: store.sends.length + 1, created_at: T0, ...entry })
    return store.sends.length
  }
  sendsDb.countSentSince = async (ruleId, since) =>
    store.sends.filter((s) => s.rule_id === ruleId && s.status === 'sent' && s.created_at >= since).length

  const activeIds = () => [...store.users.values()].filter((u) => u.status === 'active').map((u) => u.id)
  recipients.active = async () => activeIds()
  recipients.staff = async (roles) =>
    [...store.users.values()].filter((u) => u.status === 'active' && roles.includes(u.role)).map((u) => u.id)
  recipients.subscribers = async (streamId, defaultOn = []) =>
    activeIds().filter((id) => {
      const rows = [...store.prefs.entries()].filter(([k]) => k.startsWith(`${id} ${streamId} `))
      if (rows.some(([, mode]) => mode !== 'off')) return true
      const named = new Set(rows.map(([k]) => k.split(' ')[2]))
      return defaultOn.some((c) => !named.has(c))
    })
  recipients.filterActive = async (ids) =>
    [...new Set(ids)].filter((id) => store.users.get(id)?.status === 'active')
  recipients.storedModes = async (userIds, streamId, channel) =>
    new Map(
      userIds
        .filter((id) => store.prefs.has(`${id} ${streamId} ${channel}`))
        .map((id) => [id, store.prefs.get(`${id} ${streamId} ${channel}`)]),
    )
}

// ── Fixtures ───────────────────────────────────────────────────────────────

const addUser = (id, over = {}) => store.users.set(id, { id, role: 'player', status: 'active', ...over })
const optIn = (userId, streamId, channel, mode = 'instant') =>
  store.prefs.set(`${userId} ${streamId} ${channel}`, mode)

let nextRuleId = 1
function addRule(over = {}) {
  const id = nextRuleId++
  const rule = {
    id,
    trigger_id: 'uo.house.idoc_warning',
    name: `rule ${id}`,
    enabled: true,
    audience: 'owner',
    audience_segment_id: null,
    max_sends_per_hour: 100,
    channels: ['email'],
    template_keys: {},
    conditions: null,
    cooldown_seconds: 0,
    delay_seconds: 0,
    cancel_on: [],
    ...over,
  }
  store.rules.set(id, rule)
  return rule
}

/** A validated event envelope, the shape `engagementEmit.emit` builds. */
const event = (over = {}) => ({
  triggerId: 'uo.house.idoc_warning',
  owner: 'uo',
  version: 1,
  subject: 'house-4001',
  ownerUserId: 10,
  dedupeKey: null,
  occurredAt: T0.toISOString(),
  data: { house: 'The Silver Anvil', decayStatus: 'IDOC' },
  ...over,
})

/** Register a batch, the way the loader's second pass commits one. */
function register(owner, fn) {
  const api = registries.stage(owner)
  fn(api)
  registries.apply(api.staged)
}

const IDOC_TRIGGER = {
  id: 'uo.house.idoc_warning',
  label: 'House approaching collapse',
  ceiling: 'owner',
  audience: 'owner',
  subjectKey: 'house',
  variables: [
    { name: 'house', type: 'string', required: true, example: 'The Silver Anvil' },
    { name: 'decayStatus', type: 'string', required: false, example: 'IDOC' },
  ],
}

function registerUoTrigger(over = {}) {
  register('uo', (api) => api.registerEventTriggers([{ ...IDOC_TRIGGER, ...over }]))
}

const outboxRows = (filter = () => true) => [...store.outbox.values()].filter(filter)
const scheduled = () => outboxRows((r) => r.status === 'scheduled')

// Core's channels register through `coreChannels`, the way app.js does.
// Requiring `channels` alone gets the empty map — that is the design, and the
// engine dropping every rule because no channel is registered is what a
// boot-order regression would look like.
function registerChannels() {
  channels._reset()
  delete require.cache[require.resolve('../src/engagement/coreChannels')]
  // eslint-disable-next-line global-require
  require('../src/engagement/coreChannels')
}

beforeEach(() => {
  registries._reset()
  registerChannels()
  installStubs()
  nextRuleId = 1
  registerUoTrigger()
  addUser(10)
  // Every channel is opt-IN (§7.1 Q1, and channels.js `defaultMode: 'off'`), so
  // a fixture that wants mail to happen has to say so. The test below that turns
  // this off again is the one asserting exactly that.
  optIn(10, 'uo.house.idoc_warning', 'email')
})

afterEach(() => {
  registries._reset()
  restoreOriginals()
})

// ── Acceptance: cooldowns ──────────────────────────────────────────────────

test('a trigger fired twice inside cooldown_seconds for the same (rule, user, subject) sends once', async () => {
  addRule({ cooldown_seconds: 3600 })

  await engine.dispatch(event(), T0)
  await engine.dispatch(event(), later(60_000))

  assert.equal(outboxRows().length, 1)
})

test('the same trigger for a DIFFERENT subject sends again — the multi-house case (§4.1)', async () => {
  // The rule §4.1 warns about is "one IDOC mail per player per day": a player
  // with four houses decaying should hear about all four, once each. Cooling on
  // (rule, user) alone silently drops three of them, and this is the test that
  // would fail if `subject_key` were ever dropped from the primary key.
  addRule({ cooldown_seconds: 86_400 })

  await engine.dispatch(event({ subject: 'house-4001' }), T0)
  await engine.dispatch(event({ subject: 'house-4002' }), later(1000))
  await engine.dispatch(event({ subject: 'house-4003' }), later(2000))
  // …and the first house again, still inside the day.
  await engine.dispatch(event({ subject: 'house-4001' }), later(3000))

  assert.deepEqual(outboxRows().map((r) => r.subject_key).sort(), ['house-4001', 'house-4002', 'house-4003'])
})

test('a cooldown that has expired lets the same subject through again', async () => {
  addRule({ cooldown_seconds: 60 })

  await engine.dispatch(event(), T0)
  await engine.dispatch(event(), later(61_000))

  assert.equal(outboxRows().length, 2)
})

test('a cooldown does not stop a rule delivering on its OTHER channels', async () => {
  // Phase 11b's live walk. The claim runs inside the per-channel loop, so a key
  // without the channel let the FIRST channel claim the cooldown and reported
  // every later one as cooled — and `inapp` is ranked ahead of `email` on
  // purpose, so a two-channel rule delivered the inbox item and silently never
  // the mail. Decision 8 requires both, and this is the assertion that says so.
  addUser(10)
  optIn(10, 'uo.house.idoc_warning', 'email')
  optIn(10, 'uo.house.idoc_warning', 'inapp')
  addRule({ cooldown_seconds: 86_400, channels: ['email', 'inapp'] })

  await engine.dispatch(event(), T0)
  assert.deepEqual(outboxRows().map((r) => r.channel).sort(), ['email', 'inapp'])

  // …and the cooldown still holds, on both channels, for a second event about
  // the same house inside the day. Per-delivery, not per-channel-forever.
  await engine.dispatch(event(), later(60_000))
  assert.equal(outboxRows().length, 2)
})

test('two rules on one trigger each get their own cooldown', async () => {
  addRule({ cooldown_seconds: 3600 })
  addRule({ cooldown_seconds: 3600 })

  await engine.dispatch(event(), T0)

  assert.equal(outboxRows().length, 2)
})

// ── Acceptance: dedupe ─────────────────────────────────────────────────────

test('a duplicate dedupe_key is a successful no-op, not a second row and not an error', async () => {
  addRule()

  const first = await engine.dispatch(event({ dedupeKey: 'idoc:4001:2026-08-29' }), T0)
  const replay = await engine.dispatch(event({ dedupeKey: 'idoc:4001:2026-08-29' }), later(1000))

  assert.equal(first.enqueued, 1)
  assert.equal(replay.enqueued, 0)
  assert.equal(replay.deduped, 1)
  assert.equal(outboxRows().length, 1)
})

test('one dedupe_key fans out to every recipient — the key is scoped, not global', async () => {
  // §4.2a's `UNIQUE (dedupe_key)` was a defect: a dedupe key names the EVENT, and
  // one event legitimately becomes one row per (rule, user, channel). A global
  // unique index would have let the FIRST recipient's row in and silently dropped
  // everyone else's, which is the opposite of what dedupe is for.
  addUser(11)
  addUser(12)
  for (const id of [10, 11, 12]) {
    optIn(id, 'uo.house.idoc_warning', 'email')
    optIn(id, 'uo.house.idoc_warning', 'inapp')
  }
  registries._reset()
  registerUoTrigger({ ceiling: 'subscribers', audience: 'subscribers' })
  addRule({ audience: 'subscribers', channels: ['email', 'inapp'] })

  const result = await engine.dispatch(event({ dedupeKey: 'idoc:4001' }), T0)

  // three users x two channels
  assert.equal(result.enqueued, 6)
  assert.equal(new Set(outboxRows().map((r) => r.dedupe_key)).size, 1)
})

// ── Acceptance: the grace window and cancellation ──────────────────────────

test('a scheduled row is cancelled by a cancel_on trigger and never sends', async () => {
  register('uo', (api) =>
    api.registerEventTriggers([
      { ...IDOC_TRIGGER, id: 'uo.house.repaired', label: 'House repaired', subjectKey: 'house' },
    ]),
  )
  const rule = addRule({ delay_seconds: 1800, cancel_on: ['uo.house.repaired'] })

  await engine.dispatch(event(), T0)
  assert.equal(scheduled().length, 1)

  const cancelled = await engine.dispatch(
    event({ triggerId: 'uo.house.repaired', subject: 'house-4001' }),
    later(60_000),
  )
  assert.equal(cancelled.cancelled, 1)

  // The window has passed; the worker finds nothing to do.
  await worker.tick(later(1_900_000))
  assert.equal(store.outbox.get(1).status, 'cancelled')
  assert.equal(store.sends.length, 0)
  assert.equal(rule.id, 1)
})

test('a resolving event with no owner cancels every recipient queued about that subject', async () => {
  register('uo', (api) =>
    api.registerEventTriggers([
      { ...IDOC_TRIGGER, id: 'uo.house.repaired', label: 'House repaired', ceiling: 'authenticated', audience: 'authenticated' },
    ]),
  )
  addUser(11)
  registries._reset()
  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
  register('uo', (api) =>
    api.registerEventTriggers([
      { ...IDOC_TRIGGER, id: 'uo.house.repaired', label: 'House repaired', ceiling: 'authenticated', audience: 'authenticated' },
    ]),
  )
  optIn(10, 'uo.house.idoc_warning', 'email')
  optIn(11, 'uo.house.idoc_warning', 'email')
  addRule({ audience: 'authenticated', delay_seconds: 600, cancel_on: ['uo.house.repaired'] })

  await engine.dispatch(event(), T0)
  assert.equal(scheduled().length, 2)

  await engine.dispatch(
    event({ triggerId: 'uo.house.repaired', subject: 'house-4001', ownerUserId: null }),
    later(1000),
  )
  assert.equal(scheduled().length, 0)
})

test('cancellation leaves an in-flight row alone', async () => {
  register('uo', (api) =>
    api.registerEventTriggers([{ ...IDOC_TRIGGER, id: 'uo.house.repaired', label: 'House repaired' }]),
  )
  addRule({ delay_seconds: 600, cancel_on: ['uo.house.repaired'] })
  await engine.dispatch(event(), T0)

  // A worker has claimed it: cancelling now would leave two writers on one row.
  await outboxDb.claim(1)
  const result = await engine.dispatch(event({ triggerId: 'uo.house.repaired' }), later(1000))

  assert.equal(result.cancelled, 0)
  assert.equal(store.outbox.get(1).status, 'sending')
})

// ── Acceptance: exactly once across a restart ──────────────────────────────

test('a restart mid-window still sends exactly once', async () => {
  addRule({ delay_seconds: 600 })
  await engine.dispatch(event(), T0)

  // "Restart" is the engine losing its process between enqueue and due_at. The
  // outbox is the durable half, so the only question is whether the sweep after
  // the restart double-delivers — and the CAS claim is what says it cannot.
  await worker.tick(later(500_000)) // not yet due
  assert.equal(store.sends.length, 0)

  await worker.tick(later(700_000))
  await worker.tick(later(700_001)) // a second instance, or the next tick
  assert.equal(store.sends.length, 1)
})

test('two sweepers racing one due row: exactly one claim wins', async () => {
  addRule()
  await engine.dispatch(event(), T0)

  // Both see the same candidate — findDue does not claim — and then disagree
  // harmlessly about which of them owns it. §7.1 Q2's answer, as a test.
  const [a] = await outboxDb.findDue(later(1000))
  const [b] = await outboxDb.findDue(later(1000))
  assert.equal(a.id, b.id)

  assert.equal(await outboxDb.claim(a.id), true)
  assert.equal(await outboxDb.claim(b.id), false)
})

// ── The send log ───────────────────────────────────────────────────────────

test('a row whose channel has no deliver() finishes failed, and the send log says why', async () => {
  // **A channel registered for this test, because as of Phase 7 all three of
  // core's deliver.** It used to name `inapp` (and `email` before that), which
  // meant the assertion moved every time a phase gave a channel behaviour. The
  // property under test was never about a particular channel: it is that the
  // worker does not record 'sent' for a sink it cannot reach, because that would
  // be a lie in the one table whose purpose is answering "did they get it".
  channels.registerDeliveryChannel({
    id: 'nosink',
    label: 'No sink',
    carriesContent: true,
    defaultMode: 'off',
    supportsDigest: false,
  })
  addRule({ channels: ['nosink'] })
  optIn(10, 'uo.house.idoc_warning', 'nosink')
  await engine.dispatch(event(), T0)
  await worker.tick(later(1000))

  assert.equal(store.outbox.get(1).status, 'failed')
  assert.equal(store.sends.length, 1)
  assert.equal(store.sends[0].status, 'failed')
  assert.match(store.sends[0].detail, /no delivery implementation/)
  assert.equal(store.sends[0].user_id, 10)
})

test('a transient failure is retried, and then given up on', async () => {
  addRule()
  await engine.dispatch(event(), T0)

  let attempts = 0
  const failing = async () => {
    attempts += 1
    return { outcome: 'retry', detail: 'smtp timeout' }
  }

  let at = later(1000)
  for (let i = 0; i < worker.MAX_ATTEMPTS + 2; i += 1) {
    // eslint-disable-next-line no-await-in-loop
    const [row] = await outboxDb.findDue(at)
    if (!row) break
    // eslint-disable-next-line no-await-in-loop
    await worker.processRow(row, at, failing)
    at = new Date(at.getTime() + worker.RETRY_MS + 1000)
  }

  // Tried MAX_ATTEMPTS times and then stopped, rather than retrying forever.
  assert.equal(attempts, worker.MAX_ATTEMPTS)
  assert.equal(store.outbox.get(1).status, 'failed')
  assert.equal(store.sends.length, 1)
  assert.match(store.sends[0].detail, /smtp timeout/)
})

// ── Preferences ────────────────────────────────────────────────────────────

test("a user whose mode is 'off' for the channel is not enqueued", async () => {
  registries._reset()
  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
  addUser(11)
  optIn(10, 'uo.house.idoc_warning', 'email', 'instant')
  optIn(11, 'uo.house.idoc_warning', 'email', 'off')
  addRule({ audience: 'authenticated' })

  await engine.dispatch(event(), T0)

  assert.deepEqual(outboxRows().map((r) => r.user_id), [10])
})

test('absence means the CHANNEL default, and all three of core default off', async () => {
  registries._reset()
  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
  store.prefs.clear()
  addRule({ audience: 'authenticated' })

  // Nobody has expressed anything, and email defaults 'off' (§3.1) — so an
  // `authenticated` rule reaches nobody until people opt in. That is opt-IN
  // working, not the engine failing.
  const result = await engine.dispatch(event(), T0)
  assert.equal(result.enqueued, 0)
  assert.equal(channels.defaultMode('email'), 'off')
})

// **This reverses what Phase 4a asserted here**, and the reversal is Phase 6's
// §4.2b decision rather than a change of mind about queues. A digest is
// re-derived from the source tables at send time — that is what makes a hidden
// post absent from it and a user who lost access unreachable by it — so an outbox
// row for a digest recipient would be a second copy of the content with none of
// those properties. Nothing drains it, so nothing writes it.
test("a 'digest' preference does NOT enqueue — the digest re-derives at send time", async () => {
  registries._reset()
  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
  optIn(10, 'uo.house.idoc_warning', 'email', 'digest')
  addRule({ audience: 'authenticated' })

  await engine.dispatch(event(), T0)
  assert.equal(outboxRows().length, 0)
})

// ── The hourly ceiling (§7.1 Q3) ───────────────────────────────────────────

test('a rule stops at its hourly send ceiling', async () => {
  registries._reset()
  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
  for (let i = 20; i < 30; i += 1) {
    addUser(i)
    optIn(i, 'uo.house.idoc_warning', 'email')
  }
  const rule = addRule({ audience: 'authenticated', max_sends_per_hour: 4 })

  const result = await engine.dispatch(event(), T0)

  // Eleven eligible recipients (the ten here plus the fixture's user 10), and a
  // ceiling of four: four rows, and the rest are counted and dropped rather than
  // queued for later - a rule at its ceiling is a rule an operator has to fix.
  assert.equal(result.enqueued, 4)
  assert.equal(result.enqueued + result.capped, 11)
  assert.equal(rule.max_sends_per_hour, 4)
})

test('the hourly ceiling counts sends, not attempts', async () => {
  // A broken transport must not silently consume a rule's whole budget and mute
  // it: only rows the log records as 'sent' count against the ceiling.
  const rule = addRule({ max_sends_per_hour: 2 })
  store.sends.push({ rule_id: rule.id, status: 'failed', created_at: T0 })
  store.sends.push({ rule_id: rule.id, status: 'suppressed', created_at: T0 })

  const result = await engine.dispatch(event(), later(1000))
  assert.equal(result.enqueued, 1)
})

// ── templateKeys: what a channel is, and what `digest` is ──────────────────

test('a rule may name a `digest` body, which is a template slot rather than a channel', async () => {
  // The defect Phase 13's acceptance walk found. `registries.js` `checkSeedRule`
  // permits `digest` in as many words — it is the body `teamDigestWorker`
  // renders for a rule whose email channel an individual set to digest mode, so
  // it never appears in `channels` and never could — and core's own Team and
  // news rules ship one, as do sixteen of module-uo's. This validator rejected
  // it, so every one of those rules answered an operator who opened it and
  // pressed Save with a 400 naming a key they had never typed, and the only way
  // to save was to delete the digest body.
  const checked = await rules.validate({
    triggerId: 'uo.house.idoc_warning',
    name: 'IDOC warning',
    channels: ['email'],
    audience: 'owner',
    templateKeys: { email: 'notify.event', digest: 'notify.digest' },
  })

  assert.equal(checked.ok, true, checked.errors && checked.errors.join(' '))
  assert.equal(checked.rule.template_keys.digest, 'notify.digest')
})

test('a templateKeys entry that is neither a channel nor `digest` is still refused', async () => {
  // The rule that was right all along, kept: `digest` is one named exception
  // with a renderer behind it, not a hole that admits any word.
  const checked = await rules.validate({
    triggerId: 'uo.house.idoc_warning',
    name: 'IDOC warning',
    channels: ['email'],
    audience: 'owner',
    templateKeys: { email: 'notify.event', carrierpigeon: 'notify.event' },
  })

  assert.equal(checked.ok, false)
  assert.match(checked.errors.join(' '), /carrierpigeon/)
})

// ── Ceilings: the security boundary, both halves ───────────────────────────

test('a rule may not be SAVED with an audience wider than its trigger permits', async () => {
  const checked = await rules.validate({
    triggerId: 'uo.house.idoc_warning', // ceiling: owner
    name: 'IDOC warning',
    channels: ['email'],
    audience: 'authenticated',
  })

  assert.equal(checked.ok, false)
  assert.match(checked.errors.join(' '), /wider than trigger/)
})

test('the ceiling is re-checked at SEND time, so a module narrowing its declaration stops a saved rule', async () => {
  // The only way this can fail is the case it exists for: the rule was saved
  // when the trigger permitted `authenticated`, and a module upgrade has since
  // narrowed the declaration to `owner`. A save-time check alone would keep
  // mailing the wider set forever.
  registries._reset()
  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
  addUser(11)
  optIn(10, 'uo.house.idoc_warning', 'email')
  optIn(11, 'uo.house.idoc_warning', 'email')
  addRule({ audience: 'authenticated' })

  const before = await engine.dispatch(event(), T0)
  assert.equal(before.enqueued, 2)

  registries._reset()
  registerUoTrigger({ ceiling: 'owner', audience: 'owner' }) // the upgrade

  const after2 = await engine.dispatch(event({ subject: 'house-9' }), later(1000))
  assert.equal(after2.enqueued, 0)
})

test('a FIRING may narrow the ceiling, and every rule wider than that is refused', async () => {
  // EVENTS.md §I, and the case that forced it: a rehearsal fires exactly the
  // same trigger as the real thing, so without a per-firing bound, rehearsing a
  // published event mails every subscriber it. The declaration is a property of
  // the KIND of event; this is a property of the occasion.
  registries._reset()
  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
  addUser(11)
  optIn(10, 'uo.house.idoc_warning', 'email')
  optIn(11, 'uo.house.idoc_warning', 'email')
  addRule({ audience: 'authenticated' })

  const wide = await engine.dispatch(event(), T0)
  assert.equal(wide.enqueued, 2)

  const narrowed = await engine.dispatch(event({ subject: 'house-9', ceiling: 'staff' }), later(1000))
  assert.equal(narrowed.enqueued, 0)
})

test('a firing may only ever NARROW — a wider ceiling than the declaration changes nothing', async () => {
  registries._reset()
  registerUoTrigger({ ceiling: 'owner', audience: 'owner' })
  addRule({ audience: 'owner' })

  // `everyone` is the widest value the lattice has. `meet(owner, everyone)` is
  // `owner`, so the declaration still governs and the rule still fires.
  const out = await engine.dispatch(event({ ceiling: 'everyone' }), T0)
  assert.equal(out.enqueued, 1)
})

test('two INCOMPARABLE ceilings meet to nothing and the gate refuses rather than guessing', async () => {
  // `owner` and `staff` have no common descendant — "fewer people" is not "less
  // exposure", which is the whole argument `modules/ceilings.js` is built on.
  // Picking one would be the guess §5.1a rule 3 exists to refuse.
  registries._reset()
  registerUoTrigger({ ceiling: 'owner', audience: 'owner' })
  addRule({ audience: 'owner' })

  const out = await engine.dispatch(event({ ceiling: 'staff' }), T0)
  assert.equal(out.enqueued, 0)
})

test('a rule for an unregistered trigger is dormant, not deleted and not an error', async () => {
  const rule = addRule({ trigger_id: 'uo.gone.away' })
  const listed = await rules.listAnnotated()
  const found = listed.find((r) => r.id === rule.id)

  assert.equal(found.dormant, true)
  assert.match(found.dormantReasons.join(' '), /not registered/)
})

// ── Segments (§5.1a) ───────────────────────────────────────────────────────

function registerAudiences() {
  register('uo', (api) =>
    api.registerAudiences([
      { id: 'uo.team.members', label: 'Team members', ceiling: 'members', params: [{ id: 'teamId', type: 'int', required: true }], resolve: async ({ teamId }) => (teamId === 1 ? [10, 11] : [12]) },
      { id: 'uo.governors', label: 'Governors', ceiling: 'members', resolve: async () => [11, 12] },
      { id: 'uo.watchers', label: 'Watchers', ceiling: 'authenticated', resolve: async () => [10, 13] },
      { id: 'uo.flagged', label: 'Flagged accounts', ceiling: 'staff', resolve: async () => [10] },
    ]),
  )
}

test('OR takes the TIGHTER ceiling — union-widens is the wrong implementation', async () => {
  registerAudiences()
  const checked = segments.validate({
    op: 'or',
    nodes: [{ audienceId: 'uo.governors' }, { audienceId: 'uo.watchers' }],
  })

  assert.equal(checked.ok, true)
  // members is below authenticated, so the meet is members — NOT authenticated,
  // which is what a "widest wins" reading would have given.
  assert.equal(checked.ceiling, 'members')
})

test('two incomparable ceilings are refused rather than resolved to a guess', async () => {
  registerAudiences()
  const checked = segments.validate({
    op: 'and',
    nodes: [{ audienceId: 'uo.governors' }, { audienceId: 'uo.flagged' }],
  })

  assert.equal(checked.ok, false)
  assert.match(checked.errors.join(' '), /no common ceiling/)
})

test('NOT does not constrain the ceiling — excluding people cannot widen', async () => {
  registerAudiences()
  // `members AND NOT staff` reaches strictly fewer people than `members`. If the
  // complement's ceiling were folded into the meet, meet('members','staff') is
  // null and this safe segment would be refused.
  const checked = segments.validate({
    op: 'and',
    nodes: [{ audienceId: 'uo.governors' }, { op: 'not', nodes: [{ audienceId: 'uo.flagged' }] }],
  })

  assert.equal(checked.ok, true)
  assert.equal(checked.ceiling, 'members')
})

test('NOT outside an AND is refused — a complement needs a set to take it from', async () => {
  registerAudiences()
  for (const expression of [
    { op: 'not', nodes: [{ audienceId: 'uo.governors' }] },
    { op: 'or', nodes: [{ audienceId: 'uo.governors' }, { op: 'not', nodes: [{ audienceId: 'uo.flagged' }] }] },
  ]) {
    const checked = segments.validate(expression)
    assert.equal(checked.ok, false)
    assert.match(checked.errors.join(' '), /only allowed inside an "and"/)
  }
})

test('a segment resolves through the module resolvers, and AND NOT subtracts', async () => {
  registerAudiences()
  const checked = segments.validate({
    op: 'and',
    nodes: [
      { audienceId: 'uo.team.members', params: { teamId: 1 } }, // [10, 11]
      { op: 'not', nodes: [{ audienceId: 'uo.flagged' }] }, // [10]
    ],
  })
  const resolved = await segments.resolve(checked.expression)

  assert.equal(resolved.dormant, false)
  assert.deepEqual(resolved.userIds, [11])
})

test('a segment whose module is uninstalled is DORMANT and sends to nobody', async () => {
  registerAudiences()
  const checked = segments.validate({ op: 'or', nodes: [{ audienceId: 'uo.governors' }, { audienceId: 'uo.watchers' }] })
  store.segments.set(1, { id: 1, name: 'staff-ish', expression: checked.expression, ceiling: 'members' })
  addUser(11)
  optIn(10, 'uo.house.idoc_warning', 'email')
  optIn(11, 'uo.house.idoc_warning', 'email')
  registries._reset()
  registerUoTrigger({ ceiling: 'members', audience: 'members' })
  addRule({ audience: 'members', audience_segment_id: 1 })

  // The module is gone: `resolveAudience` answers dormant + empty, and the rule
  // must NOT fall back to anything. Reaching a different population than the one
  // composed is the failure §5.1a rule 4 forbids.
  const result = await engine.dispatch(event(), T0)

  assert.equal(result.enqueued, 0)
  assert.equal(outboxRows().length, 0)
})

test('a rule pointing at a deleted segment is dormant, never a fallback to its plain audience', async () => {
  registries._reset()
  registerUoTrigger({ ceiling: 'authenticated', audience: 'authenticated' })
  optIn(10, 'uo.house.idoc_warning', 'email')
  addRule({ audience: 'authenticated', audience_segment_id: 99 }) // no such segment

  const result = await engine.dispatch(event(), T0)
  assert.equal(result.enqueued, 0)
})

test("a plain 'members' audience with no segment reaches nobody", async () => {
  registries._reset()
  registerUoTrigger({ ceiling: 'members', audience: 'members' })
  optIn(10, 'uo.house.idoc_warning', 'email')
  addRule({ audience: 'members' })

  const result = await engine.dispatch(event(), T0)
  assert.equal(result.enqueued, 0)
})

// ── Conditions ─────────────────────────────────────────────────────────────

test('a condition narrows which firings are interesting', async () => {
  addRule({
    conditions: { variable: 'decayStatus', cmp: 'in', value: ['Greatly damaged', 'IDOC'] },
  })

  await engine.dispatch(event({ data: { house: 'A', decayStatus: 'IDOC' } }), T0)
  await engine.dispatch(event({ subject: 'house-2', data: { house: 'B', decayStatus: 'LikeNew' } }), later(1000))

  assert.equal(outboxRows().length, 1)
})

test('a condition naming a variable the trigger does not declare is refused at save, with the name', async () => {
  const checked = await rules.validate({
    triggerId: 'uo.house.idoc_warning',
    name: 'typo',
    channels: ['email'],
    audience: 'owner',
    conditions: { variable: 'decaystatus', cmp: 'eq', value: 'IDOC' },
  })

  assert.equal(checked.ok, false)
  assert.match(checked.errors.join(' '), /"decaystatus" is not a variable/)
})

test('an absent variable makes every comparison false — including "is not"', async () => {
  // `ne` is the one that tempts otherwise: "not equal to IDOC" reads as satisfied
  // by nothing at all, and treating it that way would fire the rule on every
  // event that omits an optional variable.
  const c = { variable: 'decayStatus', cmp: 'ne', value: 'IDOC' }
  assert.equal(conditions.evaluate(c, { house: 'A' }), false)
  assert.equal(conditions.evaluate(c, { house: 'A', decayStatus: 'LikeNew' }), true)
  assert.equal(conditions.evaluate({ variable: 'decayStatus', cmp: 'absent' }, { house: 'A' }), true)
})

test('a condition tree that no longer parses fails CLOSED', async () => {
  // A stored condition that stops making sense must stop the mail, not decay
  // into "no conditions" and reach everyone the rule could ever reach.
  assert.equal(conditions.evaluate({ op: 'xor', nodes: [] }, {}), false)
  assert.equal(conditions.evaluate('nonsense', {}), false)
  assert.equal(conditions.evaluate(null, {}), true)
})

test('and / or / not compose', () => {
  const data = { house: 'The Silver Anvil', decayStatus: 'IDOC' }
  assert.equal(
    conditions.evaluate(
      { op: 'and', nodes: [{ variable: 'decayStatus', cmp: 'eq', value: 'IDOC' }, { variable: 'house', cmp: 'contains', value: 'Silver' }] },
      data,
    ),
    true,
  )
  assert.equal(
    conditions.evaluate({ op: 'not', nodes: [{ variable: 'decayStatus', cmp: 'eq', value: 'IDOC' }] }, data),
    false,
  )
  assert.equal(
    conditions.evaluate(
      { op: 'or', nodes: [{ variable: 'decayStatus', cmp: 'eq', value: 'LikeNew' }, { variable: 'house', cmp: 'startsWith', value: 'The' }] },
      data,
    ),
    true,
  )
})

test('an operator cannot be applied to a type it does not fit', () => {
  const declaration = registries.eventTrigger('uo.house.idoc_warning')
  const checked = conditions.validate(declaration, { variable: 'house', cmp: 'gt', value: 'x' })
  assert.equal(checked.ok, false)
  assert.match(checked.errors.join(' '), /cannot be applied to a string/)
})

// ── Rule validation, the rest ──────────────────────────────────────────────

test('a new rule is created disabled unless it says otherwise (§7.1 Q3)', async () => {
  const checked = await rules.validate({
    triggerId: 'uo.house.idoc_warning',
    name: 'IDOC warning',
    channels: ['email'],
    audience: 'owner',
  })
  assert.equal(checked.ok, true)
  assert.equal(checked.rule.enabled, false)
  assert.equal(checked.rule.max_sends_per_hour, 100)
})

test('a rule naming an unregistered channel is refused', async () => {
  const checked = await rules.validate({
    triggerId: 'uo.house.idoc_warning',
    name: 'IDOC warning',
    channels: ['carrier-pigeon'],
    audience: 'owner',
  })
  assert.equal(checked.ok, false)
  assert.match(checked.errors.join(' '), /no channel "carrier-pigeon"/)
})

test('the hourly ceiling has a hard upper bound an operator cannot type past', async () => {
  const checked = await rules.validate({
    triggerId: 'uo.house.idoc_warning',
    name: 'IDOC warning',
    channels: ['email'],
    audience: 'owner',
    maxSendsPerHour: 10_000_000,
  })
  assert.equal(checked.ok, false)
  assert.match(checked.errors.join(' '), /maxSendsPerHour/)
})

test('cancelOn without a delay is refused — there is no window to cancel in', async () => {
  const checked = await rules.validate({
    triggerId: 'uo.house.idoc_warning',
    name: 'IDOC warning',
    channels: ['email'],
    audience: 'owner',
    cancelOn: ['uo.house.repaired'],
  })
  assert.equal(checked.ok, false)
  assert.match(checked.errors.join(' '), /no effect without a delaySeconds/)
})

test('a rule whose channel was removed is dormant but still editable', async () => {
  const rule = addRule({ channels: ['email', 'carrier-pigeon'] })
  const listed = await rules.listAnnotated()
  assert.equal(listed.find((r) => r.id === rule.id).dormant, true)

  // …and only the live channel is used when it fires.
  optIn(10, 'uo.house.idoc_warning', 'email')
  await engine.dispatch(event(), T0)
  assert.deepEqual([...new Set(outboxRows().map((r) => r.channel))], ['email'])
})

// ── The dispatch contract ──────────────────────────────────────────────────

test('dispatch never throws at its caller, even when the database is gone', async () => {
  addRule()
  rulesDb.enabledForTrigger = async () => {
    throw new Error('connection lost')
  }

  const result = await engine.dispatch(event(), T0)
  assert.equal(result.enqueued, 0)
})

test('an event nobody has written a rule for is a no-op', async () => {
  const result = await engine.dispatch(event(), T0)
  assert.equal(result.rules, 0)
  assert.equal(outboxRows().length, 0)
})

test('a disabled rule does not fire', async () => {
  addRule({ enabled: false })
  const result = await engine.dispatch(event(), T0)
  assert.equal(result.rules, 0)
})