Module-Rust/server/permSync.js

// ── Keeping a game's permission store and the site's record of it equal ───
//
// R2's whole mechanism, rebuilt in protocol 13 (PLAN_REDESIGNS §1) around one
// fact: **the site owns every permission and group on the server** — those that
// were there before it, those an admin makes, and those changed in the game
// (D160). So a sync is three steps, not one:
//
//   1. READ the whole store (`perm.inventory`), in pages, with the plugin that
//      registered each permission.
//   2. RECONCILE it against the site's record and what the site last pushed
//      (`reconcile.js`): a change made in the game becomes the site's own, a
//      question for a person, or undone — the server's policy says which (D161).
//      The first read of a server imports everything it finds (D198).
//   3. PUSH the whole desired set (`perm.sync`, D32) with what the site has
//      withdrawn, and record what the plugin says landed.
//
// Step 2 writes to the site's own tables, and a change in one game affects that
// server only (D190) — which can split a group shared with other servers. So it
// runs under ONE lock for the whole fleet: two servers' reconciles never split
// the same shared group at once. Steps 1 and 3 run in parallel across servers.
//
// ── When it runs ──────────────────────────────────────────────────────────
//
// Every tick asks a cheap question — does the digest of the desired set still
// equal what this server last confirmed — and does nothing when the answer is
// yes. A sync therefore happens when:
//
//   • an operator changed something (the dirty flag, and the digest behind it)
//   • the game restarted or wiped (a new boot id or wipe id)
//   • a permission hook fired in the game that we did not cause (`ingest.js`
//     marks the server dirty; the inventory then says what changed)
//   • the audit interval elapsed — the backstop that finds a hand edit on a quiet
//     server
//   • the last attempt failed, after a backoff
//
// ── What it never does ────────────────────────────────────────────────────
//
// It never judges a permission the server has not registered right now (a
// plugin unloaded for a minute is not a revocation), never touches what an event
// lease holds, never invents a permission the server has not registered (D33),
// and never treats a silent sidecar as a reason to forget anything.

const core = require('./core')

const apply = require('./model/permissions/permissions.apply')
const db = require('./model/permissions/permissions.db')
const model = require('./model/permissions/permissions.model')
const reconcile = require('./model/permissions/reconcile')
const servers = require('./model/servers/servers.model')
const serversDb = require('./model/servers/servers.db')
const sidecar = require('./sidecarClient')

const log = core.logger('permissions')

/** How often the loop asks whether anything needs pushing. */
const TICK_MS = 30 * 1000

/**
 * How long a server may go without a full reconciliation, however quiet it is:
 * the interval at which a hand edit on a quiet server is found anyway.
 */
const AUDIT_MS = 15 * 60 * 1000

/** How long to leave a failing server alone before trying again. */
const FAIL_BACKOFF_MS = 2 * 60 * 1000

/**
 * The most rows one sync may carry. Below the sidecar's line cap and the
 * plugin's operation ceiling, so the refusal happens here, where it can name the
 * server and reach an operator.
 */
const MAX_ROWS = 15000

/** The policies a server may have (D161). */
const POLICIES = ['auto-adopt', 'adopt', 'revoke']

let timer = null

/** The fleet-wide lock the reconcile step runs under (see the header). */
let lock = Promise.resolve()

function withLock(fn) {
  const run = lock.then(fn, fn)
  lock = run.catch(() => {})
  return run
}

function start() {
  if (timer) return

  timer = setInterval(() => {
    tick().catch((err) => log.error('permission sync tick failed', { error: err.message }))
  }, TICK_MS)

  if (timer.unref) timer.unref()
}

function stop() {
  if (!timer) return

  clearInterval(timer)
  timer = null
}

/**
 * One pass over every enabled server.
 *
 * The authored set is read once here, for the cheap "does anything need doing"
 * digest. The reconcile re-reads it under the lock, because another server's
 * reconcile may have changed it in between.
 */
async function tick({ force = null } = {}) {
  await db.ensureSyncRows()

  const [rows, state, sync, authored, policyRows] = await Promise.all([
    servers.listForPolling(),
    serversDb.listState(),
    db.listSync(),
    model.readAuthored(),
    db.listPolicies(),
  ])

  const syncById = new Map(sync.map((row) => [row.serverId, row]))
  const stateById = new Map(state.map((row) => [row.serverId, row]))
  const policies = new Map(policyRows.map((row) => [row.serverId, row.policy]))

  // `allSettled`: one unreachable host must not stop the others.
  await Promise.allSettled(
    rows
      .filter((server) => force === null || force === server.id)
      .map((server) =>
        syncOne(server, {
          authored,
          sync: syncById.get(server.id) || null,
          state: stateById.get(server.id) || null,
          force: force !== null,
          policy: policies.get(server.id) || 'auto-adopt',
        }),
      ),
  )
}

