Phase 6 abgeschlossen: Aktenchat, Kanäle, Echtzeit, Benachrichtigungen

Ein Ereignisstrom mit Themen (/api/stream), in dem der Dateistrom aus Phase 5
aufgegangen ist. Zwei Prüfungen: beim Verbinden, welche Themen der Nutzer
abonnieren darf, und je Ereignis wird der Inhalt mit seinen Rechten geladen.
Das zweite ist nicht nur Vorsicht — eine Umklassifizierung wirkt dadurch in
einer offenen Verbindung sofort. In der pg_notify-Nutzlast steht nie ein
Inhalt, nur Thema, Art und Kennung.

Der Aktenchat sendet optimistisch mit clientseitiger UUIDv7, die zugleich
Idempotenzschlüssel ist. Antworten, vier feste Reaktionen, Bearbeiten mit
sichtbarem editedAt, Zurückziehen mit Platzhalter — Löschen gibt es nicht, und
eine Regel in der Datenbank hält das fest. Eine zurückgezogene Nachricht kommt
ohne Text an, nicht ausgeblendet.

Aktenzeichen im Text werden zu Vorschaukarten, serverseitig aufgelöst. Eine
Akte, die der Leser nicht sehen darf, sieht darin genauso aus wie eine, die es
gar nicht gibt.

"Als Vermerk übernehmen" steht sichtbar an jeder Nachricht. Der Vermerk gehört
dem Verfasser, nicht dem Übernehmenden; ein Systemeintrag nennt den Ursprung,
damit der Wortlaut später nicht für die Akte geschrieben wirkt.

Bei den Discord-Benachrichtigungen entscheidet die Web-App, wer was erfährt,
und der Worker liefert nur aus. Die Alternative hätte geheissen, die
Sichtbarkeitsprüfung im Worker ein zweites Mal zu bauen. In meta steht ein
Titel nur, wenn er hinaus darf — ab Stufe 3 keiner. Postausgang als Tabelle,
angestossen über LISTEN, mit Takt als Sicherheitsnetz.

Nebenbei riss das Budget der Fallakte erneut (122 kB). Drei Schritte zurück
auf 119,6: Chattexte in eine eigene i18n-Datei, ein gemeinsames AddPanel statt
viermal derselbe Aufklapp-Code, und eine eigene UUIDv7 statt des Pakets — die
Sortierbarkeit, auf der die Nachlieferung beruht, steht damit im eigenen Code
und hat Tests.

Unterwegs habe ich NotifyMode und User.notifyMode ein zweites Mal angelegt,
obwohl es beides seit dem Grundgerüst gibt. Die Migration schlug fehl und die
Datenbank musste von Hand aufgeräumt werden; die Lehre steht in docs/13.

155 Tests im Web-Paket, 15 in der Datenbank, 7 im Worker, 32 e2e.

worker/src/env.ts

@@ -19,6 +19,10 @@ export type WorkerEnv = {
   fullSyncIntervalMs: number
   heartbeatIntervalMs: number
   adminChannelId?: string
+  /** Takt des Postausgangs als Sicherheitsnetz neben `notification_queued`. */
+  outboxIntervalMs: number
+  /** Steht in den Benachrichtigungen als Link. */
+  appUrl: string
 }
 
 function required(name: string): string {
@@ -39,5 +43,7 @@ export function loadEnv(): WorkerEnv {
     fullSyncIntervalMs: Number(process.env.WORKER_FULL_SYNC_MS ?? 5 * 60_000),
     heartbeatIntervalMs: Number(process.env.WORKER_HEARTBEAT_MS ?? 30_000),
     adminChannelId: process.env.DISCORD_ADMIN_CHANNEL_ID,
+    outboxIntervalMs: Number(process.env.WORKER_OUTBOX_MS ?? 30_000),
+    appUrl: process.env.APP_URL ?? 'http://localhost:3000',
   }
 }

worker/src/index.ts

@@ -1,8 +1,10 @@
 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
@@ -116,6 +118,27 @@ async function main(): Promise<void> {
     ).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) =>
@@ -127,6 +150,9 @@ async function main(): 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()
@@ -136,6 +162,23 @@ async function main(): Promise<void> {
   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)
 }

worker/src/notify.ts

