Module-Rust/server/engagement/emit.js

// ── 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 mapDb = require('../model/map/map.db')
const worldEvents = require('../worldEvents')
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) }
}

/**
 * The headline every trigger carries (`triggers.js` HEADLINE).
 *
 * Core's generic bodies fall back to a trigger's LABEL and DESCRIPTION when the
 * payload has no `title`/`intro`, and on a multi-server site that fallback says
 * "A server came online" without ever saying which. So the sentence is written
 * here, from the payload, and core renders it. Plain register, no conditionals:
 * a missing part falls back to a neutral word rather than leaving a hole.
 */
const HEADLINES = Object.freeze({
  'rust.base.destroyed': (d) => ({
    title: `Your base on ${d.server} is being raided`,
    intro: `A ${d.structure} was destroyed${d.atGrid || ''} on ${d.server}.`,
  }),
  'rust.wipe.started': (d) => ({
    title: `${d.server} has wiped`,
    intro: `A new wipe has started on ${d.server}: a fresh map, and a fresh start for everyone.`,
  }),
  'rust.server.online': (d) => ({
    title: `${d.server} is online`,
    intro: `${d.server} is back up and talking to the website.`,
  }),
  'rust.server.offline': (d) => ({
    title: `${d.server} is offline`,
    intro: `${d.server} stopped, or stopped talking to the website.`,
  }),
  'rust.leaderboard.topped': (d) => ({
    title: `${d.leader} leads ${d.server}`,
    intro: `${d.leader} now leads this wipe's kills on ${d.server}, with ${d.kills}.`,
  }),
  'rust.player.linked': (d) => ({
    title: 'A Steam account was linked to your account',
    intro: `The Steam account ${d.player || d.steamId} was linked with an in-game code. `
      + 'If that was not you, unlink it from your Rust account page.',
  }),
  'rust.kit.entitled': (d) => ({
    title: `You earned ${d.kit} on ${d.server}`,
    intro: `An event on ${d.server} rewarded you: the ${d.kit} kit is waiting in the Kits menu, `
      + 'with one extra use. Redeem it in game.',
  }),
  'rust.clan.member.left': (d) => ({
    title: `${d.member || 'A member'} left ${d.clan}`,
    intro: `${d.member || 'A member'} left ${d.clan} on ${d.server}.`,
  }),
  'rust.clan.member.kicked': (d) => ({
    title: `${d.member || 'A member'} was removed from ${d.clan}`,
    intro: `${d.by || 'A clan leader'} removed ${d.member || 'a member'} from ${d.clan} on ${d.server}.`,
  }),
  'rust.clan.disbanded': (d) => ({
    title: `${d.clan} was disbanded`,
    intro: `${d.by || 'A clan leader'} disbanded ${d.clan} on ${d.server}.`,
  }),
  'rust.player.reported': (d) => ({
    title: `${d.player || d.steamId} was reported on ${d.server}`,
    intro: `${d.reporter || 'A player'} reported ${d.player || d.steamId}`
      + `${d.reportType ? ` (${d.reportType})` : ''}${d.topic ? `: ${d.topic}` : '.'}`,
  }),
  'rust.player.banned': (d) => ({
    title: `${d.player || d.steamId} was banned on ${d.server}`,
    intro: d.reason ? `Reason given: ${d.reason}` : 'No reason was given.',
  }),
  'rust.player.unbanned': (d) => ({
    title: `${d.player || d.steamId} was unbanned on ${d.server}`,
    intro: `The ban on ${d.player || d.steamId} (${d.steamId}) was lifted.`,
  }),
  'rust.npc.died': (d) => ({
    title: `${d.npc || d.profile} was killed on ${d.server}`,
    intro: `${d.killer ? `${d.killer} killed` : 'Something killed'} ${d.npc || 'an NPC'} (${d.profile}) on ${d.server}.`,
  }),
  'rust.npc.health': (d) => ({
    title: `${d.npc || d.profile} is below ${d.percent}% on ${d.server}`,
    intro: `${d.npc || 'An NPC'} (${d.profile}) fell to ${d.percent}% of its health on ${d.server}.`,
  }),
  'rust.boss.spawned': (d) => ({
    title: `${d.boss} has appeared on ${d.server}`,
    intro: `${d.boss} is up on ${d.server}.`,
  }),
  'rust.boss.phase': (d) => ({
    title: `${d.boss} is down to ${d.percent}% on ${d.server}`,
    intro: `${d.boss} fell to ${d.percent}% of its health on ${d.server} and entered phase ${d.phase}.`,
  }),
  // Dangerous Treasures (protocol 14, §14.3.4). A missing grid drops out.
  'rust.dt.started': (d) => ({
    title: `A treasure event has started on ${d.server}`,
    intro: `A Dangerous Treasures chest opened${d.grid ? ` at ${d.grid}` : ''} on ${d.server}. It is guarded, and it is burning.`,
  }),
  'rust.dt.won': (d) => ({
    title: `The treasure was won on ${d.server}`,
    intro: `${d.player || 'Somebody'} won the treasure${d.grid ? ` at ${d.grid}` : ''} on ${d.server}`
      + `${d.invaders > 1 ? `, with ${d.invaders} players fighting for it` : ''}.`,
  }),
  'rust.boss.killed': (d) => ({
    title: `${d.boss} was killed on ${d.server}`,
    intro: `${d.killer ? `${d.killer} landed the killing blow` : `${d.boss} fell`}`
      + `${d.damagers > 1 ? `, with ${d.damagers} players fighting it` : ''}.`,
  }),
  'rust.login.denied': (d) => ({
    title: `A login to ${d.server} was not approved`,
    intro: `${d.player || 'Someone'} (${d.steamId}) tried to join ${d.server} and was not let in within a minute.`,
  }),
})