/**
 * Whether this server needs a push right now.
 *
 * Returns a reason rather than a boolean, because the reason is worth logging:
 * "why did the website just write to my game server" is a question an operator
 * asks, and `wipe` and `drift` are very different answers.
 */
function reasonToSync({ desiredHash, sync, state, force }) {
  if (force) return 'requested'
  // Not while the world is loading (PLAN_FIXES F7). The plugin connects before
  // the save loads, and the first walk's restart sync went 35 s before "Server
  // startup complete", timed out behind the busy main thread, and was retried
  // 2.5 minutes later as "0 applied" — so nothing said what the restart had
  // restored. The plugin says `worldReady: true` in the hello it sends the moment
  // the world is up, and the sync goes on the next tick. A human's "sync now" is
  // not held: they asked, and a failure then is theirs to read.
  if (state && state.worldReady === false) return null
  // Nor while the game is not connected (F7, the step-2 walk). The stored hello is
  // the LAST one: between a restart and the poll that reads the new boot's hello,
  // it still says the old world is ready. On Carbon a due audit went out in that
  // gap, 80 s before "Server startup complete". The poll marks the server offline
  // the moment it goes away, so offline holds until a fresh hello says otherwise.
  // Unknown (`online` absent) is not held — a state row always carries it.
  if (state && state.online !== undefined && state.online !== null && !Number(state.online)) return null
  if (!sync) return 'first'
  if (sync.state !== 'ok' && sync.lastAttemptAt && age(sync.lastAttemptAt) < FAIL_BACKOFF_MS && !sync.dirty) {
    return null
  }
  if (sync.state !== 'ok') return 'retry'
  if (desiredHash !== sync.syncedHash) return 'changed'
  if (sync.dirty) return 'dirty'

  const bootId = state && state.bootId ? state.bootId : null
  const wipeId = state && state.wipeId ? state.wipeId : null

  // A restart or a wipe is the case R2 exists for: the game may have forgotten
  // everything, and the site has not.
  if (bootId && bootId !== sync.bootId) return 'restart'
  if (wipeId && wipeId !== sync.wipeId) return 'wipe'

  if (!sync.lastAttemptAt || age(sync.lastAttemptAt) >= AUDIT_MS) return 'audit'

  return null
}

function age(value) {
  const at = value instanceof Date ? value.getTime() : new Date(value).getTime()
  return Number.isFinite(at) ? Date.now() - at : Number.MAX_SAFE_INTEGER
}

/** Record a failed attempt, with the reason on the row and in the log (F7). */
async function failed(server, { reason, error, desiredHash, sync, bootId, wipeId }) {
  log.warn('permission sync failed', { server: server.id, reason, error })
  await db.putSyncResult(server.id, {
    state: 'failed',
    desiredHash,
    syncedHash: sync ? sync.syncedHash : null,
    bootId,
    wipeId,
    report: null,
    error: String(error).slice(0, 191),
  })
}

/**
 * Read, reconcile and push one server. Returns what happened, for the log and
 * the tests: null (nothing to do), 'ok', or why it stopped.
 */
