website/server/test/engagementEngineSql.test.js

// ── The engine's raw SQL, against a real MariaDB ───────────────────────────
//
// ENGAGEMENT.md Phase 4a. `engagementEngine.test.js` stubs the five tables and
// exercises the engine's logic against in-memory stand-ins, which is the right
// shape for everything the engine DECIDES. It cannot prove the three statements
// whose whole correctness is a server contract:
//
//   • the cooldown claim's answer is read out of `affectedRows`, and what that
//     number MEANS depends on the pool's `foundRows` setting. This file is what
//     found that: §4.1's single `INSERT ... ON DUPLICATE KEY UPDATE` was written,
//     was green against the stub, and always allowed the send against a real
//     server, because the connector defaults `foundRows: true` and a no-op
//     update reports 1 rather than 0. A cooldown that never cools.
//   • the outbox claim is a compare-and-set (§7.1 Q2), and "exactly one winner"
//     is `affectedRows = 1` for one caller and 0 for the other.
//   • `uq_engo_dedupe` is scoped to (rule, user, channel, dedupe_key), so one
//     event's key fans out to every recipient instead of admitting the first.
//
// A stub that reproduces those from the same reading of the manual proves the
// reading, not the server. So this file talks to a real database.
//
// **It SKIPS when there is none**, and that is deliberate rather than lax: CI
// runs the suite without a database (the harness points the pool at a dead port),
// and a file that failed there would make every PR red for a reason unrelated to
// itself. Run it against this machine's container with:
//
//   DB_HOST=127.0.0.1 DB_PORT=3307 DB_USER=... DB_PASSWORD=... DB_NAME=... \
//     node --test test/engagementEngineSql.test.js
//
// It creates its tables in a throwaway database named after the process, and
// drops it again, so it can never touch a real schema.

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

const SCHEMA = `
CREATE TABLE engagement_cooldowns (
  rule_id      INT          NOT NULL,
  user_id      INT          NOT NULL,
  subject_key  VARCHAR(190) NOT NULL DEFAULT '',
  last_fired_at DATETIME    NOT NULL,
  fire_count   INT          NOT NULL DEFAULT 1,
  PRIMARY KEY (rule_id, user_id, subject_key)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE engagement_outbox (
  id           BIGINT AUTO_INCREMENT PRIMARY KEY,
  rule_id      INT          NOT NULL,
  trigger_id   VARCHAR(96)  NOT NULL,
  user_id      INT          NOT NULL,
  channel      VARCHAR(32)  NOT NULL,
  subject_key  VARCHAR(190) NOT NULL DEFAULT '',
  payload      JSON         NOT NULL,
  dedupe_key   VARCHAR(190) NULL,
  status       ENUM('scheduled','sending','sent','failed','cancelled','suppressed') NOT NULL DEFAULT 'scheduled',
  due_at       DATETIME     NOT NULL,
  attempts     SMALLINT     NOT NULL DEFAULT 0,
  last_error   TEXT         NULL,
  created_at   DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at   DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  sent_at      DATETIME     NULL,
  UNIQUE KEY uq_engo_dedupe (rule_id, user_id, channel, dedupe_key),
  INDEX idx_engo_due (status, due_at),
  INDEX idx_engo_cancel (rule_id, user_id, subject_key, status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
`

const DB = `rg_engage_test_${process.pid}`
let pool = null
let available = false

// The statements under test, verbatim from the two `.db` files. They are
// duplicated here rather than required, because requiring the modules would drag
// in `utils/db`'s pool, which the harness has already pointed at a dead port.
//
// The pool below leaves `foundRows` at the connector's default, exactly as
// `utils/db.js` does - pinning it to `false` here would make this file agree with
// the code by construction and prove nothing about the pool the server runs.
const CLAIM_COOLDOWN_UPDATE = `
UPDATE engagement_cooldowns
   SET last_fired_at = ?, fire_count = fire_count + 1
 WHERE rule_id = ? AND user_id = ? AND subject_key = ?
   AND last_fired_at <= ? - INTERVAL ? SECOND`

const CLAIM_COOLDOWN_INSERT = `
INSERT IGNORE INTO engagement_cooldowns (rule_id, user_id, subject_key, last_fired_at, fire_count)
VALUES (?, ?, ?, ?, 1)`

const CLAIM_OUTBOX = `
UPDATE engagement_outbox SET status = 'sending', attempts = attempts + 1
 WHERE id = ? AND status = 'scheduled'`