/** Where a crate is, as a sentence ends: "on Large Oil Rig", "dropped by a Chinook". */
const CRATE_PLACES = Object.freeze({
  oilrig_large: 'on Large Oil Rig',
  oilrig_small: 'on Small Oil Rig',
  cargo: 'on the cargo ship',
  ghostship: 'on a ghost ship',
  chinook: 'dropped by a Chinook',
})

/**
 * The headline of a world event (protocol 14), one sentence per type. Plain
 * register like the rest; a part that is missing drops out rather than leaving
 * a hole, so "at G7" disappears when the server has no map geometry yet.
 */
function worldHeadline(event, d) {
  const at = d.grid ? ` at ${d.grid}` : ''
  const on = ` on ${d.server}`
  const by = d.player ? ` by ${d.player}` : ''
  const place = CRATE_PLACES[d.where] || 'somewhere on the map'
  const bradley = d.where === 'launch_site' ? "Launch Site's Bradley" : d.where === 'road' ? 'A road Bradley' : 'A Bradley'
  const lines = {
    'heli.arrived': ['A patrol helicopter is overhead', `A patrol helicopter flew onto the map${at}${on}.`],
    'heli.left': ['The patrol helicopter has left', `The patrol helicopter left the map${on}.`],
    'heli.downed': ['Patrol helicopter downed', `The patrol helicopter was downed${by}${at}${on}.`],
    'bradley.spawned': [
      d.where === 'road' ? (d.count > 1 ? `${d.count} road Bradleys are out` : 'A road Bradley is out') : 'Bradley is patrolling Launch Site',
      d.where === 'road' ? `Road Bradleys took to the roads${on}.` : `Launch Site's Bradley is back${on}.`,
    ],
    'bradley.destroyed': [`${bradley} was destroyed`, `${bradley} was destroyed${by}${at}${on}.`],
    'cargo.arrived': ['The cargo ship has arrived', `The cargo ship sailed onto the map${at}${on}.`],
    'cargo.docking': ['The cargo ship is docking', `The cargo ship is coming in to dock${d.harbour ? ` at ${d.harbour}` : ''}${on}.`],
    'cargo.docked': ['The cargo ship has docked', `The cargo ship docked${d.harbour ? ` at ${d.harbour}` : ''}${on}, with fresh crates.`],
    'cargo.undocked': ['The cargo ship cast off', `The cargo ship left${d.harbour ? ` ${d.harbour}` : ' the harbour'}${on}.`],
    'cargo.left': ['The cargo ship is leaving', `The cargo ship turned for the edge of the map${on}.`],
    'chinook.arrived': ['A Chinook is inbound', `A Chinook flew onto the map${at}${on} with a locked crate.`],
    'chinook.dropped': ['A Chinook dropped a locked crate', `A Chinook dropped a locked crate${d.monument ? ` at ${d.monument}` : at}${on}.`],
    'chinook.downed': ['A Chinook was shot down', `A Chinook was shot down${by}${at}${on}.`],
    'airdrop.called': ['An airdrop is on its way', `An airdrop is on its way${at}${on}${d.player ? `, called by ${d.player}` : ''}.`],
    'airdrop.landed': ['An airdrop has landed', `An airdrop landed${at}${on}.`],
    'airdrop.looted': ['An airdrop was looted', `${d.player || 'Somebody'} looted an airdrop${at}${on}.`],
    'crate.hack.started': ['A locked crate is being hacked', `${d.player || 'Somebody'} started hacking the locked crate ${place}${on}.`],
    'crate.unlocked': ['A locked crate is open', `The locked crate ${place} finished unlocking${on}.`],
    'crate.looted': ['A locked crate was looted', `${d.player || 'Somebody'} looted the locked crate ${place}${on}.`],
    'vendor.arrived': ['The Travelling Vendor is here', `The Travelling Vendor drove onto the map${at}${on}.`],
    'vendor.left': ['The Travelling Vendor has left', `The Travelling Vendor left the map${on}.`],
    'satellite.crashed': ['A satellite has crashed', `A satellite came down${at}${on}. Its crates are burning.`],
    'deepsea.opened': ['The Deep Sea is open', `The Deep Sea opened${on}.`],
    'deepsea.closed': ['The Deep Sea has closed', `The Deep Sea closed${on}.`],
  }
  const line = lines[event]
  return line ? { title: `${line[0]}${on}`, intro: line[1] } : {}
}

