Email becomes a DeliveryChannel driven by rules, and the Team pipeline stops being
its own thing. `teamNotify.forumPost` now emits an event; a rule decides who is
mailed, through which template, and how often at most. One walk goes forum write
-> events.emit -> rule -> outbox -> worker -> email channel -> template -> SMTP.
Seven decisions settled by the org lead before any code:
- email only moves; the push tickle and the Discord bridge stay direct calls
- the EVENT carries its access-checked audience, and `members` resolves to it
- the four Team rules are seeded DISABLED, with an admin banner and a note
- team_notification_prefs stays, read by the engine as a scoped preference
- the payload wins and a structural projection fills the gaps
- the digest keeps computing at send time; only its state generalizes
- an unsubscribe token turns off the channel it names, and nothing else
Three defects found while building it:
- `email.button` never absolutized its href, while image and itemList both
did. Every rule-driven CTA would have been a dead relative link, because a
trigger's url variables are validated site-relative by construction.
- Phase 4a enqueued digest-mode recipients for a drain that Phase 6 decided
not to build. An outbox row snapshots the payload and so has none of the
three properties the digest design exists for, including the security one.
- the digest's send-log row carried no address_hash while the instant row
beside it did, which would have made half the mail uncorrelatable in Phase 9.
Also: engagement_digest_state + a replay-safe backfill, engagement_outbox.scope_key,
a v2 unsubscribe token that still verifies v1 forever, and the canonical
/public/engagement/unsubscribe pair with the old /public/teams path kept
permanently — mail is not editable once sent.
Verified with 1464 server tests, 324 client tests, and a live rig (MariaDB +
Mailpit + a real Team) covering the instant mail, the digest, the generic
template, a pre-migration unsubscribe link and the backfill's replay-safety.
Docs: RunicGateway/docs#TBD
Co-Authored-By: Claude <noreply@anthropic.com>
640 lines
26 KiB
JavaScript
640 lines
26 KiB
JavaScript
// ── The reconciler ─────────────────────────────────────────────────────────
|
|
//
|
|
// Core's projection of the module's Teams, kept in step (TEAMS.md §2.4). This is
|
|
// the only thing that writes `team_members`, and one of only two things that
|
|
// write `teams.status`.
|
|
//
|
|
// **The four places it refuses to act** are the point of the file, and they are
|
|
// all one rule stated four ways: *a result derived from an answer core does not
|
|
// trust is never applied.* Anything less specific tends to collapse, under
|
|
// maintenance, into "a failed call means no teams" — which is invariant 1's
|
|
// failure mode and would empty every roster on the site the first time a sidecar
|
|
// restarted.
|
|
//
|
|
// 1. `getTeams()` not ok → write sync state, touch NOTHING, return.
|
|
// 2. ok but empty, core holds ≥1 → quarantine; apply only if the NEXT
|
|
// authoritative answer agrees.
|
|
// 3. `getTeamMembers()` not ok → that Team's roster untouched and stale;
|
|
// the other Teams carry on.
|
|
// 4. ok but zero members, had some → the same two-strikes quarantine, per Team.
|
|
//
|
|
// Gates 2 and 4 exist because an authoritative-looking empty answer during a cold
|
|
// start is the one failure indistinguishable from a real wipe. "Every Team on the
|
|
// shard disbanded at once" costs one interval of delay to confirm; getting it
|
|
// wrong costs every roster on the site.
|
|
//
|
|
// Events (§2.3) are an OPTIMISATION, never the source of truth. They make the
|
|
// common case immediate; reconciliation is what makes it correct. Nothing
|
|
// destructive at Team level is ever driven by one — §2.2 scopes archival to an
|
|
// authoritative full list, so a `team.disbanded` event schedules a run rather
|
|
// than archiving, and a spurious event costs a reconcile instead of a Team.
|
|
|
|
const teamsDb = require('./teams.db')
|
|
const teamProvider = require('./teamProvider')
|
|
const moderation = require('./teamModeration.model')
|
|
const activity = require('./teamActivity.model')
|
|
const teamNotify = require('../../utils/teamNotify')
|
|
const teamVoiceSync = require('../../utils/teamVoiceSync')
|
|
const { slugify, uniqueSlug } = require('./teamSlug')
|
|
const settings = require('../settings/settings.model')
|
|
const log = require('../../utils/logger')('teams')
|
|
|
|
// At most one run per 30s (§2.4), so a sidecar flapping cannot become a
|
|
// reconciliation storm — every flap publishes events, and every event asks for a
|
|
// run.
|
|
const DEBOUNCE_MS = 30_000
|
|
const DEFAULT_INTERVAL_S = 900
|
|
const MIN_INTERVAL_S = 60
|
|
const INTERVAL_KEY = 'teams_reconcile_interval_s'
|
|
|
|
// The six kinds a module may publish (§2.3). Six rather than the four a
|
|
// membership-shaped reading suggests, because leadership is its own authority
|
|
// path and a leadership change must be expressible without pretending someone
|
|
// joined or left.
|
|
const EVENT_KINDS = new Set([
|
|
'team.created', 'team.disbanded',
|
|
'team.member.added', 'team.member.removed',
|
|
'team.leader.added', 'team.leader.removed',
|
|
])
|
|
|
|
// Kinds that can only be answered by a full list. `team.created` cannot be
|
|
// applied from a delta — a Team built from one has no name, no roster and no
|
|
// leaders — and `team.disbanded` must not be, per §2.2.
|
|
const RECONCILE_ONLY = new Set(['team.created', 'team.disbanded'])
|
|
|
|
// ── Scheduling state (in-process; one provider per deployment) ─────────────
|
|
|
|
let running = false
|
|
let rerunReason = null
|
|
let lastRunAt = 0
|
|
let debounceTimer = null
|
|
let pollTimer = null
|
|
let started = false
|
|
|
|
/** Resolve the poll interval, floored so a bad setting cannot become a hot loop. */
|
|
async function intervalSeconds() {
|
|
let raw
|
|
try {
|
|
raw = await settings.get(INTERVAL_KEY)
|
|
} catch {
|
|
return DEFAULT_INTERVAL_S
|
|
}
|
|
const n = Number.parseInt(raw, 10)
|
|
if (!Number.isFinite(n) || n < MIN_INTERVAL_S) return DEFAULT_INTERVAL_S
|
|
return n
|
|
}
|
|
|
|
/**
|
|
* Backoff, capped at the poll interval (§2.4).
|
|
*
|
|
* The cap is what keeps this a backoff rather than an outage: a module down for a
|
|
* day would otherwise reach a delay measured in weeks and stay stale long after
|
|
* it recovered.
|
|
*/
|
|
function backoffSeconds(consecutiveFailures, intervalS) {
|
|
if (!consecutiveFailures) return intervalS
|
|
return Math.min(intervalS, 2 ** Math.min(consecutiveFailures, 16) * 15)
|
|
}
|
|
|
|
// ── Applying one Team ──────────────────────────────────────────────────────
|
|
|
|
/**
|
|
* Create the row for a Team core has not seen, deriving its slug and screening
|
|
* its name against the reserved list (§2.8).
|
|
*
|
|
* The row is created whatever the screening says, and hidden if it matched. Core
|
|
* cannot refuse a name: the guild already exists in the game and core is a mirror
|
|
* of it, not an authority over it. A hidden Team is absent from public surfaces
|
|
* and completely functional for its own members — the people in it are not being
|
|
* punished for a name their leader chose.
|
|
*/
|
|
async function createTeam(moduleId, team) {
|
|
const taken = await teamsDb.slugsLike(slugify(team.name) || 'team')
|
|
const slug = uniqueSlug(team.name, taken)
|
|
const screened = await moderation.screenForCreate(team.name)
|
|
const id = await teamsDb.insertTeam({ moduleId, slug, ...team, ...screened })
|
|
log.info('team created', {
|
|
moduleId, externalId: team.externalId, name: team.name, slug, hidden: Boolean(screened.hidden),
|
|
})
|
|
return id
|
|
}
|
|
|
|
/**
|
|
* The §2.2 rename rule: same id and a different name is an archive plus a create.
|
|
*
|
|
* The old row keeps its forum, its activity and its grants, all read-only, and
|
|
* points at its successor so the old slug can explain itself instead of 404ing.
|
|
* Core never decides whether this is "really" the same team — that judgement is
|
|
* the module's, expressed in whether it reuses the external id (§10.5).
|
|
*/
|
|
async function applyRename(moduleId, existing, team) {
|
|
const successorId = await createTeam(moduleId, team)
|
|
await teamsDb.archiveTeam(existing.id, 'renamed', successorId)
|
|
log.info('team renamed; previous row archived', {
|
|
externalId: team.externalId, from: existing.name, to: team.name, archivedId: existing.id, successorId,
|
|
})
|
|
// §4.2's `core.team.renamed`, written to the SUCCESSOR rather than to the row
|
|
// that was renamed: the archived row is a read-only record of what happened
|
|
// before the rename (§2.2), and the person who wants to know a Team used to be
|
|
// called something else is looking at the live page.
|
|
//
|
|
// The old name is core's own, not game-sourced text a module handed us this
|
|
// run — it is the `name` column core has been serving all along — so §2.9's
|
|
// approval gate does not apply. It can still be a name staff suppressed, which
|
|
// is why a hidden Team's feed is not served publicly (teamActivity.feedFor).
|
|
await activity.logCore({
|
|
teamId: successorId,
|
|
kind: activity.CORE_KINDS.TEAM_RENAMED,
|
|
summary: `Renamed from ${existing.display_name_override || existing.name}`,
|
|
dedupeKey: `renamed:${existing.id}`,
|
|
}).catch((err) => log.warn('rename activity not recorded', { message: err.message }))
|
|
return successorId
|
|
}
|
|
|
|
/** Never a game-internal member key on a public page: that identifier is not published (§3.2). */
|
|
const memberLabel = (row) => (row && row.display_name) || 'A member'
|
|
|
|
/**
|
|
* Core's own membership items for one roster run (§4.2).
|
|
*
|
|
* **Suppressed on a Team's FIRST roster.** Importing a 155-member guild is one
|
|
* Team arriving, not 155 people joining, and emitting a join per member would
|
|
* bury every real event under the import and blow through the row cap on day one.
|
|
* `roster_synced_at IS NULL` is exactly "core has never held a roster for this
|
|
* Team", so the same condition covers a newly created Team and a newly installed
|
|
* module adopting an existing one.
|
|
*
|
|
* Never throws: the feed is a rendering of the sync, and a feed write failing
|
|
* must not abort the sync that is the actual source of truth.
|
|
*/
|
|
async function logRosterActivity(team, { joined, left, promoted, demoted }) {
|
|
if (!team.roster_synced_at) return
|
|
|
|
const items = [
|
|
...joined.map((row) => ({ kind: activity.CORE_KINDS.MEMBER_JOINED, row, verb: 'joined' })),
|
|
...left.map((row) => ({ kind: activity.CORE_KINDS.MEMBER_LEFT, row, verb: 'left' })),
|
|
...promoted.map((row) => ({ kind: activity.CORE_KINDS.LEADER_CHANGED, row, verb: 'became a leader' })),
|
|
...demoted.map((row) => ({ kind: activity.CORE_KINDS.LEADER_CHANGED, row, verb: 'stepped down as a leader' })),
|
|
]
|
|
|
|
for (const { kind, row, verb } of items) {
|
|
try {
|
|
// eslint-disable-next-line no-await-in-loop
|
|
await activity.logCore({
|
|
teamId: team.id,
|
|
kind,
|
|
summary: `${memberLabel(row)} ${verb}`,
|
|
actorMemberKey: row.member_key,
|
|
actorUserId: row.user_id ?? null,
|
|
})
|
|
} catch (err) {
|
|
log.warn('roster activity not recorded', { teamId: team.id, kind, message: err.message })
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* The push half of the same roster run (TEAMS.md §6.2, phase 6).
|
|
*
|
|
* **At most one tickle per stream per run, not one per member.** A tickle is
|
|
* content-free — it says "something happened in this Team" and the app pulls the
|
|
* rest — so five people joining in one sweep is five identical notifications and
|
|
* one piece of information. The feed above is per-member because it is a record;
|
|
* this is per-run because it is a nudge.
|
|
*
|
|
* **Suppressed on a Team's FIRST roster, exactly as the feed is**, and this is the
|
|
* half where it matters more: importing a 155-member guild would otherwise wake
|
|
* every one of their phones. `roster_synced_at IS NULL` is the same condition, read
|
|
* from the same row before the same stamp moves.
|
|
*
|
|
* Never throws — the fan-out swallows its own failures, and this adds the guard
|
|
* for anything the surrounding read could raise. A roster sync is the source of
|
|
* truth; a notification about it is not.
|
|
*/
|
|
async function notifyRoster(team, { joined, promoted, demoted }) {
|
|
if (!team.roster_synced_at) return
|
|
try {
|
|
// The count rides along for the Discord bridge (§7.2), which has no app on
|
|
// the other end to pull the roster after a content-free nudge. The tickle
|
|
// itself is unchanged and still carries nothing.
|
|
// `names` is the engagement engine's half (ENGAGEMENT.md Phase 6): the two
|
|
// triggers declare `memberName` / `leaderName` as required single values, so
|
|
// the fan-out emits one event per person while the tickle and the bridge stay
|
|
// one per run. A member the module reported without a display name is skipped
|
|
// rather than emitted as "someone" — a required variable filled with a
|
|
// placeholder is a mail that names nobody.
|
|
if (joined.length > 0) {
|
|
await teamNotify.memberJoined(team, {
|
|
count: joined.length,
|
|
names: joined.map((m) => m.display_name).filter(Boolean),
|
|
})
|
|
}
|
|
if (promoted.length > 0 || demoted.length > 0) {
|
|
await teamNotify.leadershipChanged(team, {
|
|
names: promoted.map((m) => m.display_name).filter(Boolean),
|
|
})
|
|
}
|
|
} catch (err) {
|
|
log.warn('roster notification not sent', { teamId: team.id, message: err.message })
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Sync one Team's roster and leadership. Gates 3 and 4 live here.
|
|
*
|
|
* Returns whether the roster was applied, so the caller can tell "synced" from
|
|
* "left alone", which is the difference between fresh and stale on that Team's
|
|
* page.
|
|
*/
|
|
async function syncRoster(team) {
|
|
const answer = await teamProvider.getTeamMembers(team.external_id)
|
|
|
|
// Gate 3. One Team's unanswerable roster is not the other Teams' problem, and
|
|
// it is certainly not an empty roster.
|
|
if (!answer.ok) {
|
|
log.warn('roster left untouched; provider could not answer', {
|
|
externalId: team.external_id, reason: answer.reason,
|
|
})
|
|
return false
|
|
}
|
|
|
|
// The full rows rather than just the keys: the activity feed needs the display
|
|
// name and the prior `is_leader` of everyone who is about to change, and both
|
|
// are gone once the upsert below has run. One read either way — this replaces
|
|
// the `memberKeys` call rather than adding to it.
|
|
const knownRows = await teamsDb.membersByTeam(team.id)
|
|
const knownByKey = new Map(knownRows.map((row) => [row.member_key, row]))
|
|
const known = knownRows.map((row) => row.member_key)
|
|
|
|
// Gate 4, the per-Team twin of gate 2.
|
|
if (answer.complete && answer.members.length === 0 && known.length > 0) {
|
|
if (!team.members_empty_since) {
|
|
await teamsDb.setMembersEmptySince(team.id, new Date())
|
|
log.warn('empty roster quarantined; awaiting a second answer', {
|
|
externalId: team.external_id, had: known.length,
|
|
})
|
|
return false
|
|
}
|
|
log.warn('empty roster confirmed by a second answer; departing every member', {
|
|
externalId: team.external_id, had: known.length,
|
|
})
|
|
} else if (team.members_empty_since) {
|
|
// Any non-empty answer clears the quarantine.
|
|
await teamsDb.setMembersEmptySince(team.id, null)
|
|
}
|
|
|
|
for (const member of answer.members) {
|
|
// eslint-disable-next-line no-await-in-loop
|
|
await teamsDb.upsertMember({
|
|
teamId: team.id,
|
|
memberKey: member.memberKey,
|
|
displayName: member.displayName,
|
|
userId: member.userId,
|
|
isLeader: member.leader,
|
|
rankLabel: member.rankLabel,
|
|
online: member.online,
|
|
})
|
|
}
|
|
|
|
// Anyone the module reports that core was not already holding. Read from the
|
|
// module's shape, since a joiner has no row yet.
|
|
const joined = answer.members
|
|
.filter((m) => !knownByKey.has(m.memberKey))
|
|
.map((m) => ({ member_key: m.memberKey, display_name: m.displayName, user_id: m.userId }))
|
|
|
|
// Removals only from a COMPLETE answer. `complete: false` means "valid but
|
|
// partial", so additions and updates apply and nothing is taken away.
|
|
let left = []
|
|
if (answer.complete) {
|
|
const seen = new Set(answer.members.map((m) => m.memberKey))
|
|
const departedKeys = known.filter((key) => !seen.has(key))
|
|
left = departedKeys.map((key) => knownByKey.get(key))
|
|
await teamsDb.markDeparted(team.id, departedKeys)
|
|
}
|
|
|
|
// Leadership is a separate question with a separate answer, and a provider that
|
|
// cannot answer it leaves the synced value alone rather than demoting everyone.
|
|
const leaders = await teamProvider.getTeamLeaders(team.external_id)
|
|
let promoted = []
|
|
let demoted = []
|
|
if (leaders.ok) {
|
|
// Diffed against the PRIOR rows, before setLeaders overwrites them. A member
|
|
// who joined this run as a leader is reported as joining, not as being
|
|
// promoted — they were never anything else here.
|
|
const nowLeader = new Set(leaders.leaders)
|
|
const departed = new Set(left.map((row) => row && row.member_key))
|
|
promoted = knownRows.filter((row) => nowLeader.has(row.member_key) && !row.is_leader)
|
|
demoted = knownRows.filter((row) => !nowLeader.has(row.member_key) && row.is_leader && !departed.has(row.member_key))
|
|
await teamsDb.setLeaders(team.id, leaders.leaders)
|
|
} else {
|
|
log.warn('leadership left untouched; provider could not answer', {
|
|
externalId: team.external_id, reason: leaders.reason,
|
|
})
|
|
}
|
|
|
|
await teamsDb.recount(team.id)
|
|
// Both read before `markRosterSynced` moves the stamp their first-roster
|
|
// suppression turns on.
|
|
await logRosterActivity(team, { joined, left: left.filter(Boolean), promoted, demoted })
|
|
await notifyRoster(team, { joined, promoted, demoted })
|
|
await teamsDb.markRosterSynced(team.id)
|
|
return true
|
|
}
|
|
|
|
// ── The run ────────────────────────────────────────────────────────────────
|
|
|
|
/**
|
|
* One full reconciliation. Callers use `request()`; this is the body it guards.
|
|
*
|
|
* Never throws: a reconcile is a background job, and a rejection here would
|
|
* surface as an unhandled rejection in the poll timer rather than as anything an
|
|
* operator could act on. The failure is recorded where it can be read — in
|
|
* `team_sync_state`, which Admin → Teams shows verbatim.
|
|
*/
|
|
async function runOnce(reason) {
|
|
const moduleId = teamProvider.providerModuleId()
|
|
if (!moduleId) return { ok: false, reason: 'no team provider is registered' }
|
|
|
|
await teamsDb.recordAttempt(moduleId)
|
|
const answer = await teamProvider.getTeams()
|
|
|
|
// Gate 1.
|
|
if (!answer.ok) {
|
|
await teamsDb.recordFailure(moduleId, answer.reason)
|
|
log.warn('reconcile refused; provider could not answer', { reason: answer.reason, trigger: reason })
|
|
return { ok: false, reason: answer.reason }
|
|
}
|
|
|
|
const existing = await teamsDb.activeByModule(moduleId)
|
|
|
|
// Gate 2. Only a COMPLETE answer can mean "there are no teams" — a partial one
|
|
// removes nothing by definition.
|
|
if (answer.complete && answer.teams.length === 0 && existing.length > 0) {
|
|
const state = await teamsDb.syncState(moduleId)
|
|
if (!state || !state.pending_empty_since) {
|
|
await teamsDb.setPendingEmpty(moduleId, new Date())
|
|
await teamsDb.recordSuccess(moduleId)
|
|
log.warn('empty team list quarantined; awaiting a second answer', { held: existing.length })
|
|
return { ok: true, quarantined: true, applied: 0 }
|
|
}
|
|
const intervalS = await intervalSeconds()
|
|
const waited = (Date.now() - new Date(state.pending_empty_since).getTime()) / 1000
|
|
if (waited < intervalS) {
|
|
await teamsDb.recordSuccess(moduleId)
|
|
log.warn('empty team list still quarantined', { waitedSeconds: Math.round(waited), intervalS })
|
|
return { ok: true, quarantined: true, applied: 0 }
|
|
}
|
|
log.warn('empty team list confirmed; archiving every active team', { count: existing.length })
|
|
} else if (answer.teams.length) {
|
|
// Any non-empty answer clears the quarantine.
|
|
await teamsDb.setPendingEmpty(moduleId, null)
|
|
}
|
|
|
|
const byExternalId = new Map(existing.map((t) => [t.external_id, t]))
|
|
const seen = new Set()
|
|
let created = 0
|
|
let renamed = 0
|
|
let rosters = 0
|
|
|
|
for (const team of answer.teams) {
|
|
seen.add(team.externalId)
|
|
const current = byExternalId.get(team.externalId)
|
|
let id
|
|
if (!current) {
|
|
id = await createTeam(moduleId, team)
|
|
created += 1
|
|
} else if (current.name !== team.name) {
|
|
id = await applyRename(moduleId, current, team)
|
|
renamed += 1
|
|
} else {
|
|
id = current.id
|
|
await teamsDb.updateTeam(id, { abbr: team.abbr, meta: team.meta })
|
|
}
|
|
|
|
// Re-read rather than reusing `current`: a create or a rename has just made a
|
|
// row this loop has never seen, and syncRoster reads the quarantine stamp off
|
|
// it. Passing a stale object would drop the second strike of gate 4.
|
|
const row = await teamsDb.findById(id)
|
|
if (row && await syncRoster(row)) rosters += 1
|
|
}
|
|
|
|
// Archive what the module no longer lists — the §2.2 disband path, and the only
|
|
// one. Guarded by `complete` for the same reason removals are.
|
|
let archived = 0
|
|
if (answer.complete) {
|
|
for (const team of existing) {
|
|
if (seen.has(team.external_id)) continue
|
|
await teamsDb.archiveTeam(team.id, 'disbanded')
|
|
archived += 1
|
|
log.info('team archived; absent from an authoritative list', {
|
|
externalId: team.external_id, name: team.name,
|
|
})
|
|
}
|
|
}
|
|
|
|
// Re-screen the names no human has ruled on. Names are immutable per row, so
|
|
// this only changes an outcome when the reserved TERMS changed — an operator
|
|
// adding one, or the deployment being renamed — which is exactly the case a
|
|
// create-time-only check would miss forever.
|
|
const rehidden = await moderation.rescreen(moduleId)
|
|
|
|
await teamsDb.recordSuccess(moduleId)
|
|
|
|
// §7.3's "after a successful Team reconcile": the voice reconciler runs off the
|
|
// projection this run just refreshed. Requested rather than awaited — it makes
|
|
// Discord calls, and a roster sync must never be slowed down, failed or held
|
|
// open by an integration hanging off it. It has its own debounce and its own
|
|
// suspensions (including the stale check, which is why it re-reads the state
|
|
// this run just wrote rather than trusting that it was called from a good one).
|
|
teamVoiceSync.request({ reason: 'reconcile' })
|
|
log.info('reconcile complete', {
|
|
trigger: reason, created, renamed, archived, rosters, rehidden, total: answer.teams.length,
|
|
})
|
|
return { ok: true, created, renamed, archived, rosters, rehidden }
|
|
}
|
|
|
|
// ── The public entry points ────────────────────────────────────────────────
|
|
|
|
/**
|
|
* Run now, awaited, with the per-module lock held. Admin → Resync uses this,
|
|
* because an operator pressing a button is owed the outcome rather than a
|
|
* promise that something will happen soon.
|
|
*
|
|
* A run already in progress is JOINED rather than queued: the caller wants "the
|
|
* projection is now current", and a run that started a moment ago delivers that.
|
|
*/
|
|
async function reconcileNow(reason = 'manual') {
|
|
if (running) {
|
|
rerunReason = reason
|
|
return { ok: true, joined: true }
|
|
}
|
|
running = true
|
|
try {
|
|
const result = await runOnce(reason)
|
|
lastRunAt = Date.now()
|
|
return result
|
|
} catch (err) {
|
|
log.error('reconcile threw', { message: err.message, trigger: reason })
|
|
return { ok: false, reason: err.message }
|
|
} finally {
|
|
running = false
|
|
const queued = rerunReason
|
|
rerunReason = null
|
|
// Something asked while this run was in flight, so it saw state this run may
|
|
// have been too early to include. Ask again, through the debounce.
|
|
if (queued) request({ reason: queued })
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Ask for a reconciliation. Returns immediately and never rejects — this is what
|
|
* `ctx.teams.reconcile()` is (§2.3), and a module must not be able to make its
|
|
* own call site slow or its own errors someone else's.
|
|
*/
|
|
function request({ reason = 'module' } = {}) {
|
|
if (debounceTimer) return
|
|
const since = Date.now() - lastRunAt
|
|
if (running) {
|
|
rerunReason = reason
|
|
return
|
|
}
|
|
if (since >= DEBOUNCE_MS) {
|
|
reconcileNow(reason).catch(() => {})
|
|
return
|
|
}
|
|
debounceTimer = setTimeout(() => {
|
|
debounceTimer = null
|
|
reconcileNow(reason).catch(() => {})
|
|
}, DEBOUNCE_MS - since)
|
|
// Unreffed for the same reason the provider's deadline is: a pending debounce
|
|
// must not hold a shutdown open waiting to do background work.
|
|
if (typeof debounceTimer.unref === 'function') debounceTimer.unref()
|
|
}
|
|
|
|
/**
|
|
* Apply a module-published event (§2.3).
|
|
*
|
|
* Deltas are applied only for a Team core already knows, and only for the four
|
|
* kinds a delta can express. Everything else — an unknown Team, a create, a
|
|
* disband — asks for a reconciliation instead, because a Team invented from a
|
|
* delta has no name, no roster and no leaders, and an archive driven by one is
|
|
* destruction on the strength of a message that may simply have been repeated.
|
|
*/
|
|
async function publish(event) {
|
|
const { kind, externalId } = event || {}
|
|
if (!EVENT_KINDS.has(kind)) throw new Error(`teams.publish: unknown event kind "${kind}"`)
|
|
const id = typeof externalId === 'string' ? externalId.trim() : ''
|
|
if (!id) throw new Error(`teams.publish: ${kind} has no externalId`)
|
|
|
|
const moduleId = teamProvider.providerModuleId()
|
|
if (!moduleId) return
|
|
|
|
if (RECONCILE_ONLY.has(kind)) {
|
|
request({ reason: kind })
|
|
return
|
|
}
|
|
|
|
const team = await teamsDb.findActive(moduleId, id)
|
|
if (!team) {
|
|
request({ reason: `${kind} for an unknown team` })
|
|
return
|
|
}
|
|
|
|
const memberKey = typeof event.memberKey === 'string' ? event.memberKey.trim() : ''
|
|
if (!memberKey) throw new Error(`teams.publish: ${kind} has no memberKey`)
|
|
|
|
switch (kind) {
|
|
case 'team.member.added':
|
|
await teamsDb.upsertMember({
|
|
teamId: team.id,
|
|
memberKey,
|
|
displayName: typeof event.displayName === 'string' ? event.displayName.trim() : null,
|
|
userId: Number.isInteger(event.userId) && event.userId > 0 ? event.userId : null,
|
|
isLeader: Boolean(event.leader),
|
|
rankLabel: typeof event.rankLabel === 'string' ? event.rankLabel.trim() : null,
|
|
online: Boolean(event.online),
|
|
})
|
|
break
|
|
case 'team.member.removed':
|
|
await teamsDb.markDeparted(team.id, [memberKey])
|
|
break
|
|
case 'team.leader.added':
|
|
case 'team.leader.removed':
|
|
// A no-op when the member is unknown: the row is created by the roster, not
|
|
// by a leadership delta, and inventing one here would put a member on the
|
|
// roster whose only evidence is that someone promoted them.
|
|
await teamsDb.setMemberLeader(team.id, memberKey, kind === 'team.leader.added')
|
|
break
|
|
default:
|
|
break
|
|
}
|
|
|
|
await teamsDb.recount(team.id)
|
|
// A delta is a hint that something changed, not a claim to have applied all of
|
|
// it, so every one still asks for the run that makes it correct.
|
|
request({ reason: kind })
|
|
}
|
|
|
|
// ── The poll ───────────────────────────────────────────────────────────────
|
|
|
|
async function scheduleNextPoll() {
|
|
const intervalS = await intervalSeconds()
|
|
const moduleId = teamProvider.providerModuleId()
|
|
let delayS = intervalS
|
|
if (moduleId) {
|
|
const state = await teamsDb.syncState(moduleId).catch(() => null)
|
|
if (state) delayS = backoffSeconds(state.consecutive_failures, intervalS)
|
|
}
|
|
pollTimer = setTimeout(() => {
|
|
reconcileNow('poll').catch(() => {}).then(() => { if (started) scheduleNextPoll().catch(() => {}) })
|
|
}, delayS * 1000)
|
|
if (typeof pollTimer.unref === 'function') pollTimer.unref()
|
|
}
|
|
|
|
/**
|
|
* Start the boot reconcile and the poll. Called from the module lifecycle, after
|
|
* every module has started — the website may have been down across a whole guild
|
|
* war, so the first thing it does on the way up is ask.
|
|
*/
|
|
async function start() {
|
|
if (started) return
|
|
started = true
|
|
if (!teamProvider.providerModuleId()) {
|
|
log.info('no team provider registered; the reconciler stays idle')
|
|
return
|
|
}
|
|
await reconcileNow('boot')
|
|
await scheduleNextPoll()
|
|
}
|
|
|
|
function stop() {
|
|
started = false
|
|
if (pollTimer) clearTimeout(pollTimer)
|
|
if (debounceTimer) clearTimeout(debounceTimer)
|
|
pollTimer = null
|
|
debounceTimer = null
|
|
}
|
|
|
|
// Test-only: the scheduler is module-level state, so a test that triggers a run
|
|
// has to be able to put it back.
|
|
function _reset() {
|
|
stop()
|
|
running = false
|
|
rerunReason = null
|
|
lastRunAt = 0
|
|
}
|
|
|
|
module.exports = {
|
|
reconcileNow,
|
|
request,
|
|
publish,
|
|
start,
|
|
stop,
|
|
intervalSeconds,
|
|
backoffSeconds,
|
|
EVENT_KINDS,
|
|
DEBOUNCE_MS,
|
|
DEFAULT_INTERVAL_S,
|
|
_reset,
|
|
}
|