diff --git a/server/db/schema.sql b/server/db/schema.sql index 267a7d5..0b6763a 100644 --- a/server/db/schema.sql +++ b/server/db/schema.sql @@ -1686,3 +1686,164 @@ CREATE TABLE IF NOT EXISTS notification_channel_prefs ( -- has no digest mode to be asked into instead. INSERT IGNORE INTO notification_channel_prefs (user_id, stream_id, channel, mode) SELECT user_id, stream_id, 'push', 'instant' FROM notification_subscriptions; + +-- ── The engagement engine (ENGAGEMENT.md §4.1, §4.2a, §4.5 — Phase 4a) ────── +-- +-- Five tables and no delivery. A rule says "when this trigger fires, for these +-- people, on these channels, no more often than this"; the outbox is the queue +-- the grace window needs; the cooldown table is what makes "once per house" mean +-- once per house; and the send log is the first answer this deployment has ever +-- had to "did user X get the mail?". +-- +-- Nothing here sends anything. Core seeds no rules and `enabled` defaults to 0, +-- so on a real deployment these five tables stay empty until an operator turns a +-- rule on from the screen Phase 4b builds. + +-- What an operator actually configures: trigger -> audience -> template -> timing. +-- +-- `trigger_id` deliberately has NO foreign key and no existence check: a trigger +-- is DECLARED IN CODE (§4.3), so the set of them is whatever registered on this +-- boot. A rule naming a trigger no module currently registers is DORMANT — it is +-- listed, it never fires, and it starts working again when the module comes back +-- (§7.3). Deleting it on uninstall would silently destroy an operator's +-- configuration on the strength of a module being temporarily absent. +CREATE TABLE IF NOT EXISTS engagement_rules ( + id INT AUTO_INCREMENT PRIMARY KEY, + trigger_id VARCHAR(96) NOT NULL, + name VARCHAR(160) NOT NULL, + -- OFF by default (§7.1 Q3). A rule arrives inert and an operator turns it on, + -- so no import, seed or restore can start mailing on its own. + enabled TINYINT(1) NOT NULL DEFAULT 0, + audience VARCHAR(32) NOT NULL DEFAULT 'owner', + audience_segment_id INT NULL, + -- §7.1 Q3: the hard stop that makes operator-editable rules safe to choose over + -- code-registered ones. Counted in engagement_sends, enforced before the outbox + -- row is written, never overridable from the rule editor beyond this column. + max_sends_per_hour INT NOT NULL DEFAULT 100, + channels JSON NOT NULL, + template_keys JSON NOT NULL, + conditions JSON NULL, + cooldown_seconds INT NOT NULL DEFAULT 0, + delay_seconds INT NOT NULL DEFAULT 0, + cancel_on JSON NULL, + updated_by INT NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + CONSTRAINT fk_engr_user FOREIGN KEY (updated_by) REFERENCES users(id) ON DELETE SET NULL, + INDEX idx_engr_trigger (trigger_id, enabled) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + +-- §5.1a: an operator-composed segment over module-declared audiences. Stored as a +-- boolean tree of audience ids + params; `ceiling` is DERIVED at save time as the +-- NARROWEST ceiling in the tree (ceilings.meetAll) and re-checked against the +-- trigger's own ceiling, so composition can never widen. It is a column rather +-- than a runtime computation so an audit can read what a rule was allowed to +-- reach without re-resolving it — and so a module that has since changed its +-- audience's ceiling cannot retroactively widen a saved segment. +-- +-- `engagement_rules.audience_segment_id` above points here with NO foreign key, +-- on purpose and for the same reason `trigger_id` has none: a rule whose segment +-- has been deleted must go DORMANT, not silently fall back to its plain +-- `audience` column. ON DELETE SET NULL would be exactly that silent fallback, +-- and the fallback reaches a DIFFERENT set of people (§5.1a rule 4). +CREATE TABLE IF NOT EXISTS engagement_audience_segments ( + id INT AUTO_INCREMENT PRIMARY KEY, + name VARCHAR(160) NOT NULL, + expression JSON NOT NULL, + ceiling VARCHAR(32) NOT NULL, + updated_by INT NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + CONSTRAINT fk_engseg_user FOREIGN KEY (updated_by) REFERENCES users(id) ON DELETE SET NULL +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + +-- §4.1. NOT `settings`: cooldown state is high-cardinality (recipients x rules x +-- subjects), written on every fire, and asked "is this one pair still cooling?". +-- A JSON blob under one settings key would be a read-modify-write of the whole +-- deployment's cooldown state per event, with a lost-update race between two +-- concurrent triggers. +-- +-- `subject_key` is what makes "one IDOC mail per player per day" the right rule +-- instead of the wrong one: a player with four houses decaying should hear about +-- all four, once each. Cooling per (rule, user) alone silently drops three. +CREATE TABLE IF NOT EXISTS engagement_cooldowns ( + rule_id INT NOT NULL, + user_id INT NOT NULL, + -- The SUBJECT the cooldown is about, opaque to core: a house serial, a vendor + -- id, ''. NOT NULL with a '' default, because this is a PRIMARY KEY column and + -- MariaDB would coerce a NULL one anyway. '' is "this rule cools per user, not + -- per subject". + 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), + 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, + -- So a prune worker can drop rows older than the longest configured cooldown. + -- Without it this table grows without bound, which is the failure mode + -- teamActivityPrune was written for. + INDEX idx_engc_sweep (last_fired_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + +-- §4.2a. Modelled on announce_jobs / announce_job_legs. One row per +-- (rule, user, channel) occurrence of an event. +CREATE TABLE IF NOT EXISTS engagement_outbox ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + rule_id INT NOT NULL, + trigger_id VARCHAR(96) NOT NULL, -- denormalized; survives a rule edit + user_id INT NOT NULL, + channel VARCHAR(32) NOT NULL, -- VARCHAR, never ENUM: the channel set is data + subject_key VARCHAR(190) NOT NULL DEFAULT '', + payload JSON NOT NULL, -- the declared variables, snapshotted at emit + 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, + CONSTRAINT fk_engo_rule FOREIGN KEY (rule_id) REFERENCES engagement_rules(id) ON DELETE CASCADE, + CONSTRAINT fk_engo_user FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE, + -- **Scoped to the row's identity, and §4.2a's global `UNIQUE (dedupe_key)` is + -- a defect this phase found while building it.** A dedupe key names the EVENT + -- ("house 0x4001 entered IDOC"), and one event legitimately becomes many rows: + -- an audience of fifty users is fifty rows, a rule spanning email and in-app + -- doubles that, and two rules on one trigger double it again. Under a global + -- unique index the FIRST of those inserts wins and every other one is silently + -- ignored — ninety-nine recipients dropped by the mechanism meant to stop a + -- replayed event becoming a second mail. Scoping it to (rule, user, channel) + -- keeps exactly that guarantee and nothing more. + UNIQUE KEY uq_engo_dedupe (rule_id, user_id, channel, dedupe_key), + INDEX idx_engo_due (status, due_at), + -- What a RESOLVING event queries: a house repaired back to LikeNew cancels + -- every scheduled row for that (rule, user, house). + INDEX idx_engo_cancel (rule_id, user_id, subject_key, status) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + +-- G15: the per-message record. Today "did user X get the mail?" is unanswerable. +-- +-- It is deliberately NOT a second address book: the address is stored as a +-- sha256, which is enough to correlate a bounce (Phase 9) and useless as a +-- mailing list. `user_id` is SET NULL rather than CASCADE so the log survives an +-- account deletion — an audit of what this deployment sent must not be erasable +-- by deleting the recipient. +CREATE TABLE IF NOT EXISTS engagement_sends ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + outbox_id BIGINT NULL, + rule_id INT NULL, + trigger_id VARCHAR(96) NOT NULL, + user_id INT NULL, + channel VARCHAR(32) NOT NULL, + transport VARCHAR(32) NULL, + address_hash CHAR(64) NULL, + status ENUM('sent','failed','suppressed','bounced','complained') NOT NULL, + detail VARCHAR(500) NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT fk_engs_user FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE SET NULL, + INDEX idx_engs_trigger (trigger_id, created_at), + INDEX idx_engs_user (user_id, created_at), + -- The per-rule hourly ceiling (§7.1 Q3) is counted here, so the count has to be + -- an index range scan rather than a table scan: it runs once per rule per event. + INDEX idx_engs_rule_window (rule_id, created_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/server/src/engagement/audiences.js b/server/src/engagement/audiences.js new file mode 100644 index 0000000..2d9d465 --- /dev/null +++ b/server/src/engagement/audiences.js @@ -0,0 +1,138 @@ +// ── Resolving a rule's audience to recipients ────────────────────────────── +// +// ENGAGEMENT.md §5.1a / §4.5, Phase 4a. A rule names an audience two ways and +// only ever one at a time: a **plain ceiling name** (`owner`, `staff`, +// `subscribers`, `authenticated`, `everyone`) resolved from core's own tables, or +// an **`audience_segment_id`** pointing at an operator-composed tree of +// module-declared audiences (segments.js). This file turns either into user ids. +// +// **Three things it is careful about, all of them the same worry.** The set this +// function returns is the set that gets mailed, so: +// +// 1. Every id is checked against `users.status = 'active'` - including the ones a +// MODULE's resolver produced, which core has no reason to trust with account +// status it does not know about. +// 2. A dormant segment (its module uninstalled) resolves to EMPTY and says so. +// The caller must not send. Falling back to the rule's plain `audience` +// column would reach a different population than the one composed (§5.1a +// rule 4), which is the failure mode this whole design exists to avoid. +// 3. `members` as a PLAIN audience resolves to nobody. It is the ceiling for +// "a module-declared list", and without a segment there is no list - core +// knows no game vocabulary and cannot guess which members were meant. A rule +// saved that way is inert and visible as such, rather than quietly falling +// back to something wider. + +const registries = require('../modules/registries') +const channels = require('./channels') +const segments = require('./segments') +const segmentsDb = require('../model/engagement/engagementSegments.db') +const recipients = require('../model/engagement/engagementRecipients.db') +const ceilings = require('../modules/ceilings') +const log = require('../utils/logger')('engagement') + +/** + * Which registered channels default to something other than 'off'? + * + * Read once per resolution rather than hardcoded, because it is the difference + * between "opted in" meaning a stored row and meaning the absence of one + * (§3.1, G9). All three of core's channels default 'off' today, so this is empty + * and `subscribers` is the simple query - but the answer lives in the registry. + */ +const defaultOnChannels = () => channels.all().filter((c) => c.defaultMode !== 'off').map((c) => c.id) + +/** + * Resolve one rule against one event. + * + * @returns {{ userIds: number[], ceiling: string|null, dormant: boolean, reason: string|null }} + * `dormant` means "this rule cannot be resolved right now"; `reason` names why + * for the log and, in Phase 4b, for the admin list's dormant badge. + */ +async function resolveForRule(rule, event) { + if (rule.audience_segment_id) { + const segment = await segmentsDb.getById(rule.audience_segment_id) + if (!segment) { + // The segment was deleted out from under the rule. `audience_segment_id` + // deliberately has no ON DELETE SET NULL (see schema.sql), because that + // would silently fall back to the rule's plain `audience` column and mail + // a different set of people. + return { userIds: [], ceiling: null, dormant: true, reason: 'audience segment no longer exists' } + } + const { dormant, userIds } = await segments.resolve(segment.expression) + if (dormant) { + return { userIds: [], ceiling: segment.ceiling, dormant: true, reason: 'audience segment is dormant' } + } + return { + userIds: await recipients.filterActive(userIds), + // The STORED ceiling, not one re-derived now: a module that has since + // widened its own audience's ceiling must not widen a segment that was + // saved under the old one. + ceiling: segment.ceiling, + dormant: false, + reason: null, + } + } + + switch (rule.audience) { + case 'owner': { + if (!event.ownerUserId) { + // Not dormant: the rule is fine and this particular event simply has no + // owner to mail. A trigger that never carries one is an operator's + // mistake the rule editor should catch (Phase 4b), not a runtime error. + return { userIds: [], ceiling: 'owner', dormant: false, reason: 'event carries no ownerUserId' } + } + return { + userIds: await recipients.filterActive([event.ownerUserId]), + ceiling: 'owner', + dormant: false, + reason: null, + } + } + case 'staff': + return { + userIds: await recipients.staff(ceilings.STAFF_CEILING_ROLES), + ceiling: 'staff', + dormant: false, + reason: null, + } + case 'subscribers': + return { + userIds: await recipients.subscribers(event.triggerId, defaultOnChannels()), + ceiling: 'subscribers', + dormant: false, + reason: null, + } + case 'authenticated': + case 'everyone': + return { userIds: await recipients.active(), ceiling: rule.audience, dormant: false, reason: null } + case 'members': + return { + userIds: [], + ceiling: 'members', + dormant: false, + reason: 'a "members" audience needs a segment naming which list', + } + default: + // Fails closed on an audience name the lattice does not know - the same + // posture `ceilings.permits` takes, and for the same reason. + log.warn('rule names an unknown audience', { rule: rule.id, audience: rule.audience }) + return { userIds: [], ceiling: null, dormant: true, reason: `unknown audience "${rule.audience}"` } + } +} + +/** + * The G24 gate, re-run at SEND time and not only at save time. + * + * A rule's audience was checked against its trigger's ceiling when it was saved, + * so this can only fail when something changed underneath: a module upgraded and + * narrowed its trigger's ceiling, or a module was replaced by one declaring the + * same id more tightly. That is precisely the case where a stale rule would + * otherwise mail a population the current declaration forbids, which is what + * makes this the security boundary rather than a duplicate check. + */ +function permitted(triggerId, ceiling) { + const declaration = registries.eventTrigger(triggerId) + if (!declaration) return false + return ceilings.permits(declaration.ceiling, ceiling) +} + +module.exports = { resolveForRule, permitted, defaultOnChannels } diff --git a/server/src/engagement/conditions.js b/server/src/engagement/conditions.js new file mode 100644 index 0000000..2f99031 --- /dev/null +++ b/server/src/engagement/conditions.js @@ -0,0 +1,251 @@ +// ── Rule conditions — a predicate over a trigger's DECLARED variables ─────── +// +// ENGAGEMENT.md §4.5, Phase 4a. `engagement_rules.conditions` is the half of a +// rule that decides *whether* this particular firing is interesting: "only when +// decayStatus is IDOC", "only for threads in this Team". Without it every rule is +// all-or-nothing per trigger, and an operator's only way to narrow is to ask a +// module author for a second trigger. +// +// **It is validated against the declaration, not against a payload.** A condition +// naming a variable the trigger does not declare is refused at SAVE, with the +// variable named, for the same reason §4.3 gives the template editor: a predicate +// that silently reads `undefined` is a rule that silently never fires (or always +// does), and the day you find out is the day the mail did not go. +// +// **The grammar is small and closed on purpose.** No arbitrary expressions, no +// arithmetic, no regex. An operator composes and/or/not over comparisons of one +// declared variable against a literal, and every operator here is one a rule +// editor can render as a dropdown. Anything that needs more than this is asking +// for a condition the module should have declared as a variable. +// +// Nothing in this file reaches the database or the network. + +const registries = require('../modules/registries') + +// Comparison operators, grouped by what they may be applied to. The grouping is +// the whole of the type check: `gt` on a boolean and `startsWith` on an int are +// both refused at save rather than quietly answering false forever. +const OPERATORS = { + eq: { label: 'is', types: ['string', 'int', 'float', 'boolean', 'datetime', 'url'], arity: 1 }, + ne: { label: 'is not', types: ['string', 'int', 'float', 'boolean', 'datetime', 'url'], arity: 1 }, + in: { label: 'is one of', types: ['string', 'int', 'float', 'url'], arity: 'list' }, + nin: { label: 'is none of', types: ['string', 'int', 'float', 'url'], arity: 'list' }, + gt: { label: 'is greater than', types: ['int', 'float', 'datetime'], arity: 1 }, + gte: { label: 'is at least', types: ['int', 'float', 'datetime'], arity: 1 }, + lt: { label: 'is less than', types: ['int', 'float', 'datetime'], arity: 1 }, + lte: { label: 'is at most', types: ['int', 'float', 'datetime'], arity: 1 }, + contains: { label: 'contains', types: ['string', 'url'], arity: 1 }, + startsWith: { label: 'starts with', types: ['string', 'url'], arity: 1 }, + // The one operator that takes no value: "the emit carried this variable at + // all". It is the honest way to write a rule about an OPTIONAL variable, and + // without it `ne` would have to double as a presence test and get it wrong + // (an absent variable is not "not equal to X"; it is absent). + present: { label: 'is present', types: ['string', 'int', 'float', 'boolean', 'datetime', 'url'], arity: 0 }, + absent: { label: 'is absent', types: ['string', 'int', 'float', 'boolean', 'datetime', 'url'], arity: 0 }, +} + +const BOOLEAN_OPS = ['and', 'or', 'not'] + +// A list literal an operator may type. Bounded because it is stored in a JSON +// column an admin can write, and an unbounded IN list is an unbounded predicate +// evaluated on every event. +const MAX_LIST = 50 +// Depth of the and/or/not tree. Three levels is more nesting than any rule +// editor should offer; the bound is here so a hand-written JSON body cannot +// recurse this evaluator into a stack overflow on the emit path. +const MAX_DEPTH = 5 + +const isPlainObject = (v) => v !== null && typeof v === 'object' && !Array.isArray(v) + +/** + * Check one literal against the declared type of the variable it is compared to. + * + * `datetime` accepts anything `Date` parses and is normalised to an ISO string, + * which is what `engagementEmit.coerce` does to the payload side — so both sides + * of every comparison are the same representation of a moment, and a lexical + * `<` on two ISO strings is a chronological one. + */ +function checkLiteral(type, raw) { + switch (type) { + case 'string': + case 'url': + return typeof raw === 'string' ? { value: raw } : { error: 'expected a string' } + case 'int': + return Number.isInteger(raw) ? { value: raw } : { error: 'expected an integer' } + case 'float': + return typeof raw === 'number' && Number.isFinite(raw) + ? { value: raw } + : { error: 'expected a finite number' } + case 'boolean': + return typeof raw === 'boolean' ? { value: raw } : { error: 'expected a boolean' } + case 'datetime': { + const d = raw instanceof Date ? raw : new Date(raw) + if (Number.isNaN(d.getTime())) return { error: 'expected a date' } + return { value: d.toISOString() } + } + default: + return { error: `unsupported type "${type}"` } + } +} + +/** + * Validate a condition tree against a trigger declaration. + * + * Returns `{ ok: true, conditions }` with a NEW normalised tree — literals + * coerced, unknown keys dropped — or `{ ok: false, errors }` listing every + * problem rather than the first, the posture `validatePayload` takes and for the + * same reason: an operator fixing one clause at a time is an operator making six + * round trips through a form. + * + * `null` and `undefined` are valid and mean "no conditions" — a rule that fires + * on every occurrence of its trigger, which is the common case. + */ +function validate(declaration, raw) { + const errors = [] + const variables = new Map((declaration?.variables || []).map((v) => [v.name, v])) + + function walk(node, depth, path) { + if (depth > MAX_DEPTH) { + errors.push(`${path}: nested deeper than ${MAX_DEPTH}`) + return null + } + if (!isPlainObject(node)) { + errors.push(`${path}: expected an object`) + return null + } + + if (BOOLEAN_OPS.includes(node.op)) { + // `not` takes exactly one node; `and`/`or` take a list. Both are written + // as `nodes` so a client walks one shape. + const raws = Array.isArray(node.nodes) ? node.nodes : [] + if (!raws.length) { + errors.push(`${path}: "${node.op}" has no nodes`) + return null + } + if (node.op === 'not' && raws.length !== 1) { + errors.push(`${path}: "not" takes exactly one node`) + return null + } + const nodes = raws.map((child, i) => walk(child, depth + 1, `${path}.nodes[${i}]`)).filter(Boolean) + return nodes.length === raws.length ? { op: node.op, nodes } : null + } + + if (node.op !== undefined) { + errors.push(`${path}: unknown operator "${node.op}"`) + return null + } + + // A leaf: { variable, cmp, value }. + const variable = variables.get(node.variable) + if (!variable) { + errors.push(`${path}: "${node.variable}" is not a variable of "${declaration?.id}"`) + return null + } + const operator = OPERATORS[node.cmp] + if (!operator) { + errors.push(`${path}: unknown comparison "${node.cmp}"`) + return null + } + if (!operator.types.includes(variable.type)) { + errors.push(`${path}: "${node.cmp}" cannot be applied to a ${variable.type}`) + return null + } + + if (operator.arity === 0) return { variable: variable.name, cmp: node.cmp } + + if (operator.arity === 'list') { + if (!Array.isArray(node.value) || !node.value.length) { + errors.push(`${path}: "${node.cmp}" needs a non-empty list`) + return null + } + if (node.value.length > MAX_LIST) { + errors.push(`${path}: "${node.cmp}" list is longer than ${MAX_LIST}`) + return null + } + const value = [] + let bad = false + node.value.forEach((item, i) => { + const checked = checkLiteral(variable.type, item) + if (checked.error) { + errors.push(`${path}.value[${i}]: ${checked.error}`) + bad = true + } else value.push(checked.value) + }) + return bad ? null : { variable: variable.name, cmp: node.cmp, value } + } + + const checked = checkLiteral(variable.type, node.value) + if (checked.error) { + errors.push(`${path}: ${checked.error}`) + return null + } + return { variable: variable.name, cmp: node.cmp, value: checked.value } + } + + if (raw === null || raw === undefined) return { ok: true, conditions: null } + const conditions = walk(raw, 0, 'conditions') + return errors.length ? { ok: false, errors } : { ok: true, conditions } +} + +/** Compare one already-normalised leaf against a payload. */ +function evaluateLeaf(leaf, data) { + const present = Object.prototype.hasOwnProperty.call(data, leaf.variable) + const actual = data[leaf.variable] + + if (leaf.cmp === 'present') return present + if (leaf.cmp === 'absent') return !present + // Every other comparison against an absent variable is FALSE, never true. + // `ne` is the one that tempts otherwise — "not equal to X" reads as satisfied + // by nothing at all — and treating it as true would make an optional variable's + // absence fire the rule. + if (!present) return false + + switch (leaf.cmp) { + case 'eq': return actual === leaf.value + case 'ne': return actual !== leaf.value + case 'in': return leaf.value.includes(actual) + case 'nin': return !leaf.value.includes(actual) + case 'gt': return actual > leaf.value + case 'gte': return actual >= leaf.value + case 'lt': return actual < leaf.value + case 'lte': return actual <= leaf.value + case 'contains': return typeof actual === 'string' && actual.includes(leaf.value) + case 'startsWith': return typeof actual === 'string' && actual.startsWith(leaf.value) + default: return false + } +} + +/** + * Does this event's payload satisfy the rule's conditions? + * + * `null` conditions are satisfied — a rule with no conditions fires on every + * occurrence. A tree this evaluator does not recognise answers **false**, which + * is the fail-closed direction: a stored condition that no longer parses (a rule + * saved against an older trigger version, say) must stop the mail rather than + * become "no conditions" and mail everyone. + */ +function evaluate(conditions, data = {}) { + if (conditions === null || conditions === undefined) return true + if (!isPlainObject(conditions)) return false + + if (conditions.op === 'and') return (conditions.nodes || []).every((n) => evaluate(n, data)) + if (conditions.op === 'or') return (conditions.nodes || []).some((n) => evaluate(n, data)) + if (conditions.op === 'not') return !evaluate((conditions.nodes || [])[0], data) + if (conditions.op !== undefined) return false + + return evaluateLeaf(conditions, data) +} + +/** + * The operator vocabulary a rule editor renders, with the variable types each + * one applies to. Served with the rule surface in Phase 4b rather than hardcoded + * in the client, on the same argument the ceiling vocabulary is served with the + * trigger catalog: a second copy of a rule is a copy that drifts. + */ +const vocabulary = () => + Object.entries(OPERATORS).map(([cmp, o]) => ({ cmp, label: o.label, types: o.types, arity: o.arity })) + +/** Convenience for a caller holding only a trigger id. */ +const validateFor = (triggerId, raw) => validate(registries.eventTrigger(triggerId), raw) + +module.exports = { validate, validateFor, evaluate, vocabulary, OPERATORS, MAX_LIST, MAX_DEPTH } diff --git a/server/src/engagement/engine.js b/server/src/engagement/engine.js new file mode 100644 index 0000000..82729d7 --- /dev/null +++ b/server/src/engagement/engine.js @@ -0,0 +1,241 @@ +// ── The engagement engine ────────────────────────────────────────────────── +// +// ENGAGEMENT.md Phase 4a. `ctx.events.emit` validated a payload against a +// declaration and stopped (Phase 2); this is what it now hands the validated +// event to. The engine's whole job is to answer, for one event, **who gets told, +// on what, and not too often** - and then to write that down as outbox rows. +// It never delivers: `engagementWorker` drains the outbox, and what actually +// carries a message arrives with the channels' `deliver` in Phases 6 and 7. +// +// **The order of the gates is the design, and each one is here because skipping +// it is a way to mail the wrong people or too many of them:** +// +// 1. enabled rules for this trigger - nothing is seeded, nothing is on by default +// 2. conditions - is this particular firing interesting +// 3. audience -> user ids - core's tables, or a composed segment +// 4. ceiling re-check (G24) - re-run at SEND time, not only at save +// 5. per-channel preference - a user's own opt-in, effective mode +// 6. per-rule hourly ceiling (§7.1 Q3) - the hard stop that makes rules-as-data safe +// 7. cooldown, per (rule, user, subject) - one statement, so two emits cannot race +// 8. enqueue, deduped - a replayed event is one row, not two +// +// Steps 6 and 7 are in that order deliberately. The hourly ceiling is about the +// RULE and is the thing that stops a mail storm; the cooldown is about one +// recipient and one subject. Checking the cheap global bound before consuming a +// per-recipient cooldown slot means a rule that has hit its ceiling does not also +// silently burn everybody's cooldowns on sends that never happen. +// +// **Nothing here throws at its caller.** It is invoked from inside a game-event +// handler by way of `ctx.events.emit`, and a database problem of core's must not +// become a module's control flow (the same posture the emit validator takes). + +const rulesDb = require('../model/engagement/engagementRules.db') +const outboxDb = require('../model/engagement/engagementOutbox.db') +const cooldownsDb = require('../model/engagement/engagementCooldowns.db') +const sendsDb = require('../model/engagement/engagementSends.db') +const recipients = require('../model/engagement/engagementRecipients.db') +const conditions = require('./conditions') +const audiences = require('./audiences') +const channels = require('./channels') +const log = require('../utils/logger')('engagement') + +const HOUR_MS = 60 * 60 * 1000 + +/** + * Which of a rule's channels are actually deliverable right now? + * + * A rule stores channel ids as data (`channels JSON`), so it can name one whose + * module has been removed since. An unregistered channel is dropped rather than + * failing the rule: the other channels of that rule are still correct, and a + * dropped one is visible in the log line below. + */ +const liveChannels = (rule) => (rule.channels || []).filter((c) => channels.has(c)) + +/** + * Narrow a candidate set to the users whose EFFECTIVE mode for (id, channel) is + * not 'off'. + * + * Effective, not stored: a row exists only where a user has expressed something, + * and absence means the channel's `defaultMode` (§3.1). Reading the stored rows + * and applying the default here keeps that answer in the registry, which is the + * invariant Phase 3 established. + * + * A 'digest' preference is kept, not dropped. Digest delivery is Phase 6's, and + * an outbox row for it is still the right record of "this person should be told"; + * what changes in Phase 6 is who drains it. + */ +async function subscribedTo(userIds, streamId, channel) { + if (!userIds.length) return [] + const stored = await recipients.storedModes(userIds, streamId, channel) + const fallback = channels.defaultMode(channel) + return userIds.filter((id) => (stored.get(id) ?? fallback) !== 'off') +} + +/** + * Run one rule against one event. Returns a small summary, for the log line and + * for tests; it is not read by the caller for control flow. + */ +async function applyRule(rule, event, now) { + const summary = { ruleId: rule.id, enqueued: 0, deduped: 0, cooled: 0, capped: 0, skipped: null } + + if (!conditions.evaluate(rule.conditions, event.data)) { + summary.skipped = 'conditions' + return summary + } + + const resolved = await audiences.resolveForRule(rule, event) + if (resolved.dormant) { + summary.skipped = resolved.reason || 'dormant' + return summary + } + if (!resolved.userIds.length) { + summary.skipped = resolved.reason || 'empty audience' + return summary + } + + // G24, re-run at send time. A rule saved when its trigger permitted a wider + // audience must not keep reaching it after a module upgrade narrowed the + // declaration - and that is the only way this can fail, since the save path + // ran the same check. + if (!audiences.permitted(event.triggerId, resolved.ceiling)) { + log.warn('rule audience exceeds its trigger ceiling - refusing', { + rule: rule.id, + trigger: event.triggerId, + audience: resolved.ceiling, + }) + summary.skipped = 'ceiling' + return summary + } + + const live = liveChannels(rule) + if (!live.length) { + summary.skipped = 'no registered channel' + return summary + } + + // The per-rule hourly ceiling (§7.1 Q3). Counted once for the whole event + // rather than per channel: an operator setting "100 an hour" means a hundred + // messages, not a hundred per channel per event. + const sentThisHour = await sendsDb.countSentSince(rule.id, new Date(now.getTime() - HOUR_MS)) + let budget = Math.max(0, rule.max_sends_per_hour - sentThisHour) + if (budget === 0) { + log.warn('rule is at its hourly send ceiling', { + rule: rule.id, + trigger: event.triggerId, + ceiling: rule.max_sends_per_hour, + }) + summary.skipped = 'hourly ceiling' + return summary + } + + const subjectKey = (event.subject ?? '').toString().slice(0, 190) + const dueAt = new Date(now.getTime() + Math.max(0, rule.delay_seconds) * 1000) + + for (const channel of live) { + const eligible = await subscribedTo(resolved.userIds, event.triggerId, channel) + for (const userId of eligible) { + if (budget <= 0) { + summary.capped += 1 + continue + } + // One statement, guarded on the interval, so two concurrent emits cannot + // both pass a read-then-write check (§4.1). + const allowed = await cooldownsDb.claim(rule.id, userId, subjectKey, rule.cooldown_seconds, now) + if (!allowed) { + summary.cooled += 1 + continue + } + const id = await outboxDb.enqueue({ + rule_id: rule.id, + trigger_id: event.triggerId, + user_id: userId, + channel, + subject_key: subjectKey, + payload: event.data, + // Scoped per (rule, user, channel) by the unique index, so one event + // fanned out to fifty people is fifty rows carrying the same key. + dedupe_key: event.dedupeKey, + due_at: dueAt, + }) + if (id === null) summary.deduped += 1 + else { + summary.enqueued += 1 + budget -= 1 + } + } + } + + return summary +} + +/** + * Cancel pending rows that this event resolves (§4.2a). + * + * This is the actual point of `delay_seconds`: without cancellation a delay is + * just a late mail. A house repaired back to LikeNew fires a trigger that some + * rule names in its `cancel_on`, and every still-scheduled row for that + * (rule, subject) stops. + * + * When the resolving event names an owner, only that user's rows are cancelled; + * when it does not, every user queued about that subject is - which is the + * house-repaired case, where the event is about the house and not about any one + * of the people who were going to be told. + */ +async function applyCancellations(event, summary) { + const rules = await rulesDb.enabledCancelledBy(event.triggerId) + if (!rules.length) return + const subjectKey = (event.subject ?? '').toString().slice(0, 190) + for (const rule of rules) { + const n = await outboxDb.cancel(rule.id, subjectKey, event.ownerUserId || null) + if (n) { + summary.cancelled += n + log.info('cancelled scheduled sends', { + rule: rule.id, + by: event.triggerId, + subject: subjectKey, + rows: n, + }) + } + } +} + +/** + * Dispatch one validated event. Called by `engagementEmit.emit` after the payload + * has been checked against the declaration. + * + * @param {object} event the envelope `engagementEmit` built + * @returns {Promise<{ rules: number, enqueued: number, cancelled: number }>} + */ +async function dispatch(event, now = new Date()) { + const summary = { rules: 0, enqueued: 0, deduped: 0, cooled: 0, capped: 0, cancelled: 0 } + try { + const rules = await rulesDb.enabledForTrigger(event.triggerId) + summary.rules = rules.length + + for (const rule of rules) { + const result = await applyRule(rule, event, now) + summary.enqueued += result.enqueued + summary.deduped += result.deduped + summary.cooled += result.cooled + summary.capped += result.capped + } + + await applyCancellations(event, summary) + + // Keys and counts, never values - the same rule the emit log line follows. + // A payload carries player names, house locations and forum excerpts, and a + // log that reproduces them is a second copy of exactly the content + // `engagement_sends` is careful to keep out of the database. + if (summary.rules || summary.cancelled) { + log.info('event dispatched', { trigger: event.triggerId, ...summary }) + } + } catch (err) { + // A database problem of core's must not become the module's control flow at + // three in the morning. The emit already succeeded as a contract; what failed + // is delivery, and it is logged as core's failure. + log.error('dispatch failed', { trigger: event.triggerId, message: err.message }) + } + return summary +} + +module.exports = { dispatch, applyRule, applyCancellations, subscribedTo, liveChannels, HOUR_MS } diff --git a/server/src/engagement/segments.js b/server/src/engagement/segments.js new file mode 100644 index 0000000..390044b --- /dev/null +++ b/server/src/engagement/segments.js @@ -0,0 +1,232 @@ +// ── Audience segments — operator composition over module-declared audiences ── +// +// ENGAGEMENT.md §5.1a, Phase 4a. A module declares named audiences over its own +// data ("members of a Team", "the governors"); an operator combines them with +// and/or/not into a saved segment; a rule points at the segment. This file is the +// two halves of that: derive the segment's ceiling at save time, and resolve the +// expression to user ids at send time. +// +// **Composition must NARROW, never widen** (§5.1a rule 3), and that is the whole +// security content of this file. `A OR B` takes the TIGHTER of the two ceilings, +// not the looser - a ceiling states what an expression is *allowed* to reach, not +// what it will resolve to, so the direction of the boolean operator is +// irrelevant. Union-widens is the intuitive implementation and it is the wrong +// one; `ceilings.meetAll` is the arithmetic, settled in Phase 2, and this is its +// first consumer. +// +// The second rule that shows up in both halves is **dormancy** (§5.1a rule 4). +// An audience whose module has been uninstalled resolves to the EMPTY set and +// flags itself, never to an error and never to some other set of people. A +// segment containing one is dormant, and a rule using a dormant segment does not +// send. Resolving the rest of the tree instead would mail a DIFFERENT population +// than the one the operator composed. + +const registries = require('../modules/registries') +const ceilings = require('../modules/ceilings') + +const BOOLEAN_OPS = ['and', 'or', 'not'] +// Same bounds and the same reason as conditions.js: this tree comes out of a JSON +// column an admin can write, and it is walked on the emit path. +const MAX_DEPTH = 5 +const MAX_NODES = 50 + +const isPlainObject = (v) => v !== null && typeof v === 'object' && !Array.isArray(v) +const isNot = (node) => isPlainObject(node) && node.op === 'not' + +/** Check one audience's declared params against what the operator supplied. */ +function checkParams(declaration, raw, path, errors) { + const params = {} + const supplied = isPlainObject(raw) ? raw : {} + for (const p of declaration.params || []) { + const value = supplied[p.id] + if (value === undefined || value === null || value === '') { + if (p.required) errors.push(`${path}: "${p.id}" is required`) + continue + } + if (p.type === 'int') { + const n = Number(value) + if (!Number.isInteger(n)) { + errors.push(`${path}: "${p.id}" expected an integer`) + continue + } + params[p.id] = n + } else if (p.type === 'boolean') { + if (typeof value !== 'boolean') { + errors.push(`${path}: "${p.id}" expected a boolean`) + continue + } + params[p.id] = value + } else { + if (typeof value !== 'string') { + errors.push(`${path}: "${p.id}" expected a string`) + continue + } + params[p.id] = value + } + } + return params +} + +/** + * Validate an expression and derive its ceiling in one walk. + * + * Returns `{ ok: true, expression, ceiling }` with a normalised tree, or + * `{ ok: false, errors }`. + * + * **`not` is legal only as a child of `and`**, and that restriction is what makes + * a complement mean something. A complement needs a universe, and the only + * universe available here that does not widen is the set its siblings already + * produced: `A AND NOT B` is "A, less B", which is exactly what an operator + * wants and cannot be composed into a broadcast. A bare `NOT B`, or `A OR NOT B`, + * would have to mean "everyone except..." - a way to build the whole deployment + * out of one narrow audience, which is the widening rule 3 forbids. Refusing it + * at save is better than a semantics nobody can predict from the screen. + * + * Two failure modes, and they are different: + * + * - a leaf naming an audience nobody registers is refused AT SAVE, because an + * operator composing a segment out of a typo should hear about it now rather + * than discovering a permanently-empty rule later. (A segment that was VALID + * when saved and whose module has since gone is a different case - that is + * dormancy, handled in `resolve`, and it is not refused.) + * - two incomparable ceilings have NO meet, so the composition is refused rather + * than resolved to a guess. `staff AND owner` is not `owner`; it is a question + * the lattice declines to answer, and picking a side would be a widening. + */ +function validate(raw) { + const errors = [] + let nodes = 0 + + // `underAnd` is the only context in which a `not` is legal. + function walk(node, depth, path, underAnd) { + if (++nodes > MAX_NODES) { + errors.push(`${path}: expression has more than ${MAX_NODES} nodes`) + return null + } + if (depth > MAX_DEPTH) { + errors.push(`${path}: nested deeper than ${MAX_DEPTH}`) + return null + } + if (!isPlainObject(node)) { + errors.push(`${path}: expected an object`) + return null + } + + if (node.op === 'not') { + if (!underAnd) { + errors.push(`${path}: "not" is only allowed inside an "and" - a complement needs a set to take it from`) + return null + } + const children = Array.isArray(node.nodes) ? node.nodes : [] + if (children.length !== 1) { + errors.push(`${path}: "not" takes exactly one node`) + return null + } + const inner = walk(children[0], depth + 1, `${path}.nodes[0]`, false) + if (!inner) return null + // A `not` contributes NO ceiling. Excluding people cannot widen who the + // expression reaches, so folding the excluded audience's ceiling into the + // meet would refuse perfectly safe segments: `members AND NOT staff` would + // hit meet('members','staff') = null and be rejected, even though it + // reaches strictly fewer people than `members` alone. + return { node: { op: 'not', nodes: [inner.node] }, ceiling: null, complement: true } + } + + if (node.op === 'and' || node.op === 'or') { + const children = Array.isArray(node.nodes) ? node.nodes : [] + if (!children.length) { + errors.push(`${path}: "${node.op}" has no nodes`) + return null + } + const walked = children.map((c, i) => walk(c, depth + 1, `${path}.nodes[${i}]`, node.op === 'and')) + if (walked.some((w) => w === null)) return null + const positives = walked.filter((w) => !w.complement) + if (!positives.length) { + errors.push(`${path}: "${node.op}" has nothing but complements - there is no set to exclude from`) + return null + } + return { + node: { op: node.op, nodes: walked.map((w) => w.node) }, + ceiling: ceilings.meetAll(positives.map((w) => w.ceiling)), + } + } + + if (node.op !== undefined) { + errors.push(`${path}: unknown operator "${node.op}"`) + return null + } + + const declaration = registries.audience(node.audienceId) + if (!declaration) { + errors.push(`${path}: no audience "${node.audienceId}" is registered`) + return null + } + const params = checkParams(declaration, node.params, path, errors) + return { node: { audienceId: declaration.id, params }, ceiling: declaration.ceiling } + } + + if (!isPlainObject(raw)) return { ok: false, errors: ['expression: expected an object'] } + const walked = walk(raw, 0, 'expression', false) + if (errors.length || !walked) return { ok: false, errors: errors.length ? errors : ['expression: invalid'] } + if (!walked.ceiling) { + return { + ok: false, + errors: [ + 'expression: the audiences combined here have no common ceiling, so there is no bound this segment could be given', + ], + } + } + return { ok: true, expression: walked.node, ceiling: walked.ceiling } +} + +/** + * Resolve a validated expression to a set of user ids. + * + * Returns `{ dormant, userIds }`. `dormant` is true the moment ANY leaf names an + * audience that is no longer registered, and when it is true the caller must not + * send: `userIds` is empty, because the tree it would have come from is not the + * tree the operator composed. + * + * `and` is the intersection of its positive children, less the union of its + * complements. `or` is the union of its children, which are all positive because + * `validate` refused any other shape. + */ +async function resolve(expression) { + let dormant = false + + async function walk(node) { + if (!isPlainObject(node)) return new Set() + + if (node.op === 'and' || node.op === 'or') { + const children = Array.isArray(node.nodes) ? node.nodes : [] + const positives = children.filter((c) => !isNot(c)) + const complements = children.filter(isNot) + + let out = new Set() + for (let i = 0; i < positives.length; i += 1) { + const set = await walk(positives[i]) + if (i === 0) out = set + else if (node.op === 'and') out = new Set([...out].filter((id) => set.has(id))) + else for (const id of set) out.add(id) + } + for (const c of complements) { + const excluded = await walk((c.nodes || [])[0]) + out = new Set([...out].filter((id) => !excluded.has(id))) + } + return out + } + + // A `not` reached directly (never produced by validate, but a stored row + // predates nothing and this must not throw): no universe, so no members. + if (node.op !== undefined) return new Set() + + const { dormant: gone, userIds } = await registries.resolveAudience(node.audienceId, node.params || {}) + if (gone) dormant = true + return new Set(userIds) + } + + const set = await walk(expression) + return { dormant, userIds: dormant ? [] : [...set] } +} + +module.exports = { validate, resolve, MAX_DEPTH, MAX_NODES } diff --git a/server/src/model/engagement/engagementCooldowns.db.js b/server/src/model/engagement/engagementCooldowns.db.js new file mode 100644 index 0000000..cead939 --- /dev/null +++ b/server/src/model/engagement/engagementCooldowns.db.js @@ -0,0 +1,78 @@ +const { query } = require('../../utils/db') + +/** + * Claim a fire for (rule, user, subject), or refuse it because the pair is still + * cooling. ENGAGEMENT.md §4.1. + * + * **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 + * describes, because that version was written, tested green against an in-memory + * stub, and disproved by the first run against a real MariaDB. + * + * The one-statement form reads its answer out of `affectedRows`, on the usual + * contract: 1 for an insert, 2 for an update that changed something, and 0 for a + * duplicate key whose update changed nothing - that 0 being "the guard failed, so + * this pair is still cooling". **The mariadb Node connector sets `foundRows: true` + * by default**, which makes `affectedRows` report rows MATCHED rather than rows + * CHANGED, and `utils/db.js` does not override it. Under that pool the no-op case + * returns 1, indistinguishable from a fresh insert: every cooldown would have + * passed, always, and nothing in a stubbed test could have noticed. + * + * So the guard moves into a WHERE clause, where a row either matches or does not + * and `foundRows` has nothing to fold together: + * + * 1. UPDATE the row, guarded on the interval. `affectedRows = 1` means this + * caller moved it and owns the fire. + * 2. If that matched nothing, the row either does not exist yet or is still + * cooling. `INSERT IGNORE` separates the two: 1 means we inserted the first + * fire, 0 means the row was there and step 1 already said it is cooling. + * + * It is still race-free, and each race resolves the right way: + * - two concurrent first fires: neither UPDATEs, both INSERT IGNORE, exactly + * one gets 1 (the primary key decides). The loser is treated as cooling. + * - two concurrent fires after expiry: the row is locked by the first UPDATE, + * and the second re-evaluates its guard against the committed row - which now + * holds `now`, so it fails and is refused. + * + * `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. + */ +async function claim(ruleId, userId, subjectKey, cooldownSeconds, now = new Date()) { + const moved = await query( + `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`, + [now, ruleId, userId, subjectKey, now, cooldownSeconds], + ) + if (Number(moved?.affectedRows || 0) === 1) return true + + const inserted = await query( + `INSERT IGNORE INTO engagement_cooldowns (rule_id, user_id, subject_key, last_fired_at, fire_count) + VALUES (?, ?, ?, ?, 1)`, + [ruleId, userId, subjectKey, now], + ) + return Number(inserted?.affectedRows || 0) === 1 +} + +const get = async (ruleId, userId, subjectKey) => { + const [row] = await query( + 'SELECT * FROM engagement_cooldowns WHERE rule_id = ? AND user_id = ? AND subject_key = ?', + [ruleId, userId, subjectKey], + ) + return row || null +} + +/** + * Drop cooldown rows older than `olderThan`. + * + * `idx_engc_sweep (last_fired_at)` exists for this: the table is written on every + * fire and read once per fire, so without a prune it is the unbounded-growth + * failure `teamActivityPrune` was written for. A dropped row means the next fire + * is treated as a first fire, which is correct as long as the retention window is + * longer than the longest configured cooldown - the caller's job, not this one's. + */ +const prune = (olderThan) => + query('DELETE FROM engagement_cooldowns WHERE last_fired_at < ?', [olderThan]) + +module.exports = { claim, get, prune } diff --git a/server/src/model/engagement/engagementOutbox.db.js b/server/src/model/engagement/engagementOutbox.db.js new file mode 100644 index 0000000..06aef38 --- /dev/null +++ b/server/src/model/engagement/engagementOutbox.db.js @@ -0,0 +1,158 @@ +const { query } = require('../../utils/db') +const { parseJson } = require('./engagementRules.db') + +const hydrate = (row) => row && { ...row, payload: parseJson(row.payload, {}) } + +/** + * Enqueue one (rule, user, channel) row, idempotently. + * + * `INSERT IGNORE` rather than a plain INSERT, because `uq_engo_dedupe` is the + * replay guard (§4.2a): the sidecar feed is at-least-once and a reconnect + * backfills, so the same event arriving twice must produce one row and not two + * mails. IGNORE turns that into a silent no-op, which is what a replay should be. + * + * Returns the new id, or null when the row already existed. A null is a + * SUCCESSFUL duplicate, not a failure - the caller counts it as such. + * + * A NULL dedupe_key never collides (multiple NULLs are legal under a UNIQUE + * index), so an emit that carries no key always enqueues. That is the right + * default: dedupe is something the emitter opts into by naming a key, and core + * cannot invent one that means anything. + */ +async function enqueue(row) { + const result = await query( + `INSERT IGNORE INTO engagement_outbox + (rule_id, trigger_id, user_id, channel, subject_key, payload, dedupe_key, due_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + [ + row.rule_id, + row.trigger_id, + row.user_id, + row.channel, + row.subject_key || '', + JSON.stringify(row.payload || {}), + row.dedupe_key ?? null, + row.due_at, + ], + ) + return Number(result?.affectedRows || 0) === 1 ? result.insertId : null +} + +/** + * Rows that are due. `idx_engo_due (status, due_at)` is this query. + * + * It selects rather than claims - claiming is `claim()` below, one row at a + * time - so two instances sweeping at once both see the same candidates and then + * disagree, harmlessly, about which of them owns each. + */ +const findDue = async (now, limit = 100) => + ( + await query( + "SELECT * FROM engagement_outbox WHERE status = 'scheduled' AND due_at <= ? ORDER BY due_at, id LIMIT ?", + [now, limit], + ) + ).map(hydrate) + +/** + * Take ownership of one due row: a compare-and-set from 'scheduled' to 'sending'. + * + * **This is §7.1 Q2's answer** (settled by the org lead 2026-08-29, over + * `SELECT ... FOR UPDATE SKIP LOCKED`). The winner is whoever the server reports + * `affectedRows = 1` to; every other sweeper gets 0 and moves on. No explicit + * transaction, no MariaDB version floor, and it uses a status the ENUM already + * carried for exactly this. + * + * What it makes safe is the OUTBOX and only the outbox. `announceWorker`, + * `teamDigestWorker`, `teamForumUploadSweep` and `teamActivityPrune` are all + * still written for a single instance, so this does not make the deployment + * multi-instance - it makes the one table that will carry mail ready for the day + * it is, which is cheap now and expensive after mail has doubled once. + */ +async function claim(id) { + const result = await query( + `UPDATE engagement_outbox + SET status = 'sending', attempts = attempts + 1 + WHERE id = ? AND status = 'scheduled'`, + [id], + ) + return Number(result?.affectedRows || 0) === 1 +} + +/** + * Release a claimed row back to 'scheduled' with a later `due_at` - a transient + * failure that should be retried. The mirror of announceJobs' backoff. + */ +const reschedule = (id, dueAt, error) => + query( + "UPDATE engagement_outbox SET status = 'scheduled', due_at = ?, last_error = ? WHERE id = ? AND status = 'sending'", + [dueAt, error ? String(error).slice(0, 2000) : null, id], + ) + +/** A terminal outcome: 'sent', 'failed' or 'suppressed'. */ +const finish = (id, status, error) => + query( + `UPDATE engagement_outbox + SET status = ?, last_error = ?, sent_at = IF(? = 'sent', NOW(), sent_at) + WHERE id = ?`, + [status, error ? String(error).slice(0, 2000) : null, status, id], + ) + +/** + * Cancel every still-scheduled row for a (rule, subject) - the point of the + * grace window (§4.2a). `userId` narrows it to one recipient when the resolving + * event names one; a resolving event with no owner cancels for everyone the + * original event was queued for, which is the house-repaired case. + * + * Only 'scheduled' rows are touched: a row already claimed into 'sending' is + * somebody's in-flight send and cancelling it would leave two workers writing + * one row's outcome. + */ +async function cancel(ruleId, subjectKey, userId = null) { + const params = [ruleId, subjectKey] + let sql = "UPDATE engagement_outbox SET status = 'cancelled' WHERE rule_id = ? AND subject_key = ? AND status = 'scheduled'" + if (userId !== null && userId !== undefined) { + sql += ' AND user_id = ?' + params.push(userId) + } + const result = await query(sql, params) + return Number(result?.affectedRows || 0) +} + +/** + * Recover rows stranded in 'sending' by a crash between the claim and the + * outcome. + * + * Without this the CAS claim leaks: the claiming process died, no other sweeper + * will ever match `status = 'scheduled'`, and the row sits in 'sending' forever. + * `updated_at` is the clock (it is ON UPDATE CURRENT_TIMESTAMP, so the claim + * stamped it), and the window has to be comfortably longer than the slowest + * legitimate send or this reclaims rows that are merely slow. + */ +const reclaimStale = (before) => + query( + "UPDATE engagement_outbox SET status = 'scheduled' WHERE status = 'sending' AND updated_at < ?", + [before], + ) + +const getById = async (id) => { + const [row] = await query('SELECT * FROM engagement_outbox WHERE id = ?', [id]) + return hydrate(row) +} + +/** Admin/read surfaces (Phase 4b) and tests. */ +const listForRule = async (ruleId, limit = 100) => + ( + await query('SELECT * FROM engagement_outbox WHERE rule_id = ? ORDER BY id DESC LIMIT ?', [ruleId, limit]) + ).map(hydrate) + +module.exports = { + enqueue, + findDue, + claim, + reschedule, + finish, + cancel, + reclaimStale, + getById, + listForRule, +} diff --git a/server/src/model/engagement/engagementRecipients.db.js b/server/src/model/engagement/engagementRecipients.db.js new file mode 100644 index 0000000..3b85f80 --- /dev/null +++ b/server/src/model/engagement/engagementRecipients.db.js @@ -0,0 +1,125 @@ +const { query } = require('../../utils/db') + +// A bound on every "resolve an audience" query. `authenticated` on a large +// deployment is the whole user table, and the engine turns each id into an +// outbox row - so the read that feeds it has to have a ceiling of its own. The +// per-rule hourly cap (§7.1 Q3) is the operator-facing limit; this is the one +// that keeps a single emit from loading a hundred thousand rows into memory. +const MAX_AUDIENCE = 5000 + +const ids = (rows) => rows.map((r) => Number(r.id)).filter((n) => Number.isInteger(n) && n > 0) + +const marks = (list) => list.map(() => '?').join(', ') + +/** + * Every active user. The `authenticated` audience - and `everyone`, which has no + * distinct meaning here: a signed-out visitor has no address, no device and no + * inbox, so the widest set the engine can actually deliver to is this one. The + * ceiling lattice still distinguishes them (a trigger ceilinged `everyone` + * permits an `authenticated` rule and not the reverse); only the resolution + * coincides. + * + * `status = 'active'` on every query in this file: a banned or disabled account + * is refused at login, and mailing it engagement content would be the one + * surface that did not get the message. + */ +const active = async (limit = MAX_AUDIENCE) => + ids(await query("SELECT id FROM users WHERE status = 'active' ORDER BY id LIMIT ?", [limit])) + +/** The `staff` audience. Roles come from `ceilings.STAFF_CEILING_ROLES`. */ +const staff = async (roles, limit = MAX_AUDIENCE) => { + if (!roles.length) return [] + return ids( + await query( + `SELECT id FROM users WHERE status = 'active' AND role IN (${marks(roles)}) ORDER BY id LIMIT ?`, + [...roles, limit], + ), + ) +} + +/** + * The `subscribers` audience: active users who have opted into this id on at + * least one channel. + * + * "Opted in" is the EFFECTIVE mode, not the stored one, and that is why this is + * not simply `WHERE mode <> 'off'`. A row exists only where a user said + * something; absence means the channel's `defaultMode` (§3.1). All three of + * core's channels default 'off' today, so the second half of the WHERE matches + * nobody - but writing it means the day a channel ships with a non-off default, + * this audience is already right rather than silently excluding everyone who + * never opened the preferences screen. + * + * `defaultOnChannels` is the caller's list of channels whose defaultMode is not + * 'off'; it comes from the channel registry, so the default lives in exactly one + * place here too. + */ +const subscribers = async (streamId, defaultOnChannels = [], limit = MAX_AUDIENCE) => { + const optedIn = `EXISTS ( + SELECT 1 FROM notification_channel_prefs p + WHERE p.user_id = u.id AND p.stream_id = ? AND p.mode <> 'off')` + + if (!defaultOnChannels.length) { + return ids( + await query( + `SELECT u.id FROM users u WHERE u.status = 'active' AND ${optedIn} ORDER BY u.id LIMIT ?`, + [streamId, limit], + ), + ) + } + + // "At least one default-on channel has no row for this user" - counted rather + // than NOT EXISTS, because NOT EXISTS would mean "none of them has a row". + const defaulted = `( + SELECT COUNT(*) FROM notification_channel_prefs p2 + WHERE p2.user_id = u.id AND p2.stream_id = ? AND p2.channel IN (${marks(defaultOnChannels)}) + ) < ?` + + return ids( + await query( + `SELECT u.id FROM users u + WHERE u.status = 'active' AND (${optedIn} OR ${defaulted}) + ORDER BY u.id LIMIT ?`, + [streamId, streamId, ...defaultOnChannels, defaultOnChannels.length, limit], + ), + ) +} + +/** + * Narrow a set of user ids to the active ones. + * + * Every audience that does NOT come from a query in this file goes through here: + * `owner` is a single id off the event envelope, and a module-declared audience + * (§5.1a) is a list of ids a module's own resolver produced. Neither has any + * notion of account status, and a module must not be able to mail a banned + * account by returning its id. + */ +const filterActive = async (userIds) => { + const wanted = [...new Set(userIds.map(Number).filter((n) => Number.isInteger(n) && n > 0))] + if (!wanted.length) return [] + const capped = wanted.slice(0, MAX_AUDIENCE) + return ids( + await query( + `SELECT id FROM users WHERE status = 'active' AND id IN (${marks(capped)}) ORDER BY id`, + capped, + ), + ) +} + +/** + * The stored mode for one (id, channel) across a set of users, as a Map. + * + * The caller applies the channel's `defaultMode` to anyone missing from the map, + * which keeps the defaulting in the one place §3.1 put it. Returning stored rows + * rather than a decision is what makes that possible. + */ +const storedModes = async (userIds, streamId, channel) => { + if (!userIds.length) return new Map() + const rows = await query( + `SELECT user_id, mode FROM notification_channel_prefs + WHERE stream_id = ? AND channel = ? AND user_id IN (${marks(userIds)})`, + [streamId, channel, ...userIds], + ) + return new Map(rows.map((r) => [Number(r.user_id), r.mode])) +} + +module.exports = { active, staff, subscribers, filterActive, storedModes, MAX_AUDIENCE } diff --git a/server/src/model/engagement/engagementRules.db.js b/server/src/model/engagement/engagementRules.db.js new file mode 100644 index 0000000..f003f96 --- /dev/null +++ b/server/src/model/engagement/engagementRules.db.js @@ -0,0 +1,127 @@ +const { query } = require('../../utils/db') + +// JSON columns come back from the driver already parsed on some MariaDB/driver +// combinations and as a string on others (it depends on whether the column is a +// real JSON type or the LONGTEXT + CHECK alias MariaDB implements it as). Every +// read below goes through this, so no caller has to know which it got. +function parseJson(value, fallback) { + if (value === null || value === undefined) return fallback + if (typeof value !== 'string') return value + try { + return JSON.parse(value) + } catch { + return fallback + } +} + +const hydrate = (row) => + row && { + ...row, + enabled: Boolean(row.enabled), + channels: parseJson(row.channels, []), + template_keys: parseJson(row.template_keys, {}), + conditions: parseJson(row.conditions, null), + cancel_on: parseJson(row.cancel_on, []), + } + +const list = async () => + (await query('SELECT * FROM engagement_rules ORDER BY trigger_id, name, id')).map(hydrate) + +const getById = async (id) => { + const [row] = await query('SELECT * FROM engagement_rules WHERE id = ?', [id]) + return hydrate(row) +} + +/** + * Every ENABLED rule for one trigger. The engine's hot path: one indexed read + * per emit, and `idx_engr_trigger (trigger_id, enabled)` is exactly this query. + */ +const enabledForTrigger = async (triggerId) => + (await query('SELECT * FROM engagement_rules WHERE trigger_id = ? AND enabled = 1', [triggerId])).map(hydrate) + +/** + * Every enabled rule that names `triggerId` in its `cancel_on`. + * + * A JSON_CONTAINS rather than a scan: `cancel_on` is a small array on a small + * table, but this runs on EVERY emit — including the overwhelming majority that + * cancel nothing — so it must not be a full table read of the rule set. + */ +const enabledCancelledBy = async (triggerId) => + ( + await query( + "SELECT * FROM engagement_rules WHERE enabled = 1 AND cancel_on IS NOT NULL AND JSON_CONTAINS(cancel_on, JSON_QUOTE(?))", + [triggerId], + ) + ).map(hydrate) + +const insert = async (rule) => { + const result = await query( + `INSERT INTO engagement_rules + (trigger_id, name, enabled, audience, audience_segment_id, max_sends_per_hour, + channels, template_keys, conditions, cooldown_seconds, delay_seconds, cancel_on, updated_by) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + [ + rule.trigger_id, + rule.name, + rule.enabled ? 1 : 0, + rule.audience, + rule.audience_segment_id, + rule.max_sends_per_hour, + JSON.stringify(rule.channels), + JSON.stringify(rule.template_keys), + rule.conditions === null ? null : JSON.stringify(rule.conditions), + rule.cooldown_seconds, + rule.delay_seconds, + JSON.stringify(rule.cancel_on || []), + rule.updated_by, + ], + ) + return result.insertId +} + +const update = (id, rule) => + query( + `UPDATE engagement_rules + SET name = ?, enabled = ?, audience = ?, audience_segment_id = ?, max_sends_per_hour = ?, + channels = ?, template_keys = ?, conditions = ?, cooldown_seconds = ?, + delay_seconds = ?, cancel_on = ?, updated_by = ? + WHERE id = ?`, + [ + rule.name, + rule.enabled ? 1 : 0, + rule.audience, + rule.audience_segment_id, + rule.max_sends_per_hour, + JSON.stringify(rule.channels), + JSON.stringify(rule.template_keys), + rule.conditions === null ? null : JSON.stringify(rule.conditions), + rule.cooldown_seconds, + rule.delay_seconds, + JSON.stringify(rule.cancel_on || []), + rule.updated_by, + id, + ], + ) + +const remove = (id) => query('DELETE FROM engagement_rules WHERE id = ?', [id]) + +/** Does any rule still point at this segment? The check before a segment delete. */ +const countUsingSegment = async (segmentId) => { + const [row] = await query( + 'SELECT COUNT(*) AS n FROM engagement_rules WHERE audience_segment_id = ?', + [segmentId], + ) + return Number(row?.n || 0) +} + +module.exports = { + list, + getById, + enabledForTrigger, + enabledCancelledBy, + insert, + update, + remove, + countUsingSegment, + parseJson, +} diff --git a/server/src/model/engagement/engagementRules.model.js b/server/src/model/engagement/engagementRules.model.js new file mode 100644 index 0000000..51835e9 --- /dev/null +++ b/server/src/model/engagement/engagementRules.model.js @@ -0,0 +1,225 @@ +// ── Engagement rules — the save path ─────────────────────────────────────── +// +// ENGAGEMENT.md §4.5 / §7.1 Q3, Phase 4a. A rule is **operator-editable data**, +// not code, and that was a deliberate choice with a condition attached: it is +// safe to choose only because `enabled` defaults to 0 and every rule carries a +// hard per-hour send ceiling. Both of those live in this file's validation, not +// in the screen that calls it - Phase 4b builds a form over this, and a rule that +// arrives by any other route (a restore, a fixture, a future import) gets the +// same answer. +// +// **Every check here is a boundary, not a convenience.** The rule editor will +// re-implement some of them for the sake of a good error message, and that +// second copy is expected to drift - so this one is the one that decides. +// +// The check with teeth is the ceiling (G24): an operator may narrow a rule's +// audience as much as they like and may never widen it past what the trigger +// declared. `ceilings.permits` is that arithmetic, `segments.validate` derives +// it for a composed audience, and the engine re-runs the same check at SEND +// time in case a module upgrade narrowed the declaration underneath a saved rule. + +const db = require('./engagementRules.db') +const segmentsDb = require('./engagementSegments.db') +const registries = require('../../modules/registries') +const ceilings = require('../../modules/ceilings') +const channels = require('../../engagement/channels') +const conditions = require('../../engagement/conditions') + +// A day. Longer than this and "cooldown" is really "send once", which a rule +// expresses by being disabled rather than by a decade-long interval. +const MAX_COOLDOWN_SECONDS = 86_400 +// The grace window (§4.2a). A delay longer than a day outlives the thing it is +// about - and, more practically, a queue row that sits for a week is a row whose +// payload no longer describes the world. +const MAX_DELAY_SECONDS = 86_400 +// The upper bound on the operator-set hourly ceiling. It is not "unlimited by +// another name": the number exists so that a misconfiguration is a bad hour +// rather than an unbounded one, and a ceiling nobody can raise past a bound is +// what makes rules-as-data safe (§7.1 Q3). +const MAX_SENDS_PER_HOUR = 10_000 + +const isPlainObject = (v) => v !== null && typeof v === 'object' && !Array.isArray(v) + +/** + * Validate a rule against the registries and the lattice. + * + * Returns `{ ok: true, rule }` with a normalised row ready for insert/update, or + * `{ ok: false, errors }` listing every problem. + * + * `triggerId` may name a trigger nobody currently registers ONLY on an update of + * an existing rule - a dormant rule must stay editable (its module can come + * back), and refusing to save it would make an uninstall destructive after the + * fact. A NEW rule must name a live trigger, because there is nothing to + * preserve and a typo should be caught now. + */ +async function validate(input, { existing = null } = {}) { + const errors = [] + const raw = isPlainObject(input) ? input : {} + + const triggerId = typeof raw.triggerId === 'string' ? raw.triggerId : existing?.trigger_id + const declaration = triggerId ? registries.eventTrigger(triggerId) : null + if (!triggerId) errors.push('triggerId is required') + else if (!declaration && !existing) errors.push(`no trigger "${triggerId}" is registered`) + + const name = typeof raw.name === 'string' ? raw.name.trim() : '' + if (!name) errors.push('name is required') + else if (name.length > 160) errors.push('name is longer than 160 characters') + + // Channels are stored as data and checked against the registry, so a rule + // cannot name a sink that does not exist. Phase 4b's form offers the registered + // set; this is what makes that an affordance rather than the rule. + const wanted = Array.isArray(raw.channels) ? [...new Set(raw.channels)] : [] + if (!wanted.length) errors.push('at least one channel is required') + for (const c of wanted) if (!channels.has(c)) errors.push(`no channel "${c}" is registered`) + + // `template_keys` is { channel: templateKey }. Phase 5 owns templates, so the + // KEYS are checked for shape and not for existence - a rule may legitimately + // name a template that has not been authored yet, and Phase 5's editor is where + // that becomes resolvable. + const templateKeys = {} + if (raw.templateKeys !== undefined && !isPlainObject(raw.templateKeys)) { + errors.push('templateKeys must be an object of { channel: templateKey }') + } else { + for (const [channel, key] of Object.entries(raw.templateKeys || {})) { + if (!wanted.includes(channel)) { + errors.push(`templateKeys names "${channel}", which is not one of this rule's channels`) + continue + } + if (typeof key !== 'string' || !/^[a-z0-9][a-z0-9-]{0,63}$/.test(key)) { + errors.push(`templateKeys.${channel} is not a valid template key`) + continue + } + templateKeys[channel] = key + } + } + + const numbers = [ + ['cooldownSeconds', 'cooldown_seconds', MAX_COOLDOWN_SECONDS, 0], + ['delaySeconds', 'delay_seconds', MAX_DELAY_SECONDS, 0], + ['maxSendsPerHour', 'max_sends_per_hour', MAX_SENDS_PER_HOUR, 1], + ] + const scalars = {} + for (const [key, column, max, min] of numbers) { + const supplied = raw[key] + const fallback = existing ? existing[column] : column === 'max_sends_per_hour' ? 100 : 0 + const value = supplied === undefined || supplied === null ? fallback : Number(supplied) + if (!Number.isInteger(value) || value < min || value > max) { + errors.push(`${key} must be an integer between ${min} and ${max}`) + } else scalars[column] = value + } + + // `cancel_on` names trigger ids, and they are NOT checked for registration for + // the dormancy reason (§7.3): a resolving event whose module is temporarily + // absent should stop cancelling, not make the rule unsaveable. + const cancelOn = Array.isArray(raw.cancelOn) ? [...new Set(raw.cancelOn.filter((t) => typeof t === 'string'))] : [] + if (cancelOn.length && !scalars.delay_seconds) { + // Not an error - it is a rule that will never cancel anything, because there + // is no window in which to do it. Worth saying out loud rather than silently + // accepting a setting that cannot take effect. + errors.push('cancelOn has no effect without a delaySeconds grace window') + } + + const checked = conditions.validate(declaration, raw.conditions === undefined ? existing?.conditions : raw.conditions) + if (!checked.ok) errors.push(...checked.errors) + + // ── The audience, and the one check that is a security boundary ────────── + let audience = typeof raw.audience === 'string' ? raw.audience : existing?.audience || declaration?.audience + let segmentId = raw.audienceSegmentId === undefined ? existing?.audience_segment_id ?? null : raw.audienceSegmentId + segmentId = segmentId === null || segmentId === '' ? null : Number(segmentId) + + let effectiveCeiling = null + if (segmentId !== null) { + if (!Number.isInteger(segmentId)) errors.push('audienceSegmentId must be an integer') + else { + const segment = await segmentsDb.getById(segmentId) + if (!segment) errors.push(`no audience segment ${segmentId} exists`) + else { + // The segment's STORED ceiling, derived when it was saved by + // `segments.validate` from the narrowest audience it contains. A rule + // pointing at a segment takes that as its reach; the `audience` column + // is retained for display and is not what the engine resolves. + effectiveCeiling = segment.ceiling + audience = segment.ceiling + } + } + } else if (!ceilings.isCeiling(audience)) { + errors.push(`audience must be one of ${ceilings.CEILINGS.join(', ')}`) + } else { + effectiveCeiling = audience + } + + if (declaration && effectiveCeiling && !ceilings.permits(declaration.ceiling, effectiveCeiling)) { + errors.push( + `audience "${effectiveCeiling}" is wider than trigger "${triggerId}" permits (ceiling "${declaration.ceiling}")`, + ) + } + + if (errors.length) return { ok: false, errors } + + return { + ok: true, + rule: { + trigger_id: triggerId, + name, + enabled: raw.enabled === undefined ? Boolean(existing?.enabled) : Boolean(raw.enabled), + audience, + audience_segment_id: segmentId, + max_sends_per_hour: scalars.max_sends_per_hour, + channels: wanted, + template_keys: templateKeys, + conditions: checked.conditions, + cooldown_seconds: scalars.cooldown_seconds, + delay_seconds: scalars.delay_seconds, + cancel_on: cancelOn, + updated_by: Number.isInteger(raw.updatedBy) ? raw.updatedBy : null, + }, + } +} + +async function create(input) { + const checked = await validate(input) + if (!checked.ok) return checked + const id = await db.insert(checked.rule) + return { ok: true, rule: await db.getById(id) } +} + +async function update(id, input) { + const existing = await db.getById(id) + if (!existing) return { ok: false, errors: [`no rule ${id} exists`], notFound: true } + const checked = await validate(input, { existing }) + if (!checked.ok) return checked + await db.update(id, checked.rule) + return { ok: true, rule: await db.getById(id) } +} + +/** + * List every rule, each annotated with whether it can currently fire. + * + * Dormancy is computed rather than stored (§7.3): a rule whose trigger or + * segment is not registered right now is listed, flagged, and left alone. The + * alternative - deleting or disabling it on uninstall - destroys an operator's + * configuration on the strength of a module being temporarily absent. + */ +async function listAnnotated() { + const rows = await db.list() + const segments = new Map((await segmentsDb.list()).map((s) => [s.id, s])) + return rows.map((rule) => { + const reasons = [] + if (!registries.eventTrigger(rule.trigger_id)) reasons.push(`trigger "${rule.trigger_id}" is not registered`) + if (rule.audience_segment_id && !segments.has(rule.audience_segment_id)) { + reasons.push('its audience segment no longer exists') + } + for (const c of rule.channels || []) if (!channels.has(c)) reasons.push(`channel "${c}" is not registered`) + return { ...rule, dormant: reasons.length > 0, dormantReasons: reasons } + }) +} + +module.exports = { + validate, + create, + update, + listAnnotated, + MAX_COOLDOWN_SECONDS, + MAX_DELAY_SECONDS, + MAX_SENDS_PER_HOUR, +} diff --git a/server/src/model/engagement/engagementSegments.db.js b/server/src/model/engagement/engagementSegments.db.js new file mode 100644 index 0000000..c8fee62 --- /dev/null +++ b/server/src/model/engagement/engagementSegments.db.js @@ -0,0 +1,37 @@ +const { query } = require('../../utils/db') +const { parseJson } = require('./engagementRules.db') + +const hydrate = (row) => row && { ...row, expression: parseJson(row.expression, null) } + +const list = async () => + (await query('SELECT * FROM engagement_audience_segments ORDER BY name, id')).map(hydrate) + +const getById = async (id) => { + const [row] = await query('SELECT * FROM engagement_audience_segments WHERE id = ?', [id]) + return hydrate(row) +} + +/** + * `ceiling` is written by the caller from `segments.deriveCeiling`, never taken + * from an operator. It is a stored column rather than a runtime computation so + * an audit can read what a rule was ALLOWED to reach without re-resolving it, + * and so a module that later widens its own audience's ceiling cannot + * retroactively widen a segment that was saved under the old one. + */ +const insert = async (segment) => { + const result = await query( + 'INSERT INTO engagement_audience_segments (name, expression, ceiling, updated_by) VALUES (?, ?, ?, ?)', + [segment.name, JSON.stringify(segment.expression), segment.ceiling, segment.updated_by ?? null], + ) + return result.insertId +} + +const update = (id, segment) => + query( + 'UPDATE engagement_audience_segments SET name = ?, expression = ?, ceiling = ?, updated_by = ? WHERE id = ?', + [segment.name, JSON.stringify(segment.expression), segment.ceiling, segment.updated_by ?? null, id], + ) + +const remove = (id) => query('DELETE FROM engagement_audience_segments WHERE id = ?', [id]) + +module.exports = { list, getById, insert, update, remove } diff --git a/server/src/model/engagement/engagementSegments.model.js b/server/src/model/engagement/engagementSegments.model.js new file mode 100644 index 0000000..dac36b0 --- /dev/null +++ b/server/src/model/engagement/engagementSegments.model.js @@ -0,0 +1,87 @@ +// ── Audience segments — the save path ────────────────────────────────────── +// +// ENGAGEMENT.md §5.1a, Phase 4a. The thin model over `segments.js`: it validates, +// derives the ceiling, and writes. The composition UI is Phase 4b's; this is what +// it will call, and what any other route in must go through. +// +// The `ceiling` column is never taken from the caller. It is derived from the +// expression by `segments.validate` as the narrowest ceiling in the tree, and +// stored so an audit can read what a rule was ALLOWED to reach without +// re-resolving it. + +const db = require('./engagementSegments.db') +const rulesDb = require('./engagementRules.db') +const registries = require('../../modules/registries') +const segments = require('../../engagement/segments') + +async function save(input, { id = null } = {}) { + const errors = [] + const name = typeof input?.name === 'string' ? input.name.trim() : '' + if (!name) errors.push('name is required') + else if (name.length > 160) errors.push('name is longer than 160 characters') + + const checked = segments.validate(input?.expression) + if (!checked.ok) errors.push(...checked.errors) + if (errors.length) return { ok: false, errors } + + const row = { + name, + expression: checked.expression, + ceiling: checked.ceiling, + updated_by: Number.isInteger(input?.updatedBy) ? input.updatedBy : null, + } + + if (id) { + const existing = await db.getById(id) + if (!existing) return { ok: false, errors: [`no segment ${id} exists`], notFound: true } + await db.update(id, row) + return { ok: true, segment: await db.getById(id) } + } + const newId = await db.insert(row) + return { ok: true, segment: await db.getById(newId) } +} + +/** + * Delete a segment, refusing while a rule still points at it. + * + * There is deliberately no foreign key doing this (schema.sql): the database + * options are CASCADE, which would delete an operator's rules, and SET NULL, + * which would silently fall the rule back to its plain `audience` column and mail + * a DIFFERENT set of people. Refusing here, with the count, is the third option + * and the only safe one. + */ +async function remove(id) { + const inUse = await rulesDb.countUsingSegment(id) + if (inUse > 0) { + return { + ok: false, + inUse, + errors: [`${inUse} rule${inUse === 1 ? '' : 's'} still use this segment`], + } + } + await db.remove(id) + return { ok: true } +} + +/** + * Every segment, each annotated with whether it can currently resolve. + * + * A segment naming an audience whose module has been uninstalled is DORMANT, not + * broken: it is listed, it resolves to nobody, and it starts working again when + * the module comes back (§5.1a rule 4). + */ +async function listAnnotated() { + const rows = await db.list() + return rows.map((segment) => { + const missing = [] + const walk = (node) => { + if (!node || typeof node !== 'object') return + if (node.op) (node.nodes || []).forEach(walk) + else if (!registries.audience(node.audienceId)) missing.push(node.audienceId) + } + walk(segment.expression) + return { ...segment, dormant: missing.length > 0, missingAudiences: [...new Set(missing)] } + }) +} + +module.exports = { save, remove, listAnnotated } diff --git a/server/src/model/engagement/engagementSends.db.js b/server/src/model/engagement/engagementSends.db.js new file mode 100644 index 0000000..d9b2873 --- /dev/null +++ b/server/src/model/engagement/engagementSends.db.js @@ -0,0 +1,73 @@ +const { query } = require('../../utils/db') + +/** + * Record one attempt's outcome. G15: "did user X get the mail?" has never been + * answerable on this deployment, and this row is the answer. + * + * `address_hash` is a sha256 the CALLER computes, never an address. The log has + * to correlate a bounce back to a recipient (Phase 9) and it must not become a + * second address book, and a hash does the first without the second. + */ +const record = async (entry) => { + const result = await query( + `INSERT INTO engagement_sends + (outbox_id, rule_id, trigger_id, user_id, channel, transport, address_hash, status, detail) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + [ + entry.outbox_id ?? null, + entry.rule_id ?? null, + entry.trigger_id, + entry.user_id ?? null, + entry.channel, + entry.transport ?? null, + entry.address_hash ?? null, + entry.status, + entry.detail ? String(entry.detail).slice(0, 500) : null, + ], + ) + return result.insertId +} + +/** + * How many sends this rule has made in the last hour - the count the per-rule + * ceiling (§7.1 Q3) is enforced against. + * + * It counts 'sent' only. A refusal that never left the building (`suppressed`) + * and an attempt that failed are not sends, and counting them would let a broken + * transport silently consume a rule's whole hourly budget and mute it. + * + * `idx_engs_rule_window (rule_id, created_at)` exists for this: it runs once per + * rule per event, so it has to be an index range scan. + */ +const countSentSince = async (ruleId, since) => { + const [row] = await query( + "SELECT COUNT(*) AS n FROM engagement_sends WHERE rule_id = ? AND status = 'sent' AND created_at >= ?", + [ruleId, since], + ) + return Number(row?.n || 0) +} + +/** The admin send log (Phase 4b/5), newest first. */ +const list = ({ triggerId = null, userId = null, ruleId = null, limit = 100, offset = 0 } = {}) => { + const where = [] + const params = [] + if (triggerId) { + where.push('trigger_id = ?') + params.push(triggerId) + } + if (userId) { + where.push('user_id = ?') + params.push(userId) + } + if (ruleId) { + where.push('rule_id = ?') + params.push(ruleId) + } + const clause = where.length ? `WHERE ${where.join(' AND ')}` : '' + return query( + `SELECT * FROM engagement_sends ${clause} ORDER BY id DESC LIMIT ? OFFSET ?`, + [...params, limit, offset], + ) +} + +module.exports = { record, countSentSince, list } diff --git a/server/src/server.js b/server/src/server.js index fccd480..3efb3b7 100644 --- a/server/src/server.js +++ b/server/src/server.js @@ -12,6 +12,7 @@ const announceWorker = require('./utils/announceWorker') const teamActivityPrune = require('./utils/teamActivityPrune') const teamForumUploadSweep = require('./utils/teamForumUploadSweep') const teamDigestWorker = require('./utils/teamDigestWorker') +const engagementWorker = require('./utils/engagementWorker') const { ensureSchema, close } = require('./utils/db') const { seedDefaults, createInitialAdminFromEnv } = require('../db/seed') const settings = require('./model/settings/settings.model') @@ -160,6 +161,10 @@ async function start() { teamForumUploadSweep.start() teamDigestWorker.start() + // Drain the engagement outbox (ENGAGEMENT.md §4.2a). No-op until an operator + // enables a rule: core seeds none and `enabled` defaults to 0. + engagementWorker.start() + setupShutdown(server, internalServer) } @@ -180,6 +185,7 @@ function setupShutdown(server, internalServer) { teamActivityPrune.stop() // stop the Team activity retention timer teamForumUploadSweep.stop() // stop the forum upload sweep teamDigestWorker.stop() // stop the Team forum digest timer + engagementWorker.stop() // stop the engagement outbox worker server.close(() => log.info('http server closed')) if (internalServer) internalServer.close(() => log.info('internal http server closed')) try { diff --git a/server/src/utils/engagementEmit.js b/server/src/utils/engagementEmit.js index c756738..e3e2262 100644 --- a/server/src/utils/engagementEmit.js +++ b/server/src/utils/engagementEmit.js @@ -1,16 +1,17 @@ // ── ctx.events.emit — the validating half of the engagement seam ──────────── // -// ENGAGEMENT.md §4.3 and §5.2, Phase 2. A registrant fires a declared event with -// a payload; this checks the payload against the declaration and stops there. -// **There is no delivery in this phase** — no rules, no cooldowns, no outbox, no -// mail. Phase 4 replaces the log line at the bottom with the engine call, and -// every validation rule below is already the one it will need. +// ENGAGEMENT.md §4.3 and §5.2. A registrant fires a declared event with a +// payload; this checks the payload against the declaration and, since Phase 4a, +// hands the validated event to the engine. // -// Landing the contract a phase before the engine is deliberate, and it is the +// Landing the contract a phase before the engine was deliberate, and it is the // same argument registerCore() has always made: a seam whose first real exercise -// is the thing that depends on it is a seam that has already drifted. Phase 6 -// migrates the Team mail onto this, and it should be migrating onto a validator -// that has been running against core's own five triggers since Phase 2. +// is the thing that depends on it is a seam that has already drifted. Every +// validation rule below was written in Phase 2 for a caller that did not exist +// yet, and the engine needed none of them changed. +// +// **The engine call is deliberately not awaited** — see `emit` below. Phase 6 +// migrates the Team mail onto this. // // **Two postures, one switch.** A malformed emit THROWS in development and is // DROPPED AND LOGGED in production, which is `ctx.teams.activity.push`'s posture @@ -20,6 +21,7 @@ // silently loses a variable is a template that silently renders `undefined`. const registries = require('../modules/registries') +const engine = require('../engagement/engine') const createLogger = require('./logger') const log = createLogger('engagement') @@ -203,8 +205,6 @@ function emit(owner, triggerId, envelope = {}) { data: payload.data, } - // Phase 2 ends here: validated, recorded, and deliberately undelivered. - // // The values are NOT logged. A payload carries player names, house locations // and forum excerpts, and an event log that reproduces them is a second copy // of exactly the content §4.5 was careful to keep out of `engagement_sends` @@ -217,6 +217,19 @@ function emit(owner, triggerId, envelope = {}) { variables: Object.keys(event.data), }) + // **Not awaited, and this is the point of the whole seam.** `emit` is called + // from inside a game-event handler; the caller's job is to say the event + // happened, and it must not be made to wait on rule lookups, audience + // resolution and a dozen inserts to find out whether it is allowed to carry on. + // That is the same reason the C# side's `Emit()` enqueues and returns rather + // than touching the socket from the Core thread. `dispatch` catches everything + // internally and never rejects, and the `.catch` is the belt to that braces. + // + // The consequence a test has to know about: `emit` returns before the outbox + // rows exist. `engine.dispatch(event)` is exported for a caller that needs to + // await the delivery decision, and the tests use it directly. + engine.dispatch(event).catch((err) => log.error('dispatch rejected', { trigger: triggerId, message: err.message })) + return { ok: true, event } } diff --git a/server/src/utils/engagementWorker.js b/server/src/utils/engagementWorker.js new file mode 100644 index 0000000..bea3771 --- /dev/null +++ b/server/src/utils/engagementWorker.js @@ -0,0 +1,168 @@ +// ── Engagement outbox worker ─────────────────────────────────────────────── +// +// ENGAGEMENT.md §4.2a, Phase 4a. Every ENGAGEMENT_POLL_MS it sweeps +// `engagement_outbox` for rows whose `due_at` has passed, claims each one, hands +// it to its channel, and records the outcome in `engagement_sends`. Same +// setInterval + unref + stop() shape as `announceWorker` and the three Team +// sweepers, wired into server.js start/shutdown beside them. +// +// **Claiming is a compare-and-set, not a lock** (§7.1 Q2, settled by the org lead +// 2026-08-29 over `SELECT ... FOR UPDATE SKIP LOCKED`): an +// `UPDATE ... SET status='sending' WHERE id=? AND status='scheduled'`, and the +// instance the server reports `affectedRows = 1` to owns the row. No transaction +// to hold open, no MariaDB version floor, and it uses a status the ENUM already +// carried for exactly this. What it makes safe is the outbox; the four existing +// workers are still single-instance, so this does not by itself make the +// deployment multi-instance. +// +// **Nothing is delivered in this phase, and that is visible rather than +// pretended.** A channel's `deliver` arrives with email in Phase 6 and the in-app +// inbox in Phase 7; until then `channels.get(id)` has no such function, the row +// finishes as `failed` and the send log says why in as many words. The +// alternatives were both worse: recording 'sent' would be a lie in the one table +// whose whole purpose is answering "did they get it", and leaving the row +// scheduled would mean an IDOC warning queued today arriving three weeks later +// on the deploy that first shipped a mailer. +// +// In practice this path is unreachable on a real deployment for now: core seeds +// no rules and `enabled` defaults to 0, so the outbox stays empty until an +// operator turns a rule on from the screen Phase 4b builds. + +const outboxDb = require('../model/engagement/engagementOutbox.db') +const sendsDb = require('../model/engagement/engagementSends.db') +const channels = require('../engagement/channels') +const log = require('./logger')('engagement-worker') + +const POLL_MS = Number(process.env.ENGAGEMENT_POLL_MS) || 30_000 +// How many rows one sweep will look at. A bound rather than a target: the sweep +// runs again in POLL_MS, and an unbounded batch is how a backlog turns one tick +// into a stall. +const BATCH = Number(process.env.ENGAGEMENT_BATCH) || 100 + +// A transient failure is retried with a flat backoff, and then given up on. +// Flat rather than exponential because `due_at` is also the grace window's clock +// and a doubling backoff would push a delayed message arbitrarily far past the +// moment it was about. +const MAX_ATTEMPTS = 5 +const RETRY_MS = 5 * 60 * 1000 + +// A row claimed into 'sending' by a process that then died is invisible to every +// other sweeper - `status='scheduled'` will never match it again. This window is +// how long a claim may look alive before it is taken back; it has to be +// comfortably longer than the slowest legitimate send or a slow one gets sent +// twice. +const STALE_MS = 15 * 60 * 1000 + +/** + * Deliver one claimed row. + * + * @returns {{ outcome: 'sent'|'retry'|'terminal', detail?: string, transport?: string }} + */ +async function deliver(row) { + const channel = channels.get(row.channel) + if (!channel) { + // The channel's module was removed between enqueue and now. Terminal: there + // is nothing to retry towards, and leaving the row scheduled would make it + // sweep forever. + return { outcome: 'terminal', detail: `channel "${row.channel}" is no longer registered` } + } + if (typeof channel.deliver !== 'function') { + return { outcome: 'terminal', detail: `channel "${row.channel}" has no delivery implementation yet` } + } + try { + const result = await channel.deliver(row) + if (result && result.ok) return { outcome: 'sent', transport: result.transport, detail: result.detail } + if (result && result.retry) return { outcome: 'retry', detail: result.detail || 'transient failure' } + return { outcome: 'terminal', detail: (result && result.detail) || 'delivery refused' } + } catch (err) { + // A channel shouldn't throw, but if one does it is a transient failure + // rather than a crashed tick - announceWorker's posture with its legs. + log.error('channel deliver threw', { outbox: row.id, channel: row.channel, message: err.message }) + return { outcome: 'retry', detail: err.message } + } +} + +/** + * Claim, deliver, record. One row, start to finish. + * + * `deliverFn` is injectable so a test can drive the retry/give-up path without a + * channel that fails on demand - the alternative is registering a fake channel, + * which would make the registry, not this function, the thing under test. + */ +async function processRow(row, now = new Date(), deliverFn = deliver) { + if (!(await outboxDb.claim(row.id))) return null // another sweeper got there first + + const result = await deliverFn(row) + + if (result.outcome === 'retry' && row.attempts + 1 < MAX_ATTEMPTS) { + await outboxDb.reschedule(row.id, new Date(now.getTime() + RETRY_MS), result.detail) + return 'retry' + } + + const status = result.outcome === 'sent' ? 'sent' : 'failed' + await outboxDb.finish(row.id, status, status === 'failed' ? result.detail : null) + // The send log is written for every terminal outcome, not only success. G15's + // question is "did user X get the mail?", and "no, and here is why" is an + // answer that table has to be able to give. + await sendsDb.record({ + outbox_id: row.id, + rule_id: row.rule_id, + trigger_id: row.trigger_id, + user_id: row.user_id, + channel: row.channel, + transport: result.transport ?? null, + status, + detail: result.detail ?? null, + }) + return status +} + +async function tick(now = new Date()) { + try { + await outboxDb.reclaimStale(new Date(now.getTime() - STALE_MS)) + } catch (err) { + log.error('failed to reclaim stale rows', { message: err.message }) + } + + let due + try { + due = await outboxDb.findDue(now, BATCH) + } catch (err) { + log.error('failed to load due rows', { message: err.message }) + return + } + if (!due || !due.length) return + + const counts = { sent: 0, failed: 0, retry: 0, taken: 0 } + for (const row of due) { + try { + const outcome = await processRow(row, now) + if (outcome === null) counts.taken += 1 + else counts[outcome] += 1 + } catch (err) { + log.error('row failed', { outbox: row.id, message: err.message }) + } + } + log.info('outbox swept', counts) +} + +let timer = null + +function start() { + if (timer) return timer + timer = setInterval(() => { + tick().catch((err) => log.error('engagement tick failed', { message: err.message })) + }, POLL_MS) + if (timer.unref) timer.unref() // don't keep the event loop alive (tests, shutdown) + log.info('engagement outbox worker started', { pollMs: POLL_MS, batch: BATCH }) + return timer +} + +function stop() { + if (timer) { + clearInterval(timer) + timer = null + } +} + +module.exports = { start, stop, tick, processRow, deliver, POLL_MS, MAX_ATTEMPTS, RETRY_MS, STALE_MS } diff --git a/server/test/engagementEngine.test.js b/server/test/engagementEngine.test.js new file mode 100644 index 0000000..612e792 --- /dev/null +++ b/server/test/engagementEngine.test.js @@ -0,0 +1,908 @@ +// ── 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(), // " " -> 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, cooldownSeconds, now) => { + const key = `${ruleId}|${userId}|${subjectKey}` + 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('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 () => { + // Phase 4a delivers nothing: `deliver` arrives with email in Phase 6 and the + // inbox in Phase 7. Recording 'sent' would be a lie in the one table whose + // purpose is answering "did they get it". + addRule() + 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') +}) + +test("a 'digest' preference still enqueues — batching is the drain's job, not the enqueue's", 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, 1) +}) + +// ── 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) +}) + +// ── 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 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) +}) diff --git a/server/test/engagementEngineSql.test.js b/server/test/engagementEngineSql.test.js new file mode 100644 index 0000000..03338e5 --- /dev/null +++ b/server/test/engagementEngineSql.test.js @@ -0,0 +1,283 @@ +// ── 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) +})