const WORLD_PREFIX = 'rust.world.'

function headline(triggerId, data) {
  const make = HEADLINES[triggerId]
  if (make) return make(data || {})
  return String(triggerId).startsWith(WORLD_PREFIX) ? worldHeadline(triggerId.slice(WORLD_PREFIX.length), data || {}) : {}
}

/**
 * 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 {
    const data = envelope.data || {}
    core.emit(triggerId, { ...envelope, data: { ...headline(triggerId, data), ...data } })
    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
}

/**
 * RunicNPC's NPCs (runicnpc stage 4): a death, and a health threshold. What a
 * phase gate counts, so the window is short: a death an hour old is history, not
 * a wave falling. The dedupe key is the NPC's net id, which the game never
 * reuses within a boot.
 */
const NPC_MAX_AGE_MS = 15 * 60 * 1000

function npcVars(server, frame) {
  return {
    ...serverVars(server),
    profile: str(frame.profile),
    npc: str(frame.name),
    byEvent: Boolean(frame.runId),
    runId: str(frame.runId),
    placement: str(frame.placement),
  }
}

async function onNpc(server, item, frame) {
  const t = frameTime(item, frame)
  if (!stillNews(t, NPC_MAX_AGE_MS) || !str(frame.profile)) return 0

  if (frame.kind === 'npc.died') {
    const contributors = Array.isArray(frame.contributors) ? frame.contributors : []
    return fire(T['rust.npc.died'], {
      data: { ...npcVars(server, frame), killer: str(frame.killerName), killerSteamId: str(frame.killerId), contributors: contributors.length },
      dedupeKey: dedupeKey('npc.died', server.id, frame.netId, t),
      occurredAt: t,
    }) ? 1 : 0
  }

  const percent = Math.round((Number(frame.threshold) || 0) * 100)
  return fire(T['rust.npc.health'], {
    data: { ...npcVars(server, frame), percent },
    dedupeKey: dedupeKey('npc.health', server.id, frame.netId, percent),
    occurredAt: t,
  }) ? 1 : 0
}