const ENQUEUE = `
INSERT IGNORE INTO engagement_outbox
  (rule_id, trigger_id, user_id, channel, subject_key, payload, dedupe_key, due_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`

before(async () => {
  const admin = mariadb.createPool({
    host: process.env.DB_HOST || '127.0.0.1',
    port: Number(process.env.DB_PORT) || 3306,
    user: process.env.DB_USER || 'root',
    password: process.env.DB_PASSWORD || '',
    connectionLimit: 1,
    connectTimeout: 2000,
    initializationTimeout: 2000,
    multipleStatements: true,
  })
  try {
    await admin.query(`CREATE DATABASE ${DB}`)
    available = true
  } catch {
    available = false
  } finally {
    await admin.end().catch(() => {})
  }
  if (!available) return

  pool = mariadb.createPool({
    host: process.env.DB_HOST || '127.0.0.1',
    port: Number(process.env.DB_PORT) || 3306,
    user: process.env.DB_USER || 'root',
    password: process.env.DB_PASSWORD || '',
    database: DB,
    connectionLimit: 3,
    multipleStatements: true,
    bigIntAsNumber: true,
    insertIdAsNumber: true,
  })
  await pool.query(SCHEMA)
})

after(async () => {
  if (pool) {
    await pool.query(`DROP DATABASE IF EXISTS ${DB}`).catch(() => {})
    await pool.end().catch(() => {})
  }
})

// Checked INSIDE each test, never as a `{ skip }` option: the option is
// evaluated when the file is read, which is before `before()` has had a chance to
// find out whether there is a database. Every test skipped unconditionally is
// what that mistake looks like, and it looks exactly like a passing suite.
const SKIP = 'no database reachable - set DB_HOST/DB_PORT/DB_USER/DB_PASSWORD to run'
const needDb = (t) => {
  if (available) return false
  t.skip(SKIP)
  return true
}

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

const claimCooldown = async (ruleId, userId, subject, seconds, now) => {
  const moved = await pool.query(CLAIM_COOLDOWN_UPDATE, [now, ruleId, userId, subject, now, seconds])
  if (Number(moved.affectedRows) === 1) return true
  const inserted = await pool.query(CLAIM_COOLDOWN_INSERT, [ruleId, userId, subject, now])
  return Number(inserted.affectedRows) === 1
}

// ── The cooldown claim ─────────────────────────────────────────────────────

test('the cooldown claim: first fire inserts and is allowed', async (t) => {
  if (needDb(t)) return
  assert.equal(await claimCooldown(1, 10, 'h1', 3600, T0), true)
})

test('the cooldown claim: a second fire inside the window is REFUSED', async (t) => {
  if (needDb(t)) return
  await claimCooldown(2, 10, 'h1', 3600, T0)
  // affectedRows = 0: a duplicate key whose update changed nothing.
  assert.equal(await claimCooldown(2, 10, 'h1', 3600, later(60_000)), false)
})

test('the cooldown claim: a fire after the window is allowed, and counts', async (t) => {
  if (needDb(t)) return
  await claimCooldown(3, 10, 'h1', 60, T0)
  assert.equal(await claimCooldown(3, 10, 'h1', 60, later(61_000)), true)
  const [row] = await pool.query('SELECT fire_count FROM engagement_cooldowns WHERE rule_id = 3')
  assert.equal(Number(row.fire_count), 2)
})

test('the cooldown claim: the refusal survives foundRows — the bug this file caught', async (t) => {
  if (needDb(t)) return
  // The regression, named. `foundRows: true` (the connector's default, and what
  // `utils/db.js` gets) makes `affectedRows` count MATCHED rows, so the
  // ON DUPLICATE KEY UPDATE form's "0 means still cooling" reading returns 1 and
  // every send is allowed. Guarding in a WHERE clause is what makes the number
  // mean one thing.
  const seed = 'INSERT INTO engagement_cooldowns VALUES (7, 10, "h1", ?, 1)'
  await pool.query(seed, [T0])
  const noop = await pool.query(`${seed} ON DUPLICATE KEY UPDATE fire_count = fire_count`, [T0])
  assert.equal(Number(noop.affectedRows), 1, 'a no-op ODKU reports 1 under foundRows, not 0')

  // …and the shipped claim still refuses.
  assert.equal(await claimCooldown(7, 10, 'h1', 3600, later(60_000)), false)
})

