feat(engagement): the engagement system — cutover 3 of 7 (edge → main)
#180
@@ -1774,9 +1774,18 @@ CREATE TABLE IF NOT EXISTS engagement_cooldowns (
|
|||||||
-- MariaDB would coerce a NULL one anyway. '' is "this rule cools per user, not
|
-- MariaDB would coerce a NULL one anyway. '' is "this rule cools per user, not
|
||||||
-- per subject".
|
-- per subject".
|
||||||
subject_key VARCHAR(190) NOT NULL DEFAULT '',
|
subject_key VARCHAR(190) NOT NULL DEFAULT '',
|
||||||
|
-- The CHANNEL the cooldown is about, added in Phase 11b after the live walk.
|
||||||
|
-- Without it a rule naming two channels delivers on exactly ONE of them: the
|
||||||
|
-- claim runs inside the engine's per-channel loop, `inapp` is ranked first on
|
||||||
|
-- purpose (so push can reference its inbox row), and every later channel is
|
||||||
|
-- then reported as cooled. Phase 11b's decision 8 requires the letter and the
|
||||||
|
-- inbox item to fire together, so the cooldown is per delivery, not per
|
||||||
|
-- occasion. VARCHAR like `engagement_outbox.channel`, and for the same reason:
|
||||||
|
-- the channel set is data a module can extend.
|
||||||
|
channel VARCHAR(32) NOT NULL DEFAULT '',
|
||||||
last_fired_at DATETIME NOT NULL,
|
last_fired_at DATETIME NOT NULL,
|
||||||
fire_count INT NOT NULL DEFAULT 1,
|
fire_count INT NOT NULL DEFAULT 1,
|
||||||
PRIMARY KEY (rule_id, user_id, subject_key),
|
PRIMARY KEY (rule_id, user_id, subject_key, channel),
|
||||||
CONSTRAINT fk_engc_rule FOREIGN KEY (rule_id) REFERENCES engagement_rules(id) ON DELETE CASCADE,
|
CONSTRAINT fk_engc_rule FOREIGN KEY (rule_id) REFERENCES engagement_rules(id) ON DELETE CASCADE,
|
||||||
CONSTRAINT fk_engc_user FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
|
CONSTRAINT fk_engc_user FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
|
||||||
-- So a prune worker can drop rows older than the longest configured cooldown.
|
-- So a prune worker can drop rows older than the longest configured cooldown.
|
||||||
@@ -1785,6 +1794,29 @@ CREATE TABLE IF NOT EXISTS engagement_cooldowns (
|
|||||||
INDEX idx_engc_sweep (last_fired_at)
|
INDEX idx_engc_sweep (last_fired_at)
|
||||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||||
|
|
||||||
|
-- Widen the key on a deployment that already has the table. Two statements, and
|
||||||
|
-- the second is guarded because MariaDB has no conditional form of a PRIMARY KEY
|
||||||
|
-- change: re-running `DROP PRIMARY KEY, ADD PRIMARY KEY` on a table that already
|
||||||
|
-- carries the new one is an error, not a no-op, so replaying this file on every
|
||||||
|
-- boot would fail the whole schema after the first run. The guard reads the key
|
||||||
|
-- itself out of information_schema rather than the column's existence, because
|
||||||
|
-- `ADD COLUMN IF NOT EXISTS` above can succeed while the key change does not.
|
||||||
|
--
|
||||||
|
-- Existing rows keep `channel = ''`, which is one stale cooldown per (rule, user,
|
||||||
|
-- subject) that expires on its own interval. That is the right trade against
|
||||||
|
-- deleting them: a cooldown that outlives its rewrite costs at most one delayed
|
||||||
|
-- notification, and dropping the table would let a bounce storm through.
|
||||||
|
ALTER TABLE engagement_cooldowns ADD COLUMN IF NOT EXISTS channel VARCHAR(32) NOT NULL DEFAULT '';
|
||||||
|
SET @engc_key_has_channel := (
|
||||||
|
SELECT COUNT(*) FROM information_schema.STATISTICS
|
||||||
|
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'engagement_cooldowns'
|
||||||
|
AND INDEX_NAME = 'PRIMARY' AND COLUMN_NAME = 'channel'
|
||||||
|
);
|
||||||
|
SET @sql := IF(@engc_key_has_channel = 0,
|
||||||
|
'ALTER TABLE engagement_cooldowns DROP PRIMARY KEY, ADD PRIMARY KEY (rule_id, user_id, subject_key, channel)',
|
||||||
|
'DO 0');
|
||||||
|
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
|
||||||
|
|
||||||
-- §4.2a. Modelled on announce_jobs / announce_job_legs. One row per
|
-- §4.2a. Modelled on announce_jobs / announce_job_legs. One row per
|
||||||
-- (rule, user, channel) occurrence of an event.
|
-- (rule, user, channel) occurrence of an event.
|
||||||
CREATE TABLE IF NOT EXISTS engagement_outbox (
|
CREATE TABLE IF NOT EXISTS engagement_outbox (
|
||||||
|
|||||||
@@ -203,9 +203,18 @@ async function applyRule(rule, event, now) {
|
|||||||
summary.capped += 1
|
summary.capped += 1
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
// One statement, guarded on the interval, so two concurrent emits cannot
|
// Guarded on the interval, so two concurrent emits cannot both pass a
|
||||||
// both pass a read-then-write check (§4.1).
|
// read-then-write check (§4.1).
|
||||||
const allowed = await cooldownsDb.claim(rule.id, userId, subjectKey, rule.cooldown_seconds, now)
|
//
|
||||||
|
// **Keyed on the CHANNEL as well**, which is what makes this loop correct
|
||||||
|
// rather than what makes it work. Without the channel, the first channel of
|
||||||
|
// a rule claims the cooldown and every later one is refused as cooling —
|
||||||
|
// and `inapp` is ranked first above, so a rule naming email + in-app would
|
||||||
|
// deliver the in-app item and silently never the mail. Found on Phase 11b's
|
||||||
|
// live rig; a cooldown is per delivery, not per occasion.
|
||||||
|
const allowed = await cooldownsDb.claim(
|
||||||
|
rule.id, userId, subjectKey, channel, rule.cooldown_seconds, now,
|
||||||
|
)
|
||||||
if (!allowed) {
|
if (!allowed) {
|
||||||
summary.cooled += 1
|
summary.cooled += 1
|
||||||
continue
|
continue
|
||||||
|
|||||||
@@ -1,8 +1,14 @@
|
|||||||
const { query } = require('../../utils/db')
|
const { query } = require('../../utils/db')
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Claim a fire for (rule, user, subject), or refuse it because the pair is still
|
* Claim a fire for (rule, user, subject, channel), or refuse it because that
|
||||||
* cooling. ENGAGEMENT.md §4.1.
|
* delivery is still cooling. ENGAGEMENT.md §4.1.
|
||||||
|
*
|
||||||
|
* **`channel` is part of the key, and Phase 11b is where that was settled.** The
|
||||||
|
* engine claims inside its per-channel loop, so a key without the channel means
|
||||||
|
* the first channel of a two-channel rule claims the cooldown and every later one
|
||||||
|
* is refused as cooling - which made every in-universe email body of Phase 11b
|
||||||
|
* unreachable behind the in-app one. A cooldown is per delivery.
|
||||||
*
|
*
|
||||||
* **Two statements, each of which is its own atomic decision** - and it is worth
|
* **Two statements, each of which is its own atomic decision** - and it is worth
|
||||||
* saying why it is not the single `INSERT ... ON DUPLICATE KEY UPDATE` §4.1
|
* saying why it is not the single `INSERT ... ON DUPLICATE KEY UPDATE` §4.1
|
||||||
@@ -37,28 +43,29 @@ const { query } = require('../../utils/db')
|
|||||||
* `cooldown_seconds = 0` always passes, which is the documented meaning of a rule
|
* `cooldown_seconds = 0` always passes, which is the documented meaning of a rule
|
||||||
* with no cooldown: the guard becomes `last_fired_at <= now`, and it is.
|
* with no cooldown: the guard becomes `last_fired_at <= now`, and it is.
|
||||||
*/
|
*/
|
||||||
async function claim(ruleId, userId, subjectKey, cooldownSeconds, now = new Date()) {
|
async function claim(ruleId, userId, subjectKey, channel, cooldownSeconds, now = new Date()) {
|
||||||
const moved = await query(
|
const moved = await query(
|
||||||
`UPDATE engagement_cooldowns
|
`UPDATE engagement_cooldowns
|
||||||
SET last_fired_at = ?, fire_count = fire_count + 1
|
SET last_fired_at = ?, fire_count = fire_count + 1
|
||||||
WHERE rule_id = ? AND user_id = ? AND subject_key = ?
|
WHERE rule_id = ? AND user_id = ? AND subject_key = ? AND channel = ?
|
||||||
AND last_fired_at <= ? - INTERVAL ? SECOND`,
|
AND last_fired_at <= ? - INTERVAL ? SECOND`,
|
||||||
[now, ruleId, userId, subjectKey, now, cooldownSeconds],
|
[now, ruleId, userId, subjectKey, channel, now, cooldownSeconds],
|
||||||
)
|
)
|
||||||
if (Number(moved?.affectedRows || 0) === 1) return true
|
if (Number(moved?.affectedRows || 0) === 1) return true
|
||||||
|
|
||||||
const inserted = await query(
|
const inserted = await query(
|
||||||
`INSERT IGNORE INTO engagement_cooldowns (rule_id, user_id, subject_key, last_fired_at, fire_count)
|
`INSERT IGNORE INTO engagement_cooldowns (rule_id, user_id, subject_key, channel, last_fired_at, fire_count)
|
||||||
VALUES (?, ?, ?, ?, 1)`,
|
VALUES (?, ?, ?, ?, ?, 1)`,
|
||||||
[ruleId, userId, subjectKey, now],
|
[ruleId, userId, subjectKey, channel, now],
|
||||||
)
|
)
|
||||||
return Number(inserted?.affectedRows || 0) === 1
|
return Number(inserted?.affectedRows || 0) === 1
|
||||||
}
|
}
|
||||||
|
|
||||||
const get = async (ruleId, userId, subjectKey) => {
|
const get = async (ruleId, userId, subjectKey, channel) => {
|
||||||
const [row] = await query(
|
const [row] = await query(
|
||||||
'SELECT * FROM engagement_cooldowns WHERE rule_id = ? AND user_id = ? AND subject_key = ?',
|
`SELECT * FROM engagement_cooldowns
|
||||||
[ruleId, userId, subjectKey],
|
WHERE rule_id = ? AND user_id = ? AND subject_key = ? AND channel = ?`,
|
||||||
|
[ruleId, userId, subjectKey, channel],
|
||||||
)
|
)
|
||||||
return row || null
|
return row || null
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -874,6 +874,10 @@ function checkSeedRule(owner, entry, ownTemplateKeys, coreKeys) {
|
|||||||
name: r.name,
|
name: r.name,
|
||||||
audience: r.audience,
|
audience: r.audience,
|
||||||
audience_segment_id: null,
|
audience_segment_id: null,
|
||||||
|
// Checked above and carried here: the column is NOT NULL, so a normalizer
|
||||||
|
// that validates the ceiling and then drops it fails every insert in the
|
||||||
|
// group at boot — loudly, but only on a real database.
|
||||||
|
max_sends_per_hour: r.max_sends_per_hour,
|
||||||
channels: [...r.channels],
|
channels: [...r.channels],
|
||||||
template_keys: { ...keys },
|
template_keys: { ...keys },
|
||||||
conditions: r.conditions === undefined ? null : r.conditions,
|
conditions: r.conditions === undefined ? null : r.conditions,
|
||||||
|
|||||||
@@ -98,8 +98,11 @@ function installStubs() {
|
|||||||
// the pair is still cooling. `engagementEngineSql.test.js` is what proves the
|
// 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
|
// SQL itself - a stub can only ever agree with whoever wrote it, and in this
|
||||||
// case the first version of both was wrong together.
|
// case the first version of both was wrong together.
|
||||||
cooldownsDb.claim = async (ruleId, userId, subjectKey, cooldownSeconds, now) => {
|
cooldownsDb.claim = async (ruleId, userId, subjectKey, channel, cooldownSeconds, now) => {
|
||||||
const key = `${ruleId}|${userId}|${subjectKey}`
|
// `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)
|
const row = store.cooldowns.get(key)
|
||||||
if (!row) {
|
if (!row) {
|
||||||
store.cooldowns.set(key, { last_fired_at: now, fire_count: 1 })
|
store.cooldowns.set(key, { last_fired_at: now, fire_count: 1 })
|
||||||
@@ -327,6 +330,26 @@ test('a cooldown that has expired lets the same subject through again', async ()
|
|||||||
assert.equal(outboxRows().length, 2)
|
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 () => {
|
test('two rules on one trigger each get their own cooldown', async () => {
|
||||||
addRule({ cooldown_seconds: 3600 })
|
addRule({ cooldown_seconds: 3600 })
|
||||||
addRule({ cooldown_seconds: 3600 })
|
addRule({ cooldown_seconds: 3600 })
|
||||||
|
|||||||
@@ -149,6 +149,32 @@ test('a rule must carry a per-hour ceiling', () => {
|
|||||||
assert.match(err, /max_sends_per_hour/)
|
assert.match(err, /max_sends_per_hour/)
|
||||||
})
|
})
|
||||||
|
|
||||||
|
test('a normalized rule carries every column the insert reads', () => {
|
||||||
|
// Refusing a bad ceiling and then DROPPING a good one are different bugs, and
|
||||||
|
// the first test cannot see the second: `engagementRules.db.insert` binds a
|
||||||
|
// fixed column list, so a field validated and not carried through arrives as
|
||||||
|
// NULL and fails the whole group at boot — on a real database only. Asserted
|
||||||
|
// against the column list itself rather than one field, because the next
|
||||||
|
// field added to the declaration is the next one that can be forgotten here.
|
||||||
|
assert.equal(trySeeds('demo', {
|
||||||
|
templates: [tpl()],
|
||||||
|
ruleGroups: [{ key: 'v1', rules: [rule({ delay_seconds: 60, cancel_on: ['demo.house.refreshed'] })] }],
|
||||||
|
}), null)
|
||||||
|
|
||||||
|
const seeded = registries.engagementSeedsFor('demo').ruleGroups[0].rules[0]
|
||||||
|
for (const column of [
|
||||||
|
'trigger_id', 'name', 'enabled', 'audience', 'audience_segment_id', 'max_sends_per_hour',
|
||||||
|
'channels', 'template_keys', 'conditions', 'cooldown_seconds', 'delay_seconds', 'cancel_on',
|
||||||
|
'updated_by',
|
||||||
|
]) {
|
||||||
|
assert.ok(column in seeded, `normalized rule is missing "${column}"`)
|
||||||
|
assert.notEqual(seeded[column], undefined, `normalized rule leaves "${column}" undefined`)
|
||||||
|
}
|
||||||
|
assert.equal(seeded.max_sends_per_hour, 200)
|
||||||
|
assert.equal(seeded.delay_seconds, 60)
|
||||||
|
assert.deepEqual(seeded.cancel_on, ['demo.house.refreshed'])
|
||||||
|
})
|
||||||
|
|
||||||
test('registering twice is a collision, not an addition', () => {
|
test('registering twice is a collision, not an addition', () => {
|
||||||
assert.equal(trySeeds('demo', { templates: [tpl()] }), null)
|
assert.equal(trySeeds('demo', { templates: [tpl()] }), null)
|
||||||
assert.match(trySeeds('demo', { templates: [tpl({ key: 'demo.other' })] }), /already registered/)
|
assert.match(trySeeds('demo', { templates: [tpl({ key: 'demo.other' })] }), /already registered/)
|
||||||
|
|||||||
Reference in New Issue
Block a user