async function syncOne(server, { authored, sync, state, force, policy = 'auto-adopt' }) {
  const first = model.buildDesired(server.id, authored)
  const reason = reasonToSync({ desiredHash: first.hash, sync, state, force })

  if (!reason) return null

  const bootId = state && state.bootId ? state.bootId : null
  const wipeId = state && state.wipeId ? state.wipeId : null

  // ── 1. Read the whole store ────────────────────────────────────────────
  const read = await sidecar.readInventory(server)

  if (!read.ok) {
    await failed(server, { reason, error: read.error, desiredHash: first.hash, sync, bootId, wipeId })
    return 'inventory'
  }

  const { inventory } = read
  const registered = new Set(inventory.permissions.map((row) => model.normaliseName(row.name)).filter(Boolean))

  // The option source, from the same read: what this server registers, and which
  // plugin registered each name (PLAN_REDESIGNS §0.1).
  await db.putCatalogue(
    server.id,
    inventory.permissions
      .map((row) => ({ permission: model.normaliseName(row.name), owner: row.owner || null }))
      .filter((row) => row.permission),
  )

  const importing = !(sync && sync.importedAt)

  // ── 2. Reconcile, under the fleet lock ─────────────────────────────────
  const settled = await withLock(async () => {
    let current = await model.readAuthored()
    let desired = model.buildDesired(server.id, current)
    const pushed = await db.listPushed(server.id)

    const classes = reconcile.classify({
      present: reconcile.presentRows(inventory),
      pushed,
      desired: desired.rows,
      registered,
      leased: inventory.leased,
    })

    const planned = reconcile.plan({ classes, policy, importing, sources: desired.sources })

    if (planned.ops.length) {
      const done = await apply.applyOps(server.id, planned.ops, log)
      log.info(importing ? 'permissions imported from the game' : 'changes made in the game adopted', {
        server: server.id,
        policy,
        ops: done.length,
      })

      current = await model.readAuthored()
      desired = model.buildDesired(server.id, current)
    }

    for (const row of planned.revocations) {
      await db.queueRevocation({ serverId: server.id, ...row, requestedBy: null })
    }

    return { desired, planned, classes }
  })

  const { desired, planned } = settled
  const held = reconcile.withHold(desired, planned.hold)

  // ── 3. Push ────────────────────────────────────────────────────────────
  const [pushed, revocations] = await Promise.all([db.listPushed(server.id), db.listRevocations(server.id)])

  // Retired against the WHOLE desired set: a held row is still wanted, it is
  // only not being pushed back while a person decides.
  const retirements = model.retirements(pushed, desired.rows, planned.hold)
  const { styleRetired, sent: styleRetire } = styleRetirements(retirements)
  const retire = [
    ...retirements
      .filter((row) => row.kind !== 'chat-field')
      .map((row) => ({ kind: row.kind, subject: row.subject, object: row.object })),
    ...styleRetire,
    ...revocations.map((row) => ({ kind: row.kind, subject: row.subject, object: row.object })),
  ]

  if (held.rows.length + retire.length > MAX_ROWS) {
    await failed(server, {
      reason,
      error: `the permission set is too large to push (${held.rows.length + retire.length} rows, limit ${MAX_ROWS})`,
      desiredHash: desired.hash,
      sync,
      bootId,
      wipeId,
    })
    return 'too-large'
  }

  log.info('syncing permissions', {
    server: server.id,
    reason,
    policy,
    importing,
    rows: held.rows.length,
    retire: retire.length,
    held: planned.hold.size,
  })

  const result = await sidecar.permSync(server, {
    setId: desired.hash,
    groups: withExpect(held.payload.groups, pushed),
    grants: held.payload.grants,
    credits: held.payload.credits,
    retire,
  })

  if (!result.ok) {
    await failed(server, { reason, error: result.status, desiredHash: desired.hash, sync, bootId, wipeId })
    return result.status
  }

  const report = result.data || {}

  // `perm.error` (`busy`, `too-large`) is an answer, not a transport failure.
  if (report.kind === 'perm.error') {
    await failed(server, {
      reason,
      error: `the game refused the sync: ${report.reason || 'unknown'}`,
      desiredHash: desired.hash,
      sync,
      bootId,
      wipeId,
    })
    return report.reason || 'refused'
  }

  await applyReport(server, {
    desired: held,
    hash: desired.hash,
    retire,
    styleRetired,
    report,
    bootId,
    wipeId,
    drift: planned.drift,
  })

  if (importing) await db.markImported(server.id)

  return 'ok'
}

/**
 * The groups as the wire carries them: each style field with the value this
 * site last pushed to THIS server (`expect`), or null for one it never has.
 * The plugin writes a field only when the game still holds that, so a field
 * somebody changed by hand is reported instead of overwritten (D138).
 */
function withExpect(groups, pushed) {
  const expect = new Map(
    pushed.filter((row) => row.kind === 'chat-field').map((row) => [`${row.subject} ${row.object}`, row.value]),
  )

  return groups.map((group) => {
    if (!group.chat) return group

    const chat = {}
    for (const [field, value] of Object.entries(group.chat)) {
      const last = expect.get(`${group.name} ${field}`)
      chat[field] = { value, expect: last === undefined ? null : last }
    }

    return { ...group, chat }
  })
}

/**
 * Style fields the site pushed and no longer wants, as the wire says it: ONE
 * `chat-group` retirement per group, because a style is all twelve fields or
 * none, and the only way to take one out of BetterChat is to remove its group
 * (D139).
 *
 * **`default` is never removed.** It is BetterChat's fallback, which it warns
 * about on every line when it is missing; a style withdrawn from the site's
 * `default` group stops being pushed and is left as it stands (§33.5).
 */
function styleRetirements(retirements) {
  const styleRetired = retirements.filter((row) => row.kind === 'chat-field')
  const groups = [...new Set(styleRetired.map((row) => row.subject))].filter((name) => name !== 'default')

  return { styleRetired, sent: groups.map((name) => ({ kind: 'chat-group', subject: name, object: '' })) }
}

/**
 * Record what the game said it did, and what waits for a person.
 *
 * Every write is safe to repeat: a sync that crashes here is re-run next tick and
 * reaches the same place.
 */