/**
 * A RunicNPC boss (runicnpc stage 6, D277): spawned, a phase, killed. Public
 * news, so a replay older than a broadcast's window tells nobody (D63). The
 * dedupe key is the boss's net id, with the phase for a phase.
 */
async function onBoss(server, item, frame) {
  const t = frameTime(item, frame)
  if (!stillNews(t, BROADCAST_MAX_AGE_MS) || !str(frame.profile)) return 0
  const base = { ...serverVars(server), profile: str(frame.profile), boss: str(frame.name) || str(frame.profile), byEvent: Boolean(frame.runId) }

  if (frame.kind === 'npc.boss.spawned') {
    return fire(T['rust.boss.spawned'], { data: base, dedupeKey: dedupeKey('boss.spawned', server.id, frame.netId, t), occurredAt: t }) ? 1 : 0
  }

  if (frame.kind === 'npc.boss.phase') {
    const phase = Number(frame.phase) || 0
    return fire(T['rust.boss.phase'], {
      data: { ...base, phaseKey: `${base.profile}:${phase}`, phase, percent: Math.round((Number(frame.at) || 0) * 100) },
      dedupeKey: dedupeKey('boss.phase', server.id, frame.netId, phase),
      occurredAt: t,
    }) ? 1 : 0
  }

  return fire(T['rust.boss.killed'], {
    data: { ...base, killer: str(frame.killerName), damagers: Number(frame.damagers) || 0 },
    dedupeKey: dedupeKey('boss.killed', server.id, frame.netId, t),
    occurredAt: t,
  }) ? 1 : 0
}

/**
 * Rust's own world events (protocol 14, D355): one trigger per type. Public
 * news, so a replay older than a broadcast's window tells nobody (D63). The
 * grid comes from the map's geometry, as the feed's does; a downed helicopter,
 * Chinook or Bradley names whom the server's credit rule names, as its feed
 * line and its count do (D372). Names only, never a Steam id.
 */
async function onWorld(server, item, frame) {
  const t = frameTime(item, frame)
  const event = str(frame.event)
  if (!event || !worldEvents.isKnownType(event) || !stillNews(t, BROADCAST_MAX_AGE_MS)) return 0

  const meta = frame.x !== undefined ? await mapDb.getMeta(server.id).catch(() => null) : null
  const grid = meta ? worldEvents.gridOf({ worldSize: Number(meta.worldSize), gridCells: Number(meta.gridCells), gridCellSize: Number(meta.gridCellSize) }, frame.x, frame.z) : null
  const type = worldEvents.TYPES[event]
  const crate = event.startsWith('crate.')

  const data = {
    ...serverVars(server),
    subject: crate ? `${server.id}:${event}:${frame.where || ''}` : `${server.id}:${event}`,
    ...(grid ? { grid } : {}),
    ...(frame.where !== undefined ? { where: str(frame.where) } : {}),
    ...(event === 'bradley.spawned' ? { count: Math.max(1, Math.floor(Number(frame.count) || 1)) } : {}),
    ...(frame.harbourName || frame.harbour ? { harbour: str(frame.harbourName) || str(frame.harbour) } : {}),
    ...(frame.monumentName || frame.monument ? { monument: str(frame.monumentName) || str(frame.monument) } : {}),
  }

  if (type.player && str(frame.name)) data.player = str(frame.name)

  if (type.credit) {
    const ids = worldEvents.credited(frame, worldEvents.creditRuleOf(server.creditRule))
    const phrase = worldEvents.namesPhrase(ids.map((id) => worldEvents.nameOf(frame, id)), ids.length)
    data.credited = ids.length
    if (phrase) data.player = phrase
  }

  return fire(`${WORLD_PREFIX}${event}`, {
    data,
    dedupeKey: dedupeKey('world', server.id, event, t, frame.x, frame.z),
    occurredAt: t,
  }) ? 1 : 0
}

