Entzug greift auch im offenen Ereignisstrom

Nachgesehen, was bei Herabstufung, Rollenentzug, Rauswurf und Austritt
passiert. Für Requests war alles abgedeckt: der Worker löscht die Sitzungen,
ein Trigger verwirft die zwischengespeicherten Rechte, getCurrentUser prüft bei
jedem Request Sitzungszeile, isActive und Mitgliedschaft nach. Die Zeile in der
Datenbank ist der Nachweis, nicht das Cookie - und eine Zeile lässt sich
zurückziehen.

Eine Stelle fiel dabei durch: der Ereignisstrom stellt keine neuen Requests. Er
prüfte einmal beim Verbinden und lief danach mit einer eingefrorenen
Rechtekopie weiter. Wer mit offenem Tab hinausgeworfen wurde, bekam weiter
Chatnachrichten zugestellt, mit vollem Text - jeder Seitenaufruf wäre
abgemeldet gewesen, aber es machte ja keiner mehr einen.

/api/stream löst Sitzung und Rechte jetzt bei jedem Ereignis und bei jedem
Herzschlag neu auf. Ändern sich die Rechte, werden auch die erlaubten Themen
neu bestimmt: ein Ereignis zu einem Verlaufseintrag trägt nur eine Kennung, das
Thema ist dort die einzige Schranke. Ist die Sitzung weg, geht ein revoked
hinaus und die Verbindung wird geschlossen - Schliessen allein genügt nicht,
EventSource baut sofort wieder auf.

Fünf Tests für die Web-Hälfte der Kette; die Discord-Hälfte deckte
worker/tests/mitgliedschaft.test.ts schon ab.

web/app/api/stream/route.ts

@@ -1,4 +1,7 @@
 import { getCurrentUser, type CurrentUser } from '@/lib/auth/current'
+import { loadSession } from '@/lib/auth/session'
+import { resolvePermissions } from '@/lib/auth/effective'
+import type { EffectivePermissions } from '@/lib/auth/permissions'
 import { db } from '@/lib/db'
 import { visibleCaseWhere } from '@/lib/access/caseFilter'
 import { visibleMessageWhere } from '@/lib/access/chatFilter'
@@ -25,17 +28,25 @@ const HEARTBEAT_MS = 25_000
  * durch jeden Tunnel, und der Browser baut die Verbindung von selbst wieder
  * auf. Siehe docs/06-chat-und-realtime.md.
  *
- * Zwei Prüfungen, beide nötig:
+ * Drei Prüfungen, alle nötig:
  *  1. **Beim Verbinden**: welche der angefragten Themen darf dieser Nutzer
  *     überhaupt abonnieren. Wer ein fremdes Thema nennt, bekommt es nicht —
  *     und erfährt auch nicht, ob es existiert.
  *  2. **Je Ereignis**: der Inhalt wird mit den Rechten des Empfängers geladen.
  *     Eine Umklassifizierung mitten in einer offenen Verbindung wirkt dadurch
  *     sofort, ohne dass jemand neu laden muss.
