RunicGateway/Module-Rust
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

feat(rust): notifications and engagement (phase 10, protocol 7)

Browse Source 
Registers the engagement set R7 put in v1: thirteen triggers, four push
streams, three audiences, four bodies (two triggers, email and in-app)
and thirteen disabled rules in seven groups (PLAN.md §25, D59-D68).

The raid alert goes to everyone authorised on the tool cupboard, one
emit per linked person with ownerUserId, so the owner ceiling holds per
emit. It covers doors and walls (protocol 7), never names the raider,
alerts nobody when there is no cupboard, and carries ownerOnline so
"offline only" is the seeded rule's condition rather than code.

The fan-out runs off ingest before a frame is applied, since applying a
disband deletes the roster the notice is sent to. A replayed event is
told only while it is news: 15 minutes for broadcasts, 24 hours for
personal and staff events. Dedupe keys come from the event, not the
sidecar's row id. Server online/offline and a new kills leader are
in-memory transitions, never on first sight, and a tie is not a lead.
A login with no approval within a minute becomes a staff notice via a
query, so a restart loses nothing.

Also fixes a phase-4 gap (D68): the refresh now asks /health, so a game
that hung, or whose bridge was unloaded, while the sidecar stayed up no
longer reads as online. It stops naming players as online, and a stale
board no longer moves "last seen".

engagement-triggers.json is the committed freeze of all of it, checked
in CI with line endings normalised. The check was verified by breaking
it both ways.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01E14m6SuuY6i1vASFeGDBeY
This commit is contained in:
claude
2026-09-23 06:06:08 -05:00
parent dc3c9689b4
commit 285db0baa7
22 changed files with 3256 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

490
server/engagement/emit.js Normal file
View File