/**
 * Dangerous Treasures (protocol 14, §14.3.4): a chest opened, and one won.
 * Public news, so a replay older than a broadcast's window tells nobody (D63).
 * The grid comes from the map's geometry, as the feed's does; the winner by
 * in-game name only, never a Steam id. An ending raises nothing.
 */
async function onDangerousTreasures(server, item, frame) {
  const t = frameTime(item, frame)
  if (!stillNews(t, BROADCAST_MAX_AGE_MS)) return 0

  const meta = frame.x !== undefined ? await mapDb.getMeta(server.id).catch(() => null) : null
  const grid = meta ? worldEvents.gridOf({ worldSize: Number(meta.worldSize), gridCells: Number(meta.gridCells), gridCellSize: Number(meta.gridCellSize) }, frame.x, frame.z) : null
  const data = { ...serverVars(server), subject: `${server.id}:${frame.kind}`, ...(grid ? { grid } : {}) }

  if (frame.kind === 'dt.started') {
    return fire(T['rust.dt.started'], {
      data: { ...data, byEvent: Boolean(str(frame.key)) },
      dedupeKey: dedupeKey('dt.started', server.id, t, frame.x, frame.z),
      occurredAt: t,
    }) ? 1 : 0
  }

  const invaders = Number(frame.invaderCount)
  return fire(T['rust.dt.won'], {
    data: {
      ...data,
      ...(str(frame.name) ? { player: str(frame.name) } : {}),
      invaders: Number.isInteger(invaders) && invaders >= 0 ? invaders : Array.isArray(frame.invaders) ? frame.invaders.length : 0,
    },
    dedupeKey: dedupeKey('dt.won', server.id, t, frame.steamId, frame.x, frame.z),
    occurredAt: t,
  }) ? 1 : 0
}

const HANDLERS = Object.freeze({
  'dt.started': onDangerousTreasures,
  'dt.won': onDangerousTreasures,
  'world.event': onWorld,
  'npc.boss.spawned': onBoss,
  'npc.boss.phase': onBoss,
  'npc.boss.died': onBoss,
  'npc.died': onNpc,
  'npc.health': onNpc,
  '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
  }
}

/**
 * `rust.kit.entitled` for each user one reward step granted (phase 13b). Never
 * throws: a notification that could not be raised must not fail the grant it
 * is about. Returns how many were raised.
 */
function entitled({ userIds, kit, server, mode, runId, stepId }) {
  let raised = 0
  try {
    const rewardKey = `${runId}:${stepId}`
    for (const userId of new Set(userIds || [])) {
      const uid = Number(userId)
      if (!Number.isInteger(uid) || uid < 1) continue
      const ok = fire(T['rust.kit.entitled'], {
        data: { rewardKey, kit: String(kit), ...serverVars(server), ...(mode ? { mode: String(mode) } : {}), accountUrl: PATHS.account },
        ownerUserId: uid,
        dedupeKey: dedupeKey('entitled', runId, stepId, uid),
      })
      if (ok) raised++
    }
  } catch (err) {
    log.warn('could not raise the reward notification', { error: err.message })
  }
  return raised
}

module.exports = {
  onEvent,
  serverObserved,
  checkLeader,
  sweepLoginDenied,
  linked,
  entitled,
  reset,
  dedupeKey,
  headline,
  stillNews,
  BROADCAST_MAX_AGE_MS,
  PERSONAL_MAX_AGE_MS,
  LOGIN_APPROVAL_WINDOW_MS,
  STRUCTURE_LABELS,
}