test('the cooldown claim: repeated expiries keep counting', async (t) => {
  if (needDb(t)) return
  // `fire_count` is moved by the same guarded UPDATE that moves `last_fired_at`,
  // so a claim that succeeded and a claim that counted can never disagree.
  await claimCooldown(4, 10, 'h1', 10, T0)
  for (let i = 1; i <= 3; i += 1) await claimCooldown(4, 10, 'h1', 10, later(i * 11_000))
  const [row] = await pool.query('SELECT fire_count FROM engagement_cooldowns WHERE rule_id = 4')
  assert.equal(Number(row.fire_count), 4)
})

test('the cooldown claim: a different subject is a different row', async (t) => {
  if (needDb(t)) return
  assert.equal(await claimCooldown(5, 10, 'house-1', 86_400, T0), true)
  assert.equal(await claimCooldown(5, 10, 'house-2', 86_400, later(1000)), true)
  assert.equal(await claimCooldown(5, 10, 'house-1', 86_400, later(2000)), false)
})

test('the cooldown claim: cooldown_seconds = 0 always passes', async (t) => {
  if (needDb(t)) return
  assert.equal(await claimCooldown(6, 10, '', 0, T0), true)
  assert.equal(await claimCooldown(6, 10, '', 0, later(1)), true)
})

// ── The outbox claim and the dedupe key ────────────────────────────────────

const enqueue = async (over = {}) => {
  const row = {
    rule_id: 1, trigger_id: 'uo.house.idoc_warning', user_id: 10, channel: 'email',
    subject_key: 'h1', dedupe_key: null, due_at: T0, ...over,
  }
  const r = await pool.query(ENQUEUE, [
    row.rule_id, row.trigger_id, row.user_id, row.channel, row.subject_key,
    JSON.stringify({ house: 'A' }), row.dedupe_key, row.due_at,
  ])
  return Number(r.affectedRows) === 1 ? Number(r.insertId) : null
}

test('the outbox claim: exactly one of two callers wins (§7.1 Q2)', async (t) => {
  if (needDb(t)) return
  const id = await enqueue({ rule_id: 20 })
  const a = await pool.query(CLAIM_OUTBOX, [id])
  const b = await pool.query(CLAIM_OUTBOX, [id])
  assert.equal(Number(a.affectedRows), 1)
  assert.equal(Number(b.affectedRows), 0)
  const [row] = await pool.query('SELECT status, attempts FROM engagement_outbox WHERE id = ?', [id])
  assert.equal(row.status, 'sending')
  assert.equal(Number(row.attempts), 1)
})

test('the outbox claim under real concurrency: one winner, however many racers', async (t) => {
  if (needDb(t)) return
  const id = await enqueue({ rule_id: 21 })
  // Fired at once on separate pooled connections, so the server - not the
  // JavaScript event loop's ordering - is what serialises them.
  const results = await Promise.all([1, 2, 3, 4, 5].map(() => pool.query(CLAIM_OUTBOX, [id])))
  assert.equal(results.filter((r) => Number(r.affectedRows) === 1).length, 1)
})

test('a replayed event with the same dedupe key is IGNOREd, not duplicated', async (t) => {
  if (needDb(t)) return
  const first = await enqueue({ rule_id: 30, dedupe_key: 'idoc:4001' })
  const replay = await enqueue({ rule_id: 30, dedupe_key: 'idoc:4001' })
  assert.ok(first)
  assert.equal(replay, null)
})

test('ONE dedupe key fans out to every recipient — the scoped unique key', async (t) => {
  if (needDb(t)) return
  // §4.2a's global `UNIQUE (dedupe_key)` would have admitted the first of these
  // and silently ignored the other five: fifty recipients would have become one.
  const ids = []
  for (const user of [10, 11, 12]) {
    for (const channel of ['email', 'inapp']) {
      ids.push(await enqueue({ rule_id: 31, user_id: user, channel, dedupe_key: 'idoc:4001' }))
    }
  }
  assert.equal(ids.filter(Boolean).length, 6)
})

test('a NULL dedupe key never collides — many NULLs are legal under a UNIQUE index', async (t) => {
  if (needDb(t)) return
  const a = await enqueue({ rule_id: 32, dedupe_key: null })
  const b = await enqueue({ rule_id: 32, dedupe_key: null })
  assert.ok(a && b && a !== b)
})