// ── The retention sweep, against a real MariaDB ────────────────────────────
//
// ENGAGEMENT.md Phase 14. `engagementRetention.test.js` proves the policy and
// the worker's control flow against stubs, which is right for everything the
// worker DECIDES. It cannot prove the three statements whose whole correctness
// is what the server does with them, and this is the file for those — same shape
// and same reasoning as `engagementEngineSql.test.js`, which exists because a
// stub written from the same reading of the manual proves the reading, not the
// server (that is how the `foundRows` cooldown defect got as far as it did).
//
// It is also Phase 14's acceptance rig, stated in the phase: *"a rig run shows
// the sweep deleting terminal rows while leaving a `scheduled` outbox row and an
// in-window send-log row alone."*
//
// **It SKIPS when there is no database**, deliberately: CI runs the suite with
// the pool pointed 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=3306 DB_USER=... DB_PASSWORD=... \
//     node --test test/engagementRetentionSql.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')

// The three tables under sweep, with the two indexes Phase 14 added. Copied
// rather than required, for the reason the engine's SQL file gives: requiring the
// modules would drag in `utils/db`'s pool.
const SCHEMA = `
CREATE TABLE engagement_cooldowns (
  rule_id      INT          NOT NULL,
  user_id      INT          NOT NULL,
  subject_key  VARCHAR(190) NOT NULL DEFAULT '',
  channel      VARCHAR(32)  NOT NULL DEFAULT '',
  last_fired_at DATETIME    NOT NULL,
  fire_count   INT          NOT NULL DEFAULT 1,
  PRIMARY KEY (rule_id, user_id, subject_key, channel),
  INDEX idx_engc_sweep (last_fired_at)
) 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,
  INDEX idx_engo_due (status, due_at),
  INDEX idx_engo_sweep (status, created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE engagement_sends (
  id           BIGINT AUTO_INCREMENT PRIMARY KEY,
  rule_id      INT          NULL,
  trigger_id   VARCHAR(96)  NOT NULL,
  user_id      INT          NULL,
  channel      VARCHAR(32)  NOT NULL,
  status       ENUM('sent','failed','suppressed','bounced','complained') NOT NULL,
  created_at   DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  INDEX idx_engs_rule_window (rule_id, created_at),
  INDEX idx_engs_sweep (created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
`

// Verbatim from the three `.db` files. If one of these drifts from its source the
// test still passes and proves nothing, which is the standing cost of the copy —
// the alternative (requiring the modules) costs a live pool on a dead port.
const PRUNE_COOLDOWNS = 'DELETE FROM engagement_cooldowns WHERE last_fired_at < ? LIMIT ?'
const PRUNE_OUTBOX = `
DELETE FROM engagement_outbox
 WHERE status IN ('sent', 'failed', 'cancelled', 'suppressed')
   AND created_at < ?
 LIMIT ?`
const PRUNE_SENDS = 'DELETE FROM engagement_sends WHERE created_at < ? LIMIT ?'
const GIVE_UP = `
UPDATE engagement_outbox
   SET status = 'failed', last_error = 'gave up after repeated interruptions'
 WHERE status = 'sending' AND updated_at < ? AND attempts >= ?`
const RECLAIM = "UPDATE engagement_outbox SET status = 'scheduled' WHERE status = 'sending' AND updated_at < ?"

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

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, before `before()` has found out whether there
// is a database, and every test skipped unconditionally looks like a pass.
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 NOW = new Date('2026-09-01T12:00:00Z')
const daysAgo = (d) => new Date(NOW.getTime() - d * 86400000)

const clear = async () => {
  await pool.query('DELETE FROM engagement_cooldowns')
  await pool.query('DELETE FROM engagement_outbox')
  await pool.query('DELETE FROM engagement_sends')
}

const addOutbox = (status, createdAt, extra = {}) =>
  pool.query(
    `INSERT INTO engagement_outbox
       (rule_id, trigger_id, user_id, channel, payload, due_at, status, created_at, updated_at, attempts)
     VALUES (1, 't', 1, 'email', '{}', ?, ?, ?, ?, ?)`,
    [
      extra.dueAt || createdAt,
      status,
      createdAt,
      extra.updatedAt || createdAt,
      extra.attempts ?? 0,
    ],
  )

// ── The acceptance rig ─────────────────────────────────────────────────────

test('the outbox sweep deletes terminal rows and leaves a scheduled one alone', async (t) => {
  if (needDb(t)) return
  await clear()

  // Four terminal rows, well past the horizon.
  for (const status of ['sent', 'failed', 'cancelled', 'suppressed']) {
    await addOutbox(status, daysAgo(90))
  }
  // The row the phase names: old, and still a promise this deployment has made.
  await addOutbox('scheduled', daysAgo(90))
  // And one in flight, which is a worker's row and not the sweep's business.
  await addOutbox('sending', daysAgo(90))

  const result = await pool.query(PRUNE_OUTBOX, [daysAgo(30), 1000])
  assert.equal(Number(result.affectedRows), 4, 'exactly the four terminal rows')

  const left = await pool.query('SELECT status FROM engagement_outbox ORDER BY status')
  assert.deepEqual(left.map((r) => r.status), ['scheduled', 'sending'])
})

test('an in-window row is left alone whatever its status', async (t) => {
  if (needDb(t)) return
  await clear()
  await addOutbox('sent', daysAgo(29))
  await addOutbox('sent', daysAgo(31))

  const result = await pool.query(PRUNE_OUTBOX, [daysAgo(30), 1000])
  assert.equal(Number(result.affectedRows), 1)
  const [row] = await pool.query('SELECT created_at FROM engagement_outbox')
  assert.ok(row, 'the row inside the horizon survived')
})