async function applyReport(server, { desired, hash, retire, styleRetired = [], report, bootId, wipeId, drift = [] }) {
  const unresolved = new Set((report.unresolved || []).map(model.normaliseName))
  const pending = new Set(report.pending || [])
  // Grants the plugin made and then did not find when it read them back (D85);
  // since protocol 13 also `group:parent:name` for a parent that would not set.
  const notLanded = new Set((report.notLanded || []).map((entry) => String(entry).toLowerCase()))

  // Protocol 12. A style field landed when BetterChat was there to take it and
  // the report names it neither drift nor failed (§33.2).
  const chat = report.chat && typeof report.chat === 'object' ? report.chat : null
  const chatLoaded = Boolean(chat && chat.loaded === true)
  const fieldKey = (group, field) => `${group} ${String(field || '').toLowerCase()}`
  const chatHeld = new Set([
    ...((chatLoaded && chat.drift) || []).map((row) => fieldKey(row.group, row.field)),
    ...((chatLoaded && chat.failed) || []).filter((row) => row.field).map((row) => fieldKey(row.group, row.field)),
  ])
  const chatGroupFailed = new Set(((chatLoaded && chat.failed) || []).filter((row) => !row.field).map((row) => row.group))

  const landed = desired.rows.filter((row) => {
    if (row.kind === 'chat-field') {
      return chatLoaded && !chatGroupFailed.has(row.subject) && !chatHeld.has(fieldKey(row.subject, row.object))
    }
    if (row.kind === 'grant' || row.kind === 'group-permission') {
      return !unresolved.has(row.object) && !notLanded.has(`${row.subject}:${row.object}`.toLowerCase())
    }
    if (row.kind === 'member') return !pending.has(`${row.subject}:${row.object}`)
    if (row.kind === 'group') {
      // The value landed only if the parent did; a group whose parent would not
      // set is recorded without a value, so the next sync pushes it again.
      if ([...notLanded].some((entry) => entry.startsWith(`${row.subject}:parent:`))) {
        row.value = null
      }
    }
    return true
  })

  await db.addPushed(server.id, landed)

  // Everything retired is gone from the game whether the plugin removed it or
  // found it already absent. A `chat-group` is not a ledger row; its fields are.
  await db.removePushed(server.id, retire.filter((row) => row.kind !== 'chat-group'))

  // A style's fields leave the ledger only once BetterChat has removed the group,
  // or for `default`, which is never removed (D139).
  const removed = new Set((chatLoaded && chat.removed) || [])
  await db.removePushed(
    server.id,
    styleRetired.filter((row) => row.subject === 'default' || removed.has(row.subject)),
  )

  const revocations = await db.listRevocations(server.id)
  await db.deleteRevocations(revocations.map((row) => row.id))

  // What waits for a person: the reconcile's (D161's `adopt`, and an event's
  // grant removed in the game), and a style field somebody changed by hand (D138).
  await db.replaceDrift(server.id, [
    ...drift.map((row) => ({
      kind: String(row.kind || ''),
      subject: String(row.subject || ''),
      object: String(row.object || ''),
      detail: row.detail === undefined || row.detail === null ? null : String(row.detail).slice(0, 255),
      direction: row.direction || 'added',
    })),
    ...((chatLoaded && chat.drift) || []).map((row) => ({
      kind: 'chat-field',
      subject: String(row.group || ''),
      object: String(row.field || ''),
      detail: row.game === undefined || row.game === null ? null : String(row.game).slice(0, 255),
      direction: 'changed',
    })),
  ])

  await db.putSyncResult(server.id, {
    state: 'ok',
    desiredHash: hash,
    syncedHash: hash,
    bootId,
    wipeId,
    report: JSON.stringify(report),
    error: null,
  })

  log.info('permissions synced', {
    server: server.id,
    applied: report.applied,
    unresolved: (report.unresolved || []).length,
    pending: (report.pending || []).length,
    notLanded: (report.notLanded || []).length,
    waiting: drift.length,
    ...(chat
      ? {
          betterChat: chatLoaded,
          ...(chatLoaded
            ? { styleApplied: chat.applied, styleSaved: chat.saved, styleDrift: (chat.drift || []).length, styleFailed: (chat.failed || []).length }
            : {}),
        }
      : {}),
    ...(report.creditsApplied !== undefined
      ? { creditsApplied: report.creditsApplied, creditsWithdrawn: report.creditsWithdrawn }
      : {}),
  })
}

module.exports = {
  TICK_MS,
  AUDIT_MS,
  FAIL_BACKOFF_MS,
  MAX_ROWS,
  POLICIES,
  start,
  stop,
  tick,
  syncOne,
  reasonToSync,
  applyReport,
  withExpect,
  styleRetirements,
  withLock,
}