// ── 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) }
}

/**
 * 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.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.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.`,
  }),
})

function headline(triggerId, data) {
  const make = HEADLINES[triggerId]
  return make ? make(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
}

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,
  headline,
  stillNews,
  BROADCAST_MAX_AGE_MS,
  PERSONAL_MAX_AGE_MS,
  LOGIN_APPROVAL_WINDOW_MS,
  STRUCTURE_LABELS,
}