@@ -0,0 +1,128 @@
+import type { Client } from 'discord.js'
+import type { PrismaClient } from '@afis/db'
+
+/**
+ * Zustellung der Benachrichtigungen.
+ *
+ * Der Worker ist hier **nur Transport**. Wer etwas erfahren darf und was davon
+ * gezeigt werden darf, hat die Web-App beim Einstellen in den Postausgang
+ * entschieden — sie hat die Sichtbarkeitsprüfung, er hat sie nicht. In `meta`
+ * steht deshalb ausschliesslich, was stehen darf: ein Titel liegt dort nur,
+ * wenn er auch nach Discord darf. Siehe docs/06-chat-und-realtime.md.
+ *
+ * Der Dauerläufer holt in zwei Takten: sofort, wenn `notification_queued`
+ * kommt, und alle 30 Sekunden als Sicherheitsnetz. Kein Einmal-Container,
+ * siehe CLAUDE.md, harte Regel 8.
+ */
+
+/** Nach so vielen Fehlversuchen wird eine Zeile liegengelassen. */
+const MAX_ATTEMPTS = 5
+
+/** Wie viele Zeilen ein Durchgang höchstens abarbeitet. */
+const BATCH = 25
+
+type Meta = {
+  caseNumber?: string
+  actorName?: string
+  title?: string | null
+  channelKey?: string
+}
+
+/**
+ * Baut den Satz. Bewusst knapp und ohne jeden Inhalt aus der Akte — der Link
+ * führt zur Anwendung, und die prüft dann selbst.
+ */
+export function composeMessage(
+  kind: string,
+  meta: Meta,
+  appUrl: string,
+): string | null {
+  const wer = meta.actorName ?? 'Jemand'
+
+  if (meta.caseNumber) {
+    const link = `${appUrl.replace(/\/$/, '')}/faelle/${meta.caseNumber}`
+    const bezeichnung = meta.title
+      ? `**${meta.caseNumber}** — ${meta.title}`
+      : `**${meta.caseNumber}**`
+
+    if (kind === 'entry.new') return `Neuer Eintrag in ${bezeichnung} · ${wer} · ${link}`
+    if (kind === 'chat.mention') return `${wer} hat dich in ${bezeichnung} erwähnt · ${link}`
+    if (kind === 'chat.new') return `Neue Nachricht im Aktenchat ${bezeichnung} · ${wer} · ${link}`
+    if (kind === 'case.status') return `Status geändert in ${bezeichnung} · ${wer} · ${link}`
+    return null
+  }
+
+  if (meta.channelKey && kind === 'chat.mention') {
+    const link = `${appUrl.replace(/\/$/, '')}/kanaele/${meta.channelKey}`
+    return `${wer} hat dich in #${meta.channelKey} erwähnt · ${link}`
+  }
+
+  return null
+}
+
+export async function dispatchOutbox(
+  db: PrismaClient,
+  client: Client,
+  options: { appUrl: string; adminChannelId?: string },
+): Promise<number> {
+  const offen = await db.notificationOutbox.findMany({
+    where: { sentAt: null, attempts: { lt: MAX_ATTEMPTS } },
+    orderBy: { id: 'asc' },
+    take: BATCH,
+  })
+
+  let zugestellt = 0
+
+  for (const eintrag of offen) {
+    const text = composeMessage(eintrag.kind, (eintrag.meta ?? {}) as Meta, options.appUrl)
+
+    // Nichts zu sagen ist kein Fehler: die Zeile wird abgehakt, damit sie den
+    // Postausgang nicht verstopft.
+    if (!text) {
+      await db.notificationOutbox.update({
+        where: { id: eintrag.id },
+        data: { sentAt: new Date(), lastError: 'nichts zu melden' },
+      })
+      continue
+    }
+
+    try {
+      if (eintrag.recipientId) {
+        const nutzer = await client.users.fetch(eintrag.recipientId)
+        await nutzer.send(text)
+      } else if (options.adminChannelId) {
+        const kanal = await client.channels.fetch(options.adminChannelId)
+        if (kanal?.isTextBased() && 'send' in kanal) {
+          await kanal.send(text)
+        }
+      } else {
+        // Kein allgemeiner Kanal eingerichtet: die Zeile ist erledigt, ohne
+        // dass etwas hinausging.
+        await db.notificationOutbox.update({
+          where: { id: eintrag.id },
+          data: { sentAt: new Date(), lastError: 'kein Kanal konfiguriert' },
+        })
+        continue
+      }
+
+      await db.notificationOutbox.update({
+        where: { id: eintrag.id },
+        data: { sentAt: new Date(), attempts: { increment: 1 } },
+      })
+      zugestellt++
+    } catch (error) {
+      // Eine geschlossene Direktnachricht ist keine Störung, sondern eine
+      // Entscheidung des Empfängers. Sie wird gezählt und irgendwann
+      // aufgegeben, aber nie laut.
+      await db.notificationOutbox.update({
+        where: { id: eintrag.id },
+        data: {
+          attempts: { increment: 1 },
+          lastError: error instanceof Error ? error.message.slice(0, 200) : 'unbekannt',
+        },
+      })
+    }
+  }
+
+  return zugestellt
+}

worker/tests/nachricht.test.ts

@@ -0,0 +1,42 @@
+import { describe, expect, it } from 'vitest'
+import { composeMessage } from '../src/notify.js'
+
+/**
+ * Der Worker baut den Satz aus dem, was die Web-App freigegeben hat. Hier wird
+ * geprüft, dass er nichts hinzuerfindet — vor allem keinen Titel, wo keiner
+ * mitgeschickt wurde.
+ */
+describe('Text der Benachrichtigung', () => {
+  const url = 'https://afis.example.de/'
+
+  it('nennt Aktenzeichen und Link, den Titel nur wenn er mitkam', () => {
+    const ohne = composeMessage(
+      'entry.new',
+      { caseNumber: 'AfiS-2026-0042', actorName: 'Insp. Vogel', title: null },
+      url,
+    )
+    expect(ohne).toBe(
+      'Neuer Eintrag in **AfiS-2026-0042** · Insp. Vogel · https://afis.example.de/faelle/AfiS-2026-0042',
+    )
+    expect(ohne).not.toContain('undefined')
+
+    const mit = composeMessage(
+      'entry.new',
+      { caseNumber: 'AfiS-2026-0042', actorName: 'Insp. Vogel', title: 'Korruptionsverdacht' },
+      url,
+    )
+    expect(mit).toContain('Korruptionsverdacht')
+  })
+
+  it('kann eine Erwähnung in einem Kanal melden', () => {
+    expect(
+      composeMessage('chat.mention', { channelKey: 'leitung', actorName: 'Vogel' }, url),
+    ).toBe('Vogel hat dich in #leitung erwähnt · https://afis.example.de/kanaele/leitung')
+  })
+
+  it('schweigt, wenn nichts Verwertbares mitkam', () => {
+    // Lieber gar nichts als ein Satz mit Lücken.
+    expect(composeMessage('entry.new', {}, url)).toBeNull()
+    expect(composeMessage('unbekannt', { caseNumber: 'AfiS-2026-0001' }, url)).toBeNull()
+  })
+})