@@ -0,0 +1,490 @@
// ── What happened, told to core's engagement engine ───────────────────────
//
// The fan-out behind `triggers.js` (PLAN.md §25). Ingest calls `onEvent` for
// every frame it stores; the refresh calls `serverObserved` for every poll;
// ingest calls `checkLeader` after a batch; the prune timer calls
// `sweepLoginDenied`; the link route calls `linked`.
//
// **Nothing here decides who is told.** It says what happened and, for a
// personal or clan event, who it is ABOUT. Core applies the rule, the ceiling,
// the preference, the suppression list and the verification gate. A module
// cannot send mail (MODULE_API.md §2.7), and this file is not the back door.
//
// **Nothing here throws into its caller.** Every entry point catches, because
// its callers are the ingest cursor and the refresh loop — a malformed frame or
// a core-side contract problem must cost one notification and never the feed.
//
// ── Three rules the whole file follows ────────────────────────────────────
//
// 1. **Emit on the transition, never on the poll** (R7). A server that is
// still up is not news. Transitions are tracked in memory, and a FIRST
// sighting is never one — so a website restart announces nothing.
//
// 2. **A replayed event notifies only while it is still news** (D63). After an
// outage the cursor replays hours of frames. A broadcast older than 15
// minutes tells nobody; a personal or staff event is kept for 24 hours,
// because "your base was raided at 03:10" is still true and still wanted.
//
// 3. **Every emit carries a dedupe key made from the EVENT, not the store.**
// Core's outbox is unique on (rule, user, channel, key), so the same frame
// replayed after a crash is a no-op. The key is built from what the event
// says — its server, time and subject — rather than from the sidecar's row
// id, because a sidecar whose database is replaced starts its ids again
// and would otherwise have every new alert swallowed as a repeat of an old
// one.
const crypto = require('crypto')
const core = require('../core')
const clans = require('../model/clans/clans.model')
const clansDb = require('../model/clans/clans.db')
const eventsDb = require('../model/events/events.db')
const linksDb = require('../model/links/links.db')
const serversDb = require('../model/servers/servers.db')
const { TRIGGER_IDS: T, PATHS, serverPath, leaderboardPath, clanPath } = require('./triggers')
const log = core.logger('engagement')
/** How old a broadcast may be and still be news (D63). */
const BROADCAST_MAX_AGE_MS = 15 * 60 * 1000
/** How old a personal or staff event may be and still be worth telling (D63). */
const PERSONAL_MAX_AGE_MS = 24 * 60 * 60 * 1000
/**
* How long a login attempt waits for its approval before it counts as denied
* (D64, PLAN.md §16.5). The game raises no rejection hook, so a denial is the
* ABSENCE of an approval — which is only knowable after a wait.
*/
const LOGIN_APPROVAL_WINDOW_MS = 60 * 1000
/** How far BEFORE an attempt an approval may be stamped and still answer it — clock grain, not policy. */
const LOGIN_APPROVAL_SLACK_MS = 5 * 1000
const STRUCTURE_LABELS = Object.freeze({
block: 'building block',
door: 'door',
wall: 'external wall',
cupboard: 'tool cupboard',
})
// ── Small helpers ──────────────────────────────────────────────────────────
const str = (value) => (value === undefined || value === null || value === '' ? undefined : String(value))
/** `rust:<what>:<sha1 of the parts>` — readable prefix, bounded length. */
function dedupeKey(what, ...parts) {
const digest = crypto.createHash('sha1').update(parts.map((p) => String(p ?? '')).join('\u0000')).digest('hex')
return `rust:${what}:${digest}`
}
function frameTime(item, frame) {
const t = Number(frame && frame.t) || Number(item && item.t)
return Number.isFinite(t) && t > 0 ? t : Date.now()
}
/** D63, as a question: is an event from `t` still worth telling, for this family? */
function stillNews(t, maxAgeMs, now = Date.now()) {
return now - t <= maxAgeMs
}
function serverVars(server) {
const serverId = String(server.id)
return { serverId, server: server.name || serverId, serverUrl: serverPath(serverId) }
}
/**
* Hands one event to core. Never throws.
*
* Core throws on a contract mismatch outside production, which is how a
* declaration and an emitter drifting apart is meant to be found. It is logged
* at `error` here rather than re-thrown, because the caller is the ingest
* cursor — and an `error` line is what a rig walk reads.
*/
function fire(triggerId, envelope) {
try {
core.emit(triggerId, envelope)
return true
} catch (err) {
log.error('core refused an emit', { trigger: triggerId, error: err.message })
return false
}
}
/** Steam id -> website user id, for the ids that are linked. Unlinked ones are simply absent. */
async function usersFor(steamIds) {
const ids = [...new Set((steamIds || []).map(String).filter(Boolean))]
if (!ids.length) return new Map()
const rows = await linksDb.userIdsForSteamIds(ids)
return new Map(rows.map((r) => [String(r.steamId), Number(r.userId)]))
}
// ── Per-kind handlers ──────────────────────────────────────────────────────
/**
* The raid alert (D59-D61, D66, D67).
*
* One emit per authorised, LINKED person, each with `ownerUserId` — so the
* `owner` ceiling holds per emit and "nobody else" is structural rather than a
* filter somebody could forget. Two Steam accounts held by one website user are
* one person: they get one alert, online if either account is.
*/
async function onRaid(server, item, frame) {
const t = frameTime(item, frame)
if (!stillNews(t, PERSONAL_MAX_AGE_MS)) return 0
// D67: no cupboard, nobody to tell. Absent — not empty — is also what a
// protocol-6 plugin sends, so a half-upgraded deployment alerts nobody rather
// than guessing an owner from the placer.
if (!frame.buildingId || !Array.isArray(frame.authorized)) return 0
const authorized = frame.authorized.filter((a) => a && a.steamId)
const attacker = str(frame.attackerId)
// An authorised attacker is demolishing their own base, or a teammate's.
if (attacker && authorized.some((a) => String(a.steamId) === attacker)) return 0
const users = await usersFor(authorized.map((a) => a.steamId))
if (!users.size) return 0
const byUser = new Map()
for (const a of authorized) {
const userId = users.get(String(a.steamId))
if (!userId) continue
byUser.set(userId, byUser.get(userId) === true || a.online === true)
}
const base = {
...serverVars(server),
building: String(frame.buildingId),
structure: STRUCTURE_LABELS[frame.structure] || 'structure',
grid: str(frame.grid),
atGrid: str(frame.grid) ? ` in ${frame.grid}` : undefined,
}
const key = dedupeKey('raid', server.id, frame.buildingId, t, frame.prefab)
let sent = 0
for (const [userId, online] of byUser) {
if (fire(T['rust.base.destroyed'], {
data: { ...base, ownerOnline: online },
ownerUserId: userId,
dedupeKey: key,
occurredAt: t,
})) sent += 1
}
return sent
}
async function onWipe(server, item, frame) {
const t = frameTime(item, frame)
if (!stillNews(t, BROADCAST_MAX_AGE_MS)) return 0
const wipeId = str(frame.wipeId)
if (!wipeId) return 0
return fire(T['rust.wipe.started'], {
data: { ...serverVars(server), wipeId },
dedupeKey: dedupeKey('wipe', server.id, wipeId),
occurredAt: t,
}) ? 1 : 0
}
/**
* Clan departures and disbands. Recipients travel on the envelope, because
* "the clan this was about" is a different set every firing.
*
* Nobody is told about what they did themselves: the leaver is not told they
* left, the one who kicked is not told they kicked, the one who disbanded is not
* told they disbanded. The one KICKED is told — it happened to them.
*/
async function onClan(server, item, frame) {
const t = frameTime(item, frame)
if (!stillNews(t, PERSONAL_MAX_AGE_MS)) return 0
const externalId = await clans.resolveExternalId(server.id, frame)
if (!externalId) return 0
const kind = frame.kind
const subject = str(frame.steamId)
let steamIds
let actor
if (kind === 'clan.disbanded') {
// From the frame (protocol 7): by the time this runs the next board may
// already have removed the roster the store would answer with.
steamIds = Array.isArray(frame.members) ? frame.members.map(String) : null
if (!steamIds) steamIds = (await clansDb.listMembers(externalId)).map((m) => String(m.steamId))
actor = subject
} else {
steamIds = (await clansDb.listMembers(externalId)).map((m) => String(m.steamId))
if (kind === 'clan.member.kicked') {
if (subject) steamIds.push(subject)
actor = str(frame.bySteamId)
} else {
actor = subject
}
}
const users = await usersFor(steamIds.filter((id) => id !== actor))
const recipientUserIds = [...new Set(users.values())]
if (!recipientUserIds.length) return 0
const data = {
...serverVars(server),
clanKey: externalId,
clan: str(frame.clanName) || 'your clan',
clanUrl: clanPath(externalId),
}
if (kind === 'clan.member.left') data.member = str(frame.name)
if (kind === 'clan.member.kicked') {
data.member = str(frame.name)
data.by = str(frame.byName)
}
if (kind === 'clan.disbanded') data.by = str(frame.name)
const triggerId = kind === 'clan.disbanded' ? T['rust.clan.disbanded'] : T[`rust.${kind}`]
return fire(triggerId, {
data,
recipientUserIds,
dedupeKey: dedupeKey(kind, server.id, externalId, subject, t),
occurredAt: t,
}) ? 1 : 0
}
async function onReported(server, item, frame) {
const t = frameTime(item, frame)
if (!stillNews(t, PERSONAL_MAX_AGE_MS)) return 0
const steamId = str(frame.targetId)
if (!steamId) return 0
return fire(T['rust.player.reported'], {
data: {
...serverVars(server),
steamId,
player: str(frame.targetName),
reporter: str(frame.reporterName),
reportType: str(frame.reportType),
topic: str(frame.subject),
message: str(frame.message),
},
dedupeKey: dedupeKey('reported', server.id, steamId, frame.reporterId, t),
occurredAt: t,
}) ? 1 : 0
}
async function onBan(server, item, frame) {
const t = frameTime(item, frame)
if (!stillNews(t, PERSONAL_MAX_AGE_MS)) return 0
const steamId = str(frame.steamId)
if (!steamId) return 0
const banned = frame.kind === 'player.banned'
const data = { ...serverVars(server), steamId, player: str(frame.name) }
// The address the frame carries is deliberately NOT copied: no trigger
// declares one, so no template can ever put it in a mail.
if (banned) data.reason = str(frame.reason)
return fire(banned ? T['rust.player.banned'] : T['rust.player.unbanned'], {
data,
dedupeKey: dedupeKey(frame.kind, server.id, steamId, t),
occurredAt: t,
}) ? 1 : 0
}
const HANDLERS = Object.freeze({
'entity.destroyed': onRaid,
'server.wipe': onWipe,
'clan.member.left': onClan,
'clan.member.kicked': onClan,
'clan.disbanded': onClan,
'player.reported': onReported,
'player.banned': onBan,
'player.unbanned': onBan,
})
/**
* One stored frame. Called by ingest BEFORE the frame is applied, because
* applying a disband deletes the roster a clan notification is sent to.
*
* @returns {Promise<number>} emits handed to core, for the log and the tests
*/
async function onEvent(server, item) {
const frame = (item && item.frame) || {}
const kind = (item && item.kind) || frame.kind
const handler = HANDLERS[kind]
if (!handler || !server) return 0
try {
return await handler(server, item, { ...frame, kind })
} catch (err) {
log.warn('could not raise a notification', { server: server.id, kind, error: err.message })
return 0
}
}
// ── Transitions tracked in memory ──────────────────────────────────────────
//
// Deliberately NOT persisted. The question each one answers is "has THIS
// process seen a previous value", and a value restored from the database would
// make the first poll after a restart a transition against state the game may
// have left hours ago.
const tracker = { online: new Map(), leader: new Map() }
/** Forget every tracked value. For the tests. */
function reset() {
tracker.online.clear()
tracker.leader.clear()
}
/**
* One poll's verdict on one server (D68): is its game connected now?
*
* Synchronous and fire-and-forget — the refresh must not wait on core.
*/
function serverObserved(server, connected) {
try {
if (!server) return 0
const id = String(server.id)
const now = Boolean(connected)
const before = tracker.online.get(id)
tracker.online.set(id, now)
// First sight is never a transition: a restart announces nothing.
if (before === undefined || before === now) return 0
return fire(now ? T['rust.server.online'] : T['rust.server.offline'], {
data: serverVars(server),
// A transition observed by a poll is observed NOW, so it needs no age check;
// the key is per minute so that one real flap is one event even if two
// polls land either side of a restart of this process.
dedupeKey: dedupeKey(now ? 'online' : 'offline', id, Math.floor(Date.now() / 60000)),
}) ? 1 : 0
} catch (err) {
log.warn('could not raise a server transition', { server: server && server.id, error: err.message })
return 0
}
}
/**
* After a batch: has somebody new taken the lead in this wipe's kills? (D64)
*
* Only a STRICT lead counts. The leaderboard breaks a tie on who was seen last,
* so two players level on kills trade the top row every time either one moves —
* and reading the top row alone would announce a new leader each time.
*/
async function checkLeader(server) {
try {
const state = await serversDb.getState(server.id)
const wipeId = state && state.wipeId
if (!wipeId) return 0
const rows = await eventsDb.leaderboard({ serverId: server.id, wipeId, sort: 'kills', limit: 2 })
const top = rows[0]
const kills = top ? Number(top.kills) || 0 : 0
const id = String(server.id)
const before = tracker.leader.get(id)
if (!top || kills <= 0) {
tracker.leader.set(id, { wipeId, steamId: null })
return 0
}
const tied = rows[1] && Number(rows[1].kills) === kills
const steamId = String(top.steamId)
// First sight, or a new wipe: remember, announce nothing.
if (!before || before.wipeId !== wipeId) {
tracker.leader.set(id, { wipeId, steamId: tied ? null : steamId })
return 0
}
if (tied || before.steamId === steamId) return 0
tracker.leader.set(id, { wipeId, steamId })
return fire(T['rust.leaderboard.topped'], {
data: {
...serverVars(server),
leader: str(top.name) || 'A player',
kills,
leaderboardUrl: leaderboardPath(id),
},
dedupeKey: dedupeKey('leader', id, wipeId, steamId, kills),
}) ? 1 : 0
} catch (err) {
log.warn('could not check the leaderboard', { server: server && server.id, error: err.message })
return 0
}
}
/**
* Login attempts that were never approved (D64).
*
* A query over what is stored rather than a timer per attempt, so a restart
* loses nothing and running it twice is a no-op (the key is the attempt's own
* server, Steam id and time). Bounded by D63's personal age: an attempt a day
* old is not worth a staff mail.
*/
async function sweepLoginDenied(servers, now = Date.now()) {
let sent = 0
for (const server of servers || []) {
try {
const rows = await eventsDb.unapprovedLogins({
serverId: server.id,
from: now - PERSONAL_MAX_AGE_MS,
to: now - LOGIN_APPROVAL_WINDOW_MS,
windowMs: LOGIN_APPROVAL_WINDOW_MS,
slackMs: LOGIN_APPROVAL_SLACK_MS,
})
for (const row of rows) {
const t = Number(row.t)
if (fire(T['rust.login.denied'], {
data: {
...serverVars(server),
steamId: String(row.steamId),
player: str(row.name),
attemptedAt: new Date(t),
},
dedupeKey: dedupeKey('login-denied', server.id, row.steamId, t),
occurredAt: t,
})) sent += 1
}
} catch (err) {
log.warn('could not sweep login attempts', { server: server && server.id, error: err.message })
}
}
return sent
}
/** A Steam account was just linked (R1). Called by the link route, once, on a NEW link. */
function linked({ userId, steamId, name }) {
try {
const uid = Number(userId)
if (!Number.isInteger(uid) || uid < 1 || !steamId) return 0
return fire(T['rust.player.linked'], {
data: { steamId: String(steamId), player: str(name), accountUrl: PATHS.account },
ownerUserId: uid,
dedupeKey: dedupeKey('linked', steamId, uid),
}) ? 1 : 0
} catch (err) {
log.warn('could not raise the link notification', { error: err.message })
return 0
}
}
module.exports = {
onEvent,
serverObserved,
checkLeader,
sweepLoginDenied,
linked,
reset,
dedupeKey,
stillNews,
BROADCAST_MAX_AGE_MS,
PERSONAL_MAX_AGE_MS,
LOGIN_APPROVAL_WINDOW_MS,
STRUCTURE_LABELS,
}