diff --git a/client/src/App.jsx b/client/src/App.jsx index 48aa783..2b0e598 100644 --- a/client/src/App.jsx +++ b/client/src/App.jsx @@ -48,6 +48,7 @@ import EngagementTemplates from './routes/admin/views/EngagementTemplates.jsx' import EngagementTriggers from './routes/admin/views/EngagementTriggers.jsx' import EngagementSendLog from './routes/admin/views/EngagementSendLog.jsx' import EngagementSuppressions from './routes/admin/views/EngagementSuppressions.jsx' +import EngagementRetention from './routes/admin/views/EngagementRetention.jsx' import TeamsAdmin from './routes/admin/views/TeamsAdmin.jsx' import AccountAdmin from './routes/admin/views/AccountAdmin.jsx' import Moderation from './routes/admin/views/Moderation.jsx' @@ -210,6 +211,7 @@ export default function App() { } /> } /> } /> + } /> } /> {/* Staff have an inbox and channel preferences like anyone else — diff --git a/client/src/api/client.js b/client/src/api/client.js index d732f14..966adc4 100644 --- a/client/src/api/client.js +++ b/client/src/api/client.js @@ -437,8 +437,14 @@ export const api = { // Suppressions (Phase 9). `unsuppressAddress` sends the address in the BODY // of a DELETE rather than in the path, and that is not style: a path // parameter lands in the access log, the browser history and every proxy in - // front of the deployment, and this one is a real person's address. The list - // never returns a hash to use instead. + // front of the deployment, and this one is a real person's address. + // + // **Phase 14 added the second form, and it is the one the row uses.** The + // list now returns each row's `address_hash`, so the Lift button on a row + // needs no address at all — the operator is looking at a mask and has never + // been told the address. `unsuppressAddress` stays for the address the + // operator types, which is the only way to reach a row that is not on the + // page in front of them. listEngagementSuppressions: ({ limit, offset, reason, channel, search } = {}) => { const qs = new URLSearchParams() if (limit) qs.set('limit', String(limit)) @@ -452,6 +458,19 @@ export const api = { req('/admin/engagement/suppressions', { method: 'POST', body: { address, detail } }), unsuppressAddress: (address, channel) => req('/admin/engagement/suppressions', { method: 'DELETE', body: { address, channel } }), + unsuppressByHash: (hash, channel) => { + const qs = new URLSearchParams() + if (channel) qs.set('channel', channel) + return req(`/admin/engagement/suppressions/by-hash/${hash}${withQs(qs.toString())}`, { + method: 'DELETE', + }) + }, + + // Retention (Phase 14). Three horizons, one screen; `engagement_suppressions` + // is not among them because a suppression does not expire. + getEngagementRetention: () => req('/admin/engagement/retention'), + setEngagementRetention: (body) => + req('/admin/engagement/retention', { method: 'PUT', body }), // Teams (docs/website/TEAMS.md §2.11). Three of these mean something // different depending on who calls them: for a moderator, unhide and diff --git a/client/src/routes/admin/AdminLayout.jsx b/client/src/routes/admin/AdminLayout.jsx index e8ec10c..3010eac 100644 --- a/client/src/routes/admin/AdminLayout.jsx +++ b/client/src/routes/admin/AdminLayout.jsx @@ -113,6 +113,11 @@ export const NAV = [ // "did that message go out", and this answers "why is this person not // getting any" - and it is the only screen that can lift a suppression. { to: '/admin/engagement/suppressions', label: 'Suppressions', icon: IconLog, roles: ['admin'] }, + // Last in the group because it is the one screen nobody visits weekly, and + // beside Suppressions on purpose: it is where the reader is told that the + // fourth engagement table does NOT expire, which is otherwise a silence + // that reads as an oversight. + { to: '/admin/engagement/retention', label: 'Retention', icon: IconGear, roles: ['admin'] }, ], }, { @@ -196,6 +201,7 @@ const TITLES = { '/admin/engagement/triggers': 'Triggers', '/admin/engagement/suppressions': 'Suppressions', '/admin/engagement/sends': 'Send Log', + '/admin/engagement/retention': 'Retention', } // An installed module's admin pages are not in TITLES and cannot be — core does diff --git a/client/src/routes/admin/views/EngagementRetention.jsx b/client/src/routes/admin/views/EngagementRetention.jsx new file mode 100644 index 0000000..2c22e56 --- /dev/null +++ b/client/src/routes/admin/views/EngagementRetention.jsx @@ -0,0 +1,230 @@ +import { useCallback, useEffect, useState } from 'react' +import { Loading, ErrorState } from '../../../components/PageState.jsx' +import { api } from '../../../api/client.js' + +// Admin → Engagement → Retention (ENGAGEMENT.md Phase 14). +// +// Three of the four engagement tables grew on every fire and nothing had ever +// deleted from any of them. This screen is the policy: how long the deployment +// keeps a cooldown row, a finished outbox row and a send-log entry. +// +// **Why it is a screen, when the other two retention workers in this codebase +// (`team_activity`, `user_notifications`) are invisible settings rows.** The +// send-log horizon changes what an operator-facing page is *able to show* — the +// Send Log is the only answer to "was this person told" — so an operator has to +// be able to see it and set it, not discover it by finding rows missing. Having +// made one visible, hiding the other two would be the worse split: "what does +// this deployment keep" is one question and deserves one answer. +// +// **The fourth table is on this page as prose, not as a control.** Suppressions +// do not expire (org lead, 2026-09-01), and saying so here is the point: an +// operator reading a retention screen that lists three tables would reasonably +// assume the fourth was an oversight. + +const FIELDS = [ + { + name: 'sends', + label: 'Send log', + table: 'engagement_sends', + // The one horizon the org lead asked to be pickable rather than typed — + // and `custom` stays, because a deployment with a compliance answer to + // give should not be limited to three numbers somebody chose. + presets: [90, 180, 365], + help: + 'One row per delivery attempt. This is what Admin → Engagement → Send Log reads, so the ' + + 'horizon is also how far back "was this person told" can be answered. The per-rule hourly ' + + 'ceiling counts this table too, which is why it can never go below a week.', + }, + { + name: 'cooldowns', + label: 'Cooldowns', + table: 'engagement_cooldowns', + presets: [7, 30, 90], + help: + 'One row per rule, user, subject and channel, written on every fire. Deleting a row that ' + + 'is still in force makes the next fire count as a first fire — that is a duplicate ' + + 'message — so this must stay longer than the longest cooldown on any enabled rule.', + }, + { + name: 'outbox', + label: 'Outbox', + table: 'engagement_outbox', + presets: [7, 30, 90], + help: + 'Only finished rows are ever removed: sent, failed, cancelled and not-sent. A scheduled ' + + 'row is a message this deployment still intends to send and is never swept, however old ' + + 'the horizon.', + }, +] + +export default function EngagementRetention() { + const [policy, setPolicy] = useState(null) + const [limits, setLimits] = useState({}) + const [warnings, setWarnings] = useState([]) + const [longestCooldown, setLongestCooldown] = useState(0) + const [draft, setDraft] = useState({}) + const [saving, setSaving] = useState(false) + const [note, setNote] = useState(null) + const [loading, setLoading] = useState(true) + const [error, setError] = useState(null) + + const apply = useCallback((result) => { + setPolicy(result.retention) + setDraft(result.retention) + setLimits(result.limits || {}) + setWarnings(result.warnings || []) + setLongestCooldown(result.longestCooldownSeconds || 0) + }, []) + + useEffect(() => { + let alive = true + ;(async () => { + try { + const result = await api.admin.getEngagementRetention() + if (alive) apply(result) + } catch (err) { + if (alive) setError(err.message) + } finally { + if (alive) setLoading(false) + } + })() + return () => { alive = false } + }, [apply]) + + async function save() { + setSaving(true) + setNote(null) + try { + // The whole draft, not the changed field: this screen is the one place the + // three are set together, and a partial save would leave the warning line + // (which is computed from the cooldown horizon) describing a policy that is + // half saved. The route itself is sparse, so sending three is legal. + const result = await api.admin.setEngagementRetention(draft) + apply(result) + setNote('Saved.') + } catch (err) { + setNote(err.message) + } finally { + setSaving(false) + } + } + + if (loading) return + if (error) return + + const dirty = policy && FIELDS.some((f) => Number(draft[f.name]) !== Number(policy[f.name])) + + return ( +
+

+ How long this deployment keeps the engagement system’s own records. A nightly sweep + removes anything older, in batches, and skips a table it cannot read rather than failing + the run. +

+ + {warnings.map((w) => ( +

+ {w} +

+ ))} + +
+ {FIELDS.map((f) => { + const spec = limits[f.name] || {} + const value = draft[f.name] ?? '' + const isPreset = f.presets.includes(Number(value)) + return ( +
+
+ {f.label} + {f.table} +
+

+ {f.help} +

+
+ + + {spec.min !== undefined && ( + + {spec.min}–{spec.max} days + + )} +
+
+ ) + })} +
+ +
+ + {dirty && ( + + )} + {note && {note}} +
+ +
+

+ Suppressed addresses do not expire +

+

+ A suppression is a standing decision, not a record of something that happened. Ageing one + out would re-mail an address that already hard-bounced or asked to be left alone, which is + how a sender loses a domain’s reputation. The way out of that list stays a + deliberate act:{' '} + Lift on the row, in Admin → Engagement → Suppressions. +

+ {longestCooldown > 0 && ( +

+ The longest cooldown on an enabled rule right now is {longestCooldown} seconds. +

+ )} +
+
+ ) +} diff --git a/client/src/routes/admin/views/EngagementSendLog.jsx b/client/src/routes/admin/views/EngagementSendLog.jsx index bba6231..8de2904 100644 --- a/client/src/routes/admin/views/EngagementSendLog.jsx +++ b/client/src/routes/admin/views/EngagementSendLog.jsx @@ -43,6 +43,11 @@ export default function EngagementSendLog() { const [offset, setOffset] = useState(0) const [status, setStatus] = useState('') const [testTrigger, setTestTrigger] = useState('') + // Phase 14. `total` is now a truncated number, and a screen that shows a total + // without saying so is quietly wrong about the deployment's own history — this + // is the fix for that, and the reason the horizon got an operator-facing + // control rather than the invisible settings row the other two sweeps use. + const [retainDays, setRetainDays] = useState(null) const [loading, setLoading] = useState(true) const [error, setError] = useState(null) @@ -55,6 +60,15 @@ export default function EngagementSendLog() { setRows(result.sends || []) setTotal(result.total || 0) setTestTrigger(result.testSendTrigger || '') + // Best-effort and non-blocking: the log is worth showing even if the policy + // cannot be read, so a failure here leaves the note off rather than the + // screen empty. + try { + const policy = await api.admin.getEngagementRetention() + setRetainDays(policy?.retention?.sends ?? null) + } catch { + setRetainDays(null) + } }, []) useEffect(() => { @@ -153,6 +167,7 @@ export default function EngagementSendLog() {
{offset + 1}–{to} of {total} + {retainDays ? ` · entries older than ${retainDays} days are removed automatically` : ''}
+ ))} diff --git a/server/db/schema.sql b/server/db/schema.sql index f6ec2eb..f7fa01b 100644 --- a/server/db/schema.sql +++ b/server/db/schema.sql @@ -1850,7 +1850,10 @@ CREATE TABLE IF NOT EXISTS engagement_outbox ( 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) + INDEX idx_engo_cancel (rule_id, user_id, subject_key, status), + -- Phase 14. The sweep is `status IN (terminal) AND created_at < ?`, and + -- `idx_engo_due` cannot serve it: its second column is `due_at`. + INDEX idx_engo_sweep (status, created_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- G15: the per-message record. Today "did user X get the mail?" is unanswerable. @@ -1877,7 +1880,11 @@ CREATE TABLE IF NOT EXISTS engagement_sends ( 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) + INDEX idx_engs_rule_window (rule_id, created_at), + -- Phase 14's retention sweep deletes by age alone, so it needs `created_at` + -- LEADING. Every index above has it in second position, which serves a + -- per-rule or per-user window and is useless to a whole-table horizon. + INDEX idx_engs_sweep (created_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- §4.4. The mail (and, from Phase 7, in-app) bodies an operator can edit, stored @@ -1937,6 +1944,14 @@ CREATE TABLE IF NOT EXISTS engagement_templates ( -- link in it. Same vocabulary as engagement_digest_state.scope_key below. ALTER TABLE engagement_outbox ADD COLUMN IF NOT EXISTS scope_key VARCHAR(190) NULL; +-- Phase 14 (retention). The two sweep indexes, for deployments whose tables +-- predate them. `IF NOT EXISTS` on an index is MariaDB-only and already used +-- above (`idx_wiki_search`), so this needs no INFORMATION_SCHEMA guard like the +-- cooldown primary-key change did -- that one needed one because MariaDB has no +-- conditional form of a PRIMARY KEY change, not because indexes lack one. +ALTER TABLE engagement_outbox ADD INDEX IF NOT EXISTS idx_engo_sweep (status, created_at); +ALTER TABLE engagement_sends ADD INDEX IF NOT EXISTS idx_engs_sweep (created_at); + -- §4.2b: digest state, and DELIBERATELY not a digest queue. -- -- The generic engine enqueues an outbox row per (rule, user, channel) at emit diff --git a/server/routes.guards.json b/server/routes.guards.json index 65b2ac2..56551b2 100644 --- a/server/routes.guards.json +++ b/server/routes.guards.json @@ -194,6 +194,24 @@ "requireAuth" ] }, + { + "method": "GET", + "path": "/api/v1/admin/engagement/retention", + "handlers": 2, + "gates": [ + "noindex", + "requireAuth" + ] + }, + { + "method": "PUT", + "path": "/api/v1/admin/engagement/retention", + "handlers": 2, + "gates": [ + "noindex", + "requireAuth" + ] + }, { "method": "GET", "path": "/api/v1/admin/engagement/rules", @@ -320,6 +338,15 @@ "requireAuth" ] }, + { + "method": "DELETE", + "path": "/api/v1/admin/engagement/suppressions/by-hash/:hash", + "handlers": 2, + "gates": [ + "noindex", + "requireAuth" + ] + }, { "method": "GET", "path": "/api/v1/admin/engagement/templates", diff --git a/server/routes.manifest.json b/server/routes.manifest.json index 6767767..4dce8bf 100644 --- a/server/routes.manifest.json +++ b/server/routes.manifest.json @@ -85,6 +85,14 @@ "method": "GET", "path": "/api/v1/admin/engagement/channels" }, + { + "method": "GET", + "path": "/api/v1/admin/engagement/retention" + }, + { + "method": "PUT", + "path": "/api/v1/admin/engagement/retention" + }, { "method": "GET", "path": "/api/v1/admin/engagement/rules" @@ -141,6 +149,10 @@ "method": "POST", "path": "/api/v1/admin/engagement/suppressions" }, + { + "method": "DELETE", + "path": "/api/v1/admin/engagement/suppressions/by-hash/:hash" + }, { "method": "GET", "path": "/api/v1/admin/engagement/templates" diff --git a/server/src/model/engagement/engagementCooldowns.db.js b/server/src/model/engagement/engagementCooldowns.db.js index 25a0589..82668bd 100644 --- a/server/src/model/engagement/engagementCooldowns.db.js +++ b/server/src/model/engagement/engagementCooldowns.db.js @@ -78,8 +78,23 @@ const get = async (ruleId, userId, subjectKey, channel) => { * 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. + * Phase 14 gave it that caller (`engagementRetentionPrune`), which also enforces + * the horizon-versus-longest-cooldown rule this comment names. + * + * `limit` bounds one statement, for `userNotificationsPrune`'s reason: a first + * sweep after a long outage must be a series of bounded DELETEs rather than one + * that holds locks over a million rows. Omitting it keeps the original + * unbounded behaviour, so existing callers and tests are unaffected. + * + * @returns {Promise} rows deleted */ -const prune = (olderThan) => - query('DELETE FROM engagement_cooldowns WHERE last_fired_at < ?', [olderThan]) +const prune = async (olderThan, limit = 0) => { + const bounded = Number(limit) > 0 + const result = await query( + `DELETE FROM engagement_cooldowns WHERE last_fired_at < ?${bounded ? ' LIMIT ?' : ''}`, + bounded ? [olderThan, Math.floor(limit)] : [olderThan], + ) + return Number(result?.affectedRows || 0) +} module.exports = { claim, get, prune } diff --git a/server/src/model/engagement/engagementOutbox.db.js b/server/src/model/engagement/engagementOutbox.db.js index 54ddbe4..9063b03 100644 --- a/server/src/model/engagement/engagementOutbox.db.js +++ b/server/src/model/engagement/engagementOutbox.db.js @@ -132,11 +132,61 @@ async function cancel(ruleId, subjectKey, userId = null) { * 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( +const reclaimStale = async (before, maxAttempts = 0) => { + // Give up first, reclaim second, and in that order: a row that has already + // burned its attempts must leave 'sending' as `failed`, or the reclaim below + // hands it straight back to `findDue` and it is retried forever. + // + // **This is what makes `pruneTerminal` a bound at all** (Phase 14). Attempts + // are incremented by `claim`, but `MAX_ATTEMPTS` is only consulted on a + // graceful `retry` outcome — a send that kills the process mid-flight never + // reaches that branch, so before this the row cycled sending → scheduled → + // sending forever, never reached a terminal status, and was therefore never + // eligible for any retention sweep. One poisoned payload was an outbox row + // that outlived every horizon. + let failed = 0 + if (Number(maxAttempts) > 0) { + const gaveUp = await query( + `UPDATE engagement_outbox + SET status = 'failed', last_error = 'gave up after repeated interruptions' + WHERE status = 'sending' AND updated_at < ? AND attempts >= ?`, + [before, Math.floor(maxAttempts)], + ) + failed = Number(gaveUp?.affectedRows || 0) + } + const reclaimed = await query( "UPDATE engagement_outbox SET status = 'scheduled' WHERE status = 'sending' AND updated_at < ?", [before], ) + return { failed, reclaimed: Number(reclaimed?.affectedRows || 0) } +} + +/** + * Delete terminal rows older than `before` (Phase 14). + * + * **Terminal only, and the status list is the whole policy.** A `scheduled` row + * is a promise the engine has not kept yet — `delay_seconds` can legitimately + * put one up to a day out (`MAX_DELAY_SECONDS`) — and a `sending` row may be a + * worker mid-flight. Deleting either is not retention, it is cancelling a send + * nobody asked to cancel. Only `sent`, `failed`, `cancelled` and `suppressed` + * are outcomes that have already happened. + * + * `created_at` rather than `updated_at` is the clock deliberately: the horizon + * an operator sets means "how long we keep the record of a delivery", which is + * measured from when it was enqueued, not from whenever it was last touched. + * + * @returns {Promise} rows deleted + */ +const pruneTerminal = async (before, limit = 1000) => { + const result = await query( + `DELETE FROM engagement_outbox + WHERE status IN ('sent', 'failed', 'cancelled', 'suppressed') + AND created_at < ? + LIMIT ?`, + [before, Math.floor(limit)], + ) + return Number(result?.affectedRows || 0) +} const getById = async (id) => { const [row] = await query('SELECT * FROM engagement_outbox WHERE id = ?', [id]) @@ -157,6 +207,7 @@ module.exports = { finish, cancel, reclaimStale, + pruneTerminal, getById, listForRule, } diff --git a/server/src/model/engagement/engagementRetention.model.js b/server/src/model/engagement/engagementRetention.model.js new file mode 100644 index 0000000..bfa7fe6 --- /dev/null +++ b/server/src/model/engagement/engagementRetention.model.js @@ -0,0 +1,175 @@ +// ── Engagement retention policy ──────────────────────────────────────────── +// +// ENGAGEMENT.md Phase 14. Three horizons, one place to read and write them, so +// the nightly worker and Admin → Engagement → Retention cannot disagree about +// what this deployment keeps — and so `/privacy` has one thing to describe. +// +// **The fourth engagement table is deliberately absent from this file.** +// `engagement_suppressions` does not expire (org lead, 2026-09-01): a +// suppression is a standing decision, and ageing out a hard bounce re-mails an +// address that already bounced, which is how a sender loses a domain's +// reputation. The one way out of that table stays what Phase 9 built — a +// deliberate act by an admin, which Phase 14 only made reachable per row. +// +// Settings rows rather than env, for `teamActivityPrune`'s reason: an operator +// tightening a busy shard should not need a deploy. Unlike the two workers this +// copies, these three get a screen — the send log's horizon changes what an +// operator-facing page can show, so it cannot be an invisible key. + +const settings = require('../settings/settings.model') +const rulesDb = require('./engagementRules.db') +const log = require('../../utils/logger')('engagement') + +/** + * One entry per horizon. `min` is not a UI nicety — each is the point below + * which the sweep breaks something that is not retention: + * + * - **cooldowns**: a pruned row makes the next fire a FIRST fire, i.e. a + * duplicate send. `MAX_COOLDOWN_SECONDS` is a validated 86 400 (one day), so + * 2 days is the smallest provably-safe value against any rule that can be + * saved. `checkCooldownHorizon` re-checks it against the rules that exist. + * - **outbox**: `MAX_DELAY_SECONDS` is also 86 400, so no `scheduled` row is + * ever more than a day out; terminal rows younger than that are still the + * most recent thing an operator would look at. + * - **sends**: the per-rule hourly ceiling (§7.1 Q3) counts this table, so a + * horizon under an hour would silently disable it. The floor is set far + * above that, at the point the Send Log stops being worth opening. + */ +const HORIZONS = { + sends: { + key: 'engagement_sends_retain_days', + default: 180, + min: 7, + max: 3650, + label: 'Send log', + }, + cooldowns: { + key: 'engagement_cooldowns_retain_days', + default: 30, + min: 2, + max: 3650, + label: 'Cooldowns', + }, + outbox: { + key: 'engagement_outbox_retain_days', + default: 30, + min: 2, + max: 3650, + label: 'Outbox', + }, +} + +const NAMES = Object.keys(HORIZONS) + +/** + * The current policy, as `{ sends, cooldowns, outbox }` in days. + * + * Wrapped in a try like `teamActivity.retentionConfig` and for its reason: the + * nightly worker calls this with nobody watching, so a settings table that is + * briefly unavailable must yield defaults rather than an exception that kills + * the job. An unreadable, absent, non-numeric or out-of-range value all mean + * the same thing — use the default — because none of them is a horizon. + */ +async function get() { + const out = {} + for (const name of NAMES) { + const spec = HORIZONS[name] + let days = spec.default + try { + const raw = await settings.get(spec.key) + const n = Number(raw) + if (Number.isFinite(n) && n >= spec.min && n <= spec.max) days = Math.floor(n) + } catch (err) { + log.debug('retention setting unreadable; using the default', { + key: spec.key, + message: err.message, + }) + } + out[name] = days + } + return out +} + +/** + * Write one or more horizons. Unknown names are ignored rather than rejected, so + * a client sending the whole object back is not coupled to this list; an + * out-of-range value IS rejected, because silently clamping a number an operator + * typed would leave the screen showing something the deployment is not doing. + * + * @returns {Promise} the policy as it now stands + */ +async function set(patch = {}, updatedBy = null) { + for (const name of NAMES) { + if (!(name in patch) || patch[name] === undefined || patch[name] === null) continue + const spec = HORIZONS[name] + const n = Number(patch[name]) + if (!Number.isFinite(n) || Math.floor(n) !== n) { + throw Object.assign(new Error(`${spec.label} retention must be a whole number of days`), { + status: 400, + }) + } + if (n < spec.min || n > spec.max) { + throw Object.assign( + new Error(`${spec.label} retention must be between ${spec.min} and ${spec.max} days`), + { status: 400 }, + ) + } + await settings.set(spec.key, String(n), updatedBy) + } + return get() +} + +/** + * The longest cooldown any ENABLED rule configures, in seconds — see + * `engagementRules.db.maxEnabledCooldownSeconds` for why enabled only. + * + * Swallows its error rather than propagating: this is consulted by a worker + * running on a timer, and a database hiccup must degrade the WARNING, never the + * sweep. Zero reads as "no enabled rule has a cooldown", which produces no + * warning — the same answer as an unreadable table, and the safe one, because + * the alternative is a nightly alarm nobody can act on. + */ +async function longestEnabledCooldownSeconds() { + try { + return await rulesDb.maxEnabledCooldownSeconds() + } catch (err) { + log.debug('could not read the longest enabled cooldown', { message: err.message }) + return 0 + } +} + +/** + * Phase 14's acceptance line: the cooldown horizon is CHECKED against the + * longest enabled rule's cooldown rather than picked. + * + * It returns a warning rather than throwing, and the worker sweeps anyway. The + * alternative — refusing to prune — trades a bounded, describable fault (one + * rule may re-fire early once) for the unbounded one this phase exists to end. + * The screen surfaces the same warning, which is where an operator can act on it. + * + * @returns {Promise<{ ok: boolean, longestCooldownSeconds: number, message: string|null }>} + */ +async function checkCooldownHorizon(days) { + const longest = await longestEnabledCooldownSeconds() + const horizonSeconds = days * 24 * 60 * 60 + if (longest > 0 && horizonSeconds <= longest) { + return { + ok: false, + longestCooldownSeconds: longest, + message: + `Cooldown retention is ${days} day(s), but an enabled rule has a cooldown of ` + + `${longest} second(s). Pruning a cooldown row that is still in force makes the next ` + + 'fire count as a first fire, so that rule can send twice. Raise the horizon.', + } + } + return { ok: true, longestCooldownSeconds: longest, message: null } +} + +module.exports = { + HORIZONS, + NAMES, + get, + set, + longestEnabledCooldownSeconds, + checkCooldownHorizon, +} diff --git a/server/src/model/engagement/engagementRules.db.js b/server/src/model/engagement/engagementRules.db.js index 149a8f7..16afe37 100644 --- a/server/src/model/engagement/engagementRules.db.js +++ b/server/src/model/engagement/engagementRules.db.js @@ -135,9 +135,25 @@ const countUsingSegment = async (segmentId) => { return Number(row?.n || 0) } +/** + * The longest cooldown any ENABLED rule configures, in seconds (Phase 14). + * + * Enabled only, deliberately: a disabled rule fires nothing, so it writes no + * cooldown row a retention sweep could destroy, and letting a forgotten disabled + * rule with a 24-hour cooldown veto a tighter horizon would make the warning + * advice nobody can act on. + */ +const maxEnabledCooldownSeconds = async () => { + const [row] = await query( + 'SELECT MAX(cooldown_seconds) AS n FROM engagement_rules WHERE enabled = 1', + ) + return Number(row?.n || 0) +} + module.exports = { list, getById, + maxEnabledCooldownSeconds, enabledForTrigger, enabledCancelledBy, insert, diff --git a/server/src/model/engagement/engagementSends.db.js b/server/src/model/engagement/engagementSends.db.js index 9cdee75..a6ba0ce 100644 --- a/server/src/model/engagement/engagementSends.db.js +++ b/server/src/model/engagement/engagementSends.db.js @@ -107,4 +107,26 @@ const count = async (opts = {}) => { return Number(row?.n || 0) } -module.exports = { record, countSentSince, list, count, TEST_SEND_TRIGGER } +/** + * Delete send-log rows older than `before` (Phase 14). + * + * **Every row here is terminal**, which is why this has no status filter and the + * outbox's sweep does: `engagement_sends` records an attempt that has already + * resolved. The care is entirely in the horizon, because this table has two live + * readers and they pull in opposite directions — `countSentSince` implements the + * per-rule hourly ceiling (§7.1 Q3), so any horizon under an hour silently + * disables that ceiling, and Admin -> Engagement -> Send Log is the operator's + * only answer to "was this person told", so a short one blinds it. Both are the + * caller's problem, and `engagementRetention` is where that judgement lives. + * + * @returns {Promise} rows deleted + */ +const prune = async (before, limit = 1000) => { + const result = await query('DELETE FROM engagement_sends WHERE created_at < ? LIMIT ?', [ + before, + Math.floor(limit), + ]) + return Number(result?.affectedRows || 0) +} + +module.exports = { record, countSentSince, list, count, prune, TEST_SEND_TRIGGER } diff --git a/server/src/router/v1/admin/engagement.controller.js b/server/src/router/v1/admin/engagement.controller.js index a90d201..af99f45 100644 --- a/server/src/router/v1/admin/engagement.controller.js +++ b/server/src/router/v1/admin/engagement.controller.js @@ -34,6 +34,7 @@ const templates = require('../../../model/engagement/engagementTemplates.model') const sendsDb = require('../../../model/engagement/engagementSends.db') const suppressionsDb = require('../../../model/engagement/engagementSuppressions.db') const suppressions = require('../../../engagement/suppressions') +const retention = require('../../../model/engagement/engagementRetention.model') // The lattice, flattened for a client: for each ceiling, the ones a rule may // choose under it. Served with the catalog rather than hardcoded in the admin @@ -496,12 +497,18 @@ exports.listSends = async (req, res, next) => { // human in the loop, and without a way back a mistyped-then-corrected mailbox is // silenced permanently. // -// **The list returns `address_masked`, never `address_hash`.** The send log route -// above strips the hash for a stated reason — shipping a sha256 of every address -// on the deployment to a browser is an offline dictionary attack waiting to be -// run — and the same reasoning applies twice over here, where the rows are -// exactly the addresses somebody would most want to confirm. The mask is what an -// operator can act on and is not reversible. +// **The list DOES return `address_hash`, and Phase 14 reversed a Phase 9 +// decision to get there** (org lead, 2026-09-01). Phase 9 stripped it on the +// grounds that a sha256 of every address on the deployment is an offline +// dictionary attack waiting to be run, and left the only way out of the table a +// `window.prompt` asking the operator to retype the full address — which they do +// not have, because the screen shows a mask. The trade taken: this route is +// admin-only and an admin can already suppress and unsuppress any address they +// can name, so the hash grants them no capability they lack; what it buys is a +// Lift button on the row the operator is actually looking at. The send log route +// above still strips its hash, because nothing there needs to act on a row. +// +// The mask remains what is DISPLAYED. The hash is a handle, never rendered. /** GET /api/v1/admin/engagement/suppressions */ exports.listSuppressions = async (req, res, next) => { @@ -526,7 +533,7 @@ exports.listSuppressions = async (req, res, next) => { suppressionsDb.countsByReason(), ]) res.json({ - suppressions: rows.map(({ address_hash: _hash, ...row }) => row), + suppressions: rows, total, limit, offset, @@ -595,3 +602,90 @@ exports.deleteSuppression = async (req, res, next) => { next(err) } } + +/** + * DELETE /api/v1/admin/engagement/suppressions/by-hash/:hash + * + * The per-row Lift button (Phase 14). Same effect as the route above and a + * different input: the operator is looking at a masked row and does not know the + * address, so the only thing they can act on is the row's own handle. + * + * **The hash still goes in the path and that is safe where an address is not.** + * The objection to a path parameter above is that an access log, a browser + * history and every proxy in front of the deployment would capture a real + * person's address; a sha256 that is already only ever served to an admin + * session leaks nothing further by being logged. + * + * A 404 rather than a 200 when nothing matched, so a stale screen (two admins, + * one list, one already lifted) tells the operator rather than claiming success. + */ +exports.deleteSuppressionByHash = async (req, res, next) => { + try { + const hash = typeof req.params.hash === 'string' ? req.params.hash.trim().toLowerCase() : '' + // Validated in shape rather than trusted: this value reaches a WHERE clause, + // and a 64-character hex string is the only thing this column ever holds. + if (!/^[0-9a-f]{64}$/.test(hash)) { + return res.status(400).json({ message: 'Not a suppression handle' }) + } + const channel = typeof req.query.channel === 'string' && req.query.channel + ? req.query.channel + : 'email' + const removed = await suppressionsDb.remove(hash, channel) + if (!removed) return res.status(404).json({ message: 'That address is not suppressed' }) + res.json({ removed: true }) + } catch (err) { + next(err) + } +} + +// ── Retention (Phase 14) ─────────────────────────────────────────────────── +// +// Three horizons on one screen, because "what does this deployment keep" is one +// question. The send-log horizon is the reason this is a screen at all rather +// than the invisible settings row `team_activity` and `user_notifications` each +// use: it changes what an operator-facing page is able to show, so an operator +// has to be able to see and set it. + +/** GET /api/v1/admin/engagement/retention */ +exports.getRetention = async (req, res, next) => { + try { + const policy = await retention.get() + // The guard travels with the policy rather than only being logged at 3am by + // the worker: the screen that can fix a too-short cooldown horizon is the one + // that has to say it is too short. + const cooldownCheck = await retention.checkCooldownHorizon(policy.cooldowns) + res.json({ + retention: policy, + limits: retention.HORIZONS, + longestCooldownSeconds: cooldownCheck.longestCooldownSeconds, + warnings: cooldownCheck.ok ? [] : [cooldownCheck.message], + }) + } catch (err) { + next(err) + } +} + +/** + * PUT /api/v1/admin/engagement/retention + * + * A sparse PUT: only the horizons present in the body are written, so a screen + * saving one select does not have to round-trip the other two and cannot + * clobber a value another admin changed between load and save. Out-of-range is + * a 400 rather than a clamp — silently storing something other than what was + * typed would leave the screen describing a policy the deployment is not running. + */ +exports.putRetention = async (req, res, next) => { + try { + const policy = await retention.set(req.body || {}, req.user?.id ?? null) + const cooldownCheck = await retention.checkCooldownHorizon(policy.cooldowns) + res.json({ + retention: policy, + limits: retention.HORIZONS, + longestCooldownSeconds: cooldownCheck.longestCooldownSeconds, + warnings: cooldownCheck.ok ? [] : [cooldownCheck.message], + }) + } catch (err) { + if (err.status === 400) return res.status(400).json({ message: err.message }) + next(err) + } +} diff --git a/server/src/router/v1/admin/engagement.router.js b/server/src/router/v1/admin/engagement.router.js index 6ed6865..aa45aee 100644 --- a/server/src/router/v1/admin/engagement.router.js +++ b/server/src/router/v1/admin/engagement.router.js @@ -381,4 +381,48 @@ engagementRouter.delete( controller.deleteSuppression, ) +engagementRouter.delete( + '/suppressions/by-hash/:hash', + // #swagger.tags = ['Admin - Engagement'] + // #swagger.summary = 'Lift a suppression by its row handle' + // #swagger.description = 'The per-row Lift button (Phase 14). Same effect as the route above, different input: the screen shows a mask, so the operator does not know the address and can only act on the row handle the list gives them. The handle IS safe in the path where an address is not - it is a sha256 already served only to an admin session, so an access log or proxy that captures it learns nothing new. 404 rather than 200 when nothing matched, so a stale screen (two admins, one list) says so instead of claiming success.' + // #swagger.security = [{ "cookieAuth": [] }, { "bearerAuth": [] }] + // #swagger.parameters['hash'] = { in: 'path', description: 'The address_hash the list returns for that row, 64 hex characters', required: true, schema: { type: 'string' } } + // #swagger.parameters['channel'] = { in: 'query', description: 'Defaults to email', required: false, schema: { type: 'string' } } + /* #swagger.responses[200] = { description: 'Lifted', content: { "application/json": { schema: { type: "object", properties: { removed: { type: "boolean" } } } } } } */ + /* #swagger.responses[400] = { description: 'Not a suppression handle', content: { "application/json": { schema: { $ref: "#/components/schemas/Error" } } } } */ + /* #swagger.responses[403] = { description: 'Not an admin', content: { "application/json": { schema: { $ref: "#/components/schemas/Error" } } } } */ + /* #swagger.responses[404] = { description: 'That address is not suppressed', content: { "application/json": { schema: { $ref: "#/components/schemas/Error" } } } } */ + adminOnly, + controller.deleteSuppressionByHash, +) + +// ── Retention (Phase 14) ─────────────────────────────────────────────────── + +engagementRouter.get( + '/retention', + // #swagger.tags = ['Admin - Engagement'] + // #swagger.summary = 'Read the engagement retention policy' + // #swagger.description = 'The three horizons the nightly sweep uses, in days, with the bounds each is validated against. `engagement_suppressions` is deliberately absent: a suppression is a standing decision and does not expire, because ageing out a hard bounce re-mails an address that already bounced. `warnings` carries the one check that cannot be a static bound - a cooldown horizon shorter than the longest cooldown on an ENABLED rule, which would let that rule send twice.' + // #swagger.security = [{ "cookieAuth": [] }, { "bearerAuth": [] }] + /* #swagger.responses[200] = { description: 'The current policy', content: { "application/json": { schema: { type: "object", properties: { retention: { type: "object", properties: { sends: { type: "integer" }, cooldowns: { type: "integer" }, outbox: { type: "integer" } } }, limits: { type: "object", additionalProperties: true }, longestCooldownSeconds: { type: "integer" }, warnings: { type: "array", items: { type: "string" } } } } } } } */ + /* #swagger.responses[403] = { description: 'Not an admin', content: { "application/json": { schema: { $ref: "#/components/schemas/Error" } } } } */ + adminOnly, + controller.getRetention, +) + +engagementRouter.put( + '/retention', + // #swagger.tags = ['Admin - Engagement'] + // #swagger.summary = 'Set the engagement retention policy' + // #swagger.description = 'Sparse: only the horizons named in the body are written, so saving one select cannot clobber a value another admin changed between load and save. Out of range is a 400 rather than a clamp - storing something other than what was typed would leave the screen describing a policy the deployment is not running. The floors are not UI niceties: below 2 days a pruned cooldown row makes the next fire a FIRST fire (a duplicate send), and the send log is counted by the per-rule hourly ceiling.' + // #swagger.security = [{ "cookieAuth": [] }, { "bearerAuth": [] }] + /* #swagger.requestBody = { required: true, content: { "application/json": { schema: { type: "object", properties: { sends: { type: "integer", nullable: true }, cooldowns: { type: "integer", nullable: true }, outbox: { type: "integer", nullable: true } } } } } } */ + /* #swagger.responses[200] = { description: 'The policy as it now stands', content: { "application/json": { schema: { type: "object", properties: { retention: { type: "object", additionalProperties: { type: "integer" } }, limits: { type: "object", additionalProperties: true }, longestCooldownSeconds: { type: "integer" }, warnings: { type: "array", items: { type: "string" } } } } } } } */ + /* #swagger.responses[400] = { description: 'A horizon was not a whole number of days, or was out of range', content: { "application/json": { schema: { $ref: "#/components/schemas/Error" } } } } */ + /* #swagger.responses[403] = { description: 'Not an admin', content: { "application/json": { schema: { $ref: "#/components/schemas/Error" } } } } */ + adminOnly, + controller.putRetention, +) + module.exports = engagementRouter diff --git a/server/src/server.js b/server/src/server.js index fda8743..7ac2a23 100644 --- a/server/src/server.js +++ b/server/src/server.js @@ -11,6 +11,7 @@ const botScore = require('./middleware/botScore') const announceWorker = require('./utils/announceWorker') const teamActivityPrune = require('./utils/teamActivityPrune') const inboxPrune = require('./utils/userNotificationsPrune') +const engagementRetentionPrune = require('./utils/engagementRetentionPrune') const teamForumUploadSweep = require('./utils/teamForumUploadSweep') const teamDigestWorker = require('./utils/teamDigestWorker') const engagementWorker = require('./utils/engagementWorker') @@ -160,6 +161,7 @@ async function start() { // rather than after someone notices. No-op on a deployment with no Teams. teamActivityPrune.start() inboxPrune.start() + engagementRetentionPrune.start() teamForumUploadSweep.start() teamDigestWorker.start() @@ -186,6 +188,7 @@ function setupShutdown(server, internalServer) { announceWorker.stop() // stop the news-announcement dispatcher poller teamActivityPrune.stop() // stop the Team activity retention timer inboxPrune.stop() // stop the in-app inbox retention timer + engagementRetentionPrune.stop() // stop the engagement retention timer teamForumUploadSweep.stop() // stop the forum upload sweep teamDigestWorker.stop() // stop the Team forum digest timer engagementWorker.stop() // stop the engagement outbox worker diff --git a/server/src/utils/engagementRetentionPrune.js b/server/src/utils/engagementRetentionPrune.js new file mode 100644 index 0000000..0cbf483 --- /dev/null +++ b/server/src/utils/engagementRetentionPrune.js @@ -0,0 +1,134 @@ +// ── Engagement retention worker ──────────────────────────────────────────── +// +// ENGAGEMENT.md Phase 14. Three of the four engagement tables grow on every fire +// and nothing has ever deleted from any of them: `engagement_cooldowns` (one row +// per rule × user × subject × channel per fire), `engagement_outbox` (one row per +// enqueued delivery, terminal rows included) and `engagement_sends` (one row per +// attempt). `engagement_suppressions` is the fourth and does not expire — see +// `engagementRetention.model.js` for why that is a decision rather than an +// omission. +// +// **One worker, three sweeps, not three workers.** They share a timer, a batch +// discipline and one settings-backed policy object; splitting them would give an +// operator three independent nightly table-wide DELETEs to reason about and +// three places for a horizon to be read differently. +// +// Same in-process shape as `teamActivityPrune` and `userNotificationsPrune` — +// setInterval + unref + stop(), wired into server.js start/shutdown beside them, +// with the first run delayed so a table-wide DELETE never lands in front of the +// first request on a crash-looping deployment. + +const cooldownsDb = require('../model/engagement/engagementCooldowns.db') +const outboxDb = require('../model/engagement/engagementOutbox.db') +const sendsDb = require('../model/engagement/engagementSends.db') +const retention = require('../model/engagement/engagementRetention.model') +const log = require('./logger')('engagement') + +const INTERVAL_MS = Number(process.env.ENGAGEMENT_PRUNE_MS) || 24 * 60 * 60 * 1000 +const FIRST_RUN_MS = Number(process.env.ENGAGEMENT_PRUNE_DELAY_MS) || 10 * 60 * 1000 + +// A bound per statement, so one run after a long outage is a series of bounded +// DELETEs rather than one holding locks over a million rows. Each sweep repeats +// until it clears and stops early rather than looping forever; a run that hits +// the ceiling simply resumes tomorrow, which is what a horizon means anyway. +const BATCH = 1000 +const MAX_BATCHES = 50 + +const daysAgo = (days, now) => new Date(now.getTime() - days * 24 * 60 * 60 * 1000) + +/** Repeat a bounded delete until it stops filling its batch. Never throws. */ +async function sweep(name, del) { + let removed = 0 + for (let i = 0; i < MAX_BATCHES; i += 1) { + const n = await del() + removed += n + if (n < BATCH) break + } + if (removed) log.info('engagement retention swept', { table: name, removed }) + return removed +} + +/** + * One pass over all three tables. + * + * Each sweep is caught on its own: a failure in one (a lock timeout on a huge + * outbox, say) must not stop the other two from being bounded. Never throws — it + * runs on a timer with nobody to catch it. + */ +async function tick(now = new Date()) { + const result = { cooldowns: 0, outbox: 0, sends: 0, warnings: [] } + let policy + try { + policy = await retention.get() + } catch (err) { + log.error('engagement retention policy unreadable; skipping this run', { message: err.message }) + return result + } + + // Phase 14's acceptance line: checked, not picked. A horizon shorter than a + // live cooldown is logged and swept anyway — see the model for that trade. + try { + const check = await retention.checkCooldownHorizon(policy.cooldowns) + if (!check.ok) { + result.warnings.push(check.message) + log.warn('cooldown retention is shorter than a live cooldown', { + retainDays: policy.cooldowns, + longestCooldownSeconds: check.longestCooldownSeconds, + }) + } + } catch (err) { + log.debug('cooldown horizon check failed', { message: err.message }) + } + + try { + result.cooldowns = await sweep('engagement_cooldowns', () => + cooldownsDb.prune(daysAgo(policy.cooldowns, now), BATCH)) + } catch (err) { + log.error('cooldown prune failed', { message: err.message }) + } + + try { + result.outbox = await sweep('engagement_outbox', () => + outboxDb.pruneTerminal(daysAgo(policy.outbox, now), BATCH)) + } catch (err) { + log.error('outbox prune failed', { message: err.message }) + } + + try { + result.sends = await sweep('engagement_sends', () => + sendsDb.prune(daysAgo(policy.sends, now), BATCH)) + } catch (err) { + log.error('send-log prune failed', { message: err.message }) + } + + return result +} + +let timer = null +let firstRun = null + +function start() { + if (timer || firstRun) return timer + firstRun = setTimeout(() => { + firstRun = null + tick() + timer = setInterval(() => { tick() }, INTERVAL_MS) + if (timer.unref) timer.unref() + }, FIRST_RUN_MS) + if (firstRun.unref) firstRun.unref() + log.info('engagement retention started', { intervalMs: INTERVAL_MS, firstRunMs: FIRST_RUN_MS }) + return timer +} + +function stop() { + if (firstRun) { + clearTimeout(firstRun) + firstRun = null + } + if (timer) { + clearInterval(timer) + timer = null + } +} + +module.exports = { start, stop, tick, BATCH, MAX_BATCHES, INTERVAL_MS, FIRST_RUN_MS } diff --git a/server/src/utils/engagementWorker.js b/server/src/utils/engagementWorker.js index 25669e6..628d3c2 100644 --- a/server/src/utils/engagementWorker.js +++ b/server/src/utils/engagementWorker.js @@ -153,7 +153,7 @@ async function processRow(row, now = new Date(), deliverFn = deliver) { async function tick(now = new Date()) { try { - await outboxDb.reclaimStale(new Date(now.getTime() - STALE_MS)) + await outboxDb.reclaimStale(new Date(now.getTime() - STALE_MS), MAX_ATTEMPTS) } catch (err) { log.error('failed to reclaim stale rows', { message: err.message }) } diff --git a/server/swagger/swagger-output.json b/server/swagger/swagger-output.json index 7da8651..9ce1599 100644 --- a/server/swagger/swagger-output.json +++ b/server/swagger/swagger-output.json @@ -1351,6 +1351,166 @@ ] } }, + "/api/v1/admin/engagement/retention": { + "get": { + "tags": [ + "Admin - Engagement" + ], + "summary": "Read the engagement retention policy", + "description": "The three horizons the nightly sweep uses, in days, with the bounds each is validated against. `engagement_suppressions` is deliberately absent: a suppression is a standing decision and does not expire, because ageing out a hard bounce re-mails an address that already bounced. `warnings` carries the one check that cannot be a static bound - a cooldown horizon shorter than the longest cooldown on an ENABLED rule, which would let that rule send twice.", + "responses": { + "200": { + "description": "The current policy", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "retention": { + "type": "object", + "properties": { + "sends": { + "type": "integer" + }, + "cooldowns": { + "type": "integer" + }, + "outbox": { + "type": "integer" + } + } + }, + "limits": { + "type": "object", + "additionalProperties": true + }, + "longestCooldownSeconds": { + "type": "integer" + }, + "warnings": { + "type": "array", + "items": { + "type": "string" + } + } + } + } + } + } + }, + "403": { + "description": "Not an admin", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + } + } + }, + "security": [ + { + "cookieAuth": [] + }, + { + "bearerAuth": [] + } + ] + }, + "put": { + "tags": [ + "Admin - Engagement" + ], + "summary": "Set the engagement retention policy", + "description": "Sparse: only the horizons named in the body are written, so saving one select cannot clobber a value another admin changed between load and save. Out of range is a 400 rather than a clamp - storing something other than what was typed would leave the screen describing a policy the deployment is not running. The floors are not UI niceties: below 2 days a pruned cooldown row makes the next fire a FIRST fire (a duplicate send), and the send log is counted by the per-rule hourly ceiling.", + "responses": { + "200": { + "description": "The policy as it now stands", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "retention": { + "type": "object", + "additionalProperties": { + "type": "integer" + } + }, + "limits": { + "type": "object", + "additionalProperties": true + }, + "longestCooldownSeconds": { + "type": "integer" + }, + "warnings": { + "type": "array", + "items": { + "type": "string" + } + } + } + } + } + } + }, + "400": { + "description": "A horizon was not a whole number of days, or was out of range", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + } + }, + "403": { + "description": "Not an admin", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + } + } + }, + "security": [ + { + "cookieAuth": [] + }, + { + "bearerAuth": [] + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "sends": { + "type": "integer", + "nullable": true + }, + "cooldowns": { + "type": "integer", + "nullable": true + }, + "outbox": { + "type": "integer", + "nullable": true + } + } + } + } + } + } + } + }, "/api/v1/admin/engagement/rules": { "get": { "tags": [ @@ -2490,6 +2650,90 @@ } } }, + "/api/v1/admin/engagement/suppressions/by-hash/{hash}": { + "delete": { + "tags": [ + "Admin - Engagement" + ], + "summary": "Lift a suppression by its row handle", + "description": "The per-row Lift button (Phase 14). Same effect as the route above, different input: the screen shows a mask, so the operator does not know the address and can only act on the row handle the list gives them. The handle IS safe in the path where an address is not - it is a sha256 already served only to an admin session, so an access log or proxy that captures it learns nothing new. 404 rather than 200 when nothing matched, so a stale screen (two admins, one list) says so instead of claiming success.", + "parameters": [ + { + "name": "hash", + "in": "path", + "required": true, + "schema": { + "type": "string" + }, + "description": "The address_hash the list returns for that row, 64 hex characters" + }, + { + "name": "channel", + "in": "query", + "description": "Defaults to email", + "required": false, + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "Lifted", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "removed": { + "type": "boolean" + } + } + } + } + } + }, + "400": { + "description": "Not a suppression handle", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + } + }, + "403": { + "description": "Not an admin", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + } + }, + "404": { + "description": "That address is not suppressed", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Error" + } + } + } + } + }, + "security": [ + { + "cookieAuth": [] + }, + { + "bearerAuth": [] + } + ] + } + }, "/api/v1/admin/engagement/templates": { "get": { "tags": [ diff --git a/server/test/engagementRetention.test.js b/server/test/engagementRetention.test.js new file mode 100644 index 0000000..a056e73 --- /dev/null +++ b/server/test/engagementRetention.test.js @@ -0,0 +1,234 @@ +// ── Retention: the engagement schema's sweep (ENGAGEMENT.md Phase 14) ────── +// +// The phase's acceptance line is that every one of the four tables has a stated +// policy — a sweep with a horizon, or a recorded decision that it does not +// expire — and that the cooldown horizon is CHECKED against the longest enabled +// rule rather than picked. Both are pinned here, plus the three things building +// it showed are silent when wrong: +// +// • **the outbox sweep is terminal-only.** A `scheduled` row is a message this +// deployment still intends to send; `delay_seconds` can legitimately put one +// a day out. Sweeping by age alone would cancel sends nobody cancelled, and +// the operator would see only that the mail never arrived. +// • **`reclaimStale` has to give up.** `MAX_ATTEMPTS` is consulted only on a +// graceful `retry` outcome, so before Phase 14 a send that killed the +// process mid-flight cycled sending → scheduled → sending forever, never +// reached a terminal status, and was therefore never eligible for ANY +// retention sweep. The bound depends on this. +// • **an unreadable setting means the default, not an exception.** The sweep +// runs on a timer with nobody watching. +// +// 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 retention = require('../src/model/engagement/engagementRetention.model') +const prune = require('../src/utils/engagementRetentionPrune') +const cooldownsDb = require('../src/model/engagement/engagementCooldowns.db') +const outboxDb = require('../src/model/engagement/engagementOutbox.db') +const sendsDb = require('../src/model/engagement/engagementSends.db') +const settings = require('../src/model/settings/settings.model') +const rulesDb = require('../src/model/engagement/engagementRules.db') +const db = require('../src/utils/db') + +after(() => db.close()) + +const saved = new Map() +function patch(mod, name, fn) { + if (!saved.has(mod)) saved.set(mod, new Map()) + if (!saved.get(mod).has(name)) saved.get(mod).set(name, mod[name]) + mod[name] = fn +} +function restore() { + for (const [mod, names] of saved) for (const [name, fn] of names) mod[name] = fn + saved.clear() +} + +let world + +beforeEach(() => { + world = { + settings: new Map(), + longestCooldown: 0, + // Every prune call, so a test can assert on the horizon it was given rather + // than only on the fact that something was deleted. + calls: { cooldowns: [], outbox: [], sends: [] }, + deleted: { cooldowns: 0, outbox: 0, sends: 0 }, + } + patch(settings, 'get', async (key) => { + if (!world.settings.has(key)) return null + return world.settings.get(key) + }) + patch(settings, 'set', async (key, value) => { world.settings.set(key, value) }) + patch(rulesDb, 'maxEnabledCooldownSeconds', async () => world.longestCooldown) + patch(cooldownsDb, 'prune', async (before, limit) => { + world.calls.cooldowns.push({ before, limit }) + return world.deleted.cooldowns + }) + patch(outboxDb, 'pruneTerminal', async (before, limit) => { + world.calls.outbox.push({ before, limit }) + return world.deleted.outbox + }) + patch(sendsDb, 'prune', async (before, limit) => { + world.calls.sends.push({ before, limit }) + return world.deleted.sends + }) +}) + +afterEach(() => restore()) + +const daysBetween = (now, before) => Math.round((now.getTime() - before.getTime()) / 86400000) + +// ── The policy ───────────────────────────────────────────────────────────── + +test('defaults are the shipped policy when nothing is stored', async () => { + const policy = await retention.get() + assert.deepEqual(policy, { sends: 180, cooldowns: 30, outbox: 30 }) +}) + +test('a stored value is used, and written back as a string', async () => { + await retention.set({ sends: 365 }, 7) + assert.equal(world.settings.get('engagement_sends_retain_days'), '365') + assert.equal((await retention.get()).sends, 365) +}) + +test('the PUT is sparse — an absent horizon is left alone', async () => { + await retention.set({ sends: 90 }) + await retention.set({ cooldowns: 45 }) + const policy = await retention.get() + assert.equal(policy.sends, 90, 'the earlier write survives the second call') + assert.equal(policy.cooldowns, 45) + assert.equal(policy.outbox, 30, 'and the untouched one is still the default') +}) + +test('out of range is refused rather than clamped', async () => { + // Clamping would leave the screen describing a policy the deployment is not + // running, which is worse than a visible error. + await assert.rejects(() => retention.set({ cooldowns: 1 }), /between 2 and 3650/) + await assert.rejects(() => retention.set({ sends: 5 }), /between 7 and 3650/) + await assert.rejects(() => retention.set({ outbox: 99999 }), /between 2 and 3650/) + await assert.rejects(() => retention.set({ sends: 12.5 }), /whole number/) + assert.equal(world.settings.size, 0, 'nothing was written') +}) + +test('an unknown key in the body is ignored, not rejected', async () => { + // So a client that posts the whole object back is not coupled to the list. + await retention.set({ sends: 200, suppressions: 30, nonsense: 1 }) + assert.equal((await retention.get()).sends, 200) + assert.equal(world.settings.has('engagement_suppressions_retain_days'), false) +}) + +test('a stored value outside the bounds falls back to the default', async () => { + // A row written before the bounds existed, or by hand. + world.settings.set('engagement_cooldowns_retain_days', '0') + assert.equal((await retention.get()).cooldowns, 30) + world.settings.set('engagement_cooldowns_retain_days', 'soon') + assert.equal((await retention.get()).cooldowns, 30) +}) + +test('an unreadable settings table yields defaults rather than throwing', async () => { + patch(settings, 'get', async () => { throw new Error('pool is dead') }) + assert.deepEqual(await retention.get(), { sends: 180, cooldowns: 30, outbox: 30 }) +}) + +// ── The cooldown guard ───────────────────────────────────────────────────── + +test('the cooldown horizon is checked against the longest ENABLED rule', async () => { + world.longestCooldown = 86400 // the validated maximum, one day + const ok = await retention.checkCooldownHorizon(30) + assert.equal(ok.ok, true) + assert.equal(ok.message, null) + + // A horizon shorter than a live cooldown means a pruned row makes the next + // fire a FIRST fire — the rule sends twice. + const bad = await retention.checkCooldownHorizon(0.5) + assert.equal(bad.ok, false) + assert.match(bad.message, /can send twice/) + assert.equal(bad.longestCooldownSeconds, 86400) +}) + +test('the horizon equal to the longest cooldown is refused, not accepted', async () => { + // Equality is the boundary where a row is pruned exactly as it stops being in + // force; the check has to be <=, not <. + world.longestCooldown = 2 * 86400 + assert.equal((await retention.checkCooldownHorizon(2)).ok, false) + assert.equal((await retention.checkCooldownHorizon(3)).ok, true) +}) + +test('no enabled rule means no warning at any horizon', async () => { + world.longestCooldown = 0 + assert.equal((await retention.checkCooldownHorizon(2)).ok, true) +}) + +// ── The sweep ────────────────────────────────────────────────────────────── + +test('one tick sweeps all three tables at their own horizons', async () => { + const now = new Date('2026-09-01T03:00:00Z') + const result = await prune.tick(now) + + assert.equal(daysBetween(now, world.calls.sends[0].before), 180) + assert.equal(daysBetween(now, world.calls.cooldowns[0].before), 30) + assert.equal(daysBetween(now, world.calls.outbox[0].before), 30) + assert.deepEqual(result.warnings, []) +}) + +test('each statement is bounded', async () => { + await prune.tick(new Date()) + for (const table of ['cooldowns', 'outbox', 'sends']) { + assert.equal(world.calls[table][0].limit, prune.BATCH, `${table} is batched`) + } +}) + +test('a sweep repeats while its batches come back full, and stops', async () => { + world.deleted.sends = prune.BATCH + const result = await prune.tick(new Date()) + assert.equal(world.calls.sends.length, prune.MAX_BATCHES, 'it stops rather than looping forever') + assert.equal(result.sends, prune.BATCH * prune.MAX_BATCHES) +}) + +test('one failing table does not stop the other two', async () => { + patch(outboxDb, 'pruneTerminal', async () => { throw new Error('lock wait timeout') }) + const result = await prune.tick(new Date()) + assert.equal(result.outbox, 0) + assert.equal(world.calls.cooldowns.length, 1, 'cooldowns still swept') + assert.equal(world.calls.sends.length, 1, 'the send log still swept') +}) + +test('the sweep runs even when the cooldown horizon is too short, and warns', async () => { + // Deliberate: refusing to prune would trade a bounded, describable fault (one + // rule may re-fire early) for the unbounded one this phase exists to end. + world.longestCooldown = 86400 + world.settings.set('engagement_cooldowns_retain_days', '2') + const result = await prune.tick(new Date()) + assert.equal(result.warnings.length, 0, 'two days clears a one-day cooldown') + + world.longestCooldown = 30 * 86400 // longer than any rule can actually save + const second = await prune.tick(new Date()) + assert.equal(second.warnings.length, 1) + assert.equal(world.calls.cooldowns.length, 2, 'it swept anyway') +}) + +test('an unreadable policy skips the run rather than sweeping on a guess', async () => { + patch(retention, 'get', async () => { throw new Error('pool is dead') }) + const result = await prune.tick(new Date()) + assert.deepEqual(result, { cooldowns: 0, outbox: 0, sends: 0, warnings: [] }) + assert.equal(world.calls.sends.length, 0, 'nothing was deleted') +}) + +test('start/stop is idempotent and leaves no live timer', () => { + prune.start() + prune.start() + prune.stop() + prune.stop() +}) + +// The SQL these depend on — terminal-only, the give-up-before-reclaim order and +// the batch LIMIT — is proved against a real server in engagementRetentionSql.test.js. +// It cannot be proved here: the db modules destructure `query` at require time, +// so there is nothing left to stub, and a hand-rolled stand-in would prove only +// that two readings of the manual agree (which is exactly the trap Phase 4a's +// `foundRows` defect sprang). diff --git a/server/test/engagementRetentionSql.test.js b/server/test/engagementRetentionSql.test.js new file mode 100644 index 0000000..7e741e1 --- /dev/null +++ b/server/test/engagementRetentionSql.test.js @@ -0,0 +1,318 @@ +// ── The retention sweep, against a real MariaDB ──────────────────────────── +// +// ENGAGEMENT.md Phase 14. `engagementRetention.test.js` proves the policy and +// the worker's control flow against stubs, which is right for everything the +// worker DECIDES. It cannot prove the three statements whose whole correctness +// is what the server does with them, and this is the file for those — same shape +// and same reasoning as `engagementEngineSql.test.js`, which exists because a +// stub written from the same reading of the manual proves the reading, not the +// server (that is how the `foundRows` cooldown defect got as far as it did). +// +// It is also Phase 14's acceptance rig, stated in the phase: *"a rig run shows +// the sweep deleting terminal rows while leaving a `scheduled` outbox row and an +// in-window send-log row alone."* +// +// **It SKIPS when there is no database**, deliberately: CI runs the suite with +// the pool pointed at a dead port, and a file that failed there would make every +// PR red for a reason unrelated to itself. Run it against this machine's +// container with: +// +// DB_HOST=127.0.0.1 DB_PORT=3306 DB_USER=... DB_PASSWORD=... \ +// node --test test/engagementRetentionSql.test.js +// +// It creates its tables in a throwaway database named after the process and +// drops it again, so it can never touch a real schema. + +const { test, before, after } = require('node:test') +const assert = require('node:assert/strict') +const mariadb = require('mariadb') + +// The three tables under sweep, with the two indexes Phase 14 added. Copied +// rather than required, for the reason the engine's SQL file gives: requiring the +// modules would drag in `utils/db`'s pool. +const SCHEMA = ` +CREATE TABLE engagement_cooldowns ( + rule_id INT NOT NULL, + user_id INT NOT NULL, + subject_key VARCHAR(190) NOT NULL DEFAULT '', + channel VARCHAR(32) NOT NULL DEFAULT '', + last_fired_at DATETIME NOT NULL, + fire_count INT NOT NULL DEFAULT 1, + PRIMARY KEY (rule_id, user_id, subject_key, channel), + INDEX idx_engc_sweep (last_fired_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; +CREATE TABLE engagement_outbox ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + rule_id INT NOT NULL, + trigger_id VARCHAR(96) NOT NULL, + user_id INT NOT NULL, + channel VARCHAR(32) NOT NULL, + subject_key VARCHAR(190) NOT NULL DEFAULT '', + payload JSON NOT NULL, + dedupe_key VARCHAR(190) NULL, + status ENUM('scheduled','sending','sent','failed','cancelled','suppressed') NOT NULL DEFAULT 'scheduled', + due_at DATETIME NOT NULL, + attempts SMALLINT NOT NULL DEFAULT 0, + last_error TEXT NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + sent_at DATETIME NULL, + INDEX idx_engo_due (status, due_at), + INDEX idx_engo_sweep (status, created_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; +CREATE TABLE engagement_sends ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + rule_id INT NULL, + trigger_id VARCHAR(96) NOT NULL, + user_id INT NULL, + channel VARCHAR(32) NOT NULL, + status ENUM('sent','failed','suppressed','bounced','complained') NOT NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + INDEX idx_engs_rule_window (rule_id, created_at), + INDEX idx_engs_sweep (created_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; +` + +// Verbatim from the three `.db` files. If one of these drifts from its source the +// test still passes and proves nothing, which is the standing cost of the copy — +// the alternative (requiring the modules) costs a live pool on a dead port. +const PRUNE_COOLDOWNS = 'DELETE FROM engagement_cooldowns WHERE last_fired_at < ? LIMIT ?' +const PRUNE_OUTBOX = ` +DELETE FROM engagement_outbox + WHERE status IN ('sent', 'failed', 'cancelled', 'suppressed') + AND created_at < ? + LIMIT ?` +const PRUNE_SENDS = 'DELETE FROM engagement_sends WHERE created_at < ? LIMIT ?' +const GIVE_UP = ` +UPDATE engagement_outbox + SET status = 'failed', last_error = 'gave up after repeated interruptions' + WHERE status = 'sending' AND updated_at < ? AND attempts >= ?` +const RECLAIM = "UPDATE engagement_outbox SET status = 'scheduled' WHERE status = 'sending' AND updated_at < ?" + +const DB = `rg_retain_test_${process.pid}` +let pool = null +let available = false + +before(async () => { + const admin = mariadb.createPool({ + host: process.env.DB_HOST || '127.0.0.1', + port: Number(process.env.DB_PORT) || 3306, + user: process.env.DB_USER || 'root', + password: process.env.DB_PASSWORD || '', + connectionLimit: 1, + connectTimeout: 2000, + initializationTimeout: 2000, + multipleStatements: true, + }) + try { + await admin.query(`CREATE DATABASE ${DB}`) + available = true + } catch { + available = false + } finally { + await admin.end().catch(() => {}) + } + if (!available) return + + pool = mariadb.createPool({ + host: process.env.DB_HOST || '127.0.0.1', + port: Number(process.env.DB_PORT) || 3306, + user: process.env.DB_USER || 'root', + password: process.env.DB_PASSWORD || '', + database: DB, + connectionLimit: 3, + multipleStatements: true, + bigIntAsNumber: true, + insertIdAsNumber: true, + }) + await pool.query(SCHEMA) +}) + +after(async () => { + if (pool) { + await pool.query(`DROP DATABASE IF EXISTS ${DB}`).catch(() => {}) + await pool.end().catch(() => {}) + } +}) + +// Checked INSIDE each test, never as a `{ skip }` option — the option is +// evaluated when the file is read, before `before()` has found out whether there +// is a database, and every test skipped unconditionally looks like a pass. +const SKIP = 'no database reachable - set DB_HOST/DB_PORT/DB_USER/DB_PASSWORD to run' +const needDb = (t) => { + if (available) return false + t.skip(SKIP) + return true +} + +const NOW = new Date('2026-09-01T12:00:00Z') +const daysAgo = (d) => new Date(NOW.getTime() - d * 86400000) + +const clear = async () => { + await pool.query('DELETE FROM engagement_cooldowns') + await pool.query('DELETE FROM engagement_outbox') + await pool.query('DELETE FROM engagement_sends') +} + +const addOutbox = (status, createdAt, extra = {}) => + pool.query( + `INSERT INTO engagement_outbox + (rule_id, trigger_id, user_id, channel, payload, due_at, status, created_at, updated_at, attempts) + VALUES (1, 't', 1, 'email', '{}', ?, ?, ?, ?, ?)`, + [ + extra.dueAt || createdAt, + status, + createdAt, + extra.updatedAt || createdAt, + extra.attempts ?? 0, + ], + ) + +// ── The acceptance rig ───────────────────────────────────────────────────── + +test('the outbox sweep deletes terminal rows and leaves a scheduled one alone', async (t) => { + if (needDb(t)) return + await clear() + + // Four terminal rows, well past the horizon. + for (const status of ['sent', 'failed', 'cancelled', 'suppressed']) { + await addOutbox(status, daysAgo(90)) + } + // The row the phase names: old, and still a promise this deployment has made. + await addOutbox('scheduled', daysAgo(90)) + // And one in flight, which is a worker's row and not the sweep's business. + await addOutbox('sending', daysAgo(90)) + + const result = await pool.query(PRUNE_OUTBOX, [daysAgo(30), 1000]) + assert.equal(Number(result.affectedRows), 4, 'exactly the four terminal rows') + + const left = await pool.query('SELECT status FROM engagement_outbox ORDER BY status') + assert.deepEqual(left.map((r) => r.status), ['scheduled', 'sending']) +}) + +test('an in-window row is left alone whatever its status', async (t) => { + if (needDb(t)) return + await clear() + await addOutbox('sent', daysAgo(29)) + await addOutbox('sent', daysAgo(31)) + + const result = await pool.query(PRUNE_OUTBOX, [daysAgo(30), 1000]) + assert.equal(Number(result.affectedRows), 1) + const [row] = await pool.query('SELECT created_at FROM engagement_outbox') + assert.ok(row, 'the row inside the horizon survived') +}) + +test('the send-log sweep leaves an in-window row alone', async (t) => { + if (needDb(t)) return + await clear() + const add = (createdAt) => pool.query( + `INSERT INTO engagement_sends (rule_id, trigger_id, user_id, channel, status, created_at) + VALUES (1, 't', 1, 'email', 'sent', ?)`, + [createdAt], + ) + await add(daysAgo(200)) + await add(daysAgo(179)) + // The row the per-rule hourly ceiling counts. If a horizon could reach this, + // the ceiling would silently stop capping anything. + await add(new Date(NOW.getTime() - 60 * 1000)) + + const result = await pool.query(PRUNE_SENDS, [daysAgo(180), 1000]) + assert.equal(Number(result.affectedRows), 1) + const [{ n }] = await pool.query('SELECT COUNT(*) AS n FROM engagement_sends') + assert.equal(Number(n), 2) +}) + +test('the cooldown sweep respects its batch limit and repeats', async (t) => { + if (needDb(t)) return + await clear() + for (let i = 0; i < 5; i += 1) { + await pool.query( + `INSERT INTO engagement_cooldowns (rule_id, user_id, subject_key, channel, last_fired_at) + VALUES (1, ?, '', 'email', ?)`, + [i, daysAgo(90)], + ) + } + const first = await pool.query(PRUNE_COOLDOWNS, [daysAgo(30), 2]) + assert.equal(Number(first.affectedRows), 2, 'LIMIT bounds one statement') + const second = await pool.query(PRUNE_COOLDOWNS, [daysAgo(30), 10]) + assert.equal(Number(second.affectedRows), 3, 'and the rest go on the next pass') +}) + +// ── The bound the sweep depends on ───────────────────────────────────────── + +test('a stale row that has burned its attempts is failed, not handed back', async (t) => { + if (needDb(t)) return + await clear() + // Two rows stranded in 'sending' by a crash between the claim and the outcome. + // One has attempts left; the other has spent them, and before Phase 14 nothing + // could ever move it — MAX_ATTEMPTS is consulted only on a graceful `retry`, + // so it cycled sending → scheduled → sending forever, never became terminal, + // and was therefore never eligible for any sweep. + await addOutbox('sending', daysAgo(1), { updatedAt: daysAgo(1), attempts: 1 }) + await addOutbox('sending', daysAgo(1), { updatedAt: daysAgo(1), attempts: 5 }) + + const gaveUp = await pool.query(GIVE_UP, [new Date(NOW.getTime() - 15 * 60 * 1000), 5]) + assert.equal(Number(gaveUp.affectedRows), 1, 'only the exhausted row') + const reclaimed = await pool.query(RECLAIM, [new Date(NOW.getTime() - 15 * 60 * 1000)]) + assert.equal(Number(reclaimed.affectedRows), 1, 'and only the other one comes back') + + const rows = await pool.query('SELECT status, attempts FROM engagement_outbox ORDER BY attempts') + assert.deepEqual(rows.map((r) => r.status), ['scheduled', 'failed']) + + // And now it is terminal, so the sweep can bound it. + await pool.query("UPDATE engagement_outbox SET created_at = ? WHERE status = 'failed'", [daysAgo(90)]) + const swept = await pool.query(PRUNE_OUTBOX, [daysAgo(30), 1000]) + assert.equal(Number(swept.affectedRows), 1) +}) + +test('reclaiming before giving up would loop forever — the order is the fix', async (t) => { + if (needDb(t)) return + await clear() + await addOutbox('sending', daysAgo(1), { updatedAt: daysAgo(1), attempts: 5 }) + + // The wrong order, run deliberately: reclaim first, and the exhausted row is + // back in 'scheduled' where findDue will pick it up again. + await pool.query(RECLAIM, [new Date(NOW.getTime() - 15 * 60 * 1000)]) + const gaveUp = await pool.query(GIVE_UP, [new Date(NOW.getTime() - 15 * 60 * 1000), 5]) + assert.equal(Number(gaveUp.affectedRows), 0, 'nothing left in sending to fail') + const [row] = await pool.query('SELECT status FROM engagement_outbox') + assert.equal(row.status, 'scheduled', 'which is the forever-retry this ordering avoids') +}) + +// ── The indexes ──────────────────────────────────────────────────────────── + +test('each sweep is an index range scan, not a table scan', async (t) => { + if (needDb(t)) return + await clear() + // A plan on an empty table is not worth reading, so give the optimizer rows. + for (let i = 0; i < 200; i += 1) { + await addOutbox(i % 2 ? 'sent' : 'scheduled', daysAgo(i)) + await pool.query( + `INSERT INTO engagement_sends (rule_id, trigger_id, user_id, channel, status, created_at) + VALUES (1, 't', 1, 'email', 'sent', ?)`, + [daysAgo(i)], + ) + } + await pool.query('ANALYZE TABLE engagement_outbox') + await pool.query('ANALYZE TABLE engagement_sends') + + const outboxPlan = await pool.query( + `EXPLAIN SELECT id FROM engagement_outbox + WHERE status IN ('sent','failed','cancelled','suppressed') AND created_at < ?`, + [daysAgo(30)], + ) + assert.ok( + String(outboxPlan[0].key || '').includes('idx_engo'), + `expected an index, got ${JSON.stringify(outboxPlan[0])}`, + ) + + const sendsPlan = await pool.query( + 'EXPLAIN SELECT id FROM engagement_sends WHERE created_at < ?', + [daysAgo(180)], + ) + assert.equal( + sendsPlan[0].key, + 'idx_engs_sweep', + `every other index on this table has created_at in SECOND position: ${JSON.stringify(sendsPlan[0])}`, + ) +})