test('the send-log sweep leaves an in-window row alone', async (t) => {
  if (needDb(t)) return
  await clear()
  const add = (createdAt) => pool.query(
    `INSERT INTO engagement_sends (rule_id, trigger_id, user_id, channel, status, created_at)
     VALUES (1, 't', 1, 'email', 'sent', ?)`,
    [createdAt],
  )
  await add(daysAgo(200))
  await add(daysAgo(179))
  // The row the per-rule hourly ceiling counts. If a horizon could reach this,
  // the ceiling would silently stop capping anything.
  await add(new Date(NOW.getTime() - 60 * 1000))

  const result = await pool.query(PRUNE_SENDS, [daysAgo(180), 1000])
  assert.equal(Number(result.affectedRows), 1)
  const [{ n }] = await pool.query('SELECT COUNT(*) AS n FROM engagement_sends')
  assert.equal(Number(n), 2)
})

test('the cooldown sweep respects its batch limit and repeats', async (t) => {
  if (needDb(t)) return
  await clear()
  for (let i = 0; i < 5; i += 1) {
    await pool.query(
      `INSERT INTO engagement_cooldowns (rule_id, user_id, subject_key, channel, last_fired_at)
       VALUES (1, ?, '', 'email', ?)`,
      [i, daysAgo(90)],
    )
  }
  const first = await pool.query(PRUNE_COOLDOWNS, [daysAgo(30), 2])
  assert.equal(Number(first.affectedRows), 2, 'LIMIT bounds one statement')
  const second = await pool.query(PRUNE_COOLDOWNS, [daysAgo(30), 10])
  assert.equal(Number(second.affectedRows), 3, 'and the rest go on the next pass')
})

// ── The bound the sweep depends on ─────────────────────────────────────────

test('a stale row that has burned its attempts is failed, not handed back', async (t) => {
  if (needDb(t)) return
  await clear()
  // Two rows stranded in 'sending' by a crash between the claim and the outcome.
  // One has attempts left; the other has spent them, and before Phase 14 nothing
  // could ever move it — MAX_ATTEMPTS is consulted only on a graceful `retry`,
  // so it cycled sending → scheduled → sending forever, never became terminal,
  // and was therefore never eligible for any sweep.
  await addOutbox('sending', daysAgo(1), { updatedAt: daysAgo(1), attempts: 1 })
  await addOutbox('sending', daysAgo(1), { updatedAt: daysAgo(1), attempts: 5 })

  const gaveUp = await pool.query(GIVE_UP, [new Date(NOW.getTime() - 15 * 60 * 1000), 5])
  assert.equal(Number(gaveUp.affectedRows), 1, 'only the exhausted row')
  const reclaimed = await pool.query(RECLAIM, [new Date(NOW.getTime() - 15 * 60 * 1000)])
  assert.equal(Number(reclaimed.affectedRows), 1, 'and only the other one comes back')

  const rows = await pool.query('SELECT status, attempts FROM engagement_outbox ORDER BY attempts')
  assert.deepEqual(rows.map((r) => r.status), ['scheduled', 'failed'])

  // And now it is terminal, so the sweep can bound it.
  await pool.query("UPDATE engagement_outbox SET created_at = ? WHERE status = 'failed'", [daysAgo(90)])
  const swept = await pool.query(PRUNE_OUTBOX, [daysAgo(30), 1000])
  assert.equal(Number(swept.affectedRows), 1)
})

test('reclaiming before giving up would loop forever — the order is the fix', async (t) => {
  if (needDb(t)) return
  await clear()
  await addOutbox('sending', daysAgo(1), { updatedAt: daysAgo(1), attempts: 5 })

  // The wrong order, run deliberately: reclaim first, and the exhausted row is
  // back in 'scheduled' where findDue will pick it up again.
  await pool.query(RECLAIM, [new Date(NOW.getTime() - 15 * 60 * 1000)])
  const gaveUp = await pool.query(GIVE_UP, [new Date(NOW.getTime() - 15 * 60 * 1000), 5])
  assert.equal(Number(gaveUp.affectedRows), 0, 'nothing left in sending to fail')
  const [row] = await pool.query('SELECT status FROM engagement_outbox')
  assert.equal(row.status, 'scheduled', 'which is the forever-retry this ordering avoids')
})

// ── The indexes ────────────────────────────────────────────────────────────

test('each sweep is an index range scan, not a table scan', async (t) => {
  if (needDb(t)) return
  await clear()
  // A plan on an empty table is not worth reading, so give the optimizer rows.
  for (let i = 0; i < 200; i += 1) {
    await addOutbox(i % 2 ? 'sent' : 'scheduled', daysAgo(i))
    await pool.query(
      `INSERT INTO engagement_sends (rule_id, trigger_id, user_id, channel, status, created_at)
       VALUES (1, 't', 1, 'email', 'sent', ?)`,
      [daysAgo(i)],
    )
  }
  await pool.query('ANALYZE TABLE engagement_outbox')
  await pool.query('ANALYZE TABLE engagement_sends')

  const outboxPlan = await pool.query(
    `EXPLAIN SELECT id FROM engagement_outbox
      WHERE status IN ('sent','failed','cancelled','suppressed') AND created_at < ?`,
    [daysAgo(30)],
  )
  assert.ok(
    String(outboxPlan[0].key || '').includes('idx_engo'),
    `expected an index, got ${JSON.stringify(outboxPlan[0])}`,
  )

  const sendsPlan = await pool.query(
    'EXPLAIN SELECT id FROM engagement_sends WHERE created_at < ?',
    [daysAgo(180)],
  )
  assert.equal(
    sendsPlan[0].key,
    'idx_engs_sweep',
    `every other index on this table has created_at in SECOND position: ${JSON.stringify(sendsPlan[0])}`,
  )
})