+ *  3. **Laufend**: Sitzung und Rechte werden neu aufgelöst, nicht einmal beim
+ *     Verbinden eingefroren. Eine offene Verbindung ist sonst der eine Ort im
+ *     System, an dem ein Entzug nicht ankommt: der Worker löscht die Sitzung,
+ *     jeder Seitenaufruf ist abgemeldet — und der Strom läuft weiter, weil
+ *     niemand mehr fragt. Siehe docs/03-auth-und-rechte.md.
  */
 export async function GET(request: Request): Promise<Response> {
-  const user = await getCurrentUser()
-  if (!user) return new Response(null, { status: 404 })
+  const angemeldet = await getCurrentUser()
+  if (!angemeldet) return new Response(null, { status: 404 })
+  // Eigene Konstante: die Verengung auf „nicht null" überlebt den Sprung in
+  // eine Closure sonst nicht.
+  const user = angemeldet
 
   const url = new URL(request.url)
   const gewuenscht = (url.searchParams.get('topics') ?? '')
@@ -44,9 +55,38 @@ export async function GET(request: Request): Promise<Response> {
     .filter(Boolean)
     .slice(0, 20)
 
-  const erlaubt = await erlaubteThemen(user, gewuenscht)
+  let rechte = user.permissions
+  let erlaubt = await erlaubteThemen(user, gewuenscht)
   await ensureListening()
 
+  /**
+   * Ist der Nutzer noch der, der er beim Verbinden war?
+   *
+   * Gibt die **aktuellen** Rechte zurück oder `null`, wenn die Sitzung weg
+   * ist, das Konto stillgelegt wurde oder die Mitgliedschaft fehlt. Die
+   * Auflösung ist zwischengespeichert und wird bei jeder Rollenänderung über
+   * `member_changed` verworfen, kostet also im Normalfall nichts.
+   */
+  async function nochBerechtigt(): Promise<EffectivePermissions | null> {
+    const sitzung = await loadSession(user.sessionId)
+    if (!sitzung) return null
+
+    const konto = await db.user.findUnique({
+      where: { id: user.id },
+      select: { isActive: true },
+    })
+    if (!konto?.isActive) return null
+
+    const frisch = await resolvePermissions(user.id)
+    if (!frisch.hasMembership && !frisch.isAdmin) return null
+    return frisch
+  }
+
+  /** Ob sich an den Rechten etwas geändert hat, das die Themen betrifft. */
+  function abdruck(p: EffectivePermissions): string {
+    return `${p.clearanceLevel}|${p.isAdmin}|${[...p.permissions].sort().join(',')}`
+  }
+
   const encoder = new TextEncoder()
   let heartbeat: ReturnType<typeof setInterval> | null = null
   let unsubscribe: (() => void) | null = null
@@ -78,18 +118,59 @@ export async function GET(request: Request): Promise<Response> {
 
       send('ready', { topics: erlaubt })
 
+      const schliessen = () => {
+        offen = false
+        if (heartbeat) clearInterval(heartbeat)
+        unsubscribe?.()
+        try {
+          controller.close()
+        } catch {
+          // schon zu
+        }
+      }
+
       heartbeat = setInterval(() => {
         if (!offen) return
-        try {
-          controller.enqueue(encoder.encode(': ping\n\n'))
-        } catch {
-          offen = false
-        }
+        void (async () => {
+          // Der Herzschlag ist zugleich die regelmässige Rückfrage: wer
+          // hinausgeworfen wurde, verliert spätestens hier die Verbindung.
+          const frisch = await nochBerechtigt()
+          if (!frisch) {
+            send('revoked', {})
+            schliessen()
+            return
+          }
+          if (abdruck(frisch) !== abdruck(rechte)) {
+            rechte = frisch
+            erlaubt = await erlaubteThemen({ ...user, permissions: frisch }, gewuenscht)
+          }
+          try {
+            controller.enqueue(encoder.encode(': ping\n\n'))
+          } catch {
+            offen = false
+          }
+        })()
       }, HEARTBEAT_MS)
 
       unsubscribe = subscribe((event) => {
-        if (!erlaubt.includes(event.t)) return
-        void zustellen(user, event, send)
+        void (async () => {
+          const frisch = await nochBerechtigt()
+          if (!frisch) {
+            send('revoked', {})
+            schliessen()
+            return
+          }
+          // Rechte geändert: die erlaubten Themen gelten nicht mehr. Sie neu
+          // zu bestimmen ist wichtiger als die eine Zustellung — ein Ereignis
+          // trägt bei Verlaufseinträgen nur eine Kennung, das Thema ist dort
+          // die **einzige** Schranke.
+          if (abdruck(frisch) !== abdruck(rechte)) {
+            rechte = frisch
+            erlaubt = await erlaubteThemen({ ...user, permissions: frisch }, gewuenscht)
+          }
+          if (!erlaubt.includes(event.t)) return
+          await zustellen({ ...user, permissions: rechte }, event, send)
+        })()
       })
 
       request.signal.addEventListener('abort', () => {