AfiS-Aktensystem/worker/src/index.ts

import { Client, Events, GatewayIntentBits, type Guild } from 'discord.js'
import { Client as PgClient } from 'pg'
import { createPrismaClient } from '@afis/db'
import { loadEnv } from './env.js'
import { writeHeartbeat } from './heartbeat.js'
import { enforceMembership, handleDeparture, syncMember, syncRoles } from './sync.js'
import { dispatchOutbox } from './notify.js'

/**
 * Rollen-Abgleich-Worker. Hängt am Discord-Gateway und hält Guild-Rollen und
 * Mitgliedschaft in Postgres aktuell, damit die Web-App nie live bei Discord
 * anfragen muss. Siehe docs/02-architektur.md.
 *
 * Zwei Wege, absichtlich doppelt:
 *  - Gateway-Ereignisse, damit eine Änderung in unter einer Sekunde ankommt
 *  - Voll-Abgleich alle fünf Minuten als Sicherheitsnetz, falls ein Ereignis
 *    verloren geht oder der Prozess eine Weile weg war
 */
async function main(): Promise<void> {
  const env = loadEnv()
  const db = createPrismaClient(env.databaseUrl)

  const client = new Client({
    // Server Members Intent muss im Developer Portal aktiviert sein, sonst
    // liefert der Bot keine Rollen. Siehe docs/03-auth-und-rechte.md.
    intents: [GatewayIntentBits.Guilds, GatewayIntentBits.GuildMembers],
  })

  let gatewayReady = false

  async function fullSync(reason: string): Promise<void> {
    const guild = await client.guilds.fetch(env.discordGuildId)
    const roleCount = await syncRoles(db, await guild.fetch())
    const members = await guild.members.fetch()

    let changed = 0
    for (const member of members.values()) {
      if (member.user.bot) continue
      if (await syncMember(db, member)) changed++
      await enforceMembership(db, member.id, env.bootstrapAdminIds)
    }

    // Wer nicht mehr auf dem Server ist, hat auch keine Sitzung mehr.
    const known = await db.user.findMany({ where: { isActive: true }, select: { id: true } })
    for (const user of known) {
      if (!members.has(user.id)) await handleDeparture(db, user.id)
    }

    console.log(
      `[worker] Voll-Abgleich (${reason}): ${roleCount} Rollen, ${members.size} Mitglieder, ${changed} geändert`,
    )
  }

  client.once(Events.ClientReady, (ready) => {
    gatewayReady = true
    console.log(`[worker] am Gateway als ${ready.user.tag}`)
    void fullSync('Start').catch((error: unknown) =>
      console.error('[worker] erster Abgleich fehlgeschlagen', error),
    )
  })

  client.on(Events.GuildMemberAdd, (member) => {
    if (member.guild.id !== env.discordGuildId || member.user.bot) return
    void (async () => {
      await syncMember(db, member)
      await enforceMembership(db, member.id, env.bootstrapAdminIds)
    })().catch((error: unknown) => console.error('[worker] guildMemberAdd', error))
  })

  client.on(Events.GuildMemberUpdate, (_before, member) => {
    if (member.guild.id !== env.discordGuildId || member.user.bot) return
    void (async () => {
      await syncMember(db, member)
      await enforceMembership(db, member.id, env.bootstrapAdminIds)
    })().catch((error: unknown) => console.error('[worker] guildMemberUpdate', error))
  })

  client.on(Events.GuildMemberRemove, (member) => {
    if (member.guild.id !== env.discordGuildId) return
    void handleDeparture(db, member.id).catch((error: unknown) =>
      console.error('[worker] guildMemberRemove', error),
    )
  })

  // Wird eine Rolle umbenannt, verschoben oder gelöscht, muss der Adminbereich
  // das sehen. Der Trigger auf RoleMapping löst danach pg_notify aus.
  const onRoleChange = (guild: Guild): void => {
    if (guild.id !== env.discordGuildId) return
    void syncRoles(db, guild).catch((error: unknown) =>
      console.error('[worker] Rollenabgleich', error),
    )
  }

  client.on(Events.GuildRoleCreate, (role) => onRoleChange(role.guild))
  client.on(Events.GuildRoleUpdate, (_before, role) => onRoleChange(role.guild))
  client.on(Events.GuildRoleDelete, (role) => onRoleChange(role.guild))

  client.on(Events.ShardDisconnect, (_event, shardId) => {
    gatewayReady = false
    console.warn(`[worker] Verbindung zu Shard ${shardId} verloren`)
  })

  client.on(Events.ShardResume, () => {
    gatewayReady = true
    console.log('[worker] Verbindung wiederhergestellt')
    void fullSync('nach Wiederverbindung').catch((error: unknown) =>
      console.error('[worker] Abgleich nach Wiederverbindung fehlgeschlagen', error),
    )
  })

  client.on(Events.Error, (error) => console.error('[worker] Gateway-Fehler', error))

  const heartbeat = setInterval(() => {
    void writeHeartbeat(
      db,
      gatewayReady ? 'up' : 'degraded',
      gatewayReady ? undefined : 'Gateway nicht verbunden',
    ).catch((error: unknown) => console.error('[worker] Lebenszeichen fehlgeschlagen', error))
  }, env.heartbeatIntervalMs)

  // Postausgang: der Takt ist das Sicherheitsnetz, der Anstoss über
  // `notification_queued` der Regelfall. Siehe worker/src/notify.ts.
  let laeuft = false
  async function zustellen(grund: string): Promise<void> {
    if (!gatewayReady || laeuft) return
    laeuft = true
    try {
      const anzahl = await dispatchOutbox(db, client, {
        appUrl: env.appUrl,
        adminChannelId: env.adminChannelId,
      })
      if (anzahl > 0) console.log(`[worker] ${anzahl} Benachrichtigungen zugestellt (${grund})`)
    } catch (error) {
      console.error('[worker] Postausgang fehlgeschlagen', error)
    } finally {
      laeuft = false
    }
  }

  const outbox = setInterval(() => void zustellen('planmäßig'), env.outboxIntervalMs)

  const periodicSync = setInterval(() => {
    if (!gatewayReady) return
    void fullSync('planmäßig').catch((error: unknown) =>
      console.error('[worker] Voll-Abgleich fehlgeschlagen', error),
    )
  }, env.fullSyncIntervalMs)

  async function shutdown(signal: string): Promise<void> {
    console.log(`[worker] ${signal}, fahre herunter`)
    clearInterval(heartbeat)
    clearInterval(periodicSync)
    clearInterval(outbox)
    listener.removeAllListeners()
    await listener.end().catch(() => undefined)
    await writeHeartbeat(db, 'down', 'planmäßig beendet').catch(() => undefined)
    await client.destroy()
    await db.$disconnect()
    process.exit(0)
  }

  process.on('SIGTERM', () => void shutdown('SIGTERM'))
  process.on('SIGINT', () => void shutdown('SIGINT'))

  // Eigener Client für LISTEN: im Transaktionsmodus von pgbouncer überlebt
  // ein LISTEN den Pool nicht, und Prisma gibt seine Verbindung nicht her.
  const listener = new PgClient({
    connectionString: process.env.DIRECT_DATABASE_URL ?? env.databaseUrl,
  })
  listener.on('error', (error: unknown) =>
    console.error('[worker] LISTEN-Verbindung verloren', error),
  )
  listener.on('notification', () => void zustellen('angestoßen'))
  await listener
    .connect()
    .then(() => listener.query('LISTEN notification_queued'))
    .catch((error: unknown) => {
      // Ohne LISTEN bleibt der Takt. Das ist langsamer, aber nicht kaputt.
      console.error('[worker] LISTEN nicht möglich, nutze nur den Takt', error)
    })

  await writeHeartbeat(db, 'degraded', 'Start läuft')
  await client.login(env.discordBotToken)
}

main().catch((error: unknown) => {
  console.error('[worker] Start fehlgeschlagen', error)
  process.